import json
|
from pathlib import Path
|
from datetime import datetime, timedelta
|
from auto_media_publisher.config.conf_base import TASK_CONFIG_DIR
|
from auto_media_publisher.utils.logger import get_logger
|
from auto_media_publisher.utils.utils_database import publish_task_exists,publish_task_insert,publish_task_delete
|
|
# ----------------- 任务生成器 -----------------
|
|
class TaskGenerator:
|
def __init__(self):
|
self.config_dir_path = Path(TASK_CONFIG_DIR)
|
self.logger = get_logger(self.__class__.__name__,"admin")
|
self.logger.info("TaskGenerator started.")
|
|
def run(self):
|
count = 0
|
count_new = 0
|
count_updated = 0
|
self.logger.info("generate_publish_tasks_once 开始执行")
|
try:
|
for config_file in self.config_dir_path.rglob("*.json"):
|
try:
|
user_config = json.loads(config_file.read_text(encoding="utf-8"))
|
except Exception as e:
|
self.logger.warning(f"无法读取配置文件 {config_file.name}：{e}")
|
continue
|
|
if not user_config.get("enabled", True):
|
self.logger.info(f"跳过未启用用户：{config_file.name}")
|
continue
|
|
user_id = user_config.get("user")
|
proxy = user_config.get("proxy")
|
platforms = user_config.get("platforms", [])
|
|
for platform in platforms:
|
if not platform.get("enabled", True):
|
continue
|
|
platform_name = platform["name"]
|
generator_class = platform["media_generator"]
|
generator_args = platform["media_generator_args"]
|
generator_args_str = json.dumps(generator_args, ensure_ascii=False)
|
|
platform_days_ahead = platform["task_ahead_days"]
|
|
upload_times = platform.get("upload_times", [])
|
config_updated = datetime.fromtimestamp(config_file.stat().st_mtime).isoformat()
|
|
for day_offset in range(platform_days_ahead):
|
date = (datetime.now() + timedelta(days=day_offset)).date()
|
for time_str in upload_times:
|
try:
|
scheduled_dt = datetime.strptime(f"{date} {time_str}", "%Y-%m-%d %H:%M")
|
except ValueError:
|
self.logger.warning(f"非法时间格式：{time_str} in {config_file.name}")
|
continue
|
|
existing = publish_task_exists(user_id, platform_name, scheduled_dt)
|
|
if existing:
|
task_id, old_hash, task_status = existing # 本质上 还是个 dict，这样会自动解包
|
if task_status == "pending" and old_hash != config_updated:
|
publish_task_delete(task_id)
|
publish_task_insert(user_id, platform_name, generator_class,generator_args_str,scheduled_dt, config_updated,proxy)
|
count_updated = count_updated +1
|
else:
|
publish_task_insert(user_id, platform_name, generator_class,generator_args_str,scheduled_dt, config_updated,proxy)
|
count_new =count_new+1
|
count += 1
|
except Exception as e:
|
self.logger.exception(f"generate_publish_tasks_once 出错: {e}")
|
self.logger.info(f"累计检查 {count} 条任务。")
|
self.logger.info(f"{count_updated} 条任务已更新数据库")
|
self.logger.info(f"{count_new} 条任务已插入数据库")
|
