#!/usr/bin/env python3 # -*- coding: utf-8 -*- """组卷台 v2 · 本地 AI 助手代理（功能A：讲解/出答案）。 服务组卷台.html + 转发 /api/chat 到 DeepSeek（key 只在服务端，不进前端/代码/git）。 跑法：python3 组卷台AI助手.py → 浏览器开 http://localhost:8848/组卷台.html 纯 stdlib，无第三方依赖。""" import http.server, socketserver, json, os, re, urllib.request, urllib.error import hashlib, tempfile, subprocess, shutil # /api/source 源文件预览下载用 import concurrent.futures import verify_gen # AI 出题的确定性数值核验（算式跑一遍对答案） try: import gen_router # ① 确定性生成器 except Exception: gen_router = None try: import xkw_similar # ③ 学科网真题 except Exception: xkw_similar = None HERE = os.path.dirname(os.path.abspath(__file__)) KEYFILE = os.path.expanduser('~/.deepseek_pro_key') KEY = (open(KEYFILE).read().strip() if os.path.exists(KEYFILE) else os.environ.get('DEEPSEEK_API_KEY', '')).strip() MODEL = 'deepseek-v4-flash' # 出题用 flash：推理开销小、够快够好（pro 会烧大量 reasoning token 致截断） PORT = 8848 BUILD = '2026-09-05.1' # 前后端版本握手：改前端 NEED_BUILD 时同步改这里；不一致页面会横幅提醒重启 SIM_PROMPT = r'''你是资深小学奥数老师。老师选中了一道题，请你出 N 道和这道题【同一类型、同一考法、同一解题套路】的练习题，供学生针对这一类题做专项练习。 - **紧扣原题**：原题是什么类型（如"比较大小"就出比较大小、"鸡兔同笼"就出鸡兔同笼、"求周长"就出求周长），你就出什么类型；不要换成别的题型、不要发散成"这个专题下的各种题"。 - **换数字/换情境**：每道都要换一套新的数字、或换个生活情境，是真正的新题；既不许把原题原样抄一遍，也不许只改一两个数字（要有变化、有新意），N 道之间也别雷同。 - 难度贴合原题，符合该年级学生认知；条件完整、自洽、可解、答案唯一、能算出干净答案。 - 数学写法用标准 LaTeX：行内数学（算式、分数、根号、指数、循环小数）一律用 $...$ 包裹，例如 $28\div3\times54$、$\frac{3}{4}$、$\sqrt{16}$、$2^3$、$0.\dot{2}\dot{7}$；乘写 \times、除写 \div、分数写 \frac{分子}{分母}；纯叙述文字、“5个苹果”这类量词不要包 $。 **每道题必须自己先算一遍，并交出可核对的算式**（服务端会执行它来验你有没有真算，对不上的题直接丢弃）： - `answer`：最终答案，**一个数**（如 "6"、"3/4"、"2.5"）；比较大小题填 ">"、"<" 或 "="。 - `check`：一行 **Python 算式**，只用题干里出现的数字，算出来必须**正好等于** answer。只能用 + - * / // % ** 和括号，不许出现函数、字母、变量。 - `check_kind`：算式怎么核对 —— "eq"（check 的值 = answer）；"sign"（比较大小题，check 写「左式-右式」，正负号对应 > < =）；""（数图形、逻辑推理、开放题这类写不出算式的，三个字段都留空）。 - **每道题只问一个问题**（不要分小问），答案要算得出干净的数：整数，或分母是 2 3 4 5 8 10 100 这类常见分母；**除不尽、答案是 17/23 这种脏数的题不要出**；小学题不要出现负数答案。问「几只/几个/多少人」的，答案必须是整数。 - **绝对不要出需要看图的题**（不许出现「如图」「下图」「图中」，也不许用文字描述图形来顶替）—— 你没法配图，学生看不到图就做不了。这类题型请改出纯文字叙述的同类题。 - 严格只输出 JSON：{"items":[{"type":"题型名","stem":"题干","answer":"6","check":"(32-10*2)/(4-2)","check_kind":"eq"}, ...]}；注意 JSON 字符串里的反斜杠要转义（如把 \frac 写成 \\frac）；无任何多余文字、不要解析、不要代码块围栏。**answer 和 check 只给服务端核对用，绝不会展示给学生。**''' # 题图实验通道与普通 AI 出题严格分开：普通通道仍然禁止图形题。 # 当前只开放一个可确定构图、可确定验算的模板，避免模型只写“如图摆放”却不给布局。 FIGURE_SIM_PROMPT = r'''你是资深小学奥数老师。请生成 N 道六年级“圆与扇形”同类练习。 本次题图由程序确定绘制，且只允许一种图形：三个半径相同、圆心角均为 60° 的扇形，以等边三角形三个顶点为圆心，三个圆弧围成一个曲边三角形。 每道题必须： - 题干明确写“如图”，给出三个扇形的公共半径 r（20~200 的整数）、圆心角均为 60°，问封闭图形周长；并写明 π 取 3.14。 - 周长只能是三段 60° 弧长之和，即 r×π。不要问面积、不要加任何额外条件。 - `figure` 必须严格是 {"kind":"three_sector_arcs","radius":r,"angle":60}，r 与题干完全一致。 - `answer` 给出数值周长；`check` 用只含数字和 +-*/() 的 Python 算式核对答案；`check_kind` 必须为 "eq"。 - 只输出 JSON：{"items":[{"type":"圆与扇形","stem":"...","answer":"188.4","check":"60*3.14","check_kind":"eq","figure":{"kind":"three_sector_arcs","radius":60,"angle":60}}]}。无任何额外文字、不要代码块。''' # 变式组（GSM-Plus 扰动分类）：一道种子题 → 按六种预定策略各出一道变体。 # ⚠ 和「出类似题」的区别：类似题是"同型换皮"，变式组是**定向扰动** —— 每道变体 # 都标明改了什么（换数字/扩数位/逆向求解/加干扰/换情境/加一步），给老师做"举一反三"。 # 「加干扰」是 GSM-IC 的结论：插入不影响答案的多余条件，专治学生乱抓数字。 VAR_STRATEGIES = [ ('换数字', '只把题中的数字换成另一组（仍需配平、答案干净），情节一字不动'), ('扩数位', '把数字放大一个量级（两位数变三四位数），考验计算功底，其余不动'), ('逆向求解', '把原题的答案当成已知条件，反过来求原题中的某个已知量'), ('加干扰', '加入 1 个与解题无关的多余数字条件（如无关的年龄/价格），答案不变，专治乱抓数字'), ('换情境', '数量关系完全不变，把故事背景换成另一个生活场景'), ('加一步', '在原题基础上多加一步运算（如求完各自再求相差/总共），难度提一档'), ] VAR_PROMPT = r'''你是资深小学奥数老师。老师选中了一道题，要做「举一反三」变式组：请按下面给定的策略，各出 1 道变体。 - 每道变体必须严格执行它的策略，不许混用；**`vtype` 字段必填**，只能是给定的策略名之一，放在每个对象的第一个字段。 - items 数组必须按策略给出的顺序排列（第 1 道对应第 1 个策略，依此类推）。 - **每道题只问一个问题、答案是一个数** —— 即使原题问了两问（如"鸡和兔各有几只"）， 变体也只许问其中一问（如"兔有几只"）。条件完整、可解、答案唯一且干净。 - 「加干扰」那道：答案必须与你构造的主线数字算出的一致，多余条件绝不能影响答案。 - 数学写法用标准 LaTeX，行内数学一律 $...$ 包裹；乘写 \times、除写 \div、分数写 \frac{分子}{分母}。 - 绝对不要出需要看图的题（不许出现「如图」「下图」「图中」，也不许用文字描述图形顶替）。 - 每道题必须自己先算一遍并交出可核对的算式：`answer` 最终答案一个数；`check` 一行只含数字与 + - * / // % ** 和括号的 Python 算式，值必须正好等于 answer；`check_kind` 填 "eq"（写不出算式的填空字符串，三字段都留空）。 - 严格只输出 JSON：{"items":[{"vtype":"换数字","type":"题型名","stem":"...","answer":"...","check":"...","check_kind":"eq"},...]}； JSON 字符串里反斜杠要转义（\frac 写成 \\frac）；无任何多余文字、不要代码块围栏。answer/check 只给服务端核对，绝不展示给学生。''' # JSON 里的合法转义：\" \\ \/ \uXXXX，以及 \b \f \n \r \t # ⚠ 但 LaTeX 命令正好也以这些字母开头：\frac \times \begin \neq \rightarrow … # 老写法 `re.sub(r'\\(?!["\\/bfnrtu])', ...)` 把 `\frac` 当成 `\f`(换页符)+`rac`， # 而且这样的 JSON **能正常解析**，于是悄悄产出 `$rac{1}{4}$` —— 分数全废且不报错。 # 判据改成：`\` + [bfnrt] 只有在**后面不是字母**时才算 JSON 转义，否则是 LaTeX。 _ESC = re.compile(r'\\(?:(["\\/]|u[0-9a-fA-F]{4}|[bfnrt](?![a-zA-Z]))|(.))', re.S) def _fix_latex_json(t): """把 LaTeX 里的裸反斜杠补成 `\\\\`，合法 JSON 转义原样保留。 整对消费(`\\\\` 一次吃掉两个字符)，所以已正确转义的内容不会被二次破坏。""" return _ESC.sub(lambda m: m.group(0) if m.group(1) else '\\\\' + m.group(2), t) def parse_items(content): # 剥 ```json 围栏，容错抽取 items 列表（含 LaTeX 孤立反斜杠修复） t = content.strip() if t.startswith('```'): t = re.sub(r'^```[a-zA-Z]*\n?', '', t); t = re.sub(r'\n?```$', '', t).strip() m = re.search(r'\{.*\}', t, re.S) if m: t = m.group(0) fixed = _fix_latex_json(t) obj = None for cand in (fixed, t): # ⚠ 修过的**优先**，见 _fix_latex_json try: obj = json.loads(cand); break except Exception: continue if obj is None: return [] items = obj.get('items') if isinstance(obj, dict) else (obj if isinstance(obj, list) else []) out = [] for it in (items or []): if isinstance(it, dict) and (it.get('stem') or '').strip(): # answer/check 带出来给核验层用，**不进最终返回前端的题面**（零答案红线） out.append({'type': str(it.get('type', '')).strip(), 'stem': str(it['stem']).strip(), 'answer': str(it.get('answer', '') or '').strip(), 'check': str(it.get('check', '') or '').strip(), 'check_kind': str(it.get('check_kind', '') or '').strip()}) return out def _parse_json_items(content): """与 parse_items 共用同一套 LaTeX 容错，但保留 figure 原始字段供受限题图通道校验。""" t = content.strip() if t.startswith('```'): t = re.sub(r'^```[a-zA-Z]*\n?', '', t); t = re.sub(r'\n?```$', '', t).strip() m = re.search(r'\{.*\}', t, re.S) if m: t = m.group(0) for cand in (_fix_latex_json(t), t): try: obj = json.loads(cand) items = obj.get('items') if isinstance(obj, dict) else (obj if isinstance(obj, list) else []) return [it for it in (items or []) if isinstance(it, dict)] except Exception: continue return [] def gen_similar(stem, grade, diff, topic, cnt): """照一道原题调 DeepSeek 出 cnt 道同类练习；返回 items 列表；失败抛异常。线程安全，供单题/整页批量共用。""" stars = '★' * int(diff) if str(diff).isdigit() else str(diff) user = ('老师选中的原题（%s，难度 %s%s）：\n「%s」\n\n请照着这道题，出 %d 道【同一类型、同一考法】的练习题，每道换不同的数字/情境，别抄原题、也别散成别的题型。' % (grade or '小学', stars, ('，专题：' + topic) if topic else '', stem, cnt)) # max_tokens 随道数放大：固定 4000 出 10 道会被截断，末尾几道残缺 payload = {'model': MODEL, 'temperature': 0.8, 'max_tokens': min(8000, max(4000, int(cnt) * 800)), 'stream': False, 'thinking': {'type': 'disabled'}, # 关隐藏思考:实测输出83%是思考token,关掉后3道题7.9s→2.0s,题目质量/LaTeX不受影响 'messages': [{'role': 'system', 'content': SIM_PROMPT}, {'role': 'user', 'content': user}]} req = urllib.request.Request('https://api.deepseek.com/chat/completions', data=json.dumps(payload).encode('utf-8'), headers={'Content-Type': 'application/json', 'Authorization': 'Bearer ' + KEY}) with urllib.request.urlopen(req, timeout=180) as r: out = json.load(r) return parse_items(out['choices'][0]['message']['content']) def gen_variants(stem, grade, diff, topic): """变式组：按 VAR_STRATEGIES 六策略各出一道。返回 items（带 vtype）。""" stars = '★' * int(diff) if str(diff).isdigit() else str(diff) plan = '\n'.join('%d. 【%s】%s' % (i + 1, k, v) for i, (k, v) in enumerate(VAR_STRATEGIES)) user = ('老师选中的原题（%s，难度 %s%s）：\n「%s」\n\n请按以下 %d 个策略各出 1 道变体：\n%s' % (grade or '小学', stars, ('，专题：' + topic) if topic else '', stem, len(VAR_STRATEGIES), plan)) payload = {'model': MODEL, 'temperature': 0.7, 'max_tokens': 8000, 'stream': False, 'thinking': {'type': 'disabled'}, 'messages': [{'role': 'system', 'content': VAR_PROMPT}, {'role': 'user', 'content': user}]} req = urllib.request.Request('https://api.deepseek.com/chat/completions', data=json.dumps(payload).encode('utf-8'), headers={'Content-Type': 'application/json', 'Authorization': 'Bearer ' + KEY}) with urllib.request.urlopen(req, timeout=180) as r: out = json.load(r) items = parse_items(out['choices'][0]['message']['content']) # ⚠ 实测 deepseek-v4-flash 会把 vtype 字段整个漏掉（六道题全不带，但顺序 # 和策略一一对应）。提示词已强调"必填+按序"，这里再按位置兜底回填 —— # 两道保险的代价是零，掉标签的代价是六道题在前端挤成一道（按 vtype 去重）。 for i, it in enumerate(items): if not (it.get('vtype') or '').strip() and i < len(VAR_STRATEGIES): it['vtype'] = VAR_STRATEGIES[i][0] return items # ── AI 组卷：一句话 → 结构化选题 spec ──────────────────────────────── # 架构定死为「后端只做翻译+校验，前端选题」：题库只在 组卷台.html 里一份， # 服务端绝不再养第二份（否则改题库要两处同步）。前端把可选范围的 # 专题目录（catalog，≤40KB）随请求带上来，这里一次 DeepSeek 调用出 spec， # 然后**逐项硬校验**——cd 必须是目录里的原文，LLM 没有机会编造。 COMPOSE_TYPES = ['选择', '填空', '计算', '应用', '解答', '作图', '判断', '其他'] COMPOSE_PROMPT = r'''你是组卷助手。老师用一句话描述想要的练习卷，你把它翻译成 JSON 选题条件。 严格只输出一个 JSON 对象（不要围栏、不要解释）： {"tb":"教材名","gg":"年级/册名","total":8, "groups":[{"cd":"专题名","n":3},{"cd_like":"关键词","n":5},{"kp":"知识点名","n":4},{"kp_like":"关键词","n":4}], "types":["计算","应用"] 或 null, "difficulty":"易|中|难|混合","order":"easy_first|as_is", "no_figure":false,"guessed_grade":false} 规则（逐条执行）： - tb、gg、cd 只能**原样照抄**下面目录里出现过的字符串，一个字都不许改；kp 只能原样照抄知识点表。 - {"kp":…} 是**跨教材知识点轴**：只要老师话里出现「跨教材」「不限教材」「按知识点」「各套书」 任一说法，该知识点**必须**用 {"kp":…}，即使某套教材目录里有同名专题也不许改用 cd。 例：「跨教材来 8 道相遇问题」→ {"kp":"相遇问题","n":8}； 「不限教材出 10 道鸡兔同笼」→ {"kp":"鸡兔同笼问题","n":10}。 老师没提这些说法时，一律优先 cd/cd_like（教材目录轴）。kp 组照样要给 n。 - 知识点表里同一主题常拆成多个细分知识点（如「鸡兔同笼」有 5 个），名字后括号是该知识点题目所在年级。 老师**笼统**说一个主题 → 用 {"kp_like":"鸡兔同笼"} 一次拿全（服务端按子串展开）； 老师**精确指名**某个细分 → 用 {"kp":"完整名"}。kp/kp_like 的关键词必须真实出现在知识点表里。 只用 kp/kp_like 且老师没说年级时，gg 按知识点表括号里的年级填（如「牛吃草问题（5年级）」→ 五年级）。 - 老师说的知识点 → groups。先在目录里把**所有相关**的专题名找齐： 同一系列（带（一）（二）/①②③的）、包含该词的、以及**叫法不同但意思相同**的都算相关。 然后： · 相关专题都含同一个词 → 用 {"cd_like":"那个词"} 一次拿全（如「一次函数」「大数的认识」）， 不许只挑其中一个； · 目录叫法和老师说法不同 → 按**意思**选，用 {"cd":…} 逐个列出。 例：老师说「分数应用题」而胡小群六年级目录里叫「量率问题（一）」「量率问题（二）」， 就列这两个；不要因为字面含「分数」去抓「分数巧算」（那是计算不是应用题）； · 老师把几个知识点合着说（「加减巧算」「和差倍问题」）→ 必须拆成多个 group 分别找 （加减巧算=加法的巧算+减法的巧算，目录里叫「加法简算」「减法简算」就用它们）； · cd 与 cd_like 填的字符串必须在目录里真实出现过；意思相近的都找不到才不填， 绝不硬编不存在的专题名。 - 教材选择（老师没点名教材时，按此优先级，第一条命中就定，不看后面）： 1. 提到奥数/思维/竞赛 → **必选**「举一反三」或「动脑筋」，绝不选胡小群课程； 2. 提到中考/一模/二模/真题 → 「上海中考真题」； 3. 提到课本/同步/校内 → 「一课一练」； 4. 初中内容（六~九年级/预初/初一~初三）→ **必选**「初中培优讲义」， 即使胡小群课程里也有初中册； 5. 其余小学内容 → 「胡小群课程」。 - 「上海中考真题」的 gg 是年份（如 2024年）、cd 是整卷卷名；按考次取题用 cd_like（"一模"/"二模"/"中考"）。 - 老师没说年级：按知识点推断最可能的 gg，并把 guessed_grade 设为 true。 - total = 老师说的题量；没说就填 8。groups 的 n 之和应等于 total。 - types 只能取：选择、填空、计算、应用、解答、作图、判断、其他；老师没提题型就填 null。 - difficulty 只能取 易/中/难/混合，没提填 混合。 - 老师说「由易到难/从简单到难」→ order 填 easy_first，否则 as_is。 - 老师说「不要带图的题」→ no_figure true。''' def nl_to_spec(text, catalog, kps=None): """一句话 + 目录(+知识点表) → LLM 的原始 spec（dict）。失败抛异常，调用方兜底。""" lines = [] for tb, gs in catalog.items(): for gg, cds in gs.items(): lines.append('【%s · %s】%s' % (tb, gg, '｜'.join(cds))) cat = '\n'.join(lines)[:80000] user = '老师的话：「%s」\n\n可选目录（tb·gg 与其下的专题名）：\n%s' % (text, cat) if kps: user += '\n\n知识点表（kp/kp_like 轴可选值，跨教材；括号=题目所在年级）：\n' + '、'.join(kps)[:40000] payload = {'model': MODEL, 'temperature': 0, 'max_tokens': 3000, 'stream': False, 'thinking': {'type': 'disabled'}, 'response_format': {'type': 'json_object'}, 'messages': [{'role': 'system', 'content': COMPOSE_PROMPT}, {'role': 'user', 'content': user}]} req = urllib.request.Request('https://api.deepseek.com/chat/completions', data=json.dumps(payload).encode('utf-8'), headers={'Content-Type': 'application/json', 'Authorization': 'Bearer ' + KEY}) with urllib.request.urlopen(req, timeout=120) as r: out = json.load(r) t = out['choices'][0]['message']['content'] i, j = t.find('{'), t.rfind('}') return json.loads(t[i:j + 1]) def _pick_key(want, options): """在 options 里找 want：精确 → 互为子串 → difflib。找不到返回 None。""" if want in options: return want w = str(want or '').strip() if w: sub = [o for o in options if w in o or o in w] if sub: return min(sub, key=len) import difflib m = difflib.get_close_matches(w, list(options), n=1, cutoff=0.5) return m[0] if m else None def validate_spec(raw, catalog, kps=None): """LLM 的 spec → 校验后的 spec。**这里说了才算**，LLM 说了不算。 返回 (spec, fixes, unresolved)。spec.groups 统一成 {'cds':[真实cd…],'n':int,'label':str}， 前端拿到的每个 cd 都保证在目录里存在。 """ fixes, unresolved = [], [] tb = _pick_key(raw.get('tb'), catalog) or (list(catalog)[0] if catalog else None) if not tb: raise ValueError('目录为空') gs = catalog[tb] guessed = bool(raw.get('guessed_grade')) gg = _pick_key(raw.get('gg'), gs) if gg is None: gg, guessed = list(gs)[0], True cds = gs[gg] groups = [] for g in (raw.get('groups') or []): if not isinstance(g, dict): continue try: n = max(0, int(g.get('n', 0))) except (TypeError, ValueError): n = 0 if 'cd_like' in g and g.get('cd_like'): kw = str(g['cd_like']).strip() hit = [c for c in cds if kw in c] if hit: groups.append({'cds': hit, 'n': n, 'label': '含「%s」的 %d 节' % (kw, len(hit))}) else: unresolved.append(kw) elif g.get('cd'): asked = str(g['cd']).strip() got = _pick_key(asked, cds) if got: if got != asked: fixes.append({'asked': asked, 'picked': got}) groups.append({'cds': [got], 'n': n, 'label': got}) else: unresolved.append(asked) elif g.get('kp_like'): kw = str(g['kp_like']).strip() hit = [k for k in (kps or []) if kw and kw in k] if hit: groups.append({'kps': hit, 'cds': [], 'n': n, 'label': '知识点·含「%s」的 %d 个(跨教材)' % (kw, len(hit))}) else: unresolved.append(kw) elif g.get('kp'): asked = str(g['kp']).strip() got = _pick_key(asked, list(kps or [])) if got: if got != asked: fixes.append({'asked': asked, 'picked': got}) groups.append({'kps': [got], 'cds': [], 'n': n, 'label': '知识点·%s(跨教材)' % got}) else: unresolved.append(asked) try: total = max(1, min(60, int(raw.get('total') or 0))) except (TypeError, ValueError): total = 0 if not total: total = sum(x['n'] for x in groups) or 8 ssum = sum(x['n'] for x in groups) if groups and ssum != total: if ssum <= 0: base = total // len(groups) for x in groups: x['n'] = base else: for x in groups: x['n'] = max(0, round(x['n'] * total / ssum)) diff = total - sum(x['n'] for x in groups) groups[-1]['n'] = max(0, groups[-1]['n'] + diff) types = [t for t in (raw.get('types') or []) if t in COMPOSE_TYPES] or None diff_ = raw.get('difficulty') if raw.get('difficulty') in ('易', '中', '难', '混合') else '混合' order = raw.get('order') if raw.get('order') in ('easy_first', 'as_is', 'by_type') else 'as_is' spec = {'tb': tb, 'gg': gg, 'total': total, 'groups': groups, 'types': types, 'difficulty': diff_, 'order': order, 'no_figure': bool(raw.get('no_figure')), 'allow_gen': tb != '上海中考真题', 'guessed_grade': guessed} return spec, fixes, unresolved def compose_understood(spec): """「AI 的理解」这句话由**校验后的 spec** 拼出来 —— 老师看到的就是要执行的， 绝不用 LLM 的自述（自述和实际执行可以对不上）。""" parts = ['%s · %s%s' % (spec['tb'], spec['gg'], '（年级未说，按此猜）' if spec['guessed_grade'] else '')] parts.append('、'.join('「%s」%d 道' % (g['label'], g['n']) for g in spec['groups'])) if spec['types']: parts.append('题型：' + '/'.join(spec['types'])) if spec['difficulty'] != '混合': parts.append('难度：' + spec['difficulty']) if spec['order'] == 'easy_first': parts.append('由易到难') if spec['no_figure']: parts.append('不带图') return ' · '.join(p for p in parts if p) def _normalize_figure(raw): """只放行已实现的模板与安全数值，绝不把模型返回的 GeoGebra 命令直接执行。""" if not isinstance(raw, dict) or raw.get('kind') != 'three_sector_arcs': return None try: radius = int(raw.get('radius')) angle = int(raw.get('angle')) except (TypeError, ValueError): return None if not 20 <= radius <= 200 or angle != 60: return None return {'kind': 'three_sector_arcs', 'radius': radius, 'angle': 60} def gen_figure_similar(stem, grade, diff, topic, cnt): """AI 只负责改半径和题干；题图布局由 kind 的白名单模板决定。""" stars = '★' * int(diff) if str(diff).isdigit() else str(diff) user = ('老师选中的原题（%s，难度 %s，专题：%s）：\n「%s」\n\n' '请生成 %d 道同一考法的新题。' % (grade or '六年级', stars, topic or '圆与扇形', stem, cnt)) payload = {'model': MODEL, 'temperature': 0.35, 'max_tokens': min(4200, max(1600, int(cnt) * 650)), 'stream': False, 'thinking': {'type': 'disabled'}, 'messages': [{'role': 'system', 'content': FIGURE_SIM_PROMPT}, {'role': 'user', 'content': user}]} req = urllib.request.Request('https://api.deepseek.com/chat/completions', data=json.dumps(payload).encode('utf-8'), headers={'Content-Type': 'application/json', 'Authorization': 'Bearer ' + KEY}) with urllib.request.urlopen(req, timeout=180) as r: out = json.load(r) return _parse_json_items(out['choices'][0]['message']['content']) def verify_figure_items(items, want): """题图通道的确定性验算：三段 60° 弧长之和必为 radius × 3.14。""" keep, bad = [], [] for item in items or []: spec = _normalize_figure(item.get('figure')) stem = str(item.get('stem') or '').strip() if not spec: bad.append({'stem': stem[:60], 'why': '题图结构不在白名单内'}); continue if not all(token in stem for token in ('如图', str(spec['radius']), '60', '3.14')): bad.append({'stem': stem[:60], 'why': '题干缺少题图必要条件'}); continue if str(item.get('check_kind') or '') != 'eq': bad.append({'stem': stem[:60], 'why': '题图题缺少可核对算式'}); continue try: got = verify_gen.calc(str(item.get('check') or '')) answer = verify_gen.parse_num(str(item.get('answer') or '')) except Exception: got, answer = None, None expected = verify_gen.Fraction(spec['radius'] * 314, 100) if answer is None or got != answer or answer != expected: bad.append({'stem': stem[:60], 'why': '周长与三段 60° 弧长不符'}); continue keep.append({'type': str(item.get('type') or '圆与扇形'), 'stem': stem, 'verified': True, 'figure': spec}) return keep[:want], {'kept': min(len(keep), want), 'dropped': len(bad), 'verified': min(len(keep), want), 'reasons': bad[:8]} # ⚠ 出同类题从「自动挑一条路」改成**老师三选一**（2026-07-27）： # gen 生成器 —— 秒出、带答案解析、58 个题型，命不中就没有 # ai 大模型 —— 任意题型，配 verify_gen 数值核验（算式跑一遍对答案） # xkw 学科网 —— 真题、带官方答案解析；**授权内容：不落库、不去水印** # 所以不再有全局开关（原 USE_GENERATOR 已废）——走哪条由请求的 `by` 决定。 DEFAULT_BY = 'ai' # 🟢 学科网出同类题 **2026-07-28 23:30 重新上线**。 # 当天实测：9 个接口通、只有 `/xopqbm/questions`（知识点推题）仍报 402 资源包不足。 # 学科网的资源包是**按接口分开卖、分开耗尽**的，所以充值不等于喂到了那一条。 # 相似题三条腿（v1 / v2 / ai）全部可用 —— 出同类题不依赖 `/questions`，故可上线； # 奥数的竞赛难度链路仍不可用，会自动降级并在抽屉里如实标注。 # 要再下线：改回 False + 重启，前端自动置灰（前端从 /api/version 读这个开关）。 # ⚠ 别把这个开关和 `/questions` 的 402 混为一谈（我混过一次，一刀切关掉了能用的）： # 出类似题只用 v1 / v2 / ai 三个相似题接口，**它们有额度、能用**； # 402 的 `/xopqbm/questions` 只影响「给奥数锁竞赛难度」那条支线，没它也能出题。 # 所以这个开关该开着；要停的是那条支线，不是整条腿。 XKW_ENABLED = True # 每道种子最多准打几次学科网（小克 2026-07-28 的梯子：ai×1 + v1×2 = 20 道， # 第 4 次接口必然空手）。前端会置灰，这里是**服务端兜底**，别只信前端。 XKW_PER_SEED = 3 # ⚠ 别按教材封杀学科网 —— 我一度想把奥数三套整体置灰，**是错的**。 # 起因：我只测了一道四年级行程题，ai 返回 10 道得分率全在 0.65~0.85， # 按奥数阈值一道没留下，我就推广成「奥数三套都不可用」。 # 小克拿 10 个专题 100 道复测：**58 道通过难度过滤**。行程题是特例不是通例 # （「加乘原理」那种能过 9/10），按教材封杀会误杀一大片。 # # 正确做法是**按种子降级**（小克 2026-07-28 方案）： # 第 1 次点 ai → 本地难度过滤 → 存活 0 道 → 就把这道种子后面两级(v1×2)一并置灰， # 提示「同类题都落在课内难度，建议改用生成器/AI出题」。 # · 代价上限是每种子 **1 次**调用（不是 3 次） # · 行程题那种真不行的，第 1 次就拦住 # · 加乘原理那种 9/10 的不会被名单误杀 # · 额度到账后这套逻辑不用改 # 初中培优七上 0/10 是同一个病（只是 course_id=11），同样别按教材封杀。 XKW_NO_HARD = ('这道题的同类题都落在课内难度（学科网的竞赛难度接口暂无额度），' '已停掉后续尝试以免浪费额度。建议改用「生成器」或「AI」出题。') # 学科网的按种子计数 + 会话内结果缓存。 # ⚠ **只在内存，不写盘**：学科网的题面是授权内容（带 data-copyright 水印，水印串里含 # 我们的 AppId），红线是「不落库、不去水印、不并进零答案题库；要缓存只缓存 # question_id 和检索结果，不缓存题面正文」。进程一重启就全没，这是故意的。 _XKW_SEED = {} # 种子指纹 → 已经打了几次 _XKW_CACHE = {} # (种子, mode, 第几批) → (items, note) def _fp(s): """题面指纹 —— 前端没传 qid 时用它当种子标识（只用来计数，不落盘）。""" return hashlib.md5((s or '').strip()[:400].encode('utf-8')).hexdigest()[:16] def gen_verified(stem, grade, diff, topic, want, rounds=3): """出题 → **确定性数值核验** → 不够就再补一轮。返回 (通过的题, 统计)。 多要一些再筛（核验会毙掉一部分），比"要几道出几道"更容易一次凑够。 """ keep, stat = [], {'kept': 0, 'dropped': 0, 'verified': 0, 'reasons': [], 'rounds': 0} for r in range(rounds): need = want - len(keep) if need <= 0: break # 冗余出题给核验留淘汰空间。裂项相消这类模型算不明白的题型淘汰率能到 70%， # 只多要 50% 会凑不够 → 直接按 2 倍要（上限 20，成本只是输出 token）。 ask = min(20, max(need * 2, need + 3)) raw = gen_similar(stem, grade, diff, topic, ask) ok, st = verify_gen.filter_items(raw, grade) keep += ok stat['rounds'] = r + 1 stat['dropped'] += st['dropped'] stat['reasons'] += st['reasons'] keep = keep[:want] stat['kept'] = len(keep) stat['verified'] = sum(1 for x in keep if x.get('verified')) stat['reasons'] = stat['reasons'][:8] return keep, stat class Handler(http.server.SimpleHTTPRequestHandler): def end_headers(self): # 组卷台.html 常更新（并入新教材/修图等），禁掉浏览器缓存，避免打开的是旧版 # （踩过：并入初中题后浏览器仍用缓存的旧 89MB，看不到初中题） p = self.path.split('?')[0] if p.endswith('.html') or p == '/': self.send_header('Cache-Control', 'no-cache, no-store, must-revalidate') super().end_headers() def do_GET(self): if self.path == '/api/version': # 版本握手：前端开页即查，旧后端(404)或版本不符都会触发页面横幅 return self._json({'ok': True, 'build': BUILD, # 各出题来源是否可用 —— 前端据此置灰。 'sources': {'gen': bool(gen_router), 'ai': bool(KEY), 'variants': bool(KEY), 'compose': bool(KEY), 'ai_figure': bool(KEY), 'xkw': bool(xkw_similar) and XKW_ENABLED}, 'xkw_off_reason': '' if XKW_ENABLED else '学科网接口资源包已用尽（402），暂时下线', # ⚠ 曾经在这里下发过「奥数三套禁用」名单 —— **已撤掉**。 # 按教材封杀会误杀（10 专题 100 道实测 58 道能过难度过滤）， # 改成按种子降级：出不来合格难度的那道题，服务端返回 # exhausted=true，前端就把这道种子的学科网页签置灰。 'xkw_olympiad': True}) if self.path.startswith('/api/topic/files'): return self._topic_files() if self.path.startswith('/api/topic/pdf'): return self._topic_pdf() if self.path.startswith('/api/topic'): return self._topic_preview() if self.path.startswith('/api/page'): return self._serve_page_png() if self.path.startswith('/api/source'): return self._serve_unit_source() if self.path.startswith('/api/paper'): return self._serve_paper_file() if self.path.startswith('/api/page'): return self._serve_page_png() # 「对照原书」页图：随站 原书页/ 缺失时，回退到仓库原始 成果/*/pages/*.png（offset已对齐，无需PDF） from urllib.parse import unquote rel = unquote(self.path.lstrip('/')) if rel.startswith('原书页/') and not os.path.exists(os.path.join(HERE, rel)): if self._serve_source_page(rel): return return super().do_GET() # ── 初中培优讲义：源 docx 预览/下载 ──────────────────────────── _units_cache = [None] def _units(self): """{专题标题: {key, docx, book}}，来自 数学学习系统/data/v2/units.json。""" if self._units_cache[0] is None: idx = {} for up in ('..', '../..', '.'): p = os.path.normpath(os.path.join( HERE, up, '数学学习系统', 'data', 'v2', 'units.json')) if os.path.exists(p): try: for u in json.load(open(p, encoding='utf-8')): idx[(u.get('title') or '').strip()] = u except Exception: pass break self._units_cache[0] = idx return self._units_cache[0] def _serve_unit_source(self): """/api/source?title=<专题标题>[&fmt=pdf] fmt 省略 → 原 docx（浏览器直接下载）；fmt=pdf → 转 PDF 内联预览（转换结果缓存）。""" from urllib.parse import urlparse, parse_qs, unquote, quote qs = parse_qs(urlparse(self.path).query) title = unquote((qs.get('title') or [''])[0]).strip() fmt = (qs.get('fmt') or [''])[0] u = self._units().get(title) if not u or not os.path.exists(u.get('docx') or ''): return self._json({'ok': False, 'error': '未找到该专题的源文件'}, 404) src = u['docx'] if fmt != 'pdf': data = open(src, 'rb').read() self.send_response(200) self.send_header('Content-Type', 'application/vnd.openxmlformats-' 'officedocument.wordprocessingml.document') self.send_header('Content-Disposition', "attachment; filename*=UTF-8''" + quote(os.path.basename(src))) self.send_header('Content-Length', str(len(data))) self.end_headers() return self.wfile.write(data) # 预览：**直接用已渲染好的页图**，不再调 LibreOffice。 # v2 流水线早就把 381 个单元全渲染成了 data/v2/pages//pNNN.png（16518 张）， # 再去 soffice 转一次 PDF 是白等 7 秒还起个 LibreOffice 进程 —— 纯属绕远路。 import glob as _glob from urllib.parse import quote as _q V2 = os.path.dirname(os.path.dirname(u['docx'])) if False else None for up in ('..', '../..', '.'): cand = os.path.normpath(os.path.join(HERE, up, '数学学习系统', 'data', 'v2')) if os.path.exists(os.path.join(cand, 'units.json')): V2 = cand; break pngs = sorted(_glob.glob(os.path.join(V2 or '', 'pages', u['key'], 'p*.png'))) if V2 else [] if not pngs: return self._json({'ok': False, 'error': '该专题没有已渲染的原书页图'}, 404) imgs = ''.join( '
%s / %d
' % (_q(u['key']), _q(os.path.basename(f)), os.path.basename(f)[1:-4].lstrip('0') or '1', len(pngs)) for f in pngs) html = ('%s' '' '
%s · 共 %d 页⬇ 下载原始 docx
%s' % (u['title'], u['title'], len(pngs), _q(title), imgs)).encode('utf-8') self.send_response(200) self.send_header('Content-Type', 'text/html; charset=utf-8') self.send_header('Content-Length', str(len(html))) self.end_headers() return self.wfile.write(html) # ── 专题级源文件：预览页图 / 下载切好的专题 PDF ────────────── def _topic_args(self): from urllib.parse import urlparse, parse_qs, unquote q = parse_qs(urlparse(self.path).query) return (unquote((q.get('tb') or [''])[0]), unquote((q.get('g') or [''])[0]), unquote((q.get('cd') or [''])[0]), [x for x in unquote((q.get('pg') or [''])[0]).split('|') if x], unquote((q.get('src') or [''])[0]), unquote((q.get('role') or [''])[0]) or None, unquote((q.get('label') or [''])[0]) or None) def _topic_files(self): """/api/topic/files?tb=&g=&cd= → 这个专题**实际有哪几份**源文件。 ⚠ 各教材份数和轴都不一样，不能在前端写死两行： 初中培优讲义 学生版 / 教师版 （角色轴） 胡小群 讲义 / 课后 / 课后答案 （资料轴 × 角色轴，**3 份**） 一课一练 本课 / 全册答案 动脑筋·举一反三 专题 / 整册答案 踩过：先前按「学生版/教师版」写死两行，胡小群 120 个专题的**课后习题够不着**。 """ try: import topic_source as TS except Exception: return self._json({'ok': False, 'files': []}) tb, g, cd, pgs, sh, role, label = self._topic_args() out = [] for h in TS.precut_all(tb, g, cd): out.append({'label': h['label'] or h.get('role') or '源文件', 'role': h.get('role', '学生版'), 'ext': os.path.splitext(h['path'])[1].lstrip('.').lower()}) return self._json({'ok': True, 'files': out}) def _topic_preview(self): """/api/topic?tb=&g=&cd=&pg=a|b|c → 该专题所有页图的滚动预览页。""" from urllib.parse import quote try: import topic_source as TS except Exception: return self._json({'ok': False, 'error': 'topic_source 未就绪'}, 500) tb, g, cd, pgs, sh, role, label = self._topic_args() loc = TS.locate(tb, g, cd, sh) # 教材专属（初中=整单元页图） files = None # ① 优先：已拆好的专题成品（源题文件/_专题拆分/）——预览的就是下载会拿到的那个文件 pc = TS.precut(tb, g, cd, sh, role, label) if pc: # 成品文件在手 → 预览必须严格是**它自己**的页。pdf 和 docx 都交给 precut_pages # （docx 走 LibreOffice 现渲，首次约 6 秒，之后缓存）。 # ⚠ 拿不到就宁可空着走下面「只给下载」那支，**绝不退回 loc 的页图**： # 那是题库那一版（初中培优 units.json 指的是解析版），退回去就成了 # 「点学生版预览看到的是答案版」。踩过 —— 原先这里只对 .pdf 调 precut_pages， # docx 直接掉进 loc['pages']，381 个初中培优专题两行预览都是解析版的 37 页。 files = TS.precut_pages(pc, pgs, loc) else: if loc and loc.get('pages'): files = loc['pages'] if not files: # 兜底：用前端传来的题目 pg 并集 r = TS.locate_by_questions(pgs) files = r['pages'] if r else [] if not files and loc and loc.get('pdf') and loc['pdf'][0]: # 没有预渲染页图 → pdftoppm 现渲（0.15秒/页，非 LibreOffice）， # 渲的**正是下载会给的那几页**，保证预览与下载同源。 src0, lo0, hi0 = loc['pdf'] files = TS.render_pdf_pages(src0, lo0 or 1, hi0 or 0) if not files: # 实在渲不出 → 给下载入口，别 404 让用户以为坏了 if pc or (loc and loc.get('pdf')): dl = ('⬇ 下载本专题 PDF' % (quote(tb), quote(g), quote(cd), quote(sh), quote('|'.join(pgs)))) html = ('%s' '' '

%s

这个专题的源文件是 PDF，没有预渲染页图

' '

%s

' % (cd, cd, dl)).encode('utf-8') self.send_response(200) self.send_header('Content-Type', 'text/html; charset=utf-8') self.send_header('Content-Length', str(len(html))) self.end_headers() return self.wfile.write(html) return self._json({'ok': False, 'error': '该专题没有可用的原书页图'}, 404) dl = '' alts = TS.precut_all(tb, g, cd) if len(alts) > 1: # 一个课程节下同时有讲义和课后（胡小群）→ 两个都给出来，别只给一个 dl = ''.join('⬇ 下载%s' % (quote(tb), quote(g), quote(cd), quote(a['label']), quote('|'.join(pgs)), a['label'] or '本专题') for a in alts) elif (loc and loc.get('docx')) or pgs or (loc and loc.get('pdf')) or alts: dl = ('⬇ 下载本专题' % (quote(tb), quote(g), quote(cd), quote(sh), quote('|'.join(pgs)))) figs = ''.join( '
%s
' % (quote(os.path.relpath(f, HERE)), os.path.basename(f)) for f in files) html = ('%s' '' '
%s · %s · 共 %d 页%s
%s' % (cd, cd, g, len(files), dl, figs)).encode('utf-8') self.send_response(200) self.send_header('Content-Type', 'text/html; charset=utf-8') self.send_header('Content-Length', str(len(html))) self.end_headers() return self.wfile.write(html) def _topic_pdf(self): """/api/topic/pdf?... → 初中给 docx；其余用 qpdf 从整本切出该专题页范围（缓存）。""" from urllib.parse import quote try: import topic_source as TS except Exception: return self._json({'ok': False, 'error': 'topic_source 未就绪'}, 500) tb, g, cd, pgs, sh, role, label = self._topic_args() # ① 优先：已拆好的专题成品，直接送（免去每次现切 + 缓存目录） pc = TS.precut(tb, g, cd, sh, role, label) if pc: ext = os.path.splitext(pc['path'])[1].lower() ctype = ('application/vnd.openxmlformats-officedocument.wordprocessingml.document' if ext == '.docx' else 'application/pdf') # 带上这份是「讲义/课后/课后答案」——同一专题几份都叫「相遇问题.pdf」会互相盖掉。 # 整册那种（讲义答案（全册）/全册答案/整册答案）按册命名：内容与专题无关， # 15 个专题下同一份文件不该存成 15 个不同名字。 tag = pc.get('label') or '' if '全册' in tag or '整册' in tag: name = re.sub(r'[/\\:*?"<>|]', '_', f"{pc.get('book') or ''}·{tag}").strip('·')[:60] + ext else: name = re.sub(r'[/\\:*?"<>|]', '_', cd)[:60] + (f'·{tag}' if tag else '') + ext return self._send_file(pc['path'], ctype, name) loc = TS.locate(tb, g, cd, sh) if loc and loc.get('docx') and os.path.exists(loc['docx']): return self._send_file(loc['docx'], 'application/vnd.openxmlformats-' 'officedocument.wordprocessingml.document', os.path.basename(loc['docx'])) # 找整本 PDF + 页范围 src = lo = hi = None if loc and loc.get('pdf'): src, lo, hi = loc['pdf'] if not src: # 兜底：从题目页图路径反推整本 PDF+页范围 fb = TS.fallback_pdf(pgs) if fb: src, lo, hi = fb if not src: return self._json({'ok': False, 'error': '该教材暂无可切分的整本 PDF'}, 404) if not hi: # hi=0 → 源本身就是该专题的完整文件，直接送 return self._send_file(src, 'application/pdf', os.path.basename(src)) cache = os.path.join(HERE, '.cache_topic') os.makedirs(cache, exist_ok=True) sig = hashlib.md5(f'{src}|{lo}|{hi}'.encode()).hexdigest()[:16] out = os.path.join(cache, sig + '.pdf') if not os.path.exists(out) and not TS.cut_pdf(src, lo, hi, out): return self._json({'ok': False, 'error': 'PDF 切分失败'}, 500) name = re.sub(r'[/\\:*?"<>|]', '_', cd)[:60] + '.pdf' return self._send_file(out, 'application/pdf', name) def _send_file(self, path, ctype, name): from urllib.parse import quote data = open(path, 'rb').read() self.send_response(200) self.send_header('Content-Type', ctype) self.send_header('Content-Disposition', "attachment; filename*=UTF-8''" + quote(name)) self.send_header('Content-Length', str(len(data))) self.end_headers() return self.wfile.write(data) def _serve_page_png(self): """/api/page?f=<相对 组卷系统/ 的页图路径> —— 供专题预览内联页图。 只允许指向已知页图目录，防目录穿越。""" from urllib.parse import urlparse, parse_qs, unquote q = parse_qs(urlparse(self.path).query) rel = unquote((q.get('f') or [''])[0]) fp = os.path.normpath(os.path.join(HERE, rel)) ok_roots = [os.path.normpath(os.path.join(HERE, '原书页')), # ⚠ pdftoppm 现渲的预览页落在这里，漏了它现渲的预览图会全部 404 os.path.normpath(os.path.join(HERE, '.cache_pages')), os.path.normpath(os.path.join(HERE, '..', '成果')), os.path.normpath(os.path.join(HERE, '..', '数学学习系统'))] if not (fp.lower().endswith(('.png', '.jpg', '.jpeg')) and any(fp.startswith(r) for r in ok_roots) and os.path.exists(fp)): return self._json({'ok': False, 'error': 'not found'}, 404) data = open(fp, 'rb').read() self.send_response(200) self.send_header('Content-Type', 'image/png' if fp.endswith('.png') else 'image/jpeg') self.send_header('Cache-Control', 'max-age=86400') self.send_header('Content-Length', str(len(data))) self.end_headers() return self.wfile.write(data) def _serve_source_page(self, rel): # 一课一练: 原书页/ykl/{路由key}/pNNN.png -> 成果/一课一练题库/<书目录>/pages/ m = re.match(r'原书页/ykl/(1sz|1s|2sz|2s|3s|3sz|3x|3xz|4s|4sz|4x|4xz|5s|5sz|5x|5xz|6s)/p0*(\d+)\.png$', rel) if m: YKLD = {'1s': '一年级上普通2024秋', '1sz': '一年级上增强2024秋', '2s': '二年级上普通', '2sz': '二年级上增强', '3s': '三年级上', '3sz': '三年级上增强2022秋', '3x': '三年级下普通2022秋', '3xz': '三年级下增强2022秋', '4s': '四年级上普通2026秋', '4sz': '四年级上增强2026秋', '4x': '四年级下普通2023春', '4xz': '四年级下增强2023春', '5s': '五年级上普通2026秋', '5sz': '五年级上增强2026秋', '5x': '五年级下普通2023春', '5xz': '五年级下增强2023春', '6s': '六年级上普通2024秋'} png = os.path.join(HERE, '..', '成果', '一课一练题库', YKLD[m.group(1)], 'pages', 'p%03d.png' % int(m.group(2))) return self._send_png(png) # rel = 原书页/{dnj|jyf}/{1-6}/pNNN.jpg m = re.match(r'原书页/(dnj|jyf)/([1-6])/p0*(\d+)\.jpg$', rel) if not m: return False TBM = {'dnj': '动脑筋题库', 'jyf': '举一反三题库'} GC = {'1': '一', '2': '二', '3': '三', '4': '四', '5': '五', '6': '六'} png = os.path.join(HERE, '..', '成果', TBM[m.group(1)], GC[m.group(2)] + '年级', 'pages', 'p%03d.png' % int(m.group(3))) return self._send_png(png) def _send_png(self, png): if not os.path.exists(png): return False data = open(png, 'rb').read() self.send_response(200) self.send_header('Content-Type', 'image/png') self.send_header('Content-Length', str(len(data))) self.send_header('Cache-Control', 'max-age=86400') self.end_headers() self.wfile.write(data) return True def do_POST(self): if self.path == '/api/compose': return self._do_compose() if self.path == '/api/similar': return self._do_similar() if self.path == '/api/similar_batch': return self._do_similar_batch() if self.path == '/api/layout': return self._do_layout() if self.path == '/api/export_docx': return self._do_export_docx() if self.path == '/api/report': return self._do_report() if self.path == '/api/openword': return self._do_openword() self.send_error(404) # 中考真题原卷索引：卷 id -> 源文件绝对路径。第一次用时才建（读三个 jsonl，约 0.4s）。 _PAPERS = None @classmethod def _paper_index(cls): if cls._PAPERS is not None: return cls._PAPERS idx = {} try: base = os.path.abspath(os.path.join( HERE, '..', '数学学习系统', 'data', 'drafts', 'batch_exams_final')) root = os.path.abspath(os.path.join( HERE, '..', '源题文件', '中考一模二模真题')) L = lambda f: [json.loads(x) for x in open(os.path.join(base, f), encoding='utf-8')] src = {r['external_id']: r for r in L('source_documents.jsonl')} for p in L('exam_papers.jsonl'): sids = p.get('source_document_ids') or [] r = src.get(sids[0]) if sids else None if not r: continue fp = os.path.abspath(os.path.join(root, r['original_relpath'])) # ⚠ 越界检查：relpath 里若有 ../ 就能指到仓库外任意文件 if fp.startswith(root + os.sep) and os.path.exists(fp): idx[p['external_id']] = fp except Exception: pass cls._PAPERS = idx return idx def _serve_paper_file(self): """/api/paper?pid=<卷id> → 下载这份中考原卷的**原始文件**。 ⚠ 故意**不做 PDF 预览**。data/work/src_pdf 里那 608 份是审核期 LibreOffice 转的，MathType/WMF 公式在它手里会缺笔画甚至整块丢 （269 卷源文件还是老 .doc，更悬）。老师看到的应该是原件， 不是我们的转换产物 —— 所以只给「WPS 打开」和「下载原件」两条路， 两条都不经任何转换。 """ from urllib.parse import urlparse, parse_qs, unquote, quote q = parse_qs(urlparse(self.path).query) pid = unquote((q.get('pid') or [''])[0]).strip() fp = self._paper_index().get(pid) if not fp: return self._json({'ok': False, 'error': '这道题没有对应的原卷文件'}, 404) data = open(fp, 'rb').read() ext = os.path.splitext(fp)[1].lower() self.send_response(200) self.send_header('Content-Type', 'application/msword' if ext == '.doc' else 'application/vnd.openxmlformats-officedocument.' 'wordprocessingml.document') self.send_header('Content-Disposition', "attachment; filename*=UTF-8''" + quote(os.path.basename(fp))) self.send_header('Content-Length', str(len(data))) self.end_headers() self.wfile.write(data) def _do_openword(self): """用系统默认程序（WPS/Word）打开这道题所在的原卷。 ⚠ 只接受**卷 id**，路径在服务端解析 —— 前端传路径等于把 「打开任意文件」的能力交给页面。 ⚠ 原卷很多是「试卷及答案」版，**带答案**。这是本地老师工位的功能； 公网部署没有这个后端，按钮会提示「需要本地助手」，不会泄漏。 """ try: body = self._read_body() fp = self._paper_index().get(body.get('pid') or '') if not fp: return self._json({'ok': False, 'error': '这道题没有对应的原卷文件'}, 404) subprocess.Popen(['open', fp]) self._json({'ok': True, 'name': os.path.basename(fp)}) except Exception as e: self._json({'ok': False, 'error': str(e)[:300]}, 500) def _do_report(self): # 一键报告问题：把题目 id + 上下文追加到 问题报告.jsonl（服务端盖时间戳），供事后核对修图/改题 try: import datetime body = self._read_body() if not body.get('id'): return self._json({'ok': False, 'error': '缺 id'}, 400) body['ts'] = datetime.datetime.now().isoformat(timespec='seconds') with open(os.path.join(HERE, '问题报告.jsonl'), 'a', encoding='utf-8') as f: f.write(json.dumps(body, ensure_ascii=False) + '\n') self._json({'ok': True}) except Exception as e: self._json({'ok': False, 'error': str(e)[:300]}, 500) def _do_layout(self): """版式决策接口：前端把试卷篮发过来，后端回每道题该怎么排。 为什么要有这个接口：前端原本自己有一套排版逻辑（LAY 表 + isShortCalc）， 注释写着「与后端 export_lib.LAYOUT 一致」，但后端改成七类题型之后就漂了—— **预览显示的是旧版式、下载的是新版式，老师看到的和拿到的不是一回事**。 与其在 JS 里把七类逻辑再实现一遍（那注定会再次漂移，这次就是这么漂的）， 不如让**后端算、前端画**：版式决策只有一处实现，前端退化成执行者。 入参同 export_docx（items + space + cols），出参每题一条： kind 题型（choice/fill/calc/vform/solve/sub/draw/draw_on） blank_cm 这题后面留多高的书写空白 pair 是否与下一题并排成一行 fig_cm 这题的图限宽 另附 gap_cm（恒定题间距）。 """ try: import export_lib as E body = self._read_body() items = body.get('items') or [] space = body.get('space', 2) bands = body.get('bands') # 2/3/4 栏，0 或缺省=不排版 cols = bool(body.get('cols')) # 保留：双栏模式仍由前端决定要不要并排 # ⚠ 版式决策**一律走 export_lib.plan_bands** —— 预览和导出必须逐值一致。 # 以前两边各写一套，后端改题型分类后就漂了（预览一套、下载另一套）。 out = [] for kind, grp, blank_cm, brk in E.plan_bands(items, bands, space): for j, it in enumerate(grp): t, has_fig = E.classify(it) out.append({ 'kind': t, 'blank_cm': blank_cm if j == len(grp) - 1 else 0.0, 'pair': kind == 'pair', 'fig_cm': E.FIG_CM_BY_TYPE.get(t, E.FIG_CM), 'page_break_after': bool(brk) and j == len(grp) - 1, }) return self._json({'ok': True, 'gap_cm': E.GAP_CM, 'band_cm': round(E.band_height(bands), 2) if bands else 0, 'items': out}) except Exception as e: return self._json({'ok': False, 'error': str(e)[:300]}, 500) def _do_export_docx(self): # 试卷篮 -> pandoc 真公式 Word（LaTeX 转 Word 原生公式）。失败返回 JSON,前端回退纯文本。 try: import export_lib, urllib.parse body = self._read_body() if not (body.get('items')): return self._json({'ok': False, 'error': '试卷篮为空'}, 400) data = export_lib.build_paper_docx(body) fn = ((body.get('title') or '专题') + '专题练习.docx') self.send_response(200) self.send_header('Content-Type', 'application/vnd.openxmlformats-officedocument.wordprocessingml.document') self.send_header('Content-Disposition', "attachment; filename*=UTF-8''" + urllib.parse.quote(fn)) self.send_header('Content-Length', str(len(data))) self.end_headers() self.wfile.write(data) except Exception as e: self._json({'ok': False, 'error': str(e)[:300]}, 500) def _read_body(self): n = int(self.headers.get('Content-Length', 0)) return json.loads(self.rfile.read(n) or b'{}') def _do_compose(self): """AI 组卷：{text, catalog} → {spec, understood, fixes, unresolved}。 catalog 由前端从 COURSE 压出（只带候选范围，识别不出才全量 ~37KB）。 这里只翻译不选题 —— 选题在前端的 DATA 上做，服务端没有也不该有题库。 """ try: body = self._read_body() text = (body.get('text') or '').strip() catalog = body.get('catalog') or {} if not text: return self._json({'ok': False, 'error': '没有听到要求'}, 400) if not isinstance(catalog, dict) or not catalog: return self._json({'ok': False, 'error': '缺少专题目录（前端版本过旧？）'}, 400) if not KEY: return self._json({'ok': False, 'error': 'DeepSeek key 未配置'}, 503) kps, kp_show = [], [] for x in (body.get('kps') or [])[:3000]: if isinstance(x, str): kps.append(x); kp_show.append(x) elif isinstance(x, dict) and isinstance(x.get('n'), str): kps.append(x['n']) kp_show.append(x['n'] + ('（%s）' % x['g'] if x.get('g') else '')) try: raw = nl_to_spec(text, catalog, kp_show) except Exception as e: return self._json({'ok': False, 'error': 'AI 解析失败：%s' % str(e)[:120]}, 502) spec, fixes, unresolved = validate_spec(raw, catalog, kps) if not spec['groups']: return self._json({'ok': False, 'unresolved': unresolved, 'error': '没匹配到任何专题' + ('（找不到：%s）' % '、'.join(unresolved[:4]) if unresolved else '') + '，换个说法试试？'}, 422) return self._json({'ok': True, 'spec': spec, 'understood': compose_understood(spec), 'fixes': fixes, 'unresolved': unresolved}) except Exception as e: return self._json({'ok': False, 'error': str(e)[:200]}, 500) def _do_similar(self): try: body = self._read_body() stem = (body.get('stem') or '').strip() if not stem: return self._json({'ok': False, 'error': '缺少原题'}, 400) cnt = max(1, min(20, int(body.get('n', 10)))) by = (body.get('by') or DEFAULT_BY).strip() if by == 'variants': # 变式组走两轮机会：六策略一次出齐，核验毙掉谁就只补谁的策略更贵， # 干脆整组重出一次兜底（策略间无状态，重出成本 = 一次调用）。 keep, stat = [], {'dropped': 0, 'reasons': []} seen_vt = set() for _ in range(2): try: raw = gen_variants(stem, body.get('grade', ''), body.get('difficulty', ''), (body.get('topic') or '').strip()) except Exception as e: return self._json({'ok': False, 'by': 'variants', 'error': 'DeepSeek 调用失败: %s' % str(e)[:120]}, 502) ok_items, st = verify_gen.filter_items(raw, body.get('grade', '')) stat['dropped'] += st['dropped']; stat['reasons'] += st['reasons'] for it in ok_items: vt = (it.get('vtype') or '').strip() or '变式' if vt not in seen_vt: # 每策略只留一道 seen_vt.add(vt); keep.append(it) if len(keep) >= len(VAR_STRATEGIES): break if keep: # type 字段展示成「策略名」，前端现有 UI 会直接把它渲染成题型标签 for it in keep: it['type'] = it.get('vtype') or it.get('type') or '变式' return self._json({'ok': True, 'items': keep, 'by': 'variants', 'note': '按 %d/%d 个策略生成' % (len(keep), len(VAR_STRATEGIES)), 'stat': stat}) return self._json({'ok': False, 'by': 'variants', 'error': '变式没通过数值核验', 'stat': stat}, 502) if by == 'ai_figure': # 题图实验：只给已实现确定构图模板的专题开放，绝不执行模型自带命令。 if '圆与扇形' not in str(body.get('topic') or ''): return self._json({'ok': False, 'by': 'ai_figure', 'error': 'AI 配图实验当前只支持「圆与扇形」专题'}, 400) raw = gen_figure_similar(stem, body.get('grade', ''), body.get('difficulty', ''), (body.get('topic') or '').strip(), min(cnt, 3)) items, st = verify_figure_items(raw, min(cnt, 3)) if items: return self._json({'ok': True, 'items': items, 'by': 'ai_figure', 'note': '题图结构已核对；打开几何画板后可继续编辑', 'stat': st}) return self._json({'ok': False, 'by': 'ai_figure', 'error': '生成的题图题未通过结构或数值核验', 'stat': st}, 502) if by == 'gen': # ① 生成器：秒出、带答案解析 if not gen_router: return self._json({'ok': False, 'error': '生成器不可用'}, 503) hit = gen_router.route((body.get('topic') or '').strip(), body.get('difficulty', ''), cnt, body.get('grade', ''), seed=body.get('seed')) # seed 复现契约：同 seed 同题 if hit: return self._json({'ok': True, 'items': hit, 'by': 'gen', 'seed': hit[0].get('seed')}) # 命不中不要偷偷改走别的路 —— 老师点的是生成器，就如实说没有 return self._json({'ok': False, 'by': 'gen', 'error': '生成器没有这个题型（只覆盖 58 个题型），换「AI」或「学科网」试试'}, 404) if by == 'xkw': # ③ 学科网真题 if not XKW_ENABLED: return self._json({'ok': False, 'by': 'xkw', 'error': '学科网出同类题已暂时下线：接口资源包用尽（402）。' '加包后由管理员把服务端 XKW_ENABLED 改回 True 即可恢复。'}, 503) if not xkw_similar: return self._json({'ok': False, 'error': '学科网客户端未就绪'}, 503) img = (body.get('fig') or '').strip() # 带图题：题图 base64（AI 配不出图，只有这条路能走） if img.startswith('data:'): img = img.split(',', 1)[-1] mode = 'image' if img else (body.get('mode') or 'fast') # ── 每道种子最多 3 次（前端会置灰，这里是服务端兜底）── seed = body.get('qid') or _fp(stem) used = _XKW_SEED.get(seed, 0) if used >= XKW_PER_SEED: return self._json({'ok': False, 'by': 'xkw', 'exhausted': True, 'error': '这道题的同类题已取完（学科网单题上限约 20 道）。' '换一道种子，或改用「生成器」/「AI」出题。'}, 429) # ── 会话内缓存：翻回上一批不重打（⚠ 只在内存，不写盘 —— 授权红线）── ck = (seed, mode, used) st = {} if ck in _XKW_CACHE: items, err = _XKW_CACHE[ck] else: _XKW_SEED[seed] = used + 1 items, err = xkw_similar.similar(stem, body.get('grade', ''), cnt, mode=mode, image_b64=img or None, tb=body.get('tb', ''), topic=(body.get('topic') or '').strip(), stats=st) if items: _XKW_CACHE[ck] = (items, err) if len(_XKW_CACHE) > 300: # 别无限长，进程内够用就行 _XKW_CACHE.pop(next(iter(_XKW_CACHE))) if items: # note 说明走了哪条链路、实际几道 —— 前端如实显示，别让老师 # 以为点了「出10道」就一定有10道（v1 相似题上限就是 5） return self._json({'ok': True, 'items': items, 'by': 'xkw', 'note': err or '', 'copyright': xkw_similar.COPYRIGHT_NOTE}) # 接口给了题、但难度一道都不合格 → **这道种子**到此为止，后面两级不用再试了。 # 按种子降级，不按教材封杀（见文件头 XKW_NO_HARD 上面那段）。 if st.get('got') and not st.get('kept'): _XKW_SEED[seed] = XKW_PER_SEED # 直接标成用完，后续请求 429 return self._json({'ok': False, 'by': 'xkw', 'exhausted': True, 'error': XKW_NO_HARD, 'note': err or ''}, 429) return self._json({'ok': False, 'by': 'xkw', 'error': (err or '') + '学科网没找到同类题'}, 502) items, st = gen_verified(stem, body.get('grade', ''), body.get('difficulty', ''), (body.get('topic') or '').strip(), cnt) if items: self._json({'ok': True, 'items': items, 'by': 'deepseek', 'stat': st}) elif st.get('dropped'): # 出了题但一道都没过核验 —— 说明这类题模型算不明白，别硬塞给老师 self._json({'ok': False, 'error': '生成的题都没通过数值核验（%s）' % '；'.join(r['why'] for r in st.get('reasons', [])[:3]), 'stat': st}, 502) else: self._json({'ok': False, 'error': '解析生成结果失败'}, 502) except urllib.error.HTTPError as e: self._json({'ok': False, 'error': 'DeepSeek HTTP %d: %s' % (e.code, e.read().decode('utf-8', 'ignore')[:300])}, 502) except Exception as e: self._json({'ok': False, 'error': str(e)}, 500) def _do_similar_batch(self): # 整页批量：每道题各出 1 道类似题，并发调用（限流用小线程池），保序返回。 try: body = self._read_body() reqs = body.get('items') or [] if not isinstance(reqs, list) or not reqs: return self._json({'ok': False, 'error': '缺少题目列表'}, 400) reqs = reqs[:40] # 上限保护 cnt = max(1, min(3, int(body.get('n', 1)))) def work(it): stem = (it.get('stem') or '').strip() rec = {'id': it.get('id', ''), 'stem': stem} if not stem: return dict(rec, ok=False, error='空题', items=[]) by = (it.get('by') or body.get('by') or DEFAULT_BY).strip() if by == 'gen' and gen_router: hit = gen_router.route((it.get('topic') or '').strip(), it.get('difficulty', ''), cnt, it.get('grade', '')) return dict(rec, ok=bool(hit), items=hit or [], by='gen', error='' if hit else '生成器没有这个题型') # 🔴 **整页批量一律不走学科网**（小克 2026-07-28 的第一条节流，挡掉约 90% 额度风险）： # 30 题的卷子点一次「整页出类似题」= 30 次 ai 调用 = 300 道题，**一次点击**。 # `/xopqbm/questions` 就是这么烧穿的（393 次调用只有 60 次产出成果）。 # 整页只走确定性生成器（毫秒级、零额度、自带答案解析）； # 学科网只在**单题手动**三选一里触发。 if by == 'xkw': by = 'gen' if gen_router else 'ai' if by == 'gen': hit = gen_router.route((it.get('topic') or '').strip(), it.get('difficulty', ''), cnt, it.get('grade', '')) return dict(rec, ok=bool(hit), items=hit or [], by='gen', error='' if hit else '生成器没有这个题型', note='整页批量不走学科网（省额度），已改用生成器') try: items, st = gen_verified(stem, it.get('grade', ''), it.get('difficulty', ''), (it.get('topic') or '').strip(), cnt) return dict(rec, ok=bool(items), items=items, stat=st, error='' if items else '没通过数值核验') except urllib.error.HTTPError as e: return dict(rec, ok=False, error='HTTP %d' % e.code, items=[]) except Exception as e: return dict(rec, ok=False, error=str(e)[:120], items=[]) with concurrent.futures.ThreadPoolExecutor(max_workers=10) as ex: # 实测10路并发1.2s全返无限流(线程只在等HTTP,不占算力) results = list(ex.map(work, reqs)) ok_n = sum(1 for r in results if r['ok']) self._json({'ok': True, 'results': results, 'ok_n': ok_n, 'total': len(results)}) except Exception as e: self._json({'ok': False, 'error': str(e)}, 500) def _json(self, obj, code=200): b = json.dumps(obj, ensure_ascii=False).encode('utf-8') self.send_response(code) self.send_header('Content-Type', 'application/json; charset=utf-8') self.send_header('Content-Length', str(len(b))) self.end_headers() self.wfile.write(b) def log_message(self, *a): pass # 静默（key 不在 URL/请求行，但仍不打印请求日志） class Server(socketserver.ThreadingTCPServer): allow_reuse_address = True daemon_threads = True def handle_error(self, request, client_address): # 浏览器中断连接(刷新/关标签页时大文件传一半)是常态,不打吓人的traceback import sys exc = sys.exc_info()[1] if isinstance(exc, (BrokenPipeError, ConnectionResetError)): return super().handle_error(request, client_address) if __name__ == '__main__': os.chdir(HERE) print('=' * 56) print(' 组卷台 AI 助手（出类似题 · 基于选中题生成练习）') print(' 模型：%s | key：%s' % (MODEL, '已载入' if KEY else '⚠ 缺失，请检查 ~/.deepseek_pro_key')) print(' 打开浏览器访问： http://localhost:%d/组卷台.html' % PORT) print(' 停止：按 Ctrl+C') print('=' * 56) try: Server(('127.0.0.1', PORT), Handler).serve_forever() except KeyboardInterrupt: print('\n已停止。')