0

src/api/mps/mpsCommonController.js

@@ -382,7 +382,7 @@ export async function GetMPSNotice(ctx) {
         result.IsShow = true;
         result.Title = "更新通知";
         const content = [
-            "2026年上海中考招生计划目前完成自主招生、名额到区、名额到校、1-15志愿所的计划统计工作。若有问题，欢迎反馈，我们会及时做出修正。请将信息分享给身边的家长朋友们。非常谢谢您的支持！",
+            "2026年上海中考招生目前完成名额到区、名额到校的分数线输入工作，1-15志愿分数线还需要耐心等待。若有问题，欢迎反馈，我们会及时做出修正。请将信息分享给身边的家长朋友们。非常谢谢您的支持！",
             //"2026年上海中考招生计划已经全部整理完毕，包括各区各校的自主招生、名额到区、名额到校、1-15志愿等数据。请将信息分享给身边的家长朋友们。非常谢谢您的支持！",
             //"今年夏天要读的书，必须是《录取通知书》，祝孩子们中考考出最优成绩！",
             //"最新的上海16区统一招生1-15志愿各校计划整理完毕，欢迎查看！"

秒过分数线数据导入/audit_mps_score_scores_2026_sample.py

@@ -0,0 +1,251 @@
+import argparse
+import json
+import math
+import random
+from collections import defaultdict
+from decimal import Decimal
+
+import pymysql
+
+import import_mps_score_all_district_scores_2026 as importer
+
+
+SAMPLE_RATE = Decimal("0.30")
+RANDOM_SEED = 20260720
+REPORT_FILE = "mps_score_scores_2026_sample_audit.json"
+
+
+def decimal_to_json(value):
+    if isinstance(value, Decimal):
+        return str(value)
+    return value
+
+
+def fetch_actual_scores(cursor, district_id, score_type):
+    cursor.execute(
+        """
+        SELECT ID, DistrictID, ScoreType, SchoolOfGraduation, SchoolFullNameJunior,
+               SchoolTarget, SchoolFullName, PlanNum,
+               ScoreTotal, Score1, Score2, Score3, Score4, ScoreTotalDifferenceValue
+        FROM MPS_Score
+        WHERE ScoreYear = %s
+          AND ScoreType = %s
+          AND DistrictID = %s
+        """,
+        (importer.YEAR, score_type, district_id),
+    )
+    return {row["ID"]: row for row in cursor.fetchall()}
+
+
+def values_equal(actual, expected):
+    return actual == expected
+
+
+def check_update(actual_by_id, update):
+    actual = actual_by_id.get(update["ID"])
+    if not actual:
+        return [{"column": "ID", "actual": None, "expected": update["ID"]}]
+
+    mismatches = []
+    for column in importer.SCORE_COLUMNS:
+        if not values_equal(actual[column], update[column]):
+            mismatches.append(
+                {
+                    "column": column,
+                    "actual": actual[column],
+                    "expected": update[column],
+                }
+            )
+
+    total_with_eval = update["ScoreTotal"] + Decimal("50")
+    if total_with_eval != update["total_with_eval"]:
+        mismatches.append(
+            {
+                "column": "ScoreTotal_plus_50",
+                "actual": total_with_eval,
+                "expected": update["total_with_eval"],
+            }
+        )
+
+    return mismatches
+
+
+def sample_updates(district_id, score_type, updates):
+    sample_size = math.ceil(len(updates) * float(SAMPLE_RATE))
+    if updates and sample_size < 1:
+        sample_size = 1
+    rng = random.Random(f"{RANDOM_SEED}:{district_id}:{score_type}")
+    return rng.sample(updates, sample_size)
+
+
+def full_recheck(actual_by_id, updates):
+    errors = []
+    for update in updates:
+        mismatches = check_update(actual_by_id, update)
+        if mismatches:
+            errors.append(
+                {
+                    "ID": update["ID"],
+                    "SchoolFullNameJunior": update["SchoolFullNameJunior"],
+                    "SchoolFullName": update["SchoolFullName"],
+                    "page": update["page"],
+                    "mismatches": mismatches,
+                }
+            )
+    return errors
+
+
+def audit_one(cursor, names_by_school_id, district_id, score_type):
+    data = importer.collect_one(cursor, names_by_school_id, district_id, score_type)
+    actual_by_id = fetch_actual_scores(cursor, district_id, score_type)
+    sampled = sample_updates(district_id, score_type, data["updates"])
+
+    sample_errors = []
+    for update in sampled:
+        mismatches = check_update(actual_by_id, update)
+        if mismatches:
+            sample_errors.append(
+                {
+                    "ID": update["ID"],
+                    "SchoolFullNameJunior": update["SchoolFullNameJunior"],
+                    "SchoolFullName": update["SchoolFullName"],
+                    "page": update["page"],
+                    "mismatches": mismatches,
+                }
+            )
+
+    full_errors = full_recheck(actual_by_id, data["updates"]) if sample_errors else []
+
+    sample_preview = []
+    for update in sampled[:10]:
+        sample_preview.append(
+            {
+                "ID": update["ID"],
+                "SchoolFullNameJunior": update["SchoolFullNameJunior"],
+                "SchoolFullName": update["SchoolFullName"],
+                "pdf_total": update["total_with_eval"],
+                "stored_score_total": update["ScoreTotal"],
+                "page": update["page"],
+            }
+        )
+
+    return {
+        "district_id": district_id,
+        "district": importer.DISTRICTS[district_id],
+        "score_type": score_type,
+        "pdf_parsed_rows": len(data["parsed_rows"]),
+        "db_rows": len(data["db_rows"]),
+        "auditable_imported_rows": len(data["updates"]),
+        "sample_rate": str(SAMPLE_RATE),
+        "sample_size": len(sampled),
+        "sample_errors": sample_errors,
+        "full_recheck_triggered": bool(sample_errors),
+        "full_recheck_errors": full_errors,
+        "not_audited_pending_rows": len(data["missing_db"]),
+        "pending_match_problems": len(data["problems"]),
+        "pending_duplicate_target_rows": len(data["duplicates"]),
+        "sample_preview": sample_preview,
+    }
+
+
+def duplicate_key_counts(cursor):
+    results = {}
+    checks = {
+        "名额到区": "DistrictID, SchoolTarget",
+        "名额到校": "DistrictID, SchoolOfGraduation, SchoolTarget",
+    }
+    for score_type, group_expr in checks.items():
+        cursor.execute(
+            f"""
+            SELECT COUNT(*) AS duplicate_groups
+            FROM (
+              SELECT {group_expr}, COUNT(*) c
+              FROM MPS_Score
+              WHERE ScoreYear = %s AND ScoreType = %s
+              GROUP BY {group_expr}
+              HAVING c > 1
+            ) t
+            """,
+            (importer.YEAR, score_type),
+        )
+        results[score_type] = cursor.fetchone()["duplicate_groups"]
+    return results
+
+
+def score_summary(cursor):
+    summary = defaultdict(list)
+    for score_type in importer.SCORE_TYPES:
+        cursor.execute(
+            """
+            SELECT DistrictID,
+                   COUNT(*) AS db_rows,
+                   SUM(CASE WHEN COALESCE(ScoreTotal, 0) <> 0 THEN 1 ELSE 0 END) AS scored_rows,
+                   SUM(CASE WHEN COALESCE(ScoreTotal, 0) = 0 THEN 1 ELSE 0 END) AS unscored_rows,
+                   MIN(NULLIF(ScoreTotal, 0)) AS min_score,
+                   MAX(NULLIF(ScoreTotal, 0)) AS max_score
+            FROM MPS_Score
+            WHERE ScoreYear = %s AND ScoreType = %s
+            GROUP BY DistrictID
+            ORDER BY DistrictID
+            """,
+            (importer.YEAR, score_type),
+        )
+        summary[score_type] = cursor.fetchall()
+    return summary
+
+
+def parse_args():
+    parser = argparse.ArgumentParser()
+    parser.add_argument("--district", type=int, action="append", choices=sorted(importer.DISTRICTS))
+    parser.add_argument("--score-type", choices=importer.SCORE_TYPES, action="append")
+    return parser.parse_args()
+
+
+def main():
+    args = parse_args()
+    districts = args.district or sorted(importer.DISTRICTS)
+    score_types = args.score_type or importer.SCORE_TYPES
+
+    conn = pymysql.connect(**importer.DB_CONFIG)
+    try:
+        with conn.cursor() as cursor:
+            names_by_school_id = importer.load_school_names(cursor)
+            audits = []
+            for district_id in districts:
+                for score_type in score_types:
+                    result = audit_one(cursor, names_by_school_id, district_id, score_type)
+                    audits.append(result)
+                    print(
+                        result["district_id"],
+                        result["district"],
+                        result["score_type"],
+                        "auditable",
+                        result["auditable_imported_rows"],
+                        "sample",
+                        result["sample_size"],
+                        "sample_errors",
+                        len(result["sample_errors"]),
+                        "full_recheck",
+                        result["full_recheck_triggered"],
+                        "full_errors",
+                        len(result["full_recheck_errors"]),
+                        "pending",
+                        result["not_audited_pending_rows"],
+                    )
+
+            report = {
+                "sample_rate": str(SAMPLE_RATE),
+                "random_seed": RANDOM_SEED,
+                "audits": audits,
+                "duplicate_key_counts": duplicate_key_counts(cursor),
+                "score_summary": score_summary(cursor),
+            }
+            with open(REPORT_FILE, "w", encoding="utf-8") as handle:
+                json.dump(report, handle, ensure_ascii=False, indent=2, default=decimal_to_json)
+            print("report", REPORT_FILE)
+    finally:
+        conn.close()
+
+
+if __name__ == "__main__":
+    main()

秒过分数线数据导入/import_mps_score_all_district_scores_2026.py

@@ -0,0 +1,652 @@
+import argparse
+import json
+import os
+import re
+from collections import defaultdict
+from decimal import Decimal
+
+import pdfplumber
+import pymysql
+
+
+DB_CONFIG = {
+    "host": "589ae8e08493d.sh.cdb.myqcloud.com",
+    "port": 8124,
+    "user": "cdb_outerroot",
+    "password": "kylx!@#!QAZ@WSX",
+    "database": "kylx365_db",
+    "charset": "utf8mb4",
+    "connect_timeout": 10,
+    "read_timeout": 30,
+    "write_timeout": 30,
+    "cursorclass": pymysql.cursors.DictCursor,
+}
+
+YEAR = "2026"
+PREVIOUS_YEAR = "2025"
+BASE_DIR = "/Volumes/程杰外接SD盘/上海中考招生计划/2026/分数线"
+BACKUP_FILE = "mps_score_scores_2026_backup.json"
+REPORT_FILE = "mps_score_scores_2026_import_report.json"
+SCORE_TYPES = ("名额到区", "名额到校")
+
+DISTRICTS = {
+    1: "黄浦区",
+    2: "徐汇区",
+    3: "长宁区",
+    4: "静安区",
+    5: "普陀区",
+    6: "虹口区",
+    7: "杨浦区",
+    8: "闵行区",
+    9: "宝山区",
+    10: "嘉定区",
+    11: "浦东新区",
+    12: "金山区",
+    13: "松江区",
+    14: "青浦区",
+    15: "奉贤区",
+    16: "崇明区",
+}
+
+SCORE_COLUMNS = [
+    "ScoreTotal",
+    "Score1",
+    "Score2",
+    "Score3",
+    "Score4",
+    "ScoreTotalDifferenceValue",
+]
+
+
+def compact(value):
+    return re.sub(r"\s+", "", str(value or "").replace("\u3000", ""))
+
+
+def decimal_to_json(value):
+    if isinstance(value, Decimal):
+        return str(value)
+    return value
+
+
+def parse_decimal(value):
+    match = re.search(r"\d+(?:\.\d+)?", compact(value))
+    return Decimal(match.group(0)) if match else None
+
+
+def is_subsequence(needle, haystack):
+    iterator = iter(haystack)
+    return all(char in iterator for char in needle)
+
+
+def split_aliases(value):
+    aliases = []
+    for part in re.split(r"[,，;；、/|]", str(value or "")):
+        part = compact(part)
+        if part:
+            aliases.append(part)
+    return aliases
+
+
+def name_variants(name):
+    value = compact(name)
+    variants = []
+
+    def add(item, penalty):
+        item = compact(item)
+        if item and item not in [variant for variant, _penalty in variants]:
+            variants.append((item, penalty))
+
+    add(value, 0)
+    add(value.replace("上海市", "上海"), 80)
+    if value.startswith("上海市"):
+        add(value[3:], 120)
+    if value.startswith("上海"):
+        add(value[2:], 120)
+
+    base_items = list(variants)
+    for item, penalty in base_items:
+        without_brackets = re.sub(r"[（(].*?[）)]", "", item)
+        if without_brackets != item:
+            add(without_brackets, penalty + 500)
+        for bracket_part in re.findall(r"[（(]([^）)]+)[）)]", item):
+            add(bracket_part, penalty + 300)
+    return variants
+
+
+def score_variant(candidate, raw):
+    if candidate == raw:
+        return 6000 + len(candidate)
+    if candidate in raw:
+        return 5000 + len(candidate)
+    if raw in candidate:
+        return 4500 + len(raw)
+    if is_subsequence(candidate, raw):
+        return 3000 + len(candidate)
+    if is_subsequence(raw, candidate):
+        return 2500 + len(raw)
+    return 0
+
+
+def candidate_score(raw_value, names):
+    raw = compact(raw_value)
+    scores = []
+    for name in names:
+        for variant, penalty in name_variants(name):
+            score = score_variant(variant, raw) - penalty
+            if score > 0:
+                scores.append((score, name, variant))
+    return max(scores, default=(0, "", ""), key=lambda item: item[0])
+
+
+def match_name(raw_value, candidates):
+    scored = []
+    for key, display_name, names in candidates:
+        best_score, matched_name, matched_variant = candidate_score(raw_value, names)
+        if best_score:
+            scored.append((best_score, key, display_name, matched_name, matched_variant))
+    scored.sort(reverse=True, key=lambda item: item[0])
+    if not scored:
+        return None, 0, []
+    best = scored[0]
+    ties = [item for item in scored if item[0] == best[0] and item[1] != best[1]]
+    if ties:
+        return None, best[0], scored[:5]
+    return best[1], best[0], scored[:5]
+
+
+def parse_pdf(score_type, path):
+    rows = []
+    with pdfplumber.open(path) as pdf:
+        for page_number, page in enumerate(pdf.pages, 1):
+            for table in page.extract_tables():
+                for row_index, row in enumerate(table):
+                    if not row or len(row) < 9:
+                        continue
+                    if any(
+                        token in compact(cell)
+                        for cell in row[:3]
+                        for token in ("录取最低分", "招生学校", "初中学校", "区名称")
+                    ):
+                        continue
+
+                    total_with_eval = parse_decimal(row[2])
+                    eval_score = parse_decimal(row[4])
+                    score1 = parse_decimal(row[5])
+                    score2 = parse_decimal(row[6])
+                    score3 = parse_decimal(row[7])
+                    score4 = parse_decimal(row[8])
+                    if None in (total_with_eval, eval_score, score1, score2, score3, score4):
+                        continue
+                    if eval_score != Decimal("50"):
+                        raise ValueError(
+                            f"{score_type} page {page_number} row {row_index} "
+                            f"has unexpected 综合素质评价 score: {eval_score}"
+                        )
+
+                    parsed = {
+                        "page": page_number,
+                        "row_index": row_index,
+                        "raw": row,
+                        "total_with_eval": total_with_eval,
+                        "ScoreTotal": total_with_eval - Decimal("50"),
+                        "Score1": score1,
+                        "Score2": score2,
+                        "Score3": score3,
+                        "Score4": score4,
+                    }
+                    if score_type == "名额到区":
+                        parsed["high_raw"] = row[1]
+                    else:
+                        parsed["junior_raw"] = row[0]
+                        parsed["high_raw"] = row[1]
+                    rows.append(parsed)
+    return rows
+
+
+def load_school_names(cursor):
+    cursor.execute(
+        """
+        SELECT ID, SchoolFullName, SchoolShortName, SchoolOtherName
+        FROM MPS_School
+        WHERE SchoolType1 IN ('高中', '初中')
+        """
+    )
+    names_by_id = defaultdict(list)
+    for row in cursor.fetchall():
+        for field in ("SchoolFullName", "SchoolShortName"):
+            value = compact(row[field])
+            if value and value not in names_by_id[int(row["ID"])]:
+                names_by_id[int(row["ID"])].append(value)
+        for alias in split_aliases(row["SchoolOtherName"]):
+            if alias not in names_by_id[int(row["ID"])]:
+                names_by_id[int(row["ID"])].append(alias)
+    return names_by_id
+
+
+def fetch_existing(cursor, score_type, district_id, year):
+    cursor.execute(
+        """
+        SELECT ID, ScoreYear, ScoreType, DistrictID, SchoolOfGraduation,
+               SchoolFullNameJunior, SchoolTarget, SchoolFullName, PlanNum,
+               ScoreTotal, Score1, Score2, Score3, Score4,
+               ScoreTotalDifferenceValue
+        FROM MPS_Score
+        WHERE ScoreYear = %s
+          AND ScoreType = %s
+          AND DistrictID = %s
+        ORDER BY ID
+        """,
+        (year, score_type, district_id),
+    )
+    return cursor.fetchall()
+
+
+def business_key(row, score_type):
+    if score_type == "名额到校":
+        return (int(row["SchoolOfGraduation"]), str(row["SchoolTarget"]))
+    return (str(row["SchoolTarget"]),)
+
+
+def previous_scores_by_key(cursor, score_type, district_id):
+    rows = fetch_existing(cursor, score_type, district_id, PREVIOUS_YEAR)
+    return {business_key(row, score_type): row["ScoreTotal"] for row in rows}
+
+
+def names_for_school(names_by_school_id, school_id, fallback):
+    names = []
+    fallback = compact(fallback)
+    if fallback:
+        names.append(fallback)
+    for name in names_by_school_id.get(int(school_id), []):
+        if name not in names:
+            names.append(name)
+    return names
+
+
+def attach_matches(score_type, parsed_rows, db_rows, names_by_school_id):
+    problems = []
+    matched = []
+    used_ids = defaultdict(list)
+
+    if score_type == "名额到区":
+        candidates = [
+            (
+                row["ID"],
+                row["SchoolFullName"],
+                names_for_school(names_by_school_id, row["SchoolTarget"], row["SchoolFullName"]),
+            )
+            for row in db_rows
+        ]
+        row_by_id = {row["ID"]: row for row in db_rows}
+        for parsed in parsed_rows:
+            db_id, match_score, top = match_name(parsed["high_raw"], candidates)
+            if not db_id or match_score < 2500:
+                problems.append({"type": "high_match", "row": parsed, "top": top})
+                continue
+            parsed["db_row"] = row_by_id[db_id]
+            parsed["match_score"] = match_score
+            matched.append(parsed)
+            used_ids[db_id].append(parsed)
+    else:
+        junior_by_id = {}
+        rows_by_junior = defaultdict(list)
+        row_by_id = {row["ID"]: row for row in db_rows}
+        for row in db_rows:
+            junior_id = int(row["SchoolOfGraduation"])
+            junior_by_id[junior_id] = (
+                junior_id,
+                row["SchoolFullNameJunior"],
+                names_for_school(names_by_school_id, junior_id, row["SchoolFullNameJunior"]),
+            )
+            rows_by_junior[junior_id].append(
+                (
+                    row["ID"],
+                    row["SchoolFullName"],
+                    names_for_school(names_by_school_id, row["SchoolTarget"], row["SchoolFullName"]),
+                )
+            )
+        junior_candidates = sorted(junior_by_id.values(), key=lambda item: item[0])
+
+        for parsed in parsed_rows:
+            junior_id, junior_score, junior_top = match_name(parsed["junior_raw"], junior_candidates)
+            if not junior_id or junior_score < 2500:
+                problems.append({"type": "junior_match", "row": parsed, "top": junior_top})
+                continue
+            db_id, high_score, high_top = match_name(parsed["high_raw"], rows_by_junior[junior_id])
+            if not db_id or high_score < 2500:
+                problems.append({"type": "high_match", "row": parsed, "junior_id": junior_id, "top": high_top})
+                continue
+            parsed["db_row"] = row_by_id[db_id]
+            parsed["junior_match_score"] = junior_score
+            parsed["high_match_score"] = high_score
+            matched.append(parsed)
+            used_ids[db_id].append(parsed)
+
+    duplicates = {db_id: rows for db_id, rows in used_ids.items() if len(rows) > 1}
+    if duplicates:
+        duplicate_ids = set(duplicates)
+        matched = [row for row in matched if row["db_row"]["ID"] not in duplicate_ids]
+    used_after_filter = {row["db_row"]["ID"] for row in matched}
+    missing_db = [row for row in db_rows if row["ID"] not in used_after_filter]
+    return matched, problems, duplicates, missing_db
+
+
+def build_updates(score_type, matched_rows, previous_scores):
+    updates = []
+    for parsed in matched_rows:
+        db_row = parsed["db_row"]
+        key = business_key(db_row, score_type)
+        previous = previous_scores.get(key)
+        diff = None if previous is None else parsed["ScoreTotal"] - previous
+        updates.append(
+            {
+                "ID": db_row["ID"],
+                "DistrictID": db_row["DistrictID"],
+                "SchoolOfGraduation": db_row["SchoolOfGraduation"],
+                "SchoolFullNameJunior": db_row["SchoolFullNameJunior"],
+                "SchoolTarget": db_row["SchoolTarget"],
+                "SchoolFullName": db_row["SchoolFullName"],
+                "PlanNum": db_row["PlanNum"],
+                "ScoreTotal": parsed["ScoreTotal"],
+                "Score1": parsed["Score1"],
+                "Score2": parsed["Score2"],
+                "Score3": parsed["Score3"],
+                "Score4": parsed["Score4"],
+                "ScoreTotalDifferenceValue": diff,
+                "total_with_eval": parsed["total_with_eval"],
+                "page": parsed["page"],
+            }
+        )
+    return updates
+
+
+def compare_existing_scores(db_row, update):
+    for column in SCORE_COLUMNS:
+        old_value = db_row[column]
+        new_value = update[column]
+        if old_value in (None, Decimal("0.00")):
+            continue
+        if new_value != old_value:
+            return False
+    return True
+
+
+def pdf_path(score_type, district_id):
+    return os.path.join(BASE_DIR, score_type, f"{district_id}.pdf")
+
+
+def collect_one(cursor, names_by_school_id, district_id, score_type):
+    parsed_rows = parse_pdf(score_type, pdf_path(score_type, district_id))
+    db_rows = fetch_existing(cursor, score_type, district_id, YEAR)
+    previous_scores = previous_scores_by_key(cursor, score_type, district_id)
+    matched, problems, duplicates, missing_db = attach_matches(
+        score_type, parsed_rows, db_rows, names_by_school_id
+    )
+    updates = build_updates(score_type, matched, previous_scores)
+    return {
+        "district_id": district_id,
+        "district_name": DISTRICTS[district_id],
+        "score_type": score_type,
+        "parsed_rows": parsed_rows,
+        "db_rows": db_rows,
+        "matched": matched,
+        "problems": problems,
+        "duplicates": duplicates,
+        "missing_db": missing_db,
+        "updates": updates,
+    }
+
+
+def collect_all(cursor, districts, score_types):
+    names_by_school_id = load_school_names(cursor)
+    collected = {}
+    for district_id in districts:
+        for score_type in score_types:
+            collected[(district_id, score_type)] = collect_one(
+                cursor, names_by_school_id, district_id, score_type
+            )
+    return collected
+
+
+def summarize(data):
+    scores = [row["ScoreTotal"] for row in data["updates"]]
+    return {
+        "district_id": data["district_id"],
+        "district": data["district_name"],
+        "score_type": data["score_type"],
+        "parsed": len(data["parsed_rows"]),
+        "db": len(data["db_rows"]),
+        "matched_for_update": len(data["updates"]),
+        "problems": len(data["problems"]),
+        "duplicate_target_rows": len(data["duplicates"]),
+        "missing_from_pdf_or_skipped": len(data["missing_db"]),
+        "min_score_after_minus_50": min(scores) if scores else None,
+        "max_score_after_minus_50": max(scores) if scores else None,
+    }
+
+
+def printable_top(top):
+    return [
+        {
+            "score": item[0],
+            "id": item[1],
+            "name": item[2],
+            "matched_name": item[3],
+            "matched_variant": item[4],
+        }
+        for item in top
+    ]
+
+
+def problem_to_json(problem):
+    row = problem["row"]
+    return {
+        "type": problem["type"],
+        "page": row["page"],
+        "row_index": row["row_index"],
+        "raw": row["raw"],
+        "top": printable_top(problem.get("top", [])),
+    }
+
+
+def duplicate_to_json(db_id, rows):
+    return {
+        "db_id": db_id,
+        "rows": [
+            {
+                "page": row["page"],
+                "row_index": row["row_index"],
+                "raw": row["raw"],
+            }
+            for row in rows
+        ],
+    }
+
+
+def report_data(collected):
+    report = []
+    for key in sorted(collected):
+        data = collected[key]
+        report.append(
+            {
+                "summary": summarize(data),
+                "problems": [problem_to_json(problem) for problem in data["problems"]],
+                "duplicates": [
+                    duplicate_to_json(db_id, rows) for db_id, rows in data["duplicates"].items()
+                ],
+                "missing_from_pdf_or_skipped": [
+                    {
+                        "ID": row["ID"],
+                        "SchoolOfGraduation": row["SchoolOfGraduation"],
+                        "SchoolFullNameJunior": row["SchoolFullNameJunior"],
+                        "SchoolTarget": row["SchoolTarget"],
+                        "SchoolFullName": row["SchoolFullName"],
+                        "PlanNum": row["PlanNum"],
+                    }
+                    for row in data["missing_db"]
+                ],
+            }
+        )
+    return report
+
+
+def write_report(collected):
+    with open(REPORT_FILE, "w", encoding="utf-8") as handle:
+        json.dump(report_data(collected), handle, ensure_ascii=False, indent=2, default=decimal_to_json)
+
+
+def write_backup(rows):
+    with open(BACKUP_FILE, "w", encoding="utf-8") as handle:
+        json.dump(rows, handle, ensure_ascii=False, indent=2, default=decimal_to_json)
+
+
+def print_summary(collected):
+    for key in sorted(collected):
+        summary = summarize(collected[key])
+        print(
+            summary["district_id"],
+            summary["district"],
+            summary["score_type"],
+            "parsed",
+            summary["parsed"],
+            "db",
+            summary["db"],
+            "updates",
+            summary["matched_for_update"],
+            "problems",
+            summary["problems"],
+            "dups",
+            summary["duplicate_target_rows"],
+            "missing",
+            summary["missing_from_pdf_or_skipped"],
+            "score_range",
+            summary["min_score_after_minus_50"],
+            summary["max_score_after_minus_50"],
+        )
+
+
+def update_scores(cursor, collected):
+    backup_rows = []
+    attempted = 0
+    changed = 0
+    for key in sorted(collected):
+        data = collected[key]
+        db_by_id = {row["ID"]: row for row in data["db_rows"]}
+        for update in data["updates"]:
+            db_row = db_by_id[update["ID"]]
+            if not compare_existing_scores(db_row, update):
+                raise RuntimeError(
+                    f"{data['district_name']} {data['score_type']} ID {update['ID']} "
+                    "already has different score data"
+                )
+            backup_rows.append(db_row)
+            cursor.execute(
+                """
+                UPDATE MPS_Score
+                SET ScoreTotal = %s,
+                    Score1 = %s,
+                    Score2 = %s,
+                    Score3 = %s,
+                    Score4 = %s,
+                    ScoreTotalDifferenceValue = %s
+                WHERE ID = %s
+                  AND ScoreYear = %s
+                  AND ScoreType = %s
+                  AND DistrictID = %s
+                """,
+                (
+                    update["ScoreTotal"],
+                    update["Score1"],
+                    update["Score2"],
+                    update["Score3"],
+                    update["Score4"],
+                    update["ScoreTotalDifferenceValue"],
+                    update["ID"],
+                    YEAR,
+                    data["score_type"],
+                    data["district_id"],
+                ),
+            )
+            attempted += 1
+            changed += cursor.rowcount
+    write_backup(backup_rows)
+    return attempted, changed
+
+
+def verify_database(cursor, collected):
+    mismatches = []
+    for key in sorted(collected):
+        data = collected[key]
+        expected = {row["ID"]: row for row in data["updates"]}
+        cursor.execute(
+            """
+            SELECT ID, ScoreTotal, Score1, Score2, Score3, Score4, ScoreTotalDifferenceValue
+            FROM MPS_Score
+            WHERE ScoreYear = %s
+              AND ScoreType = %s
+              AND DistrictID = %s
+            """,
+            (YEAR, data["score_type"], data["district_id"]),
+        )
+        actual_by_id = {row["ID"]: row for row in cursor.fetchall()}
+        for db_id, expected_row in expected.items():
+            actual = actual_by_id.get(db_id)
+            if not actual:
+                mismatches.append((data["district_id"], data["score_type"], db_id, "missing_actual"))
+                continue
+            for column in SCORE_COLUMNS:
+                if actual[column] != expected_row[column]:
+                    mismatches.append(
+                        (
+                            data["district_id"],
+                            data["score_type"],
+                            db_id,
+                            column,
+                            actual[column],
+                            expected_row[column],
+                        )
+                    )
+    if mismatches:
+        raise RuntimeError(f"verification failed: {mismatches[:10]}")
+
+
+def parse_args():
+    parser = argparse.ArgumentParser()
+    parser.add_argument("--dry-run", action="store_true")
+    parser.add_argument("--district", type=int, action="append", choices=sorted(DISTRICTS))
+    parser.add_argument("--score-type", choices=SCORE_TYPES, action="append")
+    return parser.parse_args()
+
+
+def main():
+    args = parse_args()
+    districts = args.district or sorted(DISTRICTS)
+    score_types = args.score_type or SCORE_TYPES
+
+    conn = pymysql.connect(**DB_CONFIG)
+    try:
+        with conn.cursor() as cursor:
+            collected = collect_all(cursor, districts, score_types)
+            write_report(collected)
+            print_summary(collected)
+            if args.dry_run:
+                conn.rollback()
+                print("dry_run_ok")
+                print("report", REPORT_FILE)
+                return
+            attempted, changed = update_scores(cursor, collected)
+            verify_database(cursor, collected)
+            conn.commit()
+            print("attempted_updates", attempted)
+            print("changed_rows", changed)
+            print("backup", BACKUP_FILE)
+            print("report", REPORT_FILE)
+    except Exception:
+        conn.rollback()
+        raise
+    finally:
+        conn.close()
+
+
+if __name__ == "__main__":
+    main()

秒过分数线数据导入/import_mps_score_huangpu_scores_2026.py

@@ -0,0 +1,470 @@
+import argparse
+import json
+import re
+from collections import defaultdict
+from decimal import Decimal
+
+import pdfplumber
+import pymysql
+
+
+DB_CONFIG = {
+    "host": "589ae8e08493d.sh.cdb.myqcloud.com",
+    "port": 8124,
+    "user": "cdb_outerroot",
+    "password": "kylx!@#!QAZ@WSX",
+    "database": "kylx365_db",
+    "charset": "utf8mb4",
+    "connect_timeout": 10,
+    "read_timeout": 30,
+    "write_timeout": 30,
+    "cursorclass": pymysql.cursors.DictCursor,
+}
+
+YEAR = "2026"
+PREVIOUS_YEAR = "2025"
+DISTRICT_ID = 1
+DISTRICT_NAME = "黄浦区"
+SCORE_TYPES = ("名额到区", "名额到校")
+PDF_PATHS = {
+    "名额到区": "/Volumes/程杰外接SD盘/上海中考招生计划/2026/分数线/名额到区/1.pdf",
+    "名额到校": "/Volumes/程杰外接SD盘/上海中考招生计划/2026/分数线/名额到校/1.pdf",
+}
+BACKUP_FILE = "mps_score_huangpu_scores_2026_backup.json"
+
+SCORE_COLUMNS = [
+    "ScoreTotal",
+    "Score1",
+    "Score2",
+    "Score3",
+    "Score4",
+    "ScoreTotalDifferenceValue",
+]
+
+
+def compact(value):
+    return re.sub(r"\s+", "", str(value or "").replace("\u3000", ""))
+
+
+def decimal_to_json(value):
+    if isinstance(value, Decimal):
+        return str(value)
+    return value
+
+
+def parse_decimal(value):
+    match = re.search(r"\d+(?:\.\d+)?", compact(value))
+    return Decimal(match.group(0)) if match else None
+
+
+def is_subsequence(needle, haystack):
+    iterator = iter(haystack)
+    return all(char in iterator for char in needle)
+
+
+def name_variants(name):
+    value = compact(name)
+    variants = []
+
+    def add(item, penalty):
+        item = compact(item)
+        if item and item not in [variant for variant, _penalty in variants]:
+            variants.append((item, penalty))
+
+    add(value, 0)
+    add(value.replace("上海市", "上海"), 80)
+    add(re.sub(r"[（(].*?[）)]", "", value), 500)
+    add(re.sub(r"[（(].*?[）)]", "", value).replace("上海市", "上海"), 580)
+    return variants
+
+
+def score_variant(candidate, raw):
+    if candidate == raw:
+        return 5000 + len(candidate)
+    if candidate in raw:
+        return 4000 + len(candidate)
+    if raw in candidate:
+        return 3500 + len(raw)
+    if is_subsequence(candidate, raw):
+        return 3000 + len(candidate)
+    if is_subsequence(raw, candidate):
+        return 2500 + len(raw)
+    return 0
+
+
+def match_name(raw_value, candidates):
+    raw = compact(raw_value)
+    scored = []
+    for key, name in candidates:
+        best_score = max(score_variant(variant, raw) - penalty for variant, penalty in name_variants(name))
+        if best_score:
+            scored.append((best_score, key, name))
+    scored.sort(reverse=True, key=lambda item: item[0])
+    if not scored:
+        return None, 0, []
+    best = scored[0]
+    ties = [item for item in scored if item[0] == best[0] and item[1] != best[1]]
+    if ties:
+        return None, best[0], scored[:5]
+    return best[1], best[0], scored[:5]
+
+
+def parse_pdf(score_type, path):
+    rows = []
+    with pdfplumber.open(path) as pdf:
+        for page_number, page in enumerate(pdf.pages, 1):
+            for table in page.extract_tables():
+                for row_index, row in enumerate(table):
+                    if not row or len(row) < 9:
+                        continue
+                    if any(
+                        token in compact(cell)
+                        for cell in row[:3]
+                        for token in ("录取最低分", "招生学校", "初中学校", "区名称")
+                    ):
+                        continue
+
+                    total_with_eval = parse_decimal(row[2])
+                    eval_score = parse_decimal(row[4])
+                    score1 = parse_decimal(row[5])
+                    score2 = parse_decimal(row[6])
+                    score3 = parse_decimal(row[7])
+                    score4 = parse_decimal(row[8])
+                    if None in (total_with_eval, eval_score, score1, score2, score3, score4):
+                        continue
+                    if eval_score != Decimal("50"):
+                        raise ValueError(
+                            f"{score_type} page {page_number} row {row_index} "
+                            f"has unexpected 综合素质评价 score: {eval_score}"
+                        )
+
+                    parsed = {
+                        "page": page_number,
+                        "row_index": row_index,
+                        "raw": row,
+                        "total_with_eval": total_with_eval,
+                        "ScoreTotal": total_with_eval - Decimal("50"),
+                        "Score1": score1,
+                        "Score2": score2,
+                        "Score3": score3,
+                        "Score4": score4,
+                    }
+                    if score_type == "名额到区":
+                        parsed["high_raw"] = row[1]
+                    else:
+                        parsed["junior_raw"] = row[0]
+                        parsed["high_raw"] = row[1]
+                    rows.append(parsed)
+    return rows
+
+
+def fetch_existing(cursor, score_type, year):
+    cursor.execute(
+        """
+        SELECT ID, ScoreYear, ScoreType, DistrictID, SchoolOfGraduation,
+               SchoolFullNameJunior, SchoolTarget, SchoolFullName, PlanNum,
+               ScoreTotal, Score1, Score2, Score3, Score4,
+               ScoreTotalDifferenceValue
+        FROM MPS_Score
+        WHERE ScoreYear = %s
+          AND ScoreType = %s
+          AND DistrictID = %s
+        ORDER BY ID
+        """,
+        (year, score_type, DISTRICT_ID),
+    )
+    return cursor.fetchall()
+
+
+def business_key(row, score_type):
+    if score_type == "名额到校":
+        return (int(row["SchoolOfGraduation"]), str(row["SchoolTarget"]))
+    return (str(row["SchoolTarget"]),)
+
+
+def previous_scores_by_key(cursor, score_type):
+    rows = fetch_existing(cursor, score_type, PREVIOUS_YEAR)
+    return {business_key(row, score_type): row["ScoreTotal"] for row in rows}
+
+
+def attach_matches(score_type, parsed_rows, db_rows):
+    problems = []
+    matched = []
+    used_ids = defaultdict(list)
+
+    if score_type == "名额到区":
+        high_candidates = [(row["ID"], row["SchoolFullName"]) for row in db_rows]
+        row_by_id = {row["ID"]: row for row in db_rows}
+        for parsed in parsed_rows:
+            db_id, match_score, top = match_name(parsed["high_raw"], high_candidates)
+            if not db_id or match_score < 2500:
+                problems.append({"type": "high_match", "row": parsed, "top": top})
+                continue
+            parsed["db_row"] = row_by_id[db_id]
+            parsed["match_score"] = match_score
+            matched.append(parsed)
+            used_ids[db_id].append(parsed)
+    else:
+        junior_candidates = sorted(
+            {
+                (row["SchoolOfGraduation"], row["SchoolFullNameJunior"])
+                for row in db_rows
+                if row["SchoolOfGraduation"] is not None
+            }
+        )
+        rows_by_junior = defaultdict(list)
+        row_by_id = {row["ID"]: row for row in db_rows}
+        for row in db_rows:
+            rows_by_junior[int(row["SchoolOfGraduation"])].append((row["ID"], row["SchoolFullName"]))
+
+        for parsed in parsed_rows:
+            junior_id, junior_score, junior_top = match_name(parsed["junior_raw"], junior_candidates)
+            if not junior_id or junior_score < 2500:
+                problems.append({"type": "junior_match", "row": parsed, "top": junior_top})
+                continue
+            db_id, high_score, high_top = match_name(parsed["high_raw"], rows_by_junior[junior_id])
+            if not db_id or high_score < 2500:
+                problems.append({"type": "high_match", "row": parsed, "junior_id": junior_id, "top": high_top})
+                continue
+            parsed["db_row"] = row_by_id[db_id]
+            parsed["junior_match_score"] = junior_score
+            parsed["high_match_score"] = high_score
+            matched.append(parsed)
+            used_ids[db_id].append(parsed)
+
+    duplicates = {db_id: rows for db_id, rows in used_ids.items() if len(rows) > 1}
+    missing_db = [row for row in db_rows if row["ID"] not in used_ids]
+    return matched, problems, duplicates, missing_db
+
+
+def build_updates(score_type, matched_rows, previous_scores):
+    updates = []
+    for parsed in matched_rows:
+        db_row = parsed["db_row"]
+        key = business_key(db_row, score_type)
+        previous = previous_scores.get(key)
+        diff = None if previous is None else parsed["ScoreTotal"] - previous
+        updates.append(
+            {
+                "ID": db_row["ID"],
+                "SchoolOfGraduation": db_row["SchoolOfGraduation"],
+                "SchoolFullNameJunior": db_row["SchoolFullNameJunior"],
+                "SchoolTarget": db_row["SchoolTarget"],
+                "SchoolFullName": db_row["SchoolFullName"],
+                "PlanNum": db_row["PlanNum"],
+                "ScoreTotal": parsed["ScoreTotal"],
+                "Score1": parsed["Score1"],
+                "Score2": parsed["Score2"],
+                "Score3": parsed["Score3"],
+                "Score4": parsed["Score4"],
+                "ScoreTotalDifferenceValue": diff,
+                "total_with_eval": parsed["total_with_eval"],
+                "page": parsed["page"],
+            }
+        )
+    return updates
+
+
+def compare_existing_scores(db_row, update):
+    for column in SCORE_COLUMNS:
+        old_value = db_row[column]
+        new_value = update[column]
+        if old_value in (None, Decimal("0.00")):
+            continue
+        if new_value != old_value:
+            return False
+    return True
+
+
+def print_summary(score_type, parsed_rows, db_rows, matched, problems, duplicates, missing_db, updates):
+    totals = [row["ScoreTotal"] for row in updates]
+    print(
+        score_type,
+        "parsed",
+        len(parsed_rows),
+        "db",
+        len(db_rows),
+        "matched",
+        len(matched),
+        "missing_db",
+        len(missing_db),
+        "problems",
+        len(problems),
+        "duplicates",
+        len(duplicates),
+    )
+    if totals:
+        print(score_type, "score_total_after_minus_50", min(totals), max(totals))
+    if problems:
+        for problem in problems[:10]:
+            print("problem", score_type, problem["type"], problem["row"]["raw"], problem.get("top"))
+    if duplicates:
+        for db_id, rows in list(duplicates.items())[:10]:
+            print("duplicate", score_type, db_id, [row["raw"] for row in rows])
+    if missing_db:
+        for row in missing_db[:20]:
+            print(
+                "missing_from_pdf",
+                score_type,
+                row["ID"],
+                row["SchoolFullNameJunior"],
+                "->",
+                row["SchoolFullName"],
+                "plan",
+                row["PlanNum"],
+            )
+    for row in updates[:5]:
+        print(
+            "sample",
+            score_type,
+            row["ID"],
+            row["SchoolFullNameJunior"],
+            "->",
+            row["SchoolFullName"],
+            "pdf_total",
+            row["total_with_eval"],
+            "stored",
+            row["ScoreTotal"],
+            "diff",
+            row["ScoreTotalDifferenceValue"],
+        )
+
+
+def collect_all(cursor):
+    collected = {}
+    for score_type in SCORE_TYPES:
+        parsed_rows = parse_pdf(score_type, PDF_PATHS[score_type])
+        db_rows = fetch_existing(cursor, score_type, YEAR)
+        previous_scores = previous_scores_by_key(cursor, score_type)
+        matched, problems, duplicates, missing_db = attach_matches(score_type, parsed_rows, db_rows)
+        updates = build_updates(score_type, matched, previous_scores)
+        collected[score_type] = {
+            "parsed_rows": parsed_rows,
+            "db_rows": db_rows,
+            "matched": matched,
+            "problems": problems,
+            "duplicates": duplicates,
+            "missing_db": missing_db,
+            "updates": updates,
+        }
+    return collected
+
+
+def validate(collected):
+    failures = []
+    for score_type, data in collected.items():
+        if data["problems"]:
+            failures.append(f"{score_type} has {len(data['problems'])} match problems")
+        if data["duplicates"]:
+            failures.append(f"{score_type} has {len(data['duplicates'])} duplicate target rows")
+        if len(data["updates"]) != len(data["matched"]):
+            failures.append(f"{score_type} update/match count mismatch")
+    if failures:
+        raise RuntimeError("; ".join(failures))
+
+
+def write_backup(rows):
+    with open(BACKUP_FILE, "w", encoding="utf-8") as handle:
+        json.dump(rows, handle, ensure_ascii=False, indent=2, default=decimal_to_json)
+
+
+def update_scores(cursor, collected):
+    backup_rows = []
+    update_count = 0
+    for score_type, data in collected.items():
+        db_by_id = {row["ID"]: row for row in data["db_rows"]}
+        for update in data["updates"]:
+            db_row = db_by_id[update["ID"]]
+            if not compare_existing_scores(db_row, update):
+                raise RuntimeError(f"{score_type} ID {update['ID']} already has different score data")
+            backup_rows.append(db_row)
+            cursor.execute(
+                """
+                UPDATE MPS_Score
+                SET ScoreTotal = %s,
+                    Score1 = %s,
+                    Score2 = %s,
+                    Score3 = %s,
+                    Score4 = %s,
+                    ScoreTotalDifferenceValue = %s
+                WHERE ID = %s
+                  AND ScoreYear = %s
+                  AND ScoreType = %s
+                  AND DistrictID = %s
+                """,
+                (
+                    update["ScoreTotal"],
+                    update["Score1"],
+                    update["Score2"],
+                    update["Score3"],
+                    update["Score4"],
+                    update["ScoreTotalDifferenceValue"],
+                    update["ID"],
+                    YEAR,
+                    score_type,
+                    DISTRICT_ID,
+                ),
+            )
+            update_count += cursor.rowcount
+    write_backup(backup_rows)
+    return update_count
+
+
+def verify_database(cursor, collected):
+    mismatches = []
+    for score_type, data in collected.items():
+        expected = {row["ID"]: row for row in data["updates"]}
+        cursor.execute(
+            """
+            SELECT ID, ScoreTotal, Score1, Score2, Score3, Score4, ScoreTotalDifferenceValue
+            FROM MPS_Score
+            WHERE ScoreYear = %s
+              AND ScoreType = %s
+              AND DistrictID = %s
+            """,
+            (YEAR, score_type, DISTRICT_ID),
+        )
+        actual_by_id = {row["ID"]: row for row in cursor.fetchall()}
+        for db_id, expected_row in expected.items():
+            actual = actual_by_id.get(db_id)
+            if not actual:
+                mismatches.append((score_type, db_id, "missing_actual"))
+                continue
+            for column in SCORE_COLUMNS:
+                if actual[column] != expected_row[column]:
+                    mismatches.append((score_type, db_id, column, actual[column], expected_row[column]))
+    if mismatches:
+        raise RuntimeError(f"verification failed: {mismatches[:10]}")
+
+
+def main():
+    parser = argparse.ArgumentParser()
+    parser.add_argument("--dry-run", action="store_true")
+    args = parser.parse_args()
+
+    conn = pymysql.connect(**DB_CONFIG)
+    try:
+        with conn.cursor() as cursor:
+            collected = collect_all(cursor)
+            for score_type in SCORE_TYPES:
+                data = collected[score_type]
+                print_summary(score_type, **data)
+            validate(collected)
+            if args.dry_run:
+                conn.rollback()
+                print("dry_run_ok")
+                return
+            update_count = update_scores(cursor, collected)
+            verify_database(cursor, collected)
+            conn.commit()
+            print("updated", update_count)
+            print("backup", BACKUP_FILE)
+    except Exception:
+        conn.rollback()
+        raise
+    finally:
+        conn.close()
+
+
+if __name__ == "__main__":
+    main()