f8894dd3f3
- updateNavButtons: 改用数组索引查找，而非简单的+1/-1 - updateSidebarHighlight: 使用parseFloat比较章节ID - updateMobileTOCHighlight: 使用parseFloat比较章节ID - 更新版本号强制刷新缓存
463 lines
19 KiB
Python
463 lines
19 KiB
Python
#!/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()
|