feat: IMAP mail sync with thread-to-contact matching (PRD §6.1)
Real IMAP connection via imapflow library: - Connects to IMAP server (IMAPFlow, TLS on port 993) - Fetches recent messages (configurable limit, default 20) - Extracts: from, fromEmail, subject, date, snippet (200 chars) - Thread-to-contact matching (PRD §6.1): 1. Match sender email against persons table 2. If no person match, extract domain and match against company.website 3. Returns matchedCompany + matchedPerson for each email - Audit-logged (action: crm.mailsync:imap) - Zod validated (host, port, user, password, folder, limit) - 502 on IMAP connection failure with detail - 400 on validation failure Tested: invalid creds → 502 'IMAP connection failed', missing host → 400
This commit is contained in:
Z User committed
1 parent
7acc0c5451
commit
e45b68bb8b
1089 files changed
+260614
No files matched your search
BIN
Binary file not shown.
BIN
Binary file not shown.
BIN
Binary file not shown.
+182
@@ -0,0 +1,182 @@
|
||||
#!/usr/bin/env python3
|
||||
"""从 student.json 构建 API 请求体，将考生倾向映射到选填参数。"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import json
|
||||
import sys
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
# 意向城市 → API provinces（省份意向）
|
||||
CITY_TO_PROVINCE: dict[str, str] = {
|
||||
"北京": "北京",
|
||||
"上海": "上海",
|
||||
"天津": "天津",
|
||||
"重庆": "重庆",
|
||||
"深圳": "广东",
|
||||
"广州": "广东",
|
||||
"东莞": "广东",
|
||||
"佛山": "广东",
|
||||
"武汉": "湖北",
|
||||
"成都": "四川",
|
||||
"杭州": "浙江",
|
||||
"宁波": "浙江",
|
||||
"南京": "江苏",
|
||||
"苏州": "江苏",
|
||||
"无锡": "江苏",
|
||||
"西安": "陕西",
|
||||
"长沙": "湖南",
|
||||
"郑州": "河南",
|
||||
"济南": "山东",
|
||||
"青岛": "山东",
|
||||
"合肥": "安徽",
|
||||
"厦门": "福建",
|
||||
"福州": "福建",
|
||||
"大连": "辽宁",
|
||||
"沈阳": "辽宁",
|
||||
"哈尔滨": "黑龙江",
|
||||
"长春": "吉林",
|
||||
"昆明": "云南",
|
||||
"南宁": "广西",
|
||||
"海口": "海南",
|
||||
"石家庄": "河北",
|
||||
"太原": "山西",
|
||||
"南昌": "江西",
|
||||
"兰州": "甘肃",
|
||||
"贵阳": "贵州",
|
||||
"呼和浩特": "内蒙古",
|
||||
"乌鲁木齐": "新疆",
|
||||
"银川": "宁夏",
|
||||
"西宁": "青海",
|
||||
"拉萨": "西藏",
|
||||
}
|
||||
|
||||
DEFAULT_API_OPTIONS: dict[str, Any] = {
|
||||
"universityNum": 30,
|
||||
"majorNum": 6,
|
||||
"isAdjust": True,
|
||||
"volunteerType": "ACADEMY_GROUP",
|
||||
"returnUniversityNum": 30,
|
||||
"intentionNum": 0,
|
||||
}
|
||||
|
||||
|
||||
def _join_csv(values: list[str] | None) -> str | None:
|
||||
if not values:
|
||||
return None
|
||||
cleaned = [str(v).strip() for v in values if str(v).strip()]
|
||||
return ",".join(cleaned) if cleaned else None
|
||||
|
||||
|
||||
def cities_to_provinces(cities: list[str] | None) -> list[str]:
|
||||
"""意向城市去重映射为省份列表。"""
|
||||
if not cities:
|
||||
return []
|
||||
provinces: list[str] = []
|
||||
seen: set[str] = set()
|
||||
for city in cities:
|
||||
key = str(city).strip().replace("市", "").replace("省", "")
|
||||
if not key:
|
||||
continue
|
||||
province = CITY_TO_PROVINCE.get(key, key)
|
||||
if province not in seen:
|
||||
seen.add(province)
|
||||
provinces.append(province)
|
||||
return provinces
|
||||
|
||||
|
||||
def student_to_api_payload(student: dict[str, Any]) -> dict[str, Any]:
|
||||
"""将 student.json 转为 API 请求体。"""
|
||||
preferred_provinces = student.get("preferred_provinces") or []
|
||||
if not preferred_provinces:
|
||||
preferred_provinces = cities_to_provinces(student.get("preferred_cities"))
|
||||
|
||||
payload: dict[str, Any] = {
|
||||
"province": student.get("province"),
|
||||
"classify": student.get("classify"),
|
||||
"subjects": student.get("subjects"),
|
||||
"gradeType": student.get("gradeType"),
|
||||
"score": student.get("score"),
|
||||
"rank": student.get("rank"),
|
||||
"batch": student.get("batch"),
|
||||
"universitys": _join_csv(student.get("preferred_universities")),
|
||||
"provinces": _join_csv(preferred_provinces),
|
||||
"tags": _join_csv(student.get("preferred_tags")),
|
||||
"majorClass": _join_csv(student.get("preferred_major_classes")),
|
||||
**DEFAULT_API_OPTIONS,
|
||||
}
|
||||
|
||||
overrides = student.get("api_overrides") or {}
|
||||
for key, value in overrides.items():
|
||||
if value is not None:
|
||||
payload[key] = value
|
||||
|
||||
return payload
|
||||
|
||||
|
||||
def summarize_preferences(student: dict[str, Any], payload: dict[str, Any]) -> dict[str, Any]:
|
||||
"""返回倾向映射摘要，便于 Agent 向用户说明。"""
|
||||
return {
|
||||
"universitys": payload.get("universitys"),
|
||||
"provinces": payload.get("provinces"),
|
||||
"tags": payload.get("tags"),
|
||||
"majorClass": payload.get("majorClass"),
|
||||
"sources": {
|
||||
"preferred_universities": student.get("preferred_universities"),
|
||||
"preferred_cities": student.get("preferred_cities"),
|
||||
"preferred_provinces": student.get("preferred_provinces"),
|
||||
"preferred_tags": student.get("preferred_tags"),
|
||||
"preferred_major_classes": student.get("preferred_major_classes"),
|
||||
},
|
||||
}
|
||||
|
||||
|
||||
def main() -> int:
|
||||
parser = argparse.ArgumentParser(description="从 student.json 构建 API 请求 JSON")
|
||||
parser.add_argument("-i", "--input", type=Path, required=True, help="student.json 路径")
|
||||
parser.add_argument("-o", "--output", type=Path, required=True, help="输出 api_request.json")
|
||||
parser.add_argument(
|
||||
"--summary",
|
||||
type=Path,
|
||||
help="可选：输出倾向映射摘要 JSON（供 Agent 核对）",
|
||||
)
|
||||
args = parser.parse_args()
|
||||
|
||||
try:
|
||||
student = json.loads(args.input.read_text(encoding="utf-8"))
|
||||
except (OSError, json.JSONDecodeError) as e:
|
||||
print(f"错误: 无法读取 student.json — {e}", file=sys.stderr)
|
||||
return 1
|
||||
|
||||
payload = student_to_api_payload(student)
|
||||
try:
|
||||
from province_config import validate_classify, validate_subjects_for_volunteer_api
|
||||
|
||||
validate_classify(str(payload.get("province", "")), payload.get("classify"))
|
||||
validate_subjects_for_volunteer_api(payload)
|
||||
except ValueError as e:
|
||||
print(f"错误: {e}", file=sys.stderr)
|
||||
return 1
|
||||
|
||||
required = ["province", "classify", "score"]
|
||||
missing = [k for k in required if not payload.get(k)]
|
||||
if missing:
|
||||
print(f"错误: student.json 缺少 API 必填字段 {missing}", file=sys.stderr)
|
||||
return 1
|
||||
|
||||
args.output.parent.mkdir(parents=True, exist_ok=True)
|
||||
args.output.write_text(json.dumps(payload, ensure_ascii=False, indent=2), encoding="utf-8")
|
||||
print(f"已生成 API 请求: {args.output}")
|
||||
|
||||
if args.summary:
|
||||
summary = summarize_preferences(student, payload)
|
||||
args.summary.write_text(json.dumps(summary, ensure_ascii=False, indent=2), encoding="utf-8")
|
||||
print(f"已生成倾向映射摘要: {args.summary}")
|
||||
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
raise SystemExit(main())
|
||||
+569
@@ -0,0 +1,569 @@
|
||||
#!/usr/bin/env python3
|
||||
"""调用智能推荐志愿表 API，解析响应并输出结构化 JSON。"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import json
|
||||
import os
|
||||
import sys
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
from urllib.error import HTTPError, URLError
|
||||
from urllib.parse import urlencode
|
||||
from urllib.request import Request, urlopen
|
||||
|
||||
from province_config import (
|
||||
PROVINCES_REQUIRING_GRADE_TYPE,
|
||||
batch_list_query_params,
|
||||
sanitize_payload_for_api,
|
||||
validate_classify,
|
||||
validate_subjects_for_volunteer_api,
|
||||
)
|
||||
|
||||
# ========== 接口配置（修改此处即可切换环境）==========
|
||||
BASE_URL = "https://publicapi.chatglm.cn/chatglm_public/skill"
|
||||
API_PATH = "/zp/volunteer/intelligenceVolunteer"
|
||||
BATCH_LIST_PATH = "/zp/volunteer/batch/list"
|
||||
# =====================================================
|
||||
|
||||
TYPE_LABELS = {
|
||||
"CHONG": "冲",
|
||||
"WEN": "稳",
|
||||
"BAO": "保",
|
||||
"NAN": "难",
|
||||
"YI": "易",
|
||||
}
|
||||
|
||||
PROVINCE_BATCH_FALLBACK: dict[str, str] = {
|
||||
"天津": "本科批A段",
|
||||
"浙江": "普通类一段",
|
||||
"山东": "普通类一段",
|
||||
"四川": "本科批B段",
|
||||
"云南": "本科批B段",
|
||||
"甘肃": "本科批",
|
||||
"宁夏": "本科批B段",
|
||||
"新疆": "本科一批",
|
||||
}
|
||||
|
||||
|
||||
def resolve_batch_fallback(province: str | None, batch: str | None) -> str | None:
|
||||
"""批次 API 失败时的省份静态兜底。"""
|
||||
if not province:
|
||||
return batch
|
||||
mapped = PROVINCE_BATCH_FALLBACK.get(province)
|
||||
if not mapped:
|
||||
return batch
|
||||
if batch is None:
|
||||
return mapped
|
||||
if str(batch).strip() == "本科批":
|
||||
return mapped
|
||||
return batch
|
||||
|
||||
|
||||
def batch_list_url(
|
||||
params: dict[str, Any],
|
||||
base_url: str | None = None,
|
||||
) -> str:
|
||||
root = (base_url or BASE_URL).rstrip("/")
|
||||
path = BATCH_LIST_PATH if BATCH_LIST_PATH.startswith("/") else f"/{BATCH_LIST_PATH}"
|
||||
query = urlencode(params)
|
||||
return f"{root}{path}?{query}"
|
||||
|
||||
|
||||
def call_batch_list_api(
|
||||
params: dict[str, Any],
|
||||
base_url: str | None = None,
|
||||
timeout: int = 60,
|
||||
) -> list[dict[str, Any]]:
|
||||
"""GET 批次列表，返回 result 数组。"""
|
||||
url = batch_list_url(params, base_url)
|
||||
req = Request(url, method="GET", headers={"Accept": "*/*"})
|
||||
try:
|
||||
with urlopen(req, timeout=timeout) as resp:
|
||||
charset = resp.headers.get_content_charset() or "utf-8"
|
||||
text = resp.read().decode(charset)
|
||||
except HTTPError as e:
|
||||
detail = e.read().decode("utf-8", errors="replace")
|
||||
raise RuntimeError(f"批次接口 HTTP {e.code}: {detail}") from e
|
||||
except URLError as e:
|
||||
raise RuntimeError(f"批次接口网络失败: {e.reason}") from e
|
||||
|
||||
raw = json.loads(text)
|
||||
if raw.get("status") != 0:
|
||||
raise ValueError(f"批次接口异常: status={raw.get('status')}, message={raw.get('message')}")
|
||||
result = raw.get("result")
|
||||
if not isinstance(result, list):
|
||||
raise ValueError("批次接口返回格式异常：result 非数组")
|
||||
if not result:
|
||||
province = params.get("province")
|
||||
classify = params.get("classify")
|
||||
grade_type = params.get("gradeType")
|
||||
hint = f"classify={classify}"
|
||||
if grade_type:
|
||||
hint += f", gradeType={grade_type}"
|
||||
raise ValueError(
|
||||
f"批次列表为空（{province}，{hint}）。"
|
||||
f"请确认 classify 与省份选科模式一致（见 reference.md）"
|
||||
)
|
||||
return result
|
||||
|
||||
|
||||
def select_batch_from_list(
|
||||
items: list[dict[str, Any]],
|
||||
user_score: int,
|
||||
preferred_batch: str | None = None,
|
||||
grade_type: str | None = None,
|
||||
) -> dict[str, Any]:
|
||||
"""从批次列表中选取与分数、层次最匹配的一项。"""
|
||||
if not items:
|
||||
raise ValueError("没有可填报的批次")
|
||||
|
||||
candidates = list(items)
|
||||
|
||||
# 1. 精确批次名（非泛称「本科批」）
|
||||
if preferred_batch and preferred_batch.strip() not in ("本科批", "专科批"):
|
||||
exact = [x for x in candidates if x.get("batch") == preferred_batch]
|
||||
if exact:
|
||||
return exact[0]
|
||||
|
||||
# 2. 按 gradeType / 泛称本科批·专科批 过滤
|
||||
if grade_type:
|
||||
filtered = [x for x in candidates if x.get("gradeType") == grade_type]
|
||||
if filtered:
|
||||
candidates = filtered
|
||||
elif preferred_batch == "本科批":
|
||||
filtered = [x for x in candidates if x.get("gradeType") == "本科"]
|
||||
if filtered:
|
||||
candidates = filtered
|
||||
elif preferred_batch == "专科批":
|
||||
filtered = [x for x in candidates if x.get("gradeType") == "专科"]
|
||||
if filtered:
|
||||
candidates = filtered
|
||||
|
||||
# 3. 分数达线：取控制线下考生分数仍达标的批次中，控制线最高者（偏本科一段）
|
||||
qualified = [x for x in candidates if user_score >= int(x.get("score") or 0)]
|
||||
pool = qualified if qualified else candidates
|
||||
return max(pool, key=lambda x: int(x.get("score") or 0))
|
||||
|
||||
|
||||
def apply_batch_item_to_payload(payload: dict[str, Any], batch_item: dict[str, Any]) -> None:
|
||||
"""将选中批次写入志愿请求体。"""
|
||||
payload["batch"] = batch_item.get("batch")
|
||||
if batch_item.get("type"):
|
||||
payload["volunteerType"] = batch_item["type"]
|
||||
province = payload.get("province")
|
||||
if (
|
||||
province in PROVINCES_REQUIRING_GRADE_TYPE
|
||||
and batch_item.get("gradeType")
|
||||
and not payload.get("gradeType")
|
||||
):
|
||||
payload["gradeType"] = batch_item["gradeType"]
|
||||
|
||||
|
||||
def resolve_batch_via_api(
|
||||
payload: dict[str, Any],
|
||||
base_url: str | None = None,
|
||||
) -> tuple[str, dict[str, Any], list[dict[str, Any]]]:
|
||||
"""调用批次接口并选定 batch，返回 (batch名, 选中项, 全部项)。"""
|
||||
params = batch_list_query_params(payload)
|
||||
items = call_batch_list_api(params, base_url=base_url)
|
||||
selected = select_batch_from_list(
|
||||
items,
|
||||
user_score=int(payload["score"]),
|
||||
preferred_batch=payload.get("batch"),
|
||||
grade_type=payload.get("gradeType"),
|
||||
)
|
||||
apply_batch_item_to_payload(payload, selected)
|
||||
return selected.get("batch") or "", selected, items
|
||||
|
||||
|
||||
def _parse_json_field(value: Any) -> Any:
|
||||
if value is None or value == "":
|
||||
return None
|
||||
if isinstance(value, (dict, list)):
|
||||
return value
|
||||
if isinstance(value, str):
|
||||
text = value.strip()
|
||||
if not text:
|
||||
return None
|
||||
try:
|
||||
return json.loads(text)
|
||||
except json.JSONDecodeError:
|
||||
return value
|
||||
return value
|
||||
|
||||
|
||||
def parse_score_history(score_str: Any) -> list[dict[str, str]]:
|
||||
"""将 score / parityScore 字段解析为 [{year, score, rank, plan_num}, ...]。"""
|
||||
parsed = _parse_json_field(score_str)
|
||||
if not isinstance(parsed, list):
|
||||
return []
|
||||
|
||||
rows: list[dict[str, str]] = []
|
||||
for item in parsed:
|
||||
if not isinstance(item, dict):
|
||||
continue
|
||||
for year, triple in item.items():
|
||||
parts = str(triple).split(",")
|
||||
rows.append(
|
||||
{
|
||||
"year": str(year),
|
||||
"score": parts[0] if len(parts) > 0 else "-",
|
||||
"rank": parts[1] if len(parts) > 1 else "-",
|
||||
"plan_num": parts[2] if len(parts) > 2 else "-",
|
||||
}
|
||||
)
|
||||
return rows
|
||||
|
||||
|
||||
def parse_parity_scores(parity_str: Any) -> list[dict[str, str]]:
|
||||
parsed = _parse_json_field(parity_str)
|
||||
if not isinstance(parsed, list):
|
||||
return []
|
||||
rows: list[dict[str, str]] = []
|
||||
for item in parsed:
|
||||
if isinstance(item, dict):
|
||||
for year, score in item.items():
|
||||
rows.append({"year": str(year), "score": str(score)})
|
||||
return rows
|
||||
|
||||
|
||||
def normalize_major(major: dict[str, Any]) -> dict[str, Any]:
|
||||
type_code = major.get("type") or ""
|
||||
return {
|
||||
"major_name": major.get("majorName") or "",
|
||||
"major_code": major.get("majorCode") or "",
|
||||
"major_num": major.get("majorNum"),
|
||||
"claim": major.get("claim") or "",
|
||||
"study_year": major.get("studyYear") or "",
|
||||
"study_cost": major.get("studyCost") or "",
|
||||
"major_class": major.get("majorClass") or "",
|
||||
"major_remarks": major.get("majorRemarks") or "",
|
||||
"plan_num": major.get("planNum"),
|
||||
"enroll_probability": major.get("enrollProbability"),
|
||||
"type": type_code,
|
||||
"type_label": TYPE_LABELS.get(type_code, type_code or "—"),
|
||||
"score_history": parse_score_history(major.get("score")),
|
||||
"parity_scores": parse_parity_scores(major.get("parityScore")),
|
||||
}
|
||||
|
||||
|
||||
def normalize_school(school: dict[str, Any], index: int) -> dict[str, Any]:
|
||||
type_code = school.get("type") or ""
|
||||
majors = school.get("majorList") or []
|
||||
return {
|
||||
"index": index,
|
||||
"university_name": school.get("universityName") or "",
|
||||
"university_code": school.get("universityCode") or "",
|
||||
"major_group": school.get("universityMajorGroup") or "",
|
||||
"logo": school.get("logo") or "",
|
||||
"category_name": school.get("categoryName") or "",
|
||||
"property_name": school.get("propertyName") or "",
|
||||
"tags": school.get("tags") or "",
|
||||
"city_name": school.get("cityName") or "",
|
||||
"province_name": school.get("provinceName") or "",
|
||||
"enroll_probability": school.get("enrollProbability"),
|
||||
"lowest_score": school.get("lowestScore"),
|
||||
"type": type_code,
|
||||
"type_label": TYPE_LABELS.get(type_code, type_code or "—"),
|
||||
"plan_num": school.get("planNum"),
|
||||
"level": school.get("level") or "",
|
||||
"is_adjust": school.get("isAdjust"),
|
||||
"score_history": parse_score_history(school.get("score")),
|
||||
"parity_scores": parse_parity_scores(school.get("parityScore")),
|
||||
"majors": [normalize_major(m) for m in majors],
|
||||
}
|
||||
|
||||
|
||||
def group_schools_by_type(schools: list[dict[str, Any]]) -> dict[str, list[dict[str, Any]]]:
|
||||
order = ["CHONG", "WEN", "BAO", "YI", "NAN"]
|
||||
grouped: dict[str, list[dict[str, Any]]] = {k: [] for k in order}
|
||||
other: list[dict[str, Any]] = []
|
||||
for school in schools:
|
||||
t = school.get("type") or ""
|
||||
if t in grouped:
|
||||
grouped[t].append(school)
|
||||
else:
|
||||
other.append(school)
|
||||
if other:
|
||||
grouped["_OTHER"] = other
|
||||
return grouped
|
||||
|
||||
|
||||
def normalize_api_response(raw: dict[str, Any]) -> dict[str, Any]:
|
||||
"""兼容新旧两种响应信封：{code,msg,body} 与 {status,message,result}。"""
|
||||
if "result" in raw and "status" in raw:
|
||||
status = raw.get("status")
|
||||
msg = raw.get("message") or ""
|
||||
if status != 0:
|
||||
raise ValueError(f"API 返回异常: status={status}, message={msg}")
|
||||
return {"code": 200, "msg": msg or "OK", "body": raw.get("result") or {}}
|
||||
return raw
|
||||
|
||||
|
||||
def _empty_to_none(value: Any) -> Any:
|
||||
if value == "" or value is None:
|
||||
return None
|
||||
return value
|
||||
|
||||
|
||||
def parse_api_response(raw: dict[str, Any]) -> dict[str, Any]:
|
||||
raw = normalize_api_response(raw)
|
||||
code = raw.get("code")
|
||||
msg = raw.get("msg") or ""
|
||||
if code != 200:
|
||||
raise ValueError(f"API 返回异常: code={code}, msg={msg}")
|
||||
|
||||
body = raw.get("body") or {}
|
||||
schools_raw = body.get("schoolList") or []
|
||||
schools = [normalize_school(s, i + 1) for i, s in enumerate(schools_raw)]
|
||||
grouped = group_schools_by_type(schools)
|
||||
|
||||
stats = {
|
||||
"total": len(schools),
|
||||
"chong": len(grouped.get("CHONG", [])),
|
||||
"wen": len(grouped.get("WEN", [])),
|
||||
"bao": len(grouped.get("BAO", [])),
|
||||
}
|
||||
|
||||
return {
|
||||
"meta": {
|
||||
"generated_from": "intelligenceVolunteer",
|
||||
"api_code": code,
|
||||
"api_msg": msg,
|
||||
},
|
||||
"profile": {
|
||||
"province": body.get("province"),
|
||||
"classify": body.get("classify"),
|
||||
"subjects": _empty_to_none(body.get("subjects")),
|
||||
"grade_type": _empty_to_none(body.get("gradeType")),
|
||||
"score": body.get("score"),
|
||||
"rank": body.get("rank"),
|
||||
"batch": body.get("batch"),
|
||||
"volunteer_type": body.get("volunteerType"),
|
||||
"is_adjust": body.get("isAdjust"),
|
||||
"university_num": body.get("universityNum"),
|
||||
"major_num": body.get("majorNum"),
|
||||
"universitys": body.get("universitys"),
|
||||
"provinces": body.get("provinces"),
|
||||
"tags": body.get("tags"),
|
||||
"major_class": body.get("majorClass"),
|
||||
},
|
||||
"stats": stats,
|
||||
"schools": schools,
|
||||
"schools_by_type": {
|
||||
TYPE_LABELS.get(k, k): grouped[k]
|
||||
for k in grouped
|
||||
if grouped[k] and not k.startswith("_")
|
||||
},
|
||||
"raw_body": body,
|
||||
}
|
||||
|
||||
|
||||
def load_payload_from_student(student_path: Path) -> dict[str, Any]:
|
||||
from build_api_request import student_to_api_payload
|
||||
|
||||
student = json.loads(student_path.read_text(encoding="utf-8"))
|
||||
return student_to_api_payload(student)
|
||||
|
||||
|
||||
def build_request_payload(args: argparse.Namespace) -> dict[str, Any]:
|
||||
if args.student:
|
||||
return load_payload_from_student(args.student)
|
||||
if args.config:
|
||||
with open(args.config, encoding="utf-8") as f:
|
||||
return json.load(f)
|
||||
|
||||
payload: dict[str, Any] = {
|
||||
"province": args.province,
|
||||
"classify": args.classify,
|
||||
"subjects": args.subjects,
|
||||
"gradeType": args.grade_type,
|
||||
"score": args.score,
|
||||
"rank": args.rank,
|
||||
"batch": args.batch,
|
||||
"universitys": args.universitys,
|
||||
"provinces": args.provinces,
|
||||
"tags": args.tags,
|
||||
"majorClass": args.major_class,
|
||||
"universityNum": args.university_num,
|
||||
"majorNum": args.major_num,
|
||||
"isAdjust": args.is_adjust,
|
||||
"volunteerType": args.volunteer_type,
|
||||
"returnUniversityNum": args.return_university_num,
|
||||
"intentionNum": args.intention_num,
|
||||
}
|
||||
return payload
|
||||
|
||||
|
||||
def api_url(base_url: str | None = None) -> str:
|
||||
root = (base_url or BASE_URL).rstrip("/")
|
||||
path = API_PATH if API_PATH.startswith("/") else f"/{API_PATH}"
|
||||
return root + path
|
||||
|
||||
|
||||
def call_api(
|
||||
payload: dict[str, Any],
|
||||
base_url: str | None = None,
|
||||
timeout: int = 60,
|
||||
) -> dict[str, Any]:
|
||||
url = api_url(base_url)
|
||||
data = json.dumps(payload, ensure_ascii=False).encode("utf-8")
|
||||
req = Request(
|
||||
url,
|
||||
data=data,
|
||||
method="POST",
|
||||
headers={
|
||||
"Content-Type": "application/json",
|
||||
"Accept": "*/*",
|
||||
},
|
||||
)
|
||||
try:
|
||||
with urlopen(req, timeout=timeout) as resp:
|
||||
charset = resp.headers.get_content_charset() or "utf-8"
|
||||
text = resp.read().decode(charset)
|
||||
except HTTPError as e:
|
||||
detail = e.read().decode("utf-8", errors="replace")
|
||||
raise RuntimeError(f"HTTP {e.code}: {detail}") from e
|
||||
except URLError as e:
|
||||
raise RuntimeError(f"网络请求失败: {e.reason}") from e
|
||||
|
||||
return json.loads(text)
|
||||
|
||||
|
||||
def main() -> int:
|
||||
parser = argparse.ArgumentParser(description="获取并解析智能推荐志愿表")
|
||||
parser.add_argument(
|
||||
"--student",
|
||||
type=Path,
|
||||
help="student.json 路径（自动映射倾向到 API 选填参数，优先于 --config）",
|
||||
)
|
||||
parser.add_argument(
|
||||
"--config",
|
||||
type=Path,
|
||||
help="完整请求 JSON 文件路径",
|
||||
)
|
||||
parser.add_argument(
|
||||
"--base-url",
|
||||
default=os.environ.get("ZP_API_BASE_URL", BASE_URL),
|
||||
help=f"API 根地址（默认脚本内 BASE_URL: {BASE_URL}）",
|
||||
)
|
||||
parser.add_argument("-o", "--output", type=Path, required=True, help="解析结果 JSON 输出路径")
|
||||
parser.add_argument("--save-raw", type=Path, help="可选：保存 API 原始响应")
|
||||
|
||||
parser.add_argument("--province", help="高考省份")
|
||||
parser.add_argument("--classify", help="选科模式：文科/理科/物理/历史/综合")
|
||||
parser.add_argument("--subjects", help="选科组合，逗号分隔，如 物理,化学,生物")
|
||||
parser.add_argument("--grade-type", dest="grade_type", help="成绩类型：本科/专科（京沪津专科批）")
|
||||
parser.add_argument("--score", type=int, help="高考成绩")
|
||||
parser.add_argument("--rank", type=int, help="高考位次")
|
||||
parser.add_argument("--batch", help="填报批次，如 本科批")
|
||||
parser.add_argument(
|
||||
"--auto-batch",
|
||||
action=argparse.BooleanOptionalAction,
|
||||
default=True,
|
||||
help="是否通过 batch/list 接口自动解析批次（默认开启）",
|
||||
)
|
||||
parser.add_argument("--universitys", help="心仪高校，逗号分隔")
|
||||
parser.add_argument("--provinces", help="省份意向，逗号分隔")
|
||||
parser.add_argument("--tags", help="院校属性意向，逗号分隔")
|
||||
parser.add_argument("--major-class", dest="major_class", help="专业类意向，逗号分隔")
|
||||
parser.add_argument("--university-num", type=int, default=30)
|
||||
parser.add_argument("--major-num", type=int, default=6)
|
||||
parser.add_argument(
|
||||
"--is-adjust",
|
||||
action=argparse.BooleanOptionalAction,
|
||||
default=True,
|
||||
)
|
||||
parser.add_argument("--volunteer-type", default="ACADEMY_GROUP")
|
||||
parser.add_argument("--return-university-num", type=int, default=30)
|
||||
parser.add_argument("--intention-num", type=int, default=0)
|
||||
|
||||
args = parser.parse_args()
|
||||
|
||||
try:
|
||||
payload = build_request_payload(args)
|
||||
except (OSError, json.JSONDecodeError) as e:
|
||||
print(f"错误: 无法读取请求配置 — {e}", file=sys.stderr)
|
||||
return 1
|
||||
|
||||
if not args.student and not args.config:
|
||||
required = ["province", "classify", "score"]
|
||||
if not args.auto_batch:
|
||||
required.append("batch")
|
||||
missing = [k for k in required if payload.get(k) is None]
|
||||
if missing:
|
||||
print(
|
||||
f"错误: 缺少必填参数 {missing}，请使用 --student、--config 或补全命令行",
|
||||
file=sys.stderr,
|
||||
)
|
||||
return 1
|
||||
|
||||
# CLI 中 null 用字符串 "null" 传入时转为 None（subjects 在 sanitize 阶段按省处理）
|
||||
for key in ("gradeType", "rank", "universitys", "provinces", "tags", "majorClass"):
|
||||
if payload.get(key) == "null":
|
||||
payload[key] = None
|
||||
|
||||
try:
|
||||
validate_classify(str(payload.get("province", "")), payload.get("classify"))
|
||||
except ValueError as e:
|
||||
print(f"错误: {e}", file=sys.stderr)
|
||||
return 1
|
||||
|
||||
batch_resolution: dict[str, Any] | None = None
|
||||
if args.auto_batch:
|
||||
try:
|
||||
batch_name, selected, all_items = resolve_batch_via_api(payload, base_url=args.base_url)
|
||||
batch_resolution = {
|
||||
"source": "batch/list",
|
||||
"selected": selected,
|
||||
"available": all_items,
|
||||
}
|
||||
print(
|
||||
f"批次: {batch_name}（{selected.get('gradeType', '')} / {selected.get('type', '')}）",
|
||||
file=sys.stderr,
|
||||
)
|
||||
except (RuntimeError, ValueError, json.JSONDecodeError) as e:
|
||||
print(f"警告: 批次接口失败，使用静态兜底 — {e}", file=sys.stderr)
|
||||
payload["batch"] = resolve_batch_fallback(payload.get("province"), payload.get("batch"))
|
||||
batch_resolution = {"source": "fallback", "batch": payload.get("batch")}
|
||||
elif not payload.get("batch"):
|
||||
print("错误: 已关闭 --auto-batch 且未提供 batch", file=sys.stderr)
|
||||
return 1
|
||||
|
||||
sanitize_payload_for_api(payload)
|
||||
try:
|
||||
validate_subjects_for_volunteer_api(payload)
|
||||
except ValueError as e:
|
||||
print(f"错误: {e}", file=sys.stderr)
|
||||
return 1
|
||||
|
||||
print(f"请求: {api_url(args.base_url)}", file=sys.stderr)
|
||||
try:
|
||||
raw = call_api(payload, base_url=args.base_url)
|
||||
except (RuntimeError, json.JSONDecodeError) as e:
|
||||
print(f"错误: {e}", file=sys.stderr)
|
||||
return 1
|
||||
|
||||
if args.save_raw:
|
||||
args.save_raw.parent.mkdir(parents=True, exist_ok=True)
|
||||
args.save_raw.write_text(json.dumps(raw, ensure_ascii=False, indent=2), encoding="utf-8")
|
||||
|
||||
try:
|
||||
parsed = parse_api_response(raw)
|
||||
except ValueError as e:
|
||||
print(f"错误: {e}", file=sys.stderr)
|
||||
return 1
|
||||
|
||||
parsed["request"] = payload
|
||||
if batch_resolution:
|
||||
parsed["batch_resolution"] = batch_resolution
|
||||
args.output.parent.mkdir(parents=True, exist_ok=True)
|
||||
args.output.write_text(json.dumps(parsed, ensure_ascii=False, indent=2), encoding="utf-8")
|
||||
print(f"已保存解析结果: {args.output}（共 {parsed['stats']['total']} 所院校）")
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
raise SystemExit(main())
|
||||
+221
@@ -0,0 +1,221 @@
|
||||
#!/usr/bin/env python3
|
||||
"""各省高考模式、classify 校验与 API 参数规范化。"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from typing import Any
|
||||
|
||||
# 仅京沪津批次接口与志愿接口需要 gradeType
|
||||
PROVINCES_REQUIRING_GRADE_TYPE = frozenset({"北京", "上海", "天津"})
|
||||
|
||||
# 测试环境批次接口暂不支持
|
||||
UNSUPPORTED_PROVINCES = frozenset({"西藏"})
|
||||
|
||||
MODE_33 = frozenset({"上海", "北京", "天津", "山东", "浙江", "海南"})
|
||||
MODE_312 = frozenset({
|
||||
"广东", "江苏", "河北", "湖北", "湖南", "福建", "辽宁", "重庆", "甘肃",
|
||||
"黑龙江", "吉林", "安徽", "江西", "贵州", "广西", "云南", "内蒙古",
|
||||
"四川", "宁夏", "山西", "河南", "陕西", "青海",
|
||||
})
|
||||
MODE_OLD = frozenset({"新疆"})
|
||||
|
||||
VALID_CLASSIFY_BY_MODE: dict[str, frozenset[str]] = {
|
||||
"33": frozenset({"综合"}),
|
||||
"312": frozenset({"物理", "历史"}),
|
||||
"老高考": frozenset({"文科", "理科"}),
|
||||
}
|
||||
|
||||
MODE_LABELS: dict[str, str] = {
|
||||
"33": "3+3",
|
||||
"312": "3+1+2",
|
||||
"老高考": "老高考",
|
||||
}
|
||||
|
||||
SUBJECTS_312_ELECTIVES = frozenset({"化学", "生物", "政治", "地理"})
|
||||
SUBJECTS_33_POOL = frozenset({"物理", "历史", "化学", "生物", "政治", "地理"})
|
||||
SUBJECTS_33_ZHEJIANG = SUBJECTS_33_POOL | frozenset({"技术"})
|
||||
|
||||
|
||||
def parse_subjects_list(subjects: Any) -> list[str]:
|
||||
if subjects in (None, "", "null", "NULL"):
|
||||
return []
|
||||
return [s.strip() for s in str(subjects).split(",") if s.strip()]
|
||||
|
||||
|
||||
def get_province_mode(province: str | None) -> str:
|
||||
if not province:
|
||||
return "unknown"
|
||||
if province in UNSUPPORTED_PROVINCES:
|
||||
return "unsupported"
|
||||
if province in MODE_OLD:
|
||||
return "老高考"
|
||||
if province in MODE_33:
|
||||
return "33"
|
||||
if province in MODE_312:
|
||||
return "312"
|
||||
return "unknown"
|
||||
|
||||
|
||||
def validate_classify(province: str, classify: str | None) -> None:
|
||||
"""校验 classify 是否与省份选科模式匹配。"""
|
||||
mode = get_province_mode(province)
|
||||
if mode == "unsupported":
|
||||
raise ValueError(
|
||||
f"省份 {province} 在测试环境暂不支持志愿/批次接口（平台未接入该省数据）"
|
||||
)
|
||||
if mode == "unknown":
|
||||
return
|
||||
if not classify:
|
||||
valid = VALID_CLASSIFY_BY_MODE[mode]
|
||||
raise ValueError(
|
||||
f"缺少 classify：{province} 为{MODE_LABELS[mode]}模式，应填 {'/'.join(sorted(valid))}"
|
||||
)
|
||||
valid = VALID_CLASSIFY_BY_MODE[mode]
|
||||
if classify not in valid:
|
||||
raise ValueError(
|
||||
f"classify 与省份不匹配：{province} 为{MODE_LABELS[mode]}模式，"
|
||||
f"classify 应为 {'/'.join(sorted(valid))}，当前为 {classify!r}。"
|
||||
f"错误 classify 会导致批次接口返回空列表。"
|
||||
)
|
||||
|
||||
|
||||
def infer_grade_type(
|
||||
province: str | None,
|
||||
score: int | None,
|
||||
explicit: str | None = None,
|
||||
batch_pref: str | None = None,
|
||||
) -> str | None:
|
||||
"""京沪津推断本科/专科；其他省份返回 None。"""
|
||||
if province not in PROVINCES_REQUIRING_GRADE_TYPE:
|
||||
return None
|
||||
if explicit in ("本科", "专科"):
|
||||
return explicit
|
||||
if batch_pref == "专科批":
|
||||
return "专科"
|
||||
if batch_pref == "本科批":
|
||||
return "本科"
|
||||
if score is not None and score <= 450:
|
||||
return "专科"
|
||||
return "本科"
|
||||
|
||||
|
||||
def batch_list_query_params(payload: dict[str, Any]) -> dict[str, Any]:
|
||||
"""从请求体构建 batch/list 查询参数。"""
|
||||
province = payload.get("province")
|
||||
classify = payload.get("classify")
|
||||
score = payload.get("score")
|
||||
if not province or not classify or score is None:
|
||||
raise ValueError("批次接口需要 province、classify、score")
|
||||
|
||||
validate_classify(str(province), str(classify))
|
||||
|
||||
params: dict[str, Any] = {
|
||||
"province": province,
|
||||
"classify": classify,
|
||||
"score": int(score),
|
||||
}
|
||||
grade_type = payload.get("gradeType")
|
||||
if not grade_type and province in PROVINCES_REQUIRING_GRADE_TYPE:
|
||||
grade_type = infer_grade_type(
|
||||
str(province),
|
||||
int(score),
|
||||
batch_pref=payload.get("batch"),
|
||||
)
|
||||
if grade_type:
|
||||
payload["gradeType"] = grade_type
|
||||
if province in PROVINCES_REQUIRING_GRADE_TYPE and grade_type:
|
||||
params["gradeType"] = grade_type
|
||||
return params
|
||||
|
||||
|
||||
def normalize_subjects_for_api(payload: dict[str, Any]) -> None:
|
||||
"""按省份规则规范化 subjects 字段。"""
|
||||
province = payload.get("province")
|
||||
grade_type = payload.get("gradeType")
|
||||
subjects = payload.get("subjects")
|
||||
|
||||
if province == "新疆":
|
||||
if subjects in (None, "", "null", "NULL"):
|
||||
payload.pop("subjects", None)
|
||||
return
|
||||
|
||||
if province in PROVINCES_REQUIRING_GRADE_TYPE and grade_type == "专科":
|
||||
if subjects in (None, "", "null", "NULL"):
|
||||
payload["subjects"] = "null"
|
||||
return
|
||||
|
||||
if subjects in ("null", "NULL", ""):
|
||||
payload["subjects"] = None
|
||||
|
||||
|
||||
def sanitize_payload_for_api(payload: dict[str, Any]) -> None:
|
||||
"""发志愿接口前按省份规则清理请求体。"""
|
||||
province = payload.get("province")
|
||||
if province not in PROVINCES_REQUIRING_GRADE_TYPE:
|
||||
payload.pop("gradeType", None)
|
||||
normalize_subjects_for_api(payload)
|
||||
|
||||
|
||||
def validate_subjects_for_volunteer_api(payload: dict[str, Any]) -> None:
|
||||
"""志愿接口前校验 subjects 是否满足省份要求。"""
|
||||
province = str(payload.get("province") or "")
|
||||
mode = get_province_mode(province)
|
||||
label = MODE_LABELS.get(mode, mode)
|
||||
|
||||
if province == "新疆":
|
||||
return
|
||||
if province in PROVINCES_REQUIRING_GRADE_TYPE and payload.get("gradeType") == "专科":
|
||||
return
|
||||
|
||||
subjects = payload.get("subjects")
|
||||
if subjects in (None, "", "null", "NULL"):
|
||||
raise ValueError(
|
||||
f"缺少 subjects：{province}（{label}）需传选科组合，如 物理,化学,生物"
|
||||
)
|
||||
|
||||
parts = parse_subjects_list(subjects)
|
||||
classify = payload.get("classify")
|
||||
|
||||
if mode == "312":
|
||||
if len(parts) != 3:
|
||||
raise ValueError(
|
||||
f"subjects 格式错误：{province}（3+1+2）须传完整三科（首选科目+两门再选），"
|
||||
f"逗号分隔，如 classify=物理 时 subjects=物理,化学,生物；"
|
||||
f"不能只传两门再选科目。当前为 {subjects!r}（{len(parts)} 科）"
|
||||
)
|
||||
if classify not in parts:
|
||||
raise ValueError(
|
||||
f"subjects 与 classify 不一致：{province} classify={classify!r}，"
|
||||
f"subjects 须包含首选科目 {classify!r}，当前为 {subjects!r}"
|
||||
)
|
||||
electives = [p for p in parts if p != classify]
|
||||
if len(electives) != 2:
|
||||
raise ValueError(
|
||||
f"subjects 格式错误：{province}（3+1+2）除首选科目外还须两门再选"
|
||||
f"（化学/生物/政治/地理），当前为 {subjects!r}"
|
||||
)
|
||||
invalid = [p for p in electives if p not in SUBJECTS_312_ELECTIVES]
|
||||
if invalid:
|
||||
raise ValueError(
|
||||
f"subjects 含无效再选科目 {invalid}：3+1+2 再选仅限化学/生物/政治/地理"
|
||||
)
|
||||
return
|
||||
|
||||
if mode == "33" and province not in PROVINCES_REQUIRING_GRADE_TYPE:
|
||||
if len(parts) != 3:
|
||||
raise ValueError(
|
||||
f"subjects 格式错误：{province}（3+3）须传完整三科选考科目，"
|
||||
f"逗号分隔，如 物理,化学,生物；当前为 {subjects!r}（{len(parts)} 科）"
|
||||
)
|
||||
pool = SUBJECTS_33_ZHEJIANG if province == "浙江" else SUBJECTS_33_POOL
|
||||
invalid = [p for p in parts if p not in pool]
|
||||
if invalid:
|
||||
raise ValueError(f"subjects 含无效科目 {invalid}：{province} 选考范围见 reference.md")
|
||||
return
|
||||
|
||||
if mode == "33" and province in PROVINCES_REQUIRING_GRADE_TYPE:
|
||||
if payload.get("gradeType") == "本科" and len(parts) != 3:
|
||||
raise ValueError(
|
||||
f"subjects 格式错误：{province}（3+3 本科）须传完整三科，"
|
||||
f"如 物理,化学,生物；当前为 {subjects!r}"
|
||||
)
|
||||
+166
@@ -0,0 +1,166 @@
|
||||
#!/usr/bin/env python3
|
||||
"""批量测试 batch/list 接口：各省 × 正确/错误 classify × 高/低分。"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import sys
|
||||
import time
|
||||
from pathlib import Path
|
||||
from urllib.error import HTTPError, URLError
|
||||
from urllib.parse import urlencode
|
||||
from urllib.request import Request, urlopen
|
||||
|
||||
# 允许从 scripts/ 目录直接运行
|
||||
sys.path.insert(0, str(Path(__file__).resolve().parent))
|
||||
|
||||
from province_config import ( # noqa: E402
|
||||
MODE_312,
|
||||
MODE_33,
|
||||
MODE_OLD,
|
||||
UNSUPPORTED_PROVINCES,
|
||||
batch_list_query_params,
|
||||
infer_grade_type,
|
||||
)
|
||||
|
||||
BASE_URL = "https://publicapi.chatglm.cn/chatglm_public/skill-test"
|
||||
BATCH_LIST_PATH = "/zp/volunteer/batch/list"
|
||||
|
||||
PROVINCE_CLASSIFY: dict[str, list[str]] = {}
|
||||
for p in MODE_33:
|
||||
PROVINCE_CLASSIFY[p] = ["综合"]
|
||||
for p in MODE_312:
|
||||
PROVINCE_CLASSIFY[p] = ["物理", "历史"]
|
||||
for p in MODE_OLD:
|
||||
PROVINCE_CLASSIFY[p] = ["文科", "理科"]
|
||||
for p in UNSUPPORTED_PROVINCES:
|
||||
PROVINCE_CLASSIFY[p] = ["综合"]
|
||||
|
||||
WRONG_CLASSIFY = {
|
||||
"综合": "物理",
|
||||
"物理": "综合",
|
||||
"历史": "理科",
|
||||
"文科": "物理",
|
||||
"理科": "历史",
|
||||
}
|
||||
|
||||
HIGH_SCORE = 650
|
||||
LOW_SCORE = 280
|
||||
|
||||
|
||||
def call_batch_from_payload(payload: dict) -> dict:
|
||||
params = batch_list_query_params(dict(payload))
|
||||
query = urlencode(params)
|
||||
url = f"{BASE_URL.rstrip('/')}{BATCH_LIST_PATH}?{query}"
|
||||
req = Request(url, method="GET", headers={"Accept": "*/*"})
|
||||
try:
|
||||
with urlopen(req, timeout=30) as resp:
|
||||
text = resp.read().decode(resp.headers.get_content_charset() or "utf-8")
|
||||
except HTTPError as e:
|
||||
return {"http": e.code, "body": e.read().decode("utf-8", errors="replace")}
|
||||
except URLError as e:
|
||||
return {"error": str(e.reason)}
|
||||
try:
|
||||
return json.loads(text)
|
||||
except json.JSONDecodeError:
|
||||
return {"error": "invalid json", "body": text[:200]}
|
||||
|
||||
|
||||
def summarize_result(raw: dict) -> str:
|
||||
if "error" in raw or "http" in raw:
|
||||
return f"FAIL:{raw.get('http') or raw.get('error')}"
|
||||
status = raw.get("status")
|
||||
msg = raw.get("message", "")
|
||||
result = raw.get("result")
|
||||
if status != 0:
|
||||
return f"FAIL:status={status},{msg}"
|
||||
if not isinstance(result, list):
|
||||
return "FAIL:result not list"
|
||||
if not result:
|
||||
return "OK:empty[]"
|
||||
batches = [f"{x.get('batch')}({x.get('gradeType')}/{x.get('type')})" for x in result[:3]]
|
||||
extra = f"+{len(result)-3}" if len(result) > 3 else ""
|
||||
return f"OK:{len(result)}项 [{', '.join(batches)}{extra}]"
|
||||
|
||||
|
||||
def main() -> int:
|
||||
rows: list[dict] = []
|
||||
provinces = sorted(PROVINCE_CLASSIFY.keys(), key=lambda x: (x not in UNSUPPORTED_PROVINCES, x))
|
||||
|
||||
print("=== 正确 classify × 高分/低分（京沪津含 gradeType）===")
|
||||
for province in provinces:
|
||||
classifies = PROVINCE_CLASSIFY[province]
|
||||
primary = classifies[0]
|
||||
for score_label, score in [("高", HIGH_SCORE), ("低", LOW_SCORE)]:
|
||||
payload = {
|
||||
"province": province,
|
||||
"classify": primary,
|
||||
"score": score,
|
||||
"batch": "本科批",
|
||||
}
|
||||
if province in ("北京", "上海", "天津"):
|
||||
payload["gradeType"] = infer_grade_type(province, score, batch_pref="本科批")
|
||||
try:
|
||||
raw = call_batch_from_payload(payload)
|
||||
except ValueError as e:
|
||||
raw = {"status": -1, "message": str(e)}
|
||||
summary = summarize_result(raw)
|
||||
rows.append({
|
||||
"province": province,
|
||||
"classify": primary,
|
||||
"score": score,
|
||||
"score_label": score_label,
|
||||
"gradeType": payload.get("gradeType"),
|
||||
"wrong_classify": False,
|
||||
"summary": summary,
|
||||
"status": raw.get("status"),
|
||||
"message": raw.get("message"),
|
||||
"count": len(raw.get("result") or []) if isinstance(raw.get("result"), list) else None,
|
||||
})
|
||||
mark = "✓" if summary.startswith("OK") and "empty" not in summary else "✗"
|
||||
gt = payload.get("gradeType") or ""
|
||||
print(f" {mark} {province:4s} {primary} {gt:2s} {score_label}分({score}) → {summary}")
|
||||
time.sleep(0.12)
|
||||
|
||||
print("\n=== 错误 classify 探测（高分）===")
|
||||
for province in provinces:
|
||||
if province in UNSUPPORTED_PROVINCES:
|
||||
continue
|
||||
primary = PROVINCE_CLASSIFY[province][0]
|
||||
wrong = WRONG_CLASSIFY.get(primary, "综合" if primary != "综合" else "物理")
|
||||
ok_payload = {"province": province, "classify": primary, "score": HIGH_SCORE, "batch": "本科批"}
|
||||
wrong_payload = dict(ok_payload, classify=wrong)
|
||||
if province in ("北京", "上海", "天津"):
|
||||
ok_payload["gradeType"] = infer_grade_type(province, HIGH_SCORE, batch_pref="本科批")
|
||||
wrong_payload["gradeType"] = ok_payload["gradeType"]
|
||||
try:
|
||||
raw_ok = call_batch_from_payload(ok_payload)
|
||||
raw_wrong = call_batch_from_payload(wrong_payload)
|
||||
except ValueError as e:
|
||||
print(f" ! {province}: validate error {e}")
|
||||
continue
|
||||
ok_s = summarize_result(raw_ok)
|
||||
wrong_s = summarize_result(raw_wrong)
|
||||
if wrong_s != ok_s:
|
||||
print(f" ! {province}: 正确({primary})={ok_s} vs 错误({wrong})={wrong_s}")
|
||||
time.sleep(0.12)
|
||||
|
||||
out = Path(__file__).resolve().parent.parent / "output" / "batch_api_test_results.json"
|
||||
out.parent.mkdir(parents=True, exist_ok=True)
|
||||
out.write_text(json.dumps(rows, ensure_ascii=False, indent=2), encoding="utf-8")
|
||||
print(f"\n详细结果已保存: {out}")
|
||||
|
||||
fails = [
|
||||
r for r in rows
|
||||
if not r["summary"].startswith("OK") or "empty[]" in r["summary"]
|
||||
if not r.get("wrong_classify")
|
||||
]
|
||||
print(f"\n=== 汇总: 正确参数异常 {len(fails)} 项 ===")
|
||||
for r in fails:
|
||||
print(f" {r['province']} {r['classify']} {r['score_label']}分 → {r['summary']} | {r.get('message')}")
|
||||
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
raise SystemExit(main())
|
||||
Reference in new issue
Block a user