Files
AI_Novel/scripts/e2e_test.py
T
Administrator 4073edcc20 AI 小说生成工作台:分层流水线 + 故事圣经(一期完整实现)
- 后端:Node/Express + node:sqlite,人物/世界观/分卷/细纲/正文 8 阶段生成流水线(SSE 流式 + zod 校验 + 重试)
- 前端:React 18 + Vite + antd 三栏工作台:流水线引导、章节树、编辑器(快切章节/段落改写)、故事圣经面板
- 写作工艺注入:人物声口/内在冲突/秘密,细纲 POV/情绪曲线/功能/张力,正文专业规则
- 自动回写故事圣经(摘要/设定/人物变化/伏笔/时间线),TXT/EPUB 导出,token 成本日志
2026-09-13 20:35:47 +08:00

107 lines
5.0 KiB
Python

# -*- coding: utf-8 -*-
"""端到端回归:纯 UTF-8 中文数据走完整 CRUD + 导出链路"""
import json
import urllib.request
import zipfile
import io
import sys
BASE = 'http://127.0.0.1:5175'
FAILED = []
def req(path, method='GET', body=None, raw=False):
data = json.dumps(body, ensure_ascii=False).encode('utf-8') if body is not None else None
r = urllib.request.Request(BASE + path, data=data, method=method)
if data:
r.add_header('Content-Type', 'application/json')
out = urllib.request.urlopen(r, timeout=30).read()
return out if raw else json.loads(out.decode('utf-8'))
def check(name, cond):
print(('PASS ' if cond else 'FAIL ') + name)
if not cond:
FAILED.append(name)
# 1. 建项目
p = req('/api/projects', 'POST', {
'name': '星海谎言', 'genre': '科幻',
'premise': '世代飞船上的工程师发现航行早已结束,航行本身是统治阶层的谎言',
'style_notes': '冷峻克制,多对话', 'target_chapter_count': 12, 'target_chapter_words': 2000,
})
pid = p['id']
check('project name roundtrip', p['name'] == '星海谎言')
check('premise roundtrip', p['premise'].startswith('世代飞船'))
# 2. 人物 / 世界观
c = req(f'/api/projects/{pid}/characters', 'POST', {
'name': '林原', 'aliases': '老林', 'role': '主角',
'personality': '沉稳多疑,对机械有近乎信仰的执着', 'goals': '查明航线数据的真相',
'appearance': '瘦高,左眼有一道旧疤', 'background': '三代轮机世家,父亲死于一次"意外"检修事故',
})
check('character roundtrip', c['name'] == '林原' and c['background'].startswith('三代轮机世家'))
cid = c['id']
w = req(f'/api/projects/{pid}/world', 'POST', {
'category': '地理', 'title': '世代飞船远望号', 'content': '全长十二公里,自持循环设计寿命五百年,实际已航行四百九十七年',
})
check('world roundtrip', w['title'] == '世代飞船远望号')
# 3. 卷 / 细纲
v = req(f'/api/projects/{pid}/volumes', 'POST', {'title': '第一卷 裂缝', 'summary': '发现数据异常到确认真相', 'sort': 1})
check('volume roundtrip', v['title'] == '第一卷 裂缝')
b = req(f'/api/projects/{pid}/beats', 'POST', {
'volume_id': v['id'], 'title': '第1章 异响', 'events': '林原在检修中发现被删除四百九十七次的航线修正记录',
'involved_characters': [cid], 'foreshadow_plan': '埋下船长知情暗示',
})
check('beat involved ids parsed to array', b['involved_characters'] == [cid])
# 4. 章节 + 内容 + 字数
ch = req(f'/api/projects/{pid}/chapters', 'POST', {
'beat_id': b['id'], 'volume_id': v['id'], 'title': '第1章 异响', 'sort': 1,
})
chid = ch['id']
content = '检修通道的灯又灭了。林原把手电咬在嘴里,手指穿过线缆,摸到了那块不该存在的芯片。\n\n数据像潮水一样涌进终端——航线修正记录,五百年间,被删除了四百九十七次。'
updated = req(f'/api/chapters/{chid}', 'PATCH', {'content': content, 'status': 'edited'})
check('word count correct', updated['word_count'] == len([ch for ch in content if not ch.isspace()]))
# 5. bundle 聚合
bundle = req(f'/api/projects/{pid}/bundle')
check('bundle beat involved is array', bundle['beats'][0]['involved_characters'] == [cid])
check('bundle counts', len(bundle['characters']) == 1 and len(bundle['world']) == 1 and len(bundle['chapters']) == 1)
# 6. 项目列表统计
lst = req('/api/projects')
me = [x for x in lst if x['id'] == pid][0]
check('project stats', me['total_words'] > 0 and me['written_count'] == 1)
# 7. 导出 TXT / EPUB
txt = req(f'/api/export/{pid}/txt', raw=True).decode('utf-8-sig')
check('txt contains title/body/volume', '星海谎言' in txt and '不该存在的芯片' in txt and '第一卷 裂缝' in txt)
epub_raw = req(f'/api/export/{pid}/epub', raw=True)
z = zipfile.ZipFile(io.BytesIO(epub_raw))
check('epub valid zip', z.testzip() is None)
check('epub mimetype stored first', z.getinfo('mimetype').compress_type == zipfile.ZIP_STORED)
nav = z.read('OEBPS/nav.xhtml').decode('utf-8')
ch1 = z.read('OEBPS/chapter_0001.xhtml').decode('utf-8')
check('epub nav contains chapter', '第1章 异响' in nav)
check('epub chapter content escaped+present', '不该存在的芯片' in ch1)
# 8. 生成接口的 SSE 错误路径:临时换成无效 key,应返回规范 error 事件而非崩溃,然后恢复原配置
try:
saved = req('/api/settings')
req('/api/settings', 'PUT', {**saved, 'api_key': 'sk-invalid-key-for-e2e'})
r = urllib.request.Request(BASE + '/api/generate/prose', data=json.dumps({
'projectId': pid, 'chapterId': chid
}).encode(), method='POST', headers={'Content-Type': 'application/json'})
body = urllib.request.urlopen(r, timeout=60).read().decode('utf-8')
check('SSE error event emitted for auth failure', 'event: error' in body and 'HTTP' in body)
finally:
req('/api/settings', 'PUT', saved)
check('llm config restored', req('/api/settings')['api_key'] == saved['api_key'])
print()
print('TOTAL FAILED:', len(FAILED))
sys.exit(1 if FAILED else 0)