import asyncio
|
import json
|
from pathlib import Path
|
from datetime import datetime
|
from auto_media_publisher.config.conf_base import TASK_CONFIG_DIR,MEDIA_CONCURRENT_COUNT
|
from auto_media_publisher.utils.logger import get_logger
|
from auto_media_publisher.generators.media_generator_base import MediaGenerator
|
from auto_media_publisher.utils.utils_database import media_get_tasks,media_update_task_done
|
from auto_media_publisher.generators import * # 具体实现在 auto_media_publisher/generators/__init__.py 里
|
|
# ----------------- 任务生成器 -----------------
|
|
class TaskMediaGenerator:
|
def __init__(self):
|
self.config_dir_path = Path(TASK_CONFIG_DIR)
|
self.logger = get_logger(self.__class__.__name__,"admin")
|
self.logger.info("TaskMediaGenerator 创建成功.")
|
|
async def _async_run(self):
|
pendding_video_tasks = media_get_tasks()
|
self.logger.info(f"[TaskMediaGenerator] 待生成媒体数量: {len(pendding_video_tasks)}")
|
|
if not pendding_video_tasks:
|
self.logger.info("[TaskMediaGenerator] 无任务，自动退出。")
|
return
|
|
video_task_queue = asyncio.Queue()
|
for row in pendding_video_tasks:
|
await video_task_queue.put(row)
|
|
semaphore = asyncio.Semaphore(MEDIA_CONCURRENT_COUNT)
|
|
async def worker_wrapper(task_row):
|
async with semaphore:
|
try:
|
id = task_row["id"]
|
generator_class = task_row["generator_class"]
|
generator_args_str = task_row["generator_args"]
|
generator_args = json.loads(generator_args_str)
|
generator_args['user_id'] = task_row["user_id"]
|
generator_args['scheduled_time'] = task_row["scheduled_time"]
|
generator_args['proxy_full'] = task_row["proxy_full"]
|
|
gen = MediaGenerator.create_by_name(generator_class,**generator_args)
|
if gen.generate():
|
media_update_task_done(id,datetime.now().strftime("%Y-%m-%d %H:%M:%S"))
|
except Exception as e:
|
self.logger.error(f"[MediaGenerator] 任务处理异常: {e}")
|
|
# 创建所有任务
|
tasks = []
|
while not video_task_queue.empty():
|
row = await video_task_queue.get()
|
tasks.append(asyncio.create_task(worker_wrapper(row)))
|
|
# 等待所有任务完成
|
await asyncio.gather(*tasks)
|
|
def run(self):
|
asyncio.run(self._async_run())
|
