# -*- coding: utf-8 -*-
|
import json
|
from pathlib import Path
|
from datetime import datetime, timedelta
|
from auto_media_publisher.utils.logger import get_logger
|
from auto_media_publisher.config.conf_video_mashup import *
|
from auto_media_publisher.generators.media_generator_base import MediaGenerator
|
from auto_media_publisher.config.conf_base import DATABASE_PATH,TASK_CONFIG_DIR,VIDEO_DIR_FINAL
|
from auto_media_publisher.utils.utils_database import media_update_task_done,publish_task_exists,publish_task_insert
|
|
|
class VideoGeneratorManual(MediaGenerator):
|
"""
|
手动地生成短视频，拷贝到 media_final / user_id / 目录下 之后。
|
要求：
|
1、每个视频要有三个文件：
|
mp4 文件、 png封面文件和json配置文件(可参考其他json)。
|
2、文件名 必须为 yyyymmddhhmmss.mp4 yyyymmddhhmmss.png yyyymmddhhmmss.json
|
如 20251009120000.mp4 20251009120000.png 20251009120000.json
|
|
本类执行的动作有两个：
|
0、 参考用户的配置文件中的user 的 enabled状态和 platform 的 enabled 状态。
|
1、 自动比对数据库中已有的任务清单，如果匹配上，就修改任务的视频状态为ready，后续上传进程会自动上传。
|
2、 如果任务时间，不在数据库中，可认为是临时的上传任务，因此添加到数据的任务表中，后续上传进程会自动上传。
|
|
"""
|
def __init__(self):
|
self.config_dir_path = Path(TASK_CONFIG_DIR)
|
self.media_dir_path = Path(VIDEO_DIR_FINAL)
|
self.db_path = DATABASE_PATH
|
self.logger = get_logger(self.__class__.__name__,"admin")
|
self.logger.info("VideoGeneratorManual started.")
|
|
def _get_avalible_media_times(self,user_id):
|
valid_times = []
|
now = datetime.now()
|
user_media_path = self.media_dir_path / user_id
|
|
for media_json in user_media_path.rglob("*.json"):
|
filename = media_json.stem # 去掉 .json 后缀
|
if not filename.isdigit() or len(filename) != 14:
|
print(f"非法文件名格式，跳过: {media_json.name}")
|
continue
|
|
try:
|
file_time = datetime.strptime(filename, "%Y%m%d%H%M%S")
|
except ValueError:
|
print(f"时间格式解析失败，跳过: {media_json.name}")
|
continue
|
|
# 如果不是将来的文件，跳过
|
if file_time <= now:
|
continue
|
|
base = media_json.with_suffix("") # 去掉 .json 的 Path
|
mp4_file = base.with_suffix(".mp4")
|
png_file = base.with_suffix(".png")
|
|
if mp4_file.exists() and png_file.exists():
|
formatted_time = file_time.strftime("%Y-%m-%d %H:%M:%S")
|
valid_times.append(formatted_time)
|
|
return valid_times
|
|
|
def run(self):
|
count = 0
|
count_new = 0
|
count_updated = 0
|
self.logger.info("VideoGeneratorManual 开始执行")
|
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")
|
user_media_times = self._get_avalible_media_times(user_id)
|
if not user_media_times :
|
continue
|
|
# 尝试给每一个video文件创建 N 个 task。 N的个数取决于 enabled 的platform数量。
|
for media_time in user_media_times:
|
|
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"]
|
|
config_updated = self._get_config_updated(config_file)
|
|
future_limit = datetime.now() + timedelta(days=platform_days_ahead)
|
|
if not media_time < future_limit :
|
continue
|
|
existing = publish_task_exists(user_id, platform_name, media_time)
|
if existing:
|
task_id, old_hash, task_status = existing
|
media_update_task_done(task_id,datetime.now().strftime("%Y-%m-%d %H:%M:%S"))
|
count_updated = count_updated +1
|
else:
|
publish_task_insert(user_id, platform_name, generator_class,generator_args_str,media_time, config_updated,proxy)
|
count_new =count_new+1
|
count += 1
|
except Exception as e:
|
self.logger.exception(f"VideoGeneratorManual 出错: {e}")
|
self.logger.info(f"累计检查 {count} 条任务。")
|
self.logger.info(f"{count_updated} 条任务已更新数据库")
|
self.logger.info(f"{count_new} 条任务已插入数据库")
|
|
