Z User
e45b68bb8b
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
222 lines
8.0 KiB
Python
Executable File
222 lines
8.0 KiB
Python
Executable File
#!/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}"
|
||
)
|