forked from baskbuddy/zhangshun
You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
224 lines
9.2 KiB
224 lines
9.2 KiB
# -*- 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()
|