# -*- coding: utf-8 -*-
|
import os
|
import random
|
import subprocess
|
from pathlib import Path
|
from typing import List, Tuple
|
from concurrent.futures import ThreadPoolExecutor
|
from pydub import AudioSegment
|
import tempfile
|
import time
|
from datetime import datetime
|
import re
|
import json
|
import re
|
from glob import glob
|
|
from auto_media_publisher.utils.logger import get_logger
|
from auto_media_publisher.config.conf_video_mashup import FINAL_MEDIA_DIR
|
|
from auto_media_publisher.generators.media_generator_base import MediaGenerator
|
from auto_media_publisher.generators.docx_generator_gemini import DocxGeneratorGemini
|
|
|
class DocxGeneratorCreator(MediaGenerator):
|
"""
|
docx生成器类，用于AI生成docx，发布微信公众号。
|
Attributes:
|
**kwargs : 其中包括 user_id,platform,proxy_full 和其他用得着addon 参数。
|
本类中,addon 参数是 docx_type 具体参考 conf_docx_random.py
|
"""
|
required_keys = ['user_id', 'scheduled_time','proxy_full']
|
def __init__(self, *args, **kwargs):
|
if args:
|
raise TypeError("不支持位置参数，请全部使用关键字参数")
|
|
for key in self.required_keys:
|
if key not in kwargs:
|
raise ValueError(f"缺少必要参数：{key}")
|
setattr(self, key, kwargs[key])
|
|
# 允许设置额外字段
|
for key, value in kwargs.items():
|
if key not in self.required_keys:
|
setattr(self, key, value)
|
|
self.logger = get_logger(self.__class__.__name__,"admin")
|
|
|
def generate(self):
|
"""
|
被外部调用。
|
生成一个docx ;
|
生成一个png封面;
|
生成一个docx的配置json文件;
|
三个文件都 保存到 media_final / user_id / 下面
|
|
"""
|
try:
|
tmp_datetime = datetime.strptime(self.scheduled_time, "%Y-%m-%d %H:%M:%S")
|
scheduled_time_str = tmp_datetime.strftime("%Y%m%d%H%M%S")
|
self.logger.info(f"DocxGeneratorCreator 正在生成 media：{self.user_id},{self.scheduled_time} .docx")
|
|
os.makedirs(os.path.join(FINAL_MEDIA_DIR,self.user_id),exist_ok=True)
|
output_docx_path =f"{FINAL_MEDIA_DIR}/{self.user_id}/{scheduled_time_str}.docx"
|
output_json_path = f"{FINAL_MEDIA_DIR}/{self.user_id}/{scheduled_time_str}.json"
|
|
# 针对不同的平台，如果配置了相同的发布时间，默认是使用同一个视频。
|
# 如果某个用户在不同的平台上，需要展示不同的内容，要 人为 地 设置最终发布时间是不同的时间。
|
|
if os.path.exists(output_json_path):
|
self.logger.info(f"DocxGeneratorCreator 文件已存在，不再执行生成任务 media：{output_json_path}")
|
return True
|
|
generater = DocxGeneratorGemini(self.docx_author,complex=True)
|
generater.generate(output_docx_path)
|
|
summary_json = generater.summary_info
|
summary_json["user"] = self.user_id
|
summary_json["scheduled_time"] = self.scheduled_time
|
summary_json["proxy_full"] = self.proxy_full
|
summary_json["path_media"] = output_docx_path
|
summary_json["author"] = self.docx_author
|
|
|
# 3. 写入磁盘
|
with open(output_json_path.replace(".docx",".json"), "w", encoding="utf-8") as f:
|
json.dump(summary_json, f, ensure_ascii=False, indent=4)
|
|
time.sleep(5)
|
return True
|
except Exception as e:
|
self.logger.error(f"DocxGeneratorCreator error: {e}")
|
return False
|
|
