yu-zhi-ran/scripts/creator.py

#!/usr/bin/env python3
"""
宇之然内容创作流水线（研究 → 大纲 → 撰写 → 合规优化）v3 - DB version
"""

import json, datetime, logging, sys, subprocess
from pathlib import Path
from typing import Dict

PROJECT_ROOT = Path('/root/openclaw-workspace/projects/yu-zhi-ran')
sys.path.insert(0, str(PROJECT_ROOT))

# 导入数据库辅助模块
from db_helper import get_topic_by_id, get_next_topic, update_topic_status

DATA_DIR = PROJECT_ROOT / "automation" / "data"
LOGS_DIR = PROJECT_ROOT / "automation" / "logs"
TODAY = datetime.datetime.now().strftime("%Y-%m-%d")

logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(levelname)s - %(message)s',
    handlers=[
        logging.FileHandler(LOGS_DIR / f"creator_{TODAY}.log"),
        logging.StreamHandler()
    ]
)
logger = logging.getLogger(__name__)

def select_next_topic(topic_id: str = None) -> Dict:
    """选择并锁定要创作的选题（趋势引擎匹配 → 回退优先级）"""
    if topic_id:
        topic = get_topic_by_id(topic_id)
        if not topic:
            raise ValueError(f"Topic {topic_id} not found")
        current_status = topic.get('status')
        if current_status in ['已发布', 'published']:
            raise ValueError(f"Topic {topic_id} is already published, cannot recreate")
        update_topic_status(topic_id, 'review')
        return topic

    try:
        from topic_selector import select_best_topic as engine_select
        topic = engine_select()
        if topic:
            logger.info(f"选题引擎推荐: {topic['id']} {topic['title']}")
            update_topic_status(topic['id'], 'review')
            return topic
    except Exception as e:
        logger.warning(f"选题引擎失效，回退简单策略: {e}")

    topic = get_next_topic(priority='高') or get_next_topic()
    if not topic:
        raise ValueError("No available topics to create (all locked or wrong status)")
    update_topic_status(topic['id'], 'review')
    return topic

def run_step(script_name: str, topic_id: str) -> bool:
    """运行一个流水线步骤（research/outline/writer）"""
    script_path = PROJECT_ROOT / "scripts" / script_name
    cmd = ["python3", str(script_path), "--topic-id", topic_id]
    logger.info(f"Running: {' '.join(cmd)}")
    result = subprocess.run(cmd, cwd=str(PROJECT_ROOT), capture_output=True, text=True, timeout=1800)
    if result.returncode != 0:
        logger.error(f"{script_name} 失败: {result.stderr}")
        return False
    logger.info(f"{script_name} 完成: {result.stdout.strip()}")
    return True

def run_optimizer_step(topic_id: str) -> bool:
    """运行合规优化步骤（只针对单个选题）"""
    script_path = PROJECT_ROOT / "scripts" / "compliance_optimizer.py"
    cmd = ["python3", str(script_path), "--topic-ids", topic_id]
    logger.info(f"Running: {' '.join(cmd)}")
    result = subprocess.run(cmd, cwd=str(PROJECT_ROOT), capture_output=True, text=True, timeout=600)
    if result.returncode != 0:
        logger.error(f"compliance_optimizer 失败: {result.stderr}")
        return False
    logger.info(f"compliance_optimizer 完成: {result.stdout.strip()}")
    return True

def run_pipeline(topic_id: str = None) -> Dict:
    """运行完整流水线：研究 → 大纲 → 撰写 → 合规优化"""
    tid = None
    try:
        topic = select_next_topic(topic_id)
        tid = topic['id']
        logger.info(f"开始创作流水线: topic_id={tid}, title={topic.get('title')}")

        # 1. 研究
        if not run_step("research.py", tid):
            update_topic_status(tid, 'pending')
            return {"ok": False, "error": "research step failed"}

        # 2. 大纲
        if not run_step("outline.py", tid):
            update_topic_status(tid, 'pending')
            return {"ok": False, "error": "outline step failed"}

        # 3. 撰写
        if not run_step("writer.py", tid):
            update_topic_status(tid, 'pending')
            return {"ok": False, "error": "writer step failed"}

        # 4. 配图生成
        image_ok = run_step("image_generator.py", tid)

        # 5. 合规优化（自动审核并标记为「待发布」）
        if not run_optimizer_step(tid):
            update_topic_status(tid, 'pending')
            return {"ok": False, "error": "optimizer step failed"}

        logger.info(f"创作流水线完成: topic_id={tid}")
        return {"ok": True, "topic_id": tid, "stdout": f"SUCCESS: Topic {tid} processed through full pipeline{' (images generated)' if image_ok else ' (images skipped)'}"}
    except Exception as e:
        logger.exception("流水线执行失败")
        if tid:
            update_topic_status(tid, 'pending')
        return {"ok": False, "error": str(e)}

def main():
    import argparse
    parser = argparse.ArgumentParser(description='内容创作流水线（研究→大纲→撰写→合规优化）')
    parser.add_argument('--topic-id', help='指定选题ID，不指定则自动选择待处理选题')
    args = parser.parse_args()

    result = run_pipeline(args.topic_id)
    print(json.dumps(result, ensure_ascii=False))
    sys.exit(0 if result['ok'] else 1)

if __name__ == "__main__":
    main()