import json import os import logging from util import api, oss TEMPLATES = {} logger = logging.getLogger("template") def check_local_template(local_name): template_def = TEMPLATES[local_name] base_dir = template_def.get("local_path") for video_part in template_def.get("video_parts", []): source_file = video_part.get("source", "") if str(source_file).startswith("http"): # download file ... elif str(source_file).startswith("PLACEHOLDER_"): continue else: if not os.path.isabs(source_file): source_file = os.path.join(base_dir, source_file) if not os.path.exists(source_file): logger.error(f"{source_file} not found, please check the template definition") raise Exception(f"{source_file} not found, please check the template definition") for audio in video_part.get("audios", []): if not os.path.isabs(audio): audio = os.path.join(base_dir, audio) if not os.path.exists(audio): logger.error(f"{audio} not found, please check the template definition") raise Exception(f"{audio} not found, please check the template definition") for lut in video_part.get("luts", []): if not os.path.isabs(lut): lut = os.path.join(base_dir, lut) if not os.path.exists(lut): logger.error(f"{lut} not found, please check the template definition") raise Exception(f"{lut} not found, please check the template definition") for mask in video_part.get("overlays", []): if not os.path.isabs(mask): mask = os.path.join(base_dir, mask) if not os.path.exists(mask): logger.error(f"{mask} not found, please check the template definition") raise Exception(f"{mask} not found, please check the template definition") def load_template(template_name, local_path): global TEMPLATES logger.info(f"加载视频模板定义:【{template_name}({local_path})】") template_def_file = os.path.join(local_path, "template.json") if os.path.exists(template_def_file): TEMPLATES[template_name] = json.load(open(template_def_file, 'rb')) TEMPLATES[template_name]["local_path"] = local_path try: check_local_template(template_name) logger.info(f"完成加载【{template_name}】模板") except Exception as e: logger.error(f"模板定义文件【{template_def_file}】有误,正在尝试重新下载模板", exc_info=e) download_template(template_name) def load_local_template(): for template_name in os.listdir(os.getenv("TEMPLATE_DIR")): if template_name.startswith("_"): continue if template_name.startswith("."): continue target_path = os.path.join(os.getenv("TEMPLATE_DIR"), template_name) if os.path.isdir(target_path): load_template(template_name, target_path) def get_template_def(template_id): if template_id not in TEMPLATES: download_template(template_id) return TEMPLATES.get(template_id) def download_template(template_id): template_info = api.get_template_info(template_id) if not os.path.isdir(template_info['local_path']): os.makedirs(template_info['local_path']) # download template assets overall_template = template_info['overall_template'] video_parts = template_info['video_parts'] def _download_assets(_template): if 'source' in _template: if str(_template['source']).startswith("http"): _, _fn = os.path.split(_template['source']) oss.download_from_oss(_template['source'], os.path.join(template_info['local_path'], _fn)) if _fn.endswith(".mp4"): from util.ffmpeg import to_annexb _fn = to_annexb(os.path.join(template_info['local_path'], _fn)) _template['source'] = os.path.relpath(_fn, template_info['local_path']) if 'overlays' in _template: for i in range(len(_template['overlays'])): overlay = _template['overlays'][i] if str(overlay).startswith("http"): _, _fn = os.path.split(overlay) oss.download_from_oss(overlay, os.path.join(template_info['local_path'], _fn)) _template['overlays'][i] = _fn if 'luts' in _template: for i in range(len(_template['luts'])): lut = _template['luts'][i] if str(lut).startswith("http"): _, _fn = os.path.split(lut) oss.download_from_oss(lut, os.path.join(template_info['local_path'], _fn)) _template['luts'][i] = _fn if 'audios' in _template: for i in range(len(_template['audios'])): if str(_template['audios'][i]).startswith("http"): _, _fn = os.path.split(_template['audios'][i]) oss.download_from_oss(_template['audios'][i], os.path.join(template_info['local_path'], _fn)) _template['audios'][i] = _fn _download_assets(overall_template) for video_part in video_parts: _download_assets(video_part) with open(os.path.join(template_info['local_path'], 'template.json'), 'w', encoding='utf-8') as f: json.dump(template_info, f) load_template(template_id, template_info['local_path']) def analyze_template(template_id): ...