# -*- coding: utf-8 -*- """ V36 四维能力测评 题库权威校验脚本 两个目标： 1) 独立校验权威题库 ability-question-bank-v0.1.json 满足全部 14 项硬性约束。 2) 将 V35__ability_assessment.sql 中写入的 48 道题与权威 JSON 逐题对账， 确保题干/选项/答案键/分值/解析/难度/版本/position 语义完全一致。 任一项校验失败 => 打印 FAIL 并退出码 1（不自动修复题库内容）。 全部通过 => 打印 PASS（针对每道题）并退出码 0。 """ import json import re import sys from pathlib import Path ROOT = Path(__file__).resolve().parents[1] BANK_JSON = ROOT / "docs" / "ability-question-bank-v0.1.json" V35_SQL = ROOT / "mydreamjob-server" / "src" / "main" / "resources" / "db" / "migration" / "V35__ability_assessment.sql" DIMENSIONS = ["LOGIC", "COMMUNICATION", "EXECUTION", "LEARNING"] COMPETENCY_TAGS = { "LOGIC": ["INFORMATION_ANALYSIS", "CAUSAL_REASONING", "PROBLEM_SOLVING"], "COMMUNICATION": ["LISTENING_UNDERSTANDING", "STRUCTURED_EXPRESSION", "COLLABORATION_CONFLICT"], "EXECUTION": ["PLANNING_PRIORITIZATION", "RISK_FOLLOWUP", "DELIVERY_REVIEW"], "LEARNING": ["INFORMATION_ACQUISITION", "FEEDBACK_REFLECTION", "TRANSFER_ADAPTATION"], } SCORES = {0, 1, 2, 4} OPTION_IDS = ["A", "B", "C", "D"] def fail(msg): print(f"FAIL: {msg}") sys.exit(1) def check(cond, msg): if not cond: fail(msg) print(f" PASS: {msg}") def main(): print("=== 第 1 步：校验权威 JSON 文件 ===") bank = json.loads(BANK_JSON.read_text(encoding='utf-8')) if bank.get("bankCode") != "WORKPLACE_ABILITY": fail("bankCode != WORKPLACE_ABILITY") if bank.get("bankVersion") != "ability-v0.1": fail("bankVersion != ability-v0.1") qs = bank.get("questions", []) check(len(qs) == 48, f"总数 = 48（实际 {len(qs)}）") # 1) 四维各 12 dim_count = {d: 0 for d in DIMENSIONS} pos_set = set() tag_count = {} key_count = {} dif_count = {} for q in qs: dim_count[q["dimension"]] = dim_count.get(q["dimension"], 0) + 1 tag_count[q["competencyTag"]] = tag_count.get(q["competencyTag"], 0) + 1 key_count[q["answerKey"]] = key_count.get(q["answerKey"], 0) + 1 dif_count[q["difficulty"]] = dif_count.get(q["difficulty"], 0) + 1 pos_set.add(q["position"]) for d in DIMENSIONS: check(dim_count.get(d) == 12, f"维度 {d} = 12（实际 {dim_count.get(d)}）") for tag in sum(COMPETENCY_TAGS.values(), []): check(tag_count.get(tag) == 4, f"competencyTag {tag} = 4（实际 {tag_count.get(tag)}）") for k in OPTION_IDS: check(key_count.get(k) == 12, f"answerKey {k} = 12（实际 {key_count.get(k)}）") for d in ["EASY", "MEDIUM", "HARD"]: check(dif_count.get(d) == 16, f"difficulty {d} = 16（实际 {dif_count.get(d)}）") check(len(pos_set) == 48 and pos_set == set(range(1, 49)), "position 1-48 唯一且连续") # 2) 逐题约束 for q in qs: code = q["questionCode"] opts = q["options"] # 每题四选一 check(len(opts) == 4, f"{code} 含 4 个选项") oids = [o["optionId"] for o in opts] check(oids == OPTION_IDS, f"{code} optionId 集合 = A/B/C/D") for o in opts: if not o.get("text") or not str(o["text"]).strip(): fail(f"{code} 存在空选项文本") # 分值集合严格等于 {0,1,2,4} rule = q["scoringRule"] check(sorted(rule.keys()) == OPTION_IDS, f"{code} scoringRule 键 = A/B/C/D") check(set(rule.values()) == SCORES, f"{code} 分值集合 = {{0,1,2,4}}（实际 {sorted(set(rule.values()))}）") check(len(set(rule.values())) == 4, f"{code} 分值互不相同（每题各档唯一）") # answerKey 对应 4 分 check(rule.get(q["answerKey"]) == 4, f"{code} answerKey={q['answerKey']} 对应 4 分") # explanation 非空 check(bool(q.get("explanation")) and str(q["explanation"]).strip(), f"{code} explanation 非空") # 版本/来源/启用 check(q["questionVersion"] == "ability-v0.1", f"{code} questionVersion=ability-v0.1") check(q["source"] == "SELF_AUTHORED", f"{code} source=SELF_AUTHORED") check(q["active"] is True, f"{code} active=true") check(q["dimension"] in DIMENSIONS, f"{code} dimension 合法") check(q["competencyTag"] in COMPETENCY_TAGS.get(q["dimension"], []), f"{code} competencyTag 归属该维度") check(q["itemType"] in ("SITUATIONAL_JUDGMENT", "SEQUENCE_JUDGMENT"), f"{code} itemType 合法") print("=== 第 2 步：V35 SQL 与权威 JSON 逐题对账 ===") sql = V35_SQL.read_text(encoding='utf-8') # 抓取 definition INSERT 中 WORKPLACE_ABILITY 的系数行 defn_pat = ("'WORKPLACE_ABILITY'[\\s\\S]*?'ACTIVE',[\\s\\S]*?" "48,\\s*24,\\s*6,\\s*30,\\s*7,\\s*90,\\s*3," "[\\s\\S]*?'ability-v0.1'[\\s\\S]*?'ability-scoring-v0.1'\\s*,?\\s*\\)") defn_m = re.search(defn_pat, sql) if not defn_m: fail("V35 中缺少 WORKPLACE_ABILITY 定义种子（48/24/6/30/7/90/3）") print(" PASS: WORKPLACE_ABILITY 定义种子存在且参数一致") # 抓取 ability_questions 的整段 VALUES（一直抓到文件结尾） values_m = re.search(r"INSERT INTO ability_questions[\s\S]*?VALUES\s*(.*)$", sql, re.S) if not values_m: fail("V35 中缺少 ability_questions INSERT") body = values_m.group(1) # 每个 VALUES 行以 "('XXNN'" 起始，且独占一行；逐行拆分 rows = [] for line in body.splitlines(): m = re.match(r"\s*\(\s*'([A-Z]{2}\d{2})'(.+)\)\s*,?;?\s*$", line) if m: rows.append((m.group(1), m.group(2))) def clean(v): v = v.strip() if v.startswith("'") and v.endswith("'"): v = v[1:-1] return v.replace("''", "'") def parse_json(v): return json.loads(clean(v)) if len(rows) != 48: fail(f"V35 中题目行数 = {len(rows)}（应为 48）") # 拆分字段：code 单独；其余依次为 # def_subquery, dimension, competency_tag, item_type, stem, # options_json, answer_key_json, scoring_rule_json, explanation, # difficulty, discrimination, question_version, source, active, position by_pos = {} for code, rest in rows: parts = split_row(rest) # rest 以 "(SELECT..." 起始，首字段为 definition_id 子查询，且行首逗号会产生一个空 token； # 去除空 token 后即为 [def_subquery, dimension, ..., position] if parts and parts[0] == "": parts = parts[1:] if len(parts) != 15: fail(f"{code} 解析出 {len(parts)} 个字段（应为 15）") by_pos[int(parts[14])] = (code, parts) if set(by_pos.keys()) != set(range(1, 49)): fail("V35 中 position 不覆盖 1-48") for q in qs: code = q["questionCode"] pos = q["position"] row = by_pos.get(pos) if row is None: fail(f"position {pos}（{code}）在 V35 中缺失") sql_code, parts = row if sql_code != code: fail(f"position {pos}：V35 questionCode={sql_code}，权威 JSON={code}") # definition_id（parts[0]）为子查询，跳过比较 sql_dim, sql_tag, sql_item, sql_stem = parts[1], parts[2], parts[3], parts[4] sql_opts, sql_key, sql_rule, sql_expl = parts[5], parts[6], parts[7], parts[8] sql_dif = parts[9] sql_ver, sql_src, sql_active, sql_pos = parts[11], parts[12], parts[13], parts[14] def eq_a(actual, expect, label): if actual != expect: fail(f"{code} {label} 不一致") eq_a(clean(sql_dim), q["dimension"], "dimension") eq_a(clean(sql_tag), q["competencyTag"], "competencyTag") eq_a(clean(sql_item), q["itemType"], "itemType") eq_a(clean(sql_stem), q["stem"], "stem") eq_a(parse_json(sql_opts), q["options"], "options_json") eq_a(parse_json(sql_key), dict(answerKey=q["answerKey"]), "answer_key_json") eq_a(parse_json(sql_rule), q["scoringRule"], "scoring_rule_json") eq_a(clean(sql_expl), q["explanation"], "explanation") eq_a(clean(sql_dif), q["difficulty"], "difficulty") eq_a(clean(sql_ver), q["questionVersion"], "question_version") eq_a(clean(sql_src), q["source"], "source") eq_a(clean(sql_active), "1", "active") eq_a(clean(sql_pos), str(q["position"]), "position") print(f" PASS: {code} 与权威 JSON 完全一致") print("ALL PASS: 题库 14 项硬性约束 + V35 SQL 逐题对账全部通过") sys.exit(0) def split_row(rest): """把一行剩余部分按顶层逗号切分（忽略引号内的逗号）。""" out = [] buf = [] in_sq = False i = 0 body = rest while i < len(body): c = body[i] if c == "'": if in_sq and i + 1 < len(body) and body[i + 1] == "'": buf.append("''") i += 2 continue in_sq = not in_sq buf.append(c) elif c == "," and not in_sq: out.append("".join(buf).strip()) buf = [] else: buf.append(c) i += 1 if buf: out.append("".join(buf).strip()) return out if __name__ == "__main__": main()