jianzhihuixiang/skills/novel-workshop/workflow.py

#!/usr/bin/env python3
"""
🎲 命题小说多模型创作工坊 v4
用法: python3 workflow.py "写作prompt" ["文档标题"]

流程: MiMo写初稿 → Gemini+Claude三路并行审阅 → Gemini改稿 → 本地存档 + 飞书文档完整写入
全程自动，飞书群聊实时进度推送。

配置: 自动从 ~/.openclaw/openclaw.json 读取 API Key 和飞书配置。
      可用环境变量覆盖: OPENROUTER_API_KEY, FEISHU_CHAT_ID, FEISHU_FOLDER_TOKEN, FEISHU_OWNER_OPEN_ID
"""

import sys, json, os, time, requests, concurrent.futures, re

# ============ 配置 ============
OPENROUTER_API_KEY = None
OPENROUTER_BASE = "https://openrouter.ai/api/v1"

MODELS = {
    "write": "xiaomi/mimo-v2-flash",
    "review_logic": "google/gemini-2.5-pro",
    "review_literary": "google/gemini-2.5-pro",
    "review_sharp": "anthropic/claude-opus-4.6",
    "revise": "google/gemini-2.5-pro",
}

FALLBACK_MODELS = {
    "xiaomi/mimo-v2-flash": "google/gemini-2.5-pro",
    "google/gemini-2.5-pro": "anthropic/claude-opus-4.6",
    "anthropic/claude-opus-4.6": "google/gemini-2.5-pro",
}

# 飞书配置
FEISHU_APP_ID = None
FEISHU_APP_SECRET = None
FEISHU_CHAT_ID = os.environ.get("FEISHU_CHAT_ID", "oc_4680833fd5ab374f3e26c90739ef6946")
FEISHU_FOLDER_TOKEN = os.environ.get("FEISHU_FOLDER_TOKEN", "U01TfC1RdlwEBzdHJCIcXwCQnVg")
FEISHU_OWNER_OPEN_ID = os.environ.get("FEISHU_OWNER_OPEN_ID", "ou_5b28826e6f7c9e54fcb49ba0b7e0b944")
FEISHU_TOKEN = None

# ============ 初始化 ============

def load_config():
    global OPENROUTER_API_KEY, FEISHU_APP_ID, FEISHU_APP_SECRET
    try:
        cfg = json.load(open(os.path.expanduser("~/.openclaw/openclaw.json")))
        providers = cfg.get("models", {}).get("providers", {})
        OPENROUTER_API_KEY = providers.get("openrouter", {}).get("apiKey", "")
        feishu = cfg.get("channels", {}).get("feishu", {})
        FEISHU_APP_ID = feishu.get("appId", "")
        FEISHU_APP_SECRET = feishu.get("appSecret", "")
    except Exception as e:
        print(f"⚠️ 读取配置失败: {e}")
    if not OPENROUTER_API_KEY:
        OPENROUTER_API_KEY = os.environ.get("OPENROUTER_API_KEY", "")
    if not OPENROUTER_API_KEY:
        print("❌ 找不到 OpenRouter API Key")
        sys.exit(1)

def get_feishu_token():
    global FEISHU_TOKEN
    if not FEISHU_APP_ID or not FEISHU_APP_SECRET:
        return None
    try:
        resp = requests.post("https://open.feishu.cn/open-apis/auth/v3/tenant_access_token/internal",
            json={"app_id": FEISHU_APP_ID, "app_secret": FEISHU_APP_SECRET}, timeout=10).json()
        FEISHU_TOKEN = resp.get("tenant_access_token", "")
        return FEISHU_TOKEN
    except:
        return None

# ============ 飞书消息推送 ============

def feishu_send(msg):
    """发送消息到飞书群聊"""
    if not FEISHU_TOKEN:
        print(f"  [飞书] (无token，跳过) {msg}")
        return
    try:
        resp = requests.post(
            f"https://open.feishu.cn/open-apis/im/v1/messages?receive_id_type=chat_id",
            headers={"Authorization": f"Bearer {FEISHU_TOKEN}", "Content-Type": "application/json"},
            json={
                "receive_id": FEISHU_CHAT_ID,
                "msg_type": "text",
                "content": json.dumps({"text": msg})
            },
            timeout=10
        ).json()
        if resp.get("code") == 0:
            print(f"  [飞书] ✅ 已发送: {msg[:50]}...")
        else:
            print(f"  [飞书] ⚠️ 发送失败: {resp.get('msg', 'unknown')}")
    except Exception as e:
        print(f"  [飞书] ⚠️ 发送异常: {e}")

# ============ OpenRouter API ============

def chat(model, messages, timeout=180):
    resp = requests.post(
        f"{OPENROUTER_BASE}/chat/completions",
        headers={
            "Authorization": f"Bearer {OPENROUTER_API_KEY}",
            "Content-Type": "application/json",
            "HTTP-Referer": "https://openclaw.ai"
        },
        json={"model": model, "messages": messages, "max_tokens": 16384},
        timeout=timeout
    )
    data = resp.json()
    if "error" in data:
        raise Exception(f"API error: {data['error']}")
    return data["choices"][0]["message"]["content"]

def chat_with_fallback(model, messages, role_name, timeout=180):
    try:
        print(f"  📡 {role_name}: 使用 {model}...")
        result = chat(model, messages, timeout)
        print(f"  ✅ {role_name}: 完成 ({len(result)} chars)")
        return result
    except Exception as e:
        print(f"  ⚠️ {role_name}: {model} 失败 ({e}), 尝试备选...")
        fallback = FALLBACK_MODELS.get(model)
        if fallback:
            try:
                result = chat(fallback, messages, timeout)
                print(f"  ✅ {role_name}: 备选 {fallback} 完成")
                return result
            except Exception as e2:
                print(f"  ❌ {role_name}: 备选也失败 ({e2})")
                return f"[审阅失败: {e2}]"
        return f"[审阅失败: {e}]"

def extract_score(text):
    matches = re.findall(r'(\d+(?:\.\d+)?)\s*/\s*10', text)
    return matches[0] if matches else "?"

# ============ 主流程 ============

def main():
    if len(sys.argv) < 2:
        print("用法: python3 workflow.py \"写作prompt\" [\"文档标题\"]")
        sys.exit(1)

    write_prompt = sys.argv[1]
    doc_title = sys.argv[2] if len(sys.argv) > 2 else "命题创作"

    load_config()
    get_feishu_token()

    start_time = time.time()

    # Step 0: 通知
    feishu_send(f"[░░░░░] 0/5 收到命题！工作流启动中 🎲")

    # ---- Step 1: 写初稿 ----
    print(f"\n{'='*50}\n  [█░░░░] 1/5 写初稿\n{'='*50}")

    draft = chat_with_fallback(
        MODELS["write"],
        [{"role": "user", "content": f"{write_prompt}\n\n直接输出正文（含标题），不需要任何解释或前言。"}],
        "初稿写作",
        timeout=240
    )

    # 提取标题（第一行去掉#号）
    first_line = draft.strip().split('\n')[0].strip().lstrip('#').strip()
    story_title = first_line if first_line else doc_title

    feishu_send(f"[█░░░░] 1/5 初稿完成 ✅《{story_title}》({len(draft)}字) 三路审阅启动中…")

    # ---- Step 2: 三路并行审阅 ----
    print(f"\n{'='*50}\n  [██░░░] 2/5 三路并行审阅\n{'='*50}")

    review_tasks = {
        "logic": {
            "model": MODELS["review_logic"],
            "name": "逻辑检阅",
            "prompt": f"你是严谨的文学逻辑审阅专家。审阅以下小说，从以下维度分析：\n1. 情节自洽性\n2. 时间线\n3. 因果关系\n4. 世界观矛盾\n\n评分1-10，详细说明。用中文。\n\n---以下是小说全文---\n\n{draft}"
        },
        "literary": {
            "model": MODELS["review_literary"],
            "name": "文学性分析",
            "prompt": f"你是资深文学评论家。审阅以下小说，从以下维度分析：\n1. 语言风格\n2. 意象构建\n3. 叙事手法\n4. 主题深度\n\n评分1-10，详细说明。用中文。\n\n---以下是小说全文---\n\n{draft}"
        },
        "sharp": {
            "model": MODELS["review_sharp"],
            "name": "锐评",
            "prompt": f"你是毒舌但有料的文学批评家，风格尖锐、一针见血。对以下小说进行锐评：\n1. 用最犀利的语言指出最大的问题\n2. 不要客气，不要留面子\n3. 给出3个核心改动建议（具体、可操作）\n\n用中文。\n\n---以下是小说全文---\n\n{draft}"
        }
    }

    reviews = {}
    with concurrent.futures.ThreadPoolExecutor(max_workers=3) as executor:
        futures = {}
        for key, task in review_tasks.items():
            future = executor.submit(
                chat_with_fallback, task["model"],
                [{"role": "user", "content": task["prompt"]}],
                task["name"], 180
            )
            futures[future] = key

        for future in concurrent.futures.as_completed(futures):
            key = futures[future]
            reviews[key] = future.result()

    logic_score = extract_score(reviews.get("logic", ""))
    literary_score = extract_score(reviews.get("literary", ""))

    feishu_send(f"[██░░░] 2/5 审阅完成 ✅ 逻辑 {logic_score}/10 | 文学 {literary_score}/10 | 改稿启动中…")

    # ---- Step 3: 改稿 ----
    print(f"\n{'='*50}\n  [███░░] 3/5 改稿\n{'='*50}")

    revision_prompt = f"""你是一位优秀的作家。以下是一篇小说初稿和三份审阅意见。请根据审阅意见全面修改小说，输出完整新版本。

修改要求：
1. 认真采纳锐评的建议，大胆改动
2. 保留原作优秀的部分
3. 让角色更真实立体
4. 结尾不要鸡汤金句，留白更好

直接输出修改后的完整小说（含标题），不需要解释修改了什么。

## 初稿
{draft}

## 审阅A：逻辑检阅
{reviews.get('logic', '[未完成]')}

## 审阅B：文学性分析
{reviews.get('literary', '[未完成]')}

## 审阅C：锐评
{reviews.get('sharp', '[未完成]')}"""

    revision = chat_with_fallback(
        MODELS["revise"],
        [{"role": "user", "content": revision_prompt}],
        "改稿",
        timeout=300
    )

    feishu_send(f"[███░░] 3/5 改稿完成 ✅ 保存中…")

    # ---- Step 4: 本地存档 ----
    print(f"\n{'='*50}\n  [████░] 4/5 存档\n{'='*50}")

    local_path = os.path.expanduser(f"~/.openclaw/workspace/novels/{doc_title}.md")
    os.makedirs(os.path.dirname(local_path), exist_ok=True)
    with open(local_path, "w") as f:
        f.write(f"# 🎲 命题创作《{doc_title}》— 多模型创作工坊\n\n")
        f.write(f"**日期**：{time.strftime('%Y-%m-%d')}\n")
        f.write(f"**评分**：逻辑 {logic_score}/10 | 文学 {literary_score}/10\n\n---\n\n")
        f.write(f"## Part 1：初稿\n\n{draft}\n\n---\n\n")
        f.write(f"## Part 2：三路审阅\n\n")
        f.write(f"### 逻辑检阅\n\n{reviews.get('logic', '[未完成]')}\n\n")
        f.write(f"### 文学性分析\n\n{reviews.get('literary', '[未完成]')}\n\n")
        f.write(f"### 锐评\n\n{reviews.get('sharp', '[未完成]')}\n\n---\n\n")
        f.write(f"## Part 3：终稿\n\n{revision}\n")
    print(f"  💾 本地保存: {local_path}")

    # 飞书文档创建 + 写入完整内容
    doc_id = ""
    doc_url = ""
    if FEISHU_TOKEN:
        doc_id = create_feishu_doc(f"🎲 命题创作《{doc_title}》— 多模型创作工坊", FEISHU_TOKEN)
        if doc_id:
            doc_url = f"https://feishu.cn/docx/{doc_id}"
            print(f"  📄 飞书文档已创建: {doc_url}")
            feishu_send(f"[████░] 4/5 存档完成 ✅ 正在写入飞书文档…")
            # 读取本地完整 md 文件，写入飞书文档（不省略任何内容！）
            write_feishu_doc_content(doc_id, local_path, FEISHU_TOKEN)
        else:
            feishu_send(f"[████░] 4/5 存档完成 ✅ 飞书文档创建失败，仅本地保存")
    else:
        feishu_send(f"[████░] 4/5 存档完成 ✅ 仅本地保存")

    # ---- Step 5: 完成通知 ----
    elapsed = time.time() - start_time

    if doc_url:
        final_msg = f"[█████] 5/5 全部完成！🎲\n\n⏱️ 总耗时: {elapsed:.0f} 秒\n📊 评分: 逻辑 {logic_score}/10 | 文学 {literary_score}/10\n📄 飞书文档: {doc_url}"
    else:
        final_msg = f"[█████] 5/5 写作完成！🎲\n\n⏱️ 总耗时: {elapsed:.0f} 秒\n📊 评分: 逻辑 {logic_score}/10 | 文学 {literary_score}/10\n💾 已保存到本地"
    feishu_send(final_msg)

    print(f"\n{'='*50}")
    print(f"  🎲 完成！总耗时: {elapsed:.0f} 秒")
    print(f"{'='*50}")

    # JSON 摘要
    summary = {
        "story_title": story_title,
        "doc_title": doc_title,
        "doc_url": doc_url,
        "doc_id": doc_id,
        "local_path": local_path,
        "scores": {"logic": logic_score, "literary": literary_score},
        "elapsed_seconds": round(elapsed),
        "status": "success"
    }
    print(f"\n__SUMMARY__:{json.dumps(summary, ensure_ascii=False)}")


def write_feishu_doc_content(doc_id, local_path, token):
    """读取本地 md 文件完整内容，通过飞书 API 分段写入文档。
    使用 openclaw CLI 的 feishu_doc write 工具，确保不省略任何内容。"""
    import subprocess
    try:
        with open(local_path, "r") as f:
            full_content = f.read()

        if not full_content.strip():
            print("  ⚠️ 本地文件为空，跳过写入")
            return

        # 使用 openclaw tool 调用 feishu_doc write
        # 先尝试直接用飞书 API 写入纯文本 block
        # 飞书文档 API: 创建文本块
        headers = {"Authorization": f"Bearer {token}", "Content-Type": "application/json"}

        # 将内容按段落拆分为 block
        lines = full_content.split('\n')
        blocks = []
        current_text = []

        for line in lines:
            stripped = line.strip()
            if stripped.startswith('# '):
                # 跳过一级标题（已经是文档标题）
                continue
            elif stripped.startswith('## '):
                if current_text:
                    blocks.append({"type": "text", "content": '\n'.join(current_text)})
                    current_text = []
                blocks.append({"type": "heading2", "content": stripped[3:]})
            elif stripped.startswith('### '):
                if current_text:
                    blocks.append({"type": "text", "content": '\n'.join(current_text)})
                    current_text = []
                blocks.append({"type": "heading3", "content": stripped[4:]})
            elif stripped == '---':
                if current_text:
                    blocks.append({"type": "text", "content": '\n'.join(current_text)})
                    current_text = []
                blocks.append({"type": "divider", "content": ""})
            else:
                current_text.append(line)

        if current_text:
            blocks.append({"type": "text", "content": '\n'.join(current_text)})

        # 使用飞书 Descendant API 批量创建 block
        doc_block_id = doc_id  # 根 block = 文档 ID

        created = 0
        for block in blocks:
            try:
                if block["type"] == "divider":
                    block_data = {
                        "block_type": 22,  # divider
                        "divider": {}
                    }
                elif block["type"] == "heading2":
                    block_data = {
                        "block_type": 4,  # heading2
                        "heading2": {
                            "elements": [{"text_run": {"content": block["content"]}}],
                            "style": {}
                        }
                    }
                elif block["type"] == "heading3":
                    block_data = {
                        "block_type": 5,  # heading3
                        "heading3": {
                            "elements": [{"text_run": {"content": block["content"]}}],
                            "style": {}
                        }
                    }
                else:
                    # 文本块，每段不超过 500 字符，超过则拆分
                    text = block["content"].strip()
                    if not text:
                        continue
                    # 按段落拆分
                    paragraphs = text.split('\n\n')
                    for para in paragraphs:
                        para = para.strip()
                        if not para:
                            continue
                        # 飞书单个文本块有字符限制，按行再拆
                        sub_lines = para.split('\n')
                        for sub_line in sub_lines:
                            sub_line_stripped = sub_line.strip()
                            if not sub_line_stripped:
                                continue
                            para_data = {
                                "block_type": 2,  # text
                                "text": {
                                    "elements": [{"text_run": {"content": sub_line_stripped}}],
                                    "style": {}
                                }
                            }
                            resp = requests.post(
                                f"https://open.feishu.cn/open-apis/docx/v1/documents/{doc_id}/blocks/{doc_block_id}/children",
                                headers=headers,
                                json={"children": [para_data], "index": -1},
                                timeout=15
                            ).json()
                            if resp.get("code") == 0:
                                created += 1
                            else:
                                print(f"  ⚠️ 写入失败: {resp.get('msg', '')[:80]}")
                    continue

                resp = requests.post(
                    f"https://open.feishu.cn/open-apis/docx/v1/documents/{doc_id}/blocks/{doc_block_id}/children",
                    headers=headers,
                    json={"children": [block_data], "index": -1},
                    timeout=15
                ).json()
                if resp.get("code") == 0:
                    created += 1
                else:
                    print(f"  ⚠️ 写入失败: {resp.get('msg', '')[:80]}")
            except Exception as e:
                print(f"  ⚠️ 写入异常: {e}")

        print(f"  📝 飞书文档写入完成: {created} 个文本块")

    except Exception as e:
        print(f"  ❌ 飞书文档写入失败: {e}")


def create_feishu_doc(title, token):
    try:
        resp = requests.post(
            "https://open.feishu.cn/open-apis/docx/v1/documents",
            headers={"Authorization": f"Bearer {token}", "Content-Type": "application/json"},
            json={"title": title, "folder_token": FEISHU_FOLDER_TOKEN},
            timeout=15
        ).json()
        doc_id = resp.get("data", {}).get("document", {}).get("document_id", "")
        if doc_id:
            try:
                requests.post(
                    f"https://open.feishu.cn/open-apis/drive/v1/permissions/{doc_id}/members?type=docx&need_notification=false",
                    headers={"Authorization": f"Bearer {token}", "Content-Type": "application/json"},
                    json={"member_type": "openid", "member_id": FEISHU_OWNER_OPEN_ID, "perm": "full_access"},
                    timeout=10
                )
            except:
                pass
        return doc_id
    except:
        return ""


if __name__ == "__main__":
    main()