#!/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()