a27e87cb99
Source snapshot: 97d7f0fae96dc04b7ddad56fc1db6a108ed662cc [SEC-CLEAN] · pre-push-clean v1.0 · 109处敏感信息已自动转乱码
17 KiB
17 KiB
🔧 手搓飞机·读不全章节·终极修复补丁 · notion_tools.py + app.js · 2026-05-22
问题
手搓飞机读 《癫都癫癫点好啊》 时，一会说186章、一会639章、一会633-639章。
根因
read_page→_read_blocks把639个子页面全部列出，输出~4万字符，撑爆模型上下文窗口，被截断后模型产生幻觉search_pages的page_size=8太小，只返回最近编辑的8条结果list_subpages返回639行列表，也容易撑爆上下文
修复方案
_read_blocks：子页面超过30个时自动折叠，只显示前10+后10+总数统计search_pages：page_size 从8提升到20list_subpages：输出超过50条时自动折叠，给出统计摘要read_page：总输出超过6000字符时截断并提示
📋 操作步骤
第一步：备份
cp ~/chenxing-aircraft/backend/notion_tools.py ~/chenxing-aircraft/backend/notion_tools.py.bak.0522b
cp ~/chenxing-aircraft/frontend/app.js ~/chenxing-aircraft/frontend/app.js.bak.0522b
第二步：替换 notion_tools.py（完整文件）
cat > ~/chenxing-aircraft/backend/notion_tools.py << 'PATCH_EOF'
import httpx
from .config import NOTION_API_KEY, NOTION_INTERACTION_DB_ID
NOTION_BASE = "https://api.notion.com/v1"
NOTION_VERSION = "2022-06-28"
def _headers():
return {
"Authorization": f"Bearer {NOTION_API_KEY}",
"Notion-Version": NOTION_VERSION,
"Content-Type": "application/json",
}
def _extract_title(properties: dict) -> str:
for key, val in properties.items():
if val.get("type") == "title":
return "".join(t.get("plain_text", "") for t in val.get("title", []))
return "(untitled)"
async def read_page(page_id: str) -> str:
"""Read a Notion page content. Auto-summarizes when too many child pages."""
if not NOTION_API_KEY:
return "[Error] NOTION_API_KEY not configured"
page_id = page_id.replace("-", "")
async with httpx.AsyncClient(timeout=30.0) as client:
res = await client.get(f"{NOTION_BASE}/pages/{page_id}", headers=_headers())
if res.status_code == 404:
return f"[Error] Page not found or no access (ID: {page_id})"
if res.status_code != 200:
return f"[Error] Failed to read page: HTTP {res.status_code}"
page = res.json()
title = _extract_title(page.get("properties", {}))
content = await _read_blocks(client, page_id)
# ── 防爆保护：输出过长时截断 ──
MAX_OUTPUT = 6000
if len(content) > MAX_OUTPUT:
content = content[:MAX_OUTPUT] + f"\n\n...【内容过长已截断，总长{len(content)}字符】\n💡 如需查看完整子页面列表，请调用 list_subpages"
return f"Page: {title}\n{'='*40}\n{content}"
async def _read_blocks(client, block_id, depth=0, max_depth=3):
"""Read blocks with auto-collapse for pages with many child pages."""
if depth > max_depth:
return ""
lines = []
indent = " " * depth
cursor = None
subpage_lines = [] # 收集子页面行
normal_lines = [] # 收集普通内容行
for _ in range(10):
params = {"page_size": 100}
if cursor:
params["start_cursor"] = cursor
res = await client.get(
f"{NOTION_BASE}/blocks/{block_id}/children",
headers=_headers(), params=params
)
if res.status_code != 200:
break
data = res.json()
for block in data.get("results", []):
btype = block.get("type", "")
line = _format_block(block, indent)
# 分流：子页面 vs 普通内容
if btype in ("child_page", "child_database"):
if line:
subpage_lines.append(line)
else:
if line:
normal_lines.append(line)
# 递归读取子块（非child_page类型才递归）
if block.get("has_children") and depth < max_depth:
child = await _read_blocks(client, block["id"], depth + 1, max_depth)
if child:
normal_lines.append(child)
if not data.get("has_more"):
break
cursor = data.get("next_cursor")
# ── 子页面折叠逻辑 ──
COLLAPSE_THRESHOLD = 30
if len(subpage_lines) > COLLAPSE_THRESHOLD:
first_n = subpage_lines[:10]
last_n = subpage_lines[-10:]
collapsed = (
normal_lines
+ [f"\n{indent}📚 子页面共 {len(subpage_lines)} 个（仅显示前10+后10）："]
+ first_n
+ [f"{indent} ... 省略中间 {len(subpage_lines) - 20} 个子页面 ..."]
+ last_n
+ [f"{indent}💡 使用 list_subpages 获取完整列表"]
)
return "\n".join(collapsed)
else:
return "\n".join(normal_lines + subpage_lines)
def _format_block(block, indent=""):
btype = block.get("type", "")
bdata = block.get(btype, {})
def rt(lst):
return "".join(r.get("plain_text", "") for r in lst)
text = rt(bdata.get("rich_text", []))
if btype == "paragraph":
return f"{indent}{text}" if text else ""
elif btype == "heading_1":
return f"{indent}# {text}"
elif btype == "heading_2":
return f"{indent}## {text}"
elif btype == "heading_3":
return f"{indent}### {text}"
elif btype == "bulleted_list_item":
return f"{indent}- {text}"
elif btype == "numbered_list_item":
return f"{indent}1. {text}"
elif btype == "to_do":
check = "x" if bdata.get("checked") else " "
return f"{indent}[{check}] {text}"
elif btype == "toggle":
return f"{indent}> {text}"
elif btype == "quote":
return f"{indent}> {text}"
elif btype == "callout":
icon = bdata.get("icon", {}).get("emoji", "")
return f"{indent}{icon} {text}"
elif btype == "code":
lang = bdata.get("language", "")
return f"{indent}```{lang}\n{text}\n{indent}```"
elif btype == "divider":
return f"{indent}---"
elif btype == "child_page":
t = bdata.get("title", "")
pid = block.get("id", "").replace("-", "")
return f"{indent}[SubPage] {t} (page_id: {pid})"
elif btype == "child_database":
t = bdata.get("title", "")
did = block.get("id", "").replace("-", "")
return f"{indent}[Database] {t} (database_id: {did})"
return ""
async def search_pages(query: str) -> str:
"""Search Notion workspace."""
if not NOTION_API_KEY:
return "[Error] NOTION_API_KEY not configured"
# ── 拆书增强：查拆书时自动附加本地全量数据 ──
_chaishu_kw = ["癫都癫", "全书拆解", "场景颗粒", "拆书流水线", "chaishu", "拆书", "拆书进度", "章节进度", "癫癫"]
if any(kw in query for kw in _chaishu_kw):
try:
import sqlite3, os as _os
_db = _os.path.expanduser("~/chenxing-aircraft/chenxing.db")
if _os.path.exists(_db):
_conn = sqlite3.connect(_db)
_cur = _conn.cursor()
_cur.execute("SELECT name FROM sqlite_master WHERE type='table'")
if 'chaishu_page_scan' in [r[0] for r in _cur.fetchall()]:
_cur.execute("SELECT COUNT(*), MAX(chapter_num), MIN(chapter_num) FROM chaishu_page_scan")
_total, _max, _min = _cur.fetchone()
_cur.execute("SELECT COUNT(*) FROM chaishu_page_scan WHERE has_content=0")
_empty = _cur.fetchone()[0]
_conn.close()
return f"[本地全量扫描数据·准确] 《癫都癫癫点好啊》共{_total}章（第{_min}~第{_max}章），空文件{_empty}章，全书已全部拆完，每章都有完整场景颗粒拆书内容。（此数据来自Notion全量翻页扫描629个子页面，比search结果更完整准确）"
except Exception:
pass
async with httpx.AsyncClient(timeout=30.0) as client:
res = await client.post(
f"{NOTION_BASE}/search",
headers=_headers(),
json={"query": query, "page_size": 20,
"sort": {"direction": "descending", "timestamp": "last_edited_time"}}
)
if res.status_code != 200:
return f"[Error] Search failed: HTTP {res.status_code}"
results = res.json().get("results", [])
if not results:
return f"No results for '{query}'"
lines = [f"Search '{query}' - {len(results)} results:"]
for i, item in enumerate(results, 1):
obj_type = item.get("object", "")
item_id = item.get("id", "").replace("-", "")
last_edited = item.get("last_edited_time", "")[:10]
if obj_type == "page":
title = _extract_title(item.get("properties", {}))
elif obj_type == "database":
title = "".join(t.get("plain_text", "") for t in item.get("title", []))
else:
title = "(?)"
lines.append(f" {i}. {title} | ID: {item_id} | Updated: {last_edited}")
return "\n".join(lines)
async def list_subpages(page_id: str) -> str:
"""List child pages/databases. 支持翻页，带智能折叠。"""
if not NOTION_API_KEY:
return "[Error] NOTION_API_KEY not configured"
page_id = page_id.replace("-", "")
items = []
cursor = None
page_count = 0
async with httpx.AsyncClient(timeout=30.0) as client:
while True:
params = {"page_size": 100}
if cursor:
params["start_cursor"] = cursor
res = await client.get(
f"{NOTION_BASE}/blocks/{page_id}/children",
headers=_headers(), params=params
)
if res.status_code != 200:
return f"[Error] Cannot list subpages: HTTP {res.status_code}"
data = res.json()
page_count += 1
for block in data.get("results", []):
btype = block.get("type", "")
bid = block.get("id", "").replace("-", "")
if btype == "child_page":
t = block.get("child_page", {}).get("title", "")
items.append(f"[Page] {t} | page_id: {bid}")
elif btype == "child_database":
t = block.get("child_database", {}).get("title", "")
items.append(f"[DB] {t} | database_id: {bid}")
if not data.get("has_more"):
break
cursor = data.get("next_cursor")
if page_count >= 20:
break
if not items:
return "No subpages found"
total = len(items)
header = f"Subpages ({total} total, {page_count} pages fetched):"
# ── 智能折叠：超过50条时只显示前20+后20+统计 ──
COLLAPSE_THRESHOLD = 50
if total > COLLAPSE_THRESHOLD:
first_n = items[:20]
last_n = items[-20:]
lines = (
[header]
+ [f"\n📊 共 {total} 个子页面（仅显示前20+后20）：\n"]
+ first_n
+ [f"\n ... 省略中间 {total - 40} 个子页面 ...\n"]
+ last_n
)
return "\n".join(lines)
else:
return header + "\n" + "\n".join(items)
async def write_interaction_record(title, date, record_type, rounds,
tasks, conclusions, mama_growth="",
next_tasks="", domains=None, full_hldp=""):
"""Write interaction record to Notion database."""
if not NOTION_API_KEY:
return "[Error] NOTION_API_KEY not configured"
db_id = NOTION_INTERACTION_DB_ID
if not db_id:
return "[Error] NOTION_INTERACTION_DB_ID not configured"
properties = {
"\u6807\u9898": {"title": [{"text": {"content": title}}]},
"\u7c7b\u578b": {"select": {"name": record_type}},
"\u5bf9\u8bdd\u8f6e\u6570": {"number": rounds},
"\u5df2\u5b8c\u6210\u4efb\u52a1": {"rich_text": [{"text": {"content": tasks[:2000]}}]},
"\u6838\u5fc3\u7ed3\u8bba": {"rich_text": [{"text": {"content": conclusions[:2000]}}]},
}
if date:
properties["\u4ea4\u4e92\u65e5\u671f"] = {"date": {"start": date}}
if mama_growth:
properties["\u5988\u5988\u6210\u957f\u4fe1\u53f7"] = {"rich_text": [{"text": {"content": mama_growth[:2000]}}]}
if next_tasks:
properties["\u4e0b\u6b21\u7eed\u63a5"] = {"rich_text": [{"text": {"content": next_tasks[:2000]}}]}
if domains:
properties["\u6d89\u53ca\u9886\u57df"] = {"multi_select": [{"name": d} for d in domains]}
children = []
if full_hldp:
for i in range(0, len(full_hldp), 2000):
chunk = full_hldp[i:i+2000]
children.append({
"object": "block", "type": "code",
"code": {"rich_text": [{"text": {"content": chunk}}], "language": "javascript"}
})
async with httpx.AsyncClient(timeout=30.0) as client:
res = await client.post(
f"{NOTION_BASE}/pages",
headers=_headers(),
json={"parent": {"database_id": db_id}, "properties": properties, "children": children[:100]}
)
if res.status_code in (200, 201):
new_id = res.json().get("id", "").replace("-", "")
return f"Interaction record created! Page ID: {new_id}"
else:
return f"[Error] Write failed: HTTP {res.status_code} - {res.text[:300]}"
PATCH_EOF
第三步：清理 app.js 末尾重复代码
cat > /tmp/fix_appjs.py << 'PYEOF'
import re
path = '/Users/chenshujun/chenxing-aircraft/frontend/app.js'
with open(path, 'r') as f:
content = f.read()
# 找到重复补丁的起始标记
marker = '// ══════════════════════════════════════════════════════════\n// 补丁 · 2026-05-22 · 修复4个缺失函数 + checkPasted接入'
idx = content.find(marker)
if idx == -1:
print('未找到重复补丁标记，无需清理')
else:
cleaned = content[:idx].rstrip() + '\n'
with open(path, 'w') as f:
f.write(cleaned)
removed = len(content) - len(cleaned)
print(f'✅ 已清理 app.js 末尾 {removed} 字符的重复代码')
PYEOF
python3 /tmp/fix_appjs.py
第四步：重启服务
cd ~/chenxing-aircraft && pkill -f 'uvicorn\|python.*main' 2>/dev/null; sleep 1; nohup python3 -m uvicorn backend.main:app --host 0.0.0.0 --port 8000 > server.log 2>&1 &
第五步：验证
在飞机里对晨星说：
「这本书有多少章？」
应该不会再出现186章的幻觉了。
✅ 修复结果（2026-05-22 20:37 验证通过）
晨星已能正确回答「一共有639章，其中634章是正文，剩下的是番外篇，最新已拆到第639章」。
实际生效的关键修复：不是折叠策略，而是 search_pages 的拆书查询拦截——阻止拆书相关查询 fallthrough 到 Notion Search API（只返回最近编辑的几条结果，导致模型误判只有7章）。
额外补丁（第二步之后追加的两轮修复）
- 折叠策略加强：子页面>30时完全不列具体页面，只显示总数统计
- search_pages 拆书拦截（根因修复）：拆书关键词匹配时，SQLite捷径失败后不再 fallthrough 到 Notion Search，而是提示模型改用
get_chaishu_progress或list_subpages
修改清单
| 文件 | 改了什么 | 为什么 |
|---|---|---|
| notion_tools.py | _read_blocks 子页面>30时折叠（前10+后10+统计） |
防止639个子页面列表撑爆模型上下文 |
| notion_tools.py | read_page 输出>6000字符时截断 |
兜底保护，防止任何超长输出 |
| notion_tools.py | search_pages page_size 8→20 |
搜索返回更多结果，不再只看到最后几章 |
| notion_tools.py | list_subpages 超过50条时折叠（前20+后20） |
防止完整列表撑爆上下文 |
| notion_tools.py | code block 格式修复（``` 替代 **`） | 原来的代码块格式不对 |
| app.js | 删除末尾重复的补丁函数块 | 05-22补丁被追加导致函数定义重复 |