yu-zhi-ran/scripts/shortvideo.py

#!/usr/bin/env python3
"""
长文 → 短视频脚本改写器
读取某选题的长文（知乎/公众号 markdown），生成 N 条差异化短视频口播/分镜脚本。
"""
import argparse
import datetime
import json
import logging
import sys
from pathlib import Path
from typing import Dict, List, Optional

PROJECT_ROOT = Path(__file__).resolve().parent.parent
sys.path.insert(0, str(PROJECT_ROOT))
sys.path.insert(0, str(PROJECT_ROOT / 'platform' / 'backend'))

from db_helper import get_topic_by_id, get_articles_by_topic, save_article
from prompt_loader import get_prompt, get_prompt_params

try:
    from app.core.nvidia_client import call_llm
    HAVE_LLM = True
except ImportError:
    HAVE_LLM = False

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"shortvideo_{TODAY}.log"),
        logging.StreamHandler(),
    ],
)
logger = logging.getLogger(__name__)

SYSTEM_PROMPT = "你是短视频脚本编剧，擅长把长文改成高完播率口播脚本"

_VARIANT_ANGLES = [
    ("观点冲突型", "从一个反直觉/有争议的观点切入，制造认知冲突，让观众想反驳或认同"),
    ("干货清单型", "提炼 3-5 条可立刻上手的干货要点，节奏快、信息密度高"),
    ("故事共鸣型", "用一个真实人物/生活场景的故事开头，引发情绪共鸣，再带出观点"),
]


def _build_prompt(source_text: str, angle_name: str, angle_desc: str) -> str:
    return f"""你是一个短视频脚本编剧。下面是一篇长文的内容，请把它改写成一条短视频脚本。

改写角度：{angle_name}——{angle_desc}

长文素材：
{source_text}

请严格按以下结构输出（用 markdown，各级标题不要省略）：

## 标题（爆款钩子）
写 1 个 15-25 字的短视频标题，带强钩子（悬念/冲突/数字/结果），不要标题党到离谱。

## 口播文案
写 60-90 秒的口语化口播稿，约 200-350 字。像真人对着镜头说话，有停顿标记（用「［停顿］」标注换气/节奏点）。不要书面语，不要「首先其次最后」套路。

## 分镜建议
给出 3-5 个镜头，每个镜头包含：
- 画面：具体拍什么（景别/动作/场景）
- 字幕：屏幕上打出的关键词
- 时长：几秒

用列表或表格呈现，每个镜头一行。

## 适配平台
说明这条脚本分别适配 抖音 / 视频号 / 小红书视频 时的微调建议（各 1 句话）。

只输出上面的 markdown，不要其他说明。"""


def _template_markdown(topic_title: str, angle_name: str) -> str:
    return f"""## 标题（爆款钩子）
（待生成）{topic_title} - {angle_name}角度

## 口播文案
［LLM 不可用，使用模板占位。请稍后手动补充 60-90 秒口播稿，并用［停顿］标记节奏。］

## 分镜建议
- 画面：（占位） 字幕：（占位） 时长：（占位）

## 适配平台
- 抖音：（占位）
- 视频号：（占位）
- 小红书视频：（占位）
"""


def _load_source(topic: Dict, articles: List[Dict]) -> str:
    for platform in ("zhihu", "wechat"):
        for a in articles:
            if a.get("platform") == platform and a.get("html_content"):
                return a["html_content"]
    if topic.get("core_concept"):
        return f"# {topic.get('title', '')}\n\n{topic.get('core_concept')}"
    return topic.get("title", "")


def _generate_one(source: str, idx: int, topic_title: str) -> str:
    angle_name, angle_desc = _VARIANT_ANGLES[(idx - 1) % len(_VARIANT_ANGLES)]
    if HAVE_LLM:
        try:
            params = get_prompt_params("shortvideo_script") or {}
            temperature = params.get("temperature", 0.8)
            max_tokens = params.get("max_tokens", 2000)
            prompt = _build_prompt(source, angle_name, angle_desc)
            result = call_llm(
                prompt,
                temperature=temperature,
                max_tokens=max_tokens,
                system_prompt=SYSTEM_PROMPT,
            )
            if result and result.strip():
                return result.strip()
        except Exception as e:
            logger.warning("LLM 调用失败（variant %s）：%s", idx, e)
    logger.warning("LLM 不可用或失败，写入模板占位（variant %s）", idx)
    return _template_markdown(topic_title, angle_name)


def _extract_title(markdown: str, fallback: str) -> str:
    for line in markdown.splitlines():
        line = line.strip()
        if line.startswith("## 标题") or line.startswith("# 标题"):
            rest = line.split("标题", 1)[-1].strip("（()：: ")
            if rest:
                return rest[:60]
    first = next((l.strip("# ").strip() for l in markdown.splitlines() if l.strip()), "")
    return first[:60] if first else fallback


def main():
    parser = argparse.ArgumentParser(description="长文转短视频脚本")
    parser.add_argument("--topic-id", required=True, help="选题 ID")
    parser.add_argument("--count", type=int, default=3, help="生成脚本数量（默认 3）")
    args = parser.parse_args()

    topic_id = args.topic_id
    count = max(1, args.count)

    files: List[str] = []
    out_dir = DATA_DIR / "shortvideos" / TODAY
    out_dir.mkdir(parents=True, exist_ok=True)

    topic = get_topic_by_id(topic_id)
    if not topic:
        logger.error("选题不存在：%s", topic_id)
        print(json.dumps({"ok": False, "topic_id": topic_id, "count": 0, "files": []}, ensure_ascii=False))
        sys.exit(1)

    articles = get_articles_by_topic(topic_id)
    source = _load_source(topic, articles)
    topic_title = topic.get("title", topic_id)

    ok = True
    for i in range(1, count + 1):
        markdown = _generate_one(source, i, topic_title)
        title = _extract_title(markdown, f"{topic_title}_短视频{i}")
        file_path = out_dir / f"{topic_id}_shortvideo_{i}.md"
        try:
            file_path.write_text(markdown, encoding="utf-8")
            files.append(str(file_path))
            logger.info("已写出脚本文件：%s", file_path)
        except Exception as e:
            logger.error("写文件失败：%s", e)
            ok = False
            continue

        try:
            save_article(topic_id, f"shortvideo_{i}", html_content=None, title=title, content=markdown)
            logger.info("已写入 DB：shortvideo_%s / %s", i, topic_id)
        except Exception as e:
            logger.warning("DB 保存失败（文件已落盘）：%s", e)

    summary = {"ok": ok, "topic_id": topic_id, "count": count, "files": files}
    print(json.dumps(summary, ensure_ascii=False))
    sys.exit(0 if ok else 1)


if __name__ == "__main__":
    main()