# -*- coding: utf-8 -*- """校验问题清单 + 通知模板与手工发送 E2E（《03-接口设计》§12.1~12.3、§12.7）。 ## 这一批接口的价值在"对齐"两个字上 · **§12.1~12.3 的问题清单**不是一个"查询接口"，而是**历次校验的快照**： 每跑一次就把算出来的问题对齐进表。所以验收重点不是"能不能查"， 而是**三种情况下语义对不对**： ① 算出来了 + 表里也有 → 只更新内容，**不碰 resolved** （用户在界面上点过"已处理"，下次跑校验不能把它改回未解决 —— 那等于系统推翻用户的判断，而这会让人再也不敢点那个按钮）； ② 算出来了 + 表里没有 → 插入； ③ 表里有 + 这次没算出来 → **标成已解决**（"上轮 47 个 → 这轮 5 个"的来源）。 · **§12.7 通知模板**：库里只存"被改过的"，没改过的走代码里的内置文案。 所以"恢复默认"就是删掉库里那一行 —— 不需要另存一份原文。 ## 还要验的两件契约要求 · **越界变量**：契约写明"越界变量渲染为空**并告警**"。实现上选择**原样保留** 加 WARN —— 理由是模板由非技术人员编辑，把 `{名字}` 渲染成空之后 用户只会以为"系统没数据"，而保留原样他一眼就能看到自己写错了。 · **微信回调**：唯一一个匿名**写**入口，必须①验签②只写推送状态不读业务数据 ③凭记录 id 反查租户（不能猜）。 跑法：python scripts/e2e-m7-validation.py """ import argparse import importlib.util import json import os import subprocess import sys sys.stdout.reconfigure(encoding="utf-8") ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) PASSED = [] FAILED = [] CREATED_TASK_IDS = [] TMP_ROSTER = os.path.join(ROOT, "scripts", "_e2e-val-roster.xlsx") def _load(name, fn): spec = importlib.util.spec_from_file_location(name, os.path.join(ROOT, "scripts", fn)) mod = importlib.util.module_from_spec(spec) spec.loader.exec_module(mod) return mod R = _load("e2e_m2_roster", "e2e-m2-roster.py") S = _load("e2e_m3_submit", "e2e-m3-submit.py") A = _load("e2e_m5_audit", "e2e-m5-audit.py") BASE = "http://localhost:8080" STU = 994101 def ok(name): PASSED.append(name) print(" [OK] " + name) def fail(name, detail): FAILED.append((name, detail)) print(" [FAIL] " + name + "\n " + detail) def check(name, cond, detail=""): if cond: ok(name) else: fail(name, detail or "断言不成立") return bool(cond) def sql(stmt): _, out = R.sql(stmt) return out def body_of(method, path, token=None, payload=None): _, b = R.call(method, BASE + path, token, payload) return b def mp(uid): cfg = R.read_dev_config() return R.mint_mp_token(cfg["jwt_secret"], cfg["issuer"], uid, 1, ["STUDENT"], 4) def main(): ap = argparse.ArgumentParser(description="校验问题清单 E2E") ap.parse_args() print("=" * 74) print(" 校验问题清单 + 通知模板 E2E（§12.1~12.3 / §12.7）") print("=" * 74) _, adm = R.login(BASE, "admin", "Admin@123", "demo") if adm.get("code") != 0: print(" x 管理员登录失败：" + str(adm)) return 2 at = adm["data"]["token"] print("\n[0] 摆夹具（一个任务 + 一份有问题的填报单）") suffix = os.urandom(3).hex() _, b = R.call("POST", BASE + "/api/v1/tasks", at, { "taskName": "E2E校验 " + suffix, "batchNo": "202609", "majorName": "数控加工", "collectMode": 1, "certType": 1, "startTime": "2026-01-01", "endTime": "2026-12-31" }) if b.get("code") != 0: print(" x 建任务失败：" + str(b)) return 2 task_id = b["data"]["id"] CREATED_TASK_IDS.append(task_id) _, b = R.call("GET", BASE + "/api/v1/orgs/tree", at) classes = [] def walk(nodes): for n in nodes or []: if n.get("orgType") == 3: classes.append(n) walk(n.get("children")) walk(b.get("data") or []) org = classes[0] R.call("PUT", BASE + "/api/v1/tasks/%d/orgs" % task_id, at, {"orgs": [{"orgId": org["id"], "expectedCount": 1}]}) R.call("POST", BASE + "/api/v1/tasks/%d/materials/apply-template" % task_id, at, {"templateCode": "TPL_SCHOOL_ROLL"}) _build_roster([("校验甲", R.with_checksum("36102920080505081"))]) body = R.upload_roster(BASE, at, task_id, TMP_ROSTER, extra={"dryRun": "false", "mode": "APPEND", "orgId": org["id"]}) check("夹具：名册就绪", body.get("code") == 0, json.dumps(body, ensure_ascii=False)[:180]) R.call("POST", BASE + "/api/v1/tasks/%d/publish" % task_id, at, {"force": True}) tok = mp(STU) _, b = R.call("POST", BASE + "/api/v1/mp/roster/match", tok, {"taskId": task_id, "realName": "校验甲", "idCardNo": R.with_checksum("36102920080505081")}) check("夹具：学生门禁绑定", b.get("code") == 0, json.dumps(b, ensure_ascii=False)[:180]) _, b = R.call("POST", BASE + "/api/v1/mp/submissions", tok, {"taskId": task_id}) sid = b["data"]["id"] check("夹具：学生建了草稿", b.get("code") == 0, json.dumps(b, ensure_ascii=False)[:150]) # ── 1. §12.1 第一次跑校验 ── print("\n[1] §12.1 触发全量校验（草稿状态：必然有问题）") run1 = body_of("POST", "/api/v1/validation/run", at, {"taskId": task_id, "scope": "ALL"}) check("校验接口可用", run1.get("code") == 0, json.dumps(run1, ensure_ascii=False)[:250]) v1 = run1.get("data") or {} check("返回 jobId 与 estimatedSeconds（契约要求的两个字段）", bool(v1.get("jobId")) and v1.get("estimatedSeconds") is not None, json.dumps(v1, ensure_ascii=False)) check("⭐ 校验了 1 份填报单", v1.get("checkedSubmissions") == 1, json.dumps(v1, ensure_ascii=False)) check("⭐ 发现了问题（草稿必然缺材料）", (v1.get("issuesFound") or 0) > 0, json.dumps(v1, ensure_ascii=False)) check("⭐ 返回剩余未解决数（用户最想立刻知道的那个数）", (v1.get("unresolvedLeft") or 0) > 0, json.dumps(v1, ensure_ascii=False)) check("耗时被记下来了（一期是同步跑完的，见 F-51）", (v1.get("elapsedMs") or 0) >= 0, str(v1.get("elapsedMs"))) n_first = sql("select count(*) from collect_validation_issue " "where task_id=%d and deleted=0;" % task_id).strip() check("问题落库了", n_first != "0", n_first) # ── 2. §12.2 清单 ── print("\n[2] §12.2 问题清单") lst = body_of("GET", "/api/v1/validation/issues?taskId=%d&pageSize=50" % task_id, at) check("清单可用", lst.get("code") == 0, json.dumps(lst, ensure_ascii=False)[:200]) data = lst.get("data") or {} check("带 total", (data.get("total") or 0) > 0, json.dumps(data, ensure_ascii=False)[:150]) c = data.get("counts") or {} check("⭐ 带 counts（各级别数量）", all(k in c for k in ("severity1", "severity2", "severity3")), json.dumps(c, ensure_ascii=False)) check("⭐ counts 之和 = total", (c.get("severity1", 0) + c.get("severity2", 0) + c.get("severity3", 0)) == (data.get("total") or 0), "counts=%s total=%s" % (json.dumps(c), data.get("total"))) br = data.get("byRule") or [] check("⭐ 带 byRule（按规则分组）", len(br) >= 1, json.dumps(br, ensure_ascii=False)[:200]) check("⭐ byRule 每条都有 count 与 autoFixed", all("count" in x and "autoFixed" in x for x in br), json.dumps(br, ensure_ascii=False)[:200]) rows = data.get("list") or [] check("清单行带学生姓名（不带给前端就 N+1）", all(x.get("realName") == "校验甲" for x in rows if x.get("submissionId") == sid), json.dumps(rows[:1], ensure_ascii=False)[:250]) check("清单行带班级名", any(x.get("orgName") for x in rows), json.dumps(rows[:1], ensure_ascii=False)[:200]) sevs = [x.get("severity") for x in rows] check("⭐ 严重的问题排在前面（第一屏必须是「最该先看的」）", sevs == sorted(sevs, reverse=True), str(sevs[:8])) check("按 severity 过滤生效", all(x.get("severity") == 3 for x in (body_of( "GET", "/api/v1/validation/issues?taskId=%d&severity=3" % task_id, at ).get("data") or {}).get("list") or []), "见上面清单") # ── 3. 重跑的三种语义（这是最容易做错的地方）── print("\n[3] 重跑语义：①已有→只更新 ②新增→插入 ③消失→标已解决") first_ids = set(x["id"] for x in rows) # 人工把其中一条标成已解决，模拟"用户在界面上点了已处理" one_id = sorted(first_ids)[0] R.sql("update collect_validation_issue set resolved=1, resolved_by=2, " "resolved_time=now() where id=%d;" % one_id) run2 = body_of("POST", "/api/v1/validation/run", at, {"taskId": task_id, "scope": "ALL"}) check("重跑成功", run2.get("code") == 0, json.dumps(run2, ensure_ascii=False)[:200]) check("⭐ 重跑不会把「已处理」改回未解决（改了等于推翻用户的判断）", sql("select resolved from collect_validation_issue where id=%d;" % one_id).strip() == "1", sql("select id, resolved, resolved_by from collect_validation_issue where id=%d;" % one_id)) check("⭐ 重跑不会重复插入（同一问题只有一行）", sql("select count(*) from collect_validation_issue where task_id=%d and deleted=0;" % task_id).strip() == n_first, "第一次 %s，第二次 %s" % (n_first, sql( "select count(*) from collect_validation_issue where task_id=%d " "and deleted=0;" % task_id).strip())) # 把学生补齐并提交 → 缺材料的问题应该"消失"并被标成已解决 res = A.prepare_student(tok, sid, task_id, { "work_unit": "无", "is_retake": False, "is_student": True, "school": "示范职业学校", "major": "数控加工", "enroll_date": "2024-09-01" }, dup_first_file=False) check("学生补齐并提交", ((res.get("data") or {}).get("submitted")) is True, json.dumps(res, ensure_ascii=False)[:220]) run3 = body_of("POST", "/api/v1/validation/run", at, {"taskId": task_id, "scope": "ALL"}) v3 = run3.get("data") or {} check("第三次跑成功", run3.get("code") == 0, json.dumps(run3, ensure_ascii=False)[:200]) # ⚠️ 断言要精确到"缺材料那一条"：夹具把同一张小图用在了两项材料上， # 会触发「有内容完全相同的材料」—— 那是**真实且正确**的校验结果， # 所以笼统地看"总数下降"会被它挡住 miss_left = sql("select count(*) from collect_validation_issue where task_id=%d " "and resolved=0 and title like '%%缺少必传材料%%';" % task_id).strip() check("⭐ 补齐材料之后，「缺少必传材料」那一条未解决数归零", miss_left == "0", "还剩 " + miss_left) auto_resolved = sql("select count(*) from collect_validation_issue " "where task_id=%d and resolved=1 and resolved_by is null;" % task_id).strip() check("⭐ 「这次没再算出来」的被标成已解决，且 resolved_by 为空（= 系统判定）", auto_resolved != "0", auto_resolved) check("而人工点过的那条仍记着处理人", sql("select resolved_by from collect_validation_issue where id=%d;" % one_id).strip() == "2", sql("select resolved_by from collect_validation_issue where id=%d;" % one_id)) # ── 4. 自动修复的留痕（byRule.autoFixed 要有真实数据）── print("\n[4] byRule.autoFixed：系统替你改过什么，要有据可查") fixed = sql("select count(*) from collect_validation_issue where task_id=%d " "and auto_fixed=1;" % task_id).strip() check("⭐ 上传时改了文件名的，被记成 VAL-06 自动修复", fixed != "0", "auto_fixed=1 的行数：" + fixed + "；样本：" + sql( "select rule_code, title, expected_val, actual_val from collect_validation_issue " "where task_id=%d and auto_fixed=1 limit 3;" % task_id).replace("\n", " | ")) br2 = (body_of("GET", "/api/v1/validation/issues?taskId=%d" % task_id, at) .get("data") or {}).get("byRule") or [] val06 = [x for x in br2 if x.get("ruleCode") == "VAL-06"] check("⭐ byRule 里 VAL-06 的 autoFixed 不为 0（契约 §12.2 示例里那个数）", bool(val06) and (val06[0].get("autoFixed") or 0) > 0, json.dumps(br2, ensure_ascii=False)[:300]) # ── 5. scope=ISSUES_ONLY 与 APPOINTED ── print("\n[5] 范围：ISSUES_ONLY / APPOINTED") only = body_of("POST", "/api/v1/validation/run", at, {"taskId": task_id, "scope": "ISSUES_ONLY"}) check("ISSUES_ONLY 可用", only.get("code") == 0, json.dumps(only, ensure_ascii=False)[:200]) ap_res = body_of("POST", "/api/v1/validation/run", at, {"taskId": task_id, "scope": "APPOINTED"}) check("⭐ APPOINTED 不传 submissionIds → 明确报错（不是静默跑全部）", ap_res.get("code") != 0 and "submissionIds" in (ap_res.get("msg") or ""), "code=%s msg=%s" % (ap_res.get("code"), ap_res.get("msg"))) ap2 = body_of("POST", "/api/v1/validation/run", at, {"taskId": task_id, "scope": "APPOINTED", "submissionIds": [sid]}) check("APPOINTED + 指定 id 只校验那一份", (ap2.get("data") or {}).get("checkedSubmissions") == 1, json.dumps(ap2.get("data"), ensure_ascii=False)) rules = body_of("POST", "/api/v1/validation/run", at, {"taskId": task_id, "rules": ["VAL-05"]}) check("只跑指定规则可用", rules.get("code") == 0, json.dumps(rules, ensure_ascii=False)[:200]) # ── 6. §12.3 导出 Excel ── print("\n[6] §12.3 导出异常清单 Excel") xp = os.path.join(ROOT, "scripts", "_e2e-val-issues.xlsx") code = os.system('curl -sf -o "%s" -H "Authorization: Bearer %s" ' '"%s/api/v1/validation/issues/export?taskId=%d"' % (xp, at, BASE, task_id)) size = os.path.getsize(xp) if os.path.exists(xp) else 0 check("Excel 能下载", size > 1000, "exit=%s size=%s" % (code, size)) if size > 1000: from openpyxl import load_workbook wb = load_workbook(xp) ws = wb.active header = [c.value for c in ws[1]] check("⭐ 表头与契约 §12.3 的十列一致", header == ["姓名", "班级", "材料", "规则码", "级别", "问题", "期望值", "实际值", "是否自动修复", "处理人"], str(header)) check("有数据行", ws.max_row > 1, "max_row=%s" % ws.max_row) lv = set() for r in range(2, min(ws.max_row, 30) + 1): lv.add(ws.cell(row=r, column=5).value) check("⭐ 级别写「错误/警告/提示」而不是 1/2/3（这份是给客户看的）", lv <= {"错误", "警告", "提示"}, str(lv)) af = set() for r in range(2, min(ws.max_row, 30) + 1): af.add(ws.cell(row=r, column=9).value) check("⭐ 「是否自动修复」写是/否", af <= {"是", "否"}, str(af)) os.remove(xp) # ── 7. §12.7 通知模板 ── print("\n[7] §12.7 通知模板：内置 + 覆盖 + 恢复默认") tpls = body_of("GET", "/api/v1/notify/templates", at).get("data") or [] check("模板列表可用", len(tpls) >= 7, "共 %d 个" % len(tpls)) check("⭐ 没改过的行没有 id（= 内置），界面上据此区分「默认/已自定义」", all(x.get("id") is None and x.get("customized") is False for x in tpls if not x.get("customized")), json.dumps([(x.get("templateCode"), x.get("id")) for x in tpls], ensure_ascii=False)) check("内置行带 remark 说明（界面要显示「这条是干什么的」）", all(x.get("remark") for x in tpls), json.dumps(tpls[:1], ensure_ascii=False)[:200]) check("⭐ 越界变量被标出来（unknownVars）", all(isinstance(x.get("unknownVars"), list) for x in tpls), json.dumps(tpls[:1], ensure_ascii=False)[:250]) vars_ = body_of("GET", "/api/v1/notify/templates/variables", at).get("data") or {} check("⭐ 变量白名单由服务端给（前端抄一份迟早不一致）", "姓名" in vars_ and "驳回原因" in vars_ and "下载链接" in vars_, json.dumps(list(vars_.keys()), ensure_ascii=False)) # 直接改内置行 → 必须失败（它没有 id） bad = body_of("PUT", "/api/v1/notify/templates/999999", at, {"title": "x"}) check("改不存在的模板 → 明确报错（不是 500）", bad.get("code") != 0, "code=%s msg=%s" % (bad.get("code"), bad.get("msg"))) cz = body_of("POST", "/api/v1/notify/templates/URGE/customize", at) check("⭐ 自定义内置模板可用（契约漏了这一步，见 F-52）", cz.get("code") == 0 and (cz.get("data") or {}).get("id"), json.dumps(cz, ensure_ascii=False)[:250]) tpl_id = (cz.get("data") or {}).get("id") cz2 = body_of("POST", "/api/v1/notify/templates/URGE/customize", at) check("自定义是幂等的（重复点不造出第二行）", (cz2.get("data") or {}).get("id") == tpl_id, json.dumps(cz2, ensure_ascii=False)[:200]) upd = body_of("PUT", "/api/v1/notify/templates/%d" % tpl_id, at, {"title": "自定义催报标题", "content": "{任务名}：还差 {提交进度}，{名字} 请尽快", "enabled": True}) check("改模板成功", upd.get("code") == 0, json.dumps(upd, ensure_ascii=False)[:250]) check("⭐ 越界变量在保存时就被标出来（{名字} 不在白名单）", "名字" in ((upd.get("data") or {}).get("unknownVars") or []), json.dumps(upd.get("data"), ensure_ascii=False)[:250]) check("库里记下了自定义内容", "自定义催报标题" in sql("select title from collect_notify_template where id=%d;" % tpl_id), sql("select title, content from collect_notify_template where id=%d;" % tpl_id)) lst2 = body_of("GET", "/api/v1/notify/templates", at).get("data") or [] urge = [x for x in lst2 if x.get("templateCode") == "URGE"][0] check("列表里这一行变成 customized=true", urge.get("customized") is True, json.dumps(urge, ensure_ascii=False)[:200]) rst = body_of("DELETE", "/api/v1/notify/templates/URGE", at) check("恢复默认可用", rst.get("code") == 0, json.dumps(rst, ensure_ascii=False)[:200]) # ⚠️ 必须是**物理**删除：软删除（deleted=1）时那一行还在表里， # 而唯一索引不看 deleted → "恢复默认 → 再自定义"会撞唯一约束 check("⭐ 恢复默认是物理删除覆盖行（软删会让再次自定义撞唯一索引）", sql("select count(*) from collect_notify_template where template_code='URGE';" ).strip() == "0", sql("select id, deleted from collect_notify_template where template_code='URGE';")) cz_again = body_of("POST", "/api/v1/notify/templates/URGE/customize", at) check("⭐ 恢复默认之后还能再次自定义（这是软删除会踩的那个坑）", cz_again.get("code") == 0 and (cz_again.get("data") or {}).get("id"), json.dumps(cz_again, ensure_ascii=False)[:250]) body_of("DELETE", "/api/v1/notify/templates/URGE", at) lst3 = body_of("GET", "/api/v1/notify/templates", at).get("data") or [] urge3 = [x for x in lst3 if x.get("templateCode") == "URGE"][0] check("⭐ 恢复之后文案回到内置的那一版", urge3.get("customized") is False and "提交进度" in (urge3.get("content") or ""), json.dumps(urge3, ensure_ascii=False)[:250]) # ── 8. §12.7 手工发送 ── print("\n[8] §12.7 手工发送") before = sql("select count(*) from collect_message where user_id=%d;" % STU).strip() snd = body_of("POST", "/api/v1/notify/send", at, { "bizType": 5, "templateCode": "URGE", "taskId": task_id, "receiverIds": [STU] }) check("手工发送成功", snd.get("code") == 0, json.dumps(snd, ensure_ascii=False)[:250]) sv = snd.get("data") or {} check("⭐ sent 的口径是「站内信写入成功」", sv.get("sent") == 1, json.dumps(sv, ensure_ascii=False)) check("微信推送单独报（没配/没额度时是 0，但 sent 仍是 1）", sv.get("pushSent") is not None, json.dumps(sv, ensure_ascii=False)) after = sql("select count(*) from collect_message where user_id=%d;" % STU).strip() check("⭐ 站内信真的写进去了（不能只写通知记录）", int(after) == int(before) + 1, "前 %s → 后 %s" % (before, after)) check("⭐ 消息内容里的变量被渲染了（{任务名} 不再是字面量）", "{" not in sql("select content from collect_message where user_id=%d " "order by id desc limit 1;" % STU), sql("select title, content from collect_message where user_id=%d " "order by id desc limit 1;" % STU)) check("通知记录也写了（与消息中心是两条不同的记录，契约 §13.5 的说明）", sql("select count(*) from collect_notify_record where receiver_id=%d " "and biz_type='5';" % STU).strip() != "0", sql("select channel, biz_type, status from collect_notify_record " "where receiver_id=%d order by id desc limit 2;" % STU)) # 按名册行发 + 没绑账号的人进 skipped R.sql("update collect_roster set bind_user_id=null where task_id=%d;" % task_id) snd2 = body_of("POST", "/api/v1/notify/send", at, { "bizType": 5, "title": "标题", "content": "内容", "taskId": task_id, "all": True }) check("⭐ 整任务发送：没绑账号的人进 skipped 而不是静默漏掉", (snd2.get("data") or {}).get("skipped") == 1, json.dumps(snd2, ensure_ascii=False)[:250]) snd3 = body_of("POST", "/api/v1/notify/send", at, {"bizType": 5}) check("⭐ 没有收件人 → 明确报错（不是静默发 0 条还回成功）", snd3.get("code") != 0 and "收件人" in (snd3.get("msg") or ""), "code=%s msg=%s" % (snd3.get("code"), snd3.get("msg"))) snd4 = body_of("POST", "/api/v1/notify/send", at, {"bizType": 99, "receiverIds": [STU]}) check("bizType 越界 → 明确报错", snd4.get("code") != 0, "code=%s msg=%s" % (snd4.get("code"), snd4.get("msg"))) # ── 9. 微信回调 ── print("\n[9] 微信回调（唯一的匿名**写**入口）") cb = post_raw("/api/v1/callback/wechat/msg", json.dumps( {"recordId": 1, "Status": 1}, ensure_ascii=False)) check("⭐ 未配置微信时明确拒绝（不安静地记一条来源不明的结果）", cb.get("errcode") != 0, json.dumps(cb, ensure_ascii=False)[:200]) cb2 = post_raw("/api/v1/callback/wechat/msg", "") check("空报文被拒", cb2.get("errcode") != 0, json.dumps(cb2, ensure_ascii=False)[:200]) check("⭐ 响应体是微信格式（errcode/errmsg），不是本项目的 code/msg/data", "errcode" in cb and "code" not in cb, json.dumps(cb, ensure_ascii=False)[:120]) # ── 10. 权限 ── print("\n[10] 权限") stok2 = mp(994199) b = body_of("GET", "/api/v1/validation/issues?taskId=%d" % task_id, stok2) check("学生看问题清单 → 403", b.get("code") in (40101, 40301, 403), "code=%s" % b.get("code")) b = body_of("GET", "/api/v1/notify/templates", stok2) check("学生看通知模板 → 403", b.get("code") in (40101, 40301, 403), "code=%s" % b.get("code")) b = body_of("POST", "/api/v1/validation/run", None, {"taskId": task_id}) check("未登录跑校验 → 40101", b.get("code") == 40101, "code=%s" % b.get("code")) cleanup() print("\n" + "=" * 74) print("通过 %d　失败 %d　共 %d" % (len(PASSED), len(FAILED), len(PASSED) + len(FAILED))) print("=" * 74) for n, det in FAILED: print(" x " + n + "\n " + det) return 1 if FAILED else 0 def post_raw(path, body): """发一个**不带登录态**的原始请求（微信回调是匿名入口）。""" import urllib.request req = urllib.request.Request(BASE + path, data=(body or "").encode("utf-8"), method="POST") req.add_header("Content-Type", "application/json") try: with urllib.request.urlopen(req, timeout=30) as resp: return json.loads(resp.read().decode("utf-8")) except Exception as e: return {"errcode": -1, "errmsg": str(e)[:200]} def _build_roster(rows): from openpyxl import Workbook wb = Workbook() ws = wb.active ws.title = "校验班" header = ["序号", "姓名", "身份证号", "学历", "联系电话", "工作单位", "报考职业", "报考工种", "报考等级", "是否补考", "补考理论", "补考实操", "补考综合", "是否在校生", "在读学校", "所学专业", "入学时间"] for c, t in enumerate(header, start=1): ws.cell(row=1, column=c, value=t) for i, (name, idc) in enumerate(rows): vals = [i + 1, name, idc, "中专", "1380013550%d" % i, "无", "车工", "数控车工", "四级", "否", "", "", "", "是", "示范职业学校", "数控加工", "2024-09-01"] for c, v in enumerate(vals, start=1): if v != "": ws.cell(row=i + 2, column=c, value=v) wb.save(TMP_ROSTER) def cleanup(): for tid in CREATED_TASK_IDS: R.sql("delete from collect_validation_issue where task_id=%d;" % tid) R.sql("delete from collect_notify_record where task_id=%d;" % tid) R.sql("delete from collect_file where task_id=%d;" % tid) R.sql("delete from collect_submission where task_id=%d;" % tid) R.sql("delete from collect_roster where task_id=%d;" % tid) R.sql("delete from collect_task_org where task_id=%d;" % tid) R.sql("delete from collect_material_def where task_id=%d;" % tid) R.sql("delete from collect_task where id=%d;" % tid) R.sql("delete from collect_notify_template where template_code='URGE';") R.sql("delete from collect_message where user_id in (994101, 994199);") R.sql("delete from collect_notify_record where receiver_id in (994101, 994199);") if os.path.exists(TMP_ROSTER): os.remove(TMP_ROSTER) print(" 已清理：任务/问题清单/消息/通知记录/模板覆盖/临时 Excel") if __name__ == "__main__": raise SystemExit(main())