mcn-short-video/project/短视频脚本创作/V1.0/脚本创作技能/scripts/MCN_CYLG_API.py

#!/usr/bin/env python3
"""MCN 创意灵感接口 — Dify chat-messages 调用脚本。

两种用法:
  1. 联网搜索（search）  — 输入创作意图 + 搜索内容，联网获取最新信息
  2. 查询事件库（recall） — 输入创作意图 + 搜索标签和内容，查询事件库中的知识

用法示例:

  # 方式一：联网搜索
  $env:DIFY_MCN_CYLG_KEY="app-xxxx"
  python scripts/MCN_CYLG_API.py search \\
      --intent "为萌宠账号写一个猫咪应激的搞笑短视频" \\
      --content "抖音上'耄耋猫'是什么梗，有哪些爆款视频" \\
      --timeout 300

  # 方式二：查询事件库
  $env:DIFY_MCN_CYLG_KEY="app-xxxx"
  python scripts/MCN_CYLG_API.py recall \\
      --intent "为萌宠账号写一个猫咪应激的搞笑短视频" \\
      --tags "萌宠,搞笑,应激,反差萌" \\
      --content "猫咪应激反应的表现和常见搞笑场景" \\
      --timeout 300

接口文档: references/接口调用/MCN_CYLG_python.md
"""

import argparse
import json
import os
import sys
import urllib.request
import urllib.error

# ---------------------------------------------------------------------------
# 默认值
# ---------------------------------------------------------------------------

DIFY_URL = os.environ.get("DIFY_URL", "https://mydify.youmanvideo.com/v1")
DIFY_ENDPOINT = f"{DIFY_URL.rstrip('/')}/chat-messages"
DEFAULT_MODEL = os.environ.get("DIFY_MCN_MODEL", "gemini")
DEFAULT_TIMEOUT = 1200  # 秒


# ---------------------------------------------------------------------------
# query 构造
# ---------------------------------------------------------------------------

def build_search_query(intent: str, content: str) -> str:
    """构造联网搜索模式的 query。

    Args:
        intent:  创作意图（账号定位、视频目标等）
        content: 需要搜索的内容（热词释义、热点趋势、竞品分析等）

    Returns:
        组装后的 query 字符串
    """
    return (
        f"联网搜索\n"
        f"创作意图：{intent}\n"
        f"搜索内容：{content}"
    )


def build_recall_query(intent: str, tags: str, content: str) -> str:
    """构造查询事件库模式的 query。

    Args:
        intent:  创作意图（账号定位、视频目标等）
        tags:    搜索标签（逗号分隔的关键词，用于事件库检索匹配）
        content: 搜索内容（需要查询的具体事件/知识描述）

    Returns:
        组装后的 query 字符串
    """
    return (
        f"查询事件库\n"
        f"创作意图：{intent}\n"
        f"搜索标签：{tags}\n"
        f"搜索内容：{content}"
    )


# ---------------------------------------------------------------------------
# 核心调用
# ---------------------------------------------------------------------------

def call_dify(
    query: str,
    is_think: str = "0",
    is_online: str = "0",
    model: str = DEFAULT_MODEL,
    response_mode: str = "blocking",
    user: str = "wb-skill-001",
    timeout: int = DEFAULT_TIMEOUT,
) -> dict:
    """调用 Dify 创意灵感接口，返回解析后的 JSON 响应。

    Args:
        query:          组装后的完整 query
        is_think:       是否深度思考 ("0" 或 "1")
        is_online:      是否联网搜索 ("0" 或 "1")
        model:          模型名称
        response_mode:  "blocking" 或 "streaming"
        user:           用户唯一标识
        timeout:        HTTP 读取超时（秒）

    Returns:
        dict: API 原始响应

    Raises:
        SystemExit: 缺少密钥、网络错误、API 返回非 200 时退出
    """
    api_key = os.environ.get("DIFY_MCN_CYLG_KEY", "").strip()
    if not api_key:
        print("ERROR: 环境变量 DIFY_MCN_CYLG_KEY 未设置。", file=sys.stderr)
        print("请先执行: $env:DIFY_MCN_CYLG_KEY=\"app-xxxx\"", file=sys.stderr)
        sys.exit(1)

    payload = json.dumps({
        "inputs": {
            "is_think": is_think,
            "is_online": is_online,
            "model": model,
        },
        "query": query,
        "response_mode": response_mode,
        "user": user,
    }).encode("utf-8")

    req = urllib.request.Request(
        DIFY_ENDPOINT,
        data=payload,
        headers={
            "Authorization": f"Bearer {api_key}",
            "Content-Type": "application/json; charset=utf-8",
        },
        method="POST",
    )

    try:
        with urllib.request.urlopen(req, timeout=timeout) as resp:
            body = resp.read().decode("utf-8")
            return json.loads(body)
    except urllib.error.HTTPError as e:
        print(f"ERROR: HTTP {e.code} — {e.reason}", file=sys.stderr)
        try:
            print(e.read().decode("utf-8", errors="replace")[:500], file=sys.stderr)
        except Exception:
            pass
        sys.exit(2)
    except urllib.error.URLError as e:
        print(f"ERROR: 网络请求失败 — {e.reason}", file=sys.stderr)
        sys.exit(3)
    except Exception as e:
        print(f"ERROR: {e}", file=sys.stderr)
        sys.exit(4)


# ---------------------------------------------------------------------------
# CLI — 子命令模式
# ---------------------------------------------------------------------------

def cmd_search(args: argparse.Namespace) -> None:
    """联网搜索模式：is_online=1，提示词中说明"联网搜索"。"""
    query = build_search_query(args.intent, args.content)
    result = call_dify(
        query=query,
        is_think=args.is_think,
        is_online="1",  # 联网搜索固定开启
        model=args.model,
        response_mode=args.response_mode,
        user=args.user,
        timeout=args.timeout,
    )
    print(json.dumps(result, ensure_ascii=False, indent=2))


def cmd_recall(args: argparse.Namespace) -> None:
    """查询事件库模式：is_online=0，提示词中写明'查询事件库'。"""
    query = build_recall_query(args.intent, args.tags, args.content)
    result = call_dify(
        query=query,
        is_think=args.is_think,
        is_online="0",  # 不联网，走事件库查询
        model=args.model,
        response_mode=args.response_mode,
        user=args.user,
        timeout=args.timeout,
    )
    print(json.dumps(result, ensure_ascii=False, indent=2))


def main() -> None:
    parser = argparse.ArgumentParser(
        description="MCN 创意灵感 — Dify chat-messages 调用工具",
        formatter_class=argparse.RawDescriptionHelpFormatter,
        epilog="""
示例:
  # 联网搜索
  python MCN_CYLG_API.py search \\
      --intent "为萌宠账号写猫咪应激搞笑短视频" \\
      --content "抖音耄耋猫是什么梗，有哪些爆款" \\
      --timeout 300

  # 查询事件库
  python MCN_CYLG_API.py recall \\
      --intent "为萌宠账号写猫咪应激搞笑短视频" \\
      --tags "萌宠,搞笑,应激,反差萌" \\
      --content "猫咪应激反应的表现和搞笑场景" \\
      --timeout 300
        """,
    )

    subparsers = parser.add_subparsers(dest="mode", required=True, help="调用模式")

    # 公共参数
    def add_common_args(sub):
        sub.add_argument("--is-think", default="0", choices=["0", "1"],
                         help="是否深度思考 (0=否, 1=是)")
        sub.add_argument("--model", default=DEFAULT_MODEL, help="模型名称")
        sub.add_argument("--response-mode", default="blocking", choices=["blocking", "streaming"],
                         help="响应模式")
        sub.add_argument("--user", default="wb-skill-001", help="用户唯一标识")
        sub.add_argument("--timeout", type=int, default=DEFAULT_TIMEOUT, help="HTTP 超时（秒）")

    # 子命令: search（联网搜索）
    p_search = subparsers.add_parser(
        "search", help="联网搜索模式 — 输入创作意图+搜索内容，联网获取最新信息"
    )
    p_search.add_argument("--intent", required=True,
                          help="创作意图（账号定位、视频目标等）")
    p_search.add_argument("--content", required=True,
                          help="需要搜索的内容（热词释义、热点趋势、竞品分析等）")
    add_common_args(p_search)
    p_search.set_defaults(func=cmd_search)

    # 子命令: recall（查询事件库）
    p_recall = subparsers.add_parser(
        "recall", help="查询事件库模式 — 输入创作意图+搜索标签+搜索内容，查询事件库"
    )
    p_recall.add_argument("--intent", required=True,
                          help="创作意图（账号定位、视频目标等）")
    p_recall.add_argument("--tags", required=True,
                          help="搜索标签（逗号分隔的关键词，用于事件库检索匹配）")
    p_recall.add_argument("--content", required=True,
                          help="需要查询的具体事件/知识描述")
    add_common_args(p_recall)
    p_recall.set_defaults(func=cmd_recall)

    args = parser.parse_args()
    args.func(args)


if __name__ == "__main__":
    main()