sj_nmtybt_server/scripts/backfill-ssc-from-excel.py

# -*- coding: utf-8 -*-
"""从「已匹配200编号」明细回填人员底数社保卡号（c_social_insurance_code）。

匹配顺序：身份证号 → 银行卡号。仅补充库中空值，不覆盖已有不同卡号。
凭据只从环境变量读取，勿写入仓库。

  python -X utf8 scripts/backfill-ssc-from-excel.py
  python -X utf8 scripts/backfill-ssc-from-excel.py --apply
"""
from __future__ import annotations

import argparse
import csv
import json
import os
import re
import sys
import threading
import time
import urllib.error
import urllib.parse
import urllib.request
from collections import defaultdict
from concurrent.futures import ThreadPoolExecutor, as_completed
from pathlib import Path

from openpyxl import load_workbook

sys.stdout.reconfigure(encoding="utf-8")

HOST = os.environ.get("SJ_DMS_HOST", "121.43.55.7")
PORT = int(os.environ.get("SJ_DMS_PORT", "2101"))
BASE = f"http://{HOST}:{PORT}"
COLUMN_ID = "1847"
MODEL_ID = "1970"
DEFAULT_XLSX = Path(r"C:\Users\Liumouren\Desktop\松江区老年农民土地退养补助金发放人员明细-已匹配200编号.xlsx")
REPORT_DIR = Path(r"C:\Users\Liumouren\Desktop")

_token_lock = threading.Lock()
_token = {"value": "", "ts": 0.0}


def digits(v) -> str:
    if v is None:
        return ""
    if isinstance(v, float):
        if v != v:  # NaN
            return ""
        s = format(v, ".0f")
    elif isinstance(v, int):
        s = str(v)
    else:
        s = str(v).strip()
        if s.lower() in ("none", "null", "nan"):
            return ""
        if re.fullmatch(r"\d+\.0+", s):
            s = s.split(".", 1)[0]
    return re.sub(r"\D", "", s)


def norm_id(v) -> str:
    if v is None:
        return ""
    s = str(v).strip().replace(" ", "").replace("\u3000", "").upper()
    if re.fullmatch(r"\d+\.0+", s):
        s = s.split(".", 1)[0]
    return s


def norm_ssc(v) -> str:
    s = digits(v)
    return s if len(s) >= 10 else ""


def post_form(path: str, fields: dict, token: str | None = None, timeout: int = 90) -> dict:
    body = urllib.parse.urlencode({k: "" if v is None else str(v) for k, v in fields.items()}).encode("utf-8")
    headers = {
        "Content-Type": "application/x-www-form-urlencoded",
        "Content-Length": str(len(body)),
    }
    if token:
        headers["Token"] = token
        headers["token"] = token
    req = urllib.request.Request(BASE + path, data=body, method="POST", headers=headers)
    try:
        with urllib.request.urlopen(req, timeout=timeout) as resp:
            raw = resp.read().decode("utf-8", "replace")
    except urllib.error.HTTPError as e:
        raw = e.read().decode("utf-8", "replace")
    try:
        return json.loads(raw) if raw else {}
    except json.JSONDecodeError:
        return {"code": 599, "message": raw[:300]}


def login(force: bool = False) -> str:
    now = time.time()
    with _token_lock:
        if not force and _token["value"] and now - _token["ts"] < 20 * 60:
            return _token["value"]
        user = os.environ.get("SJ_OAUTH_USER", "user_liu")
        password = os.environ.get("SJ_OAUTH_PASSWORD", "")
        client_id = os.environ.get("SJ_OAUTH_CLIENT_ID", "1")
        if not password:
            raise SystemExit("缺少 SJ_OAUTH_PASSWORD")
        res = post_form(
            "/proxy_oauth/user/login",
            {"userName": user, "password": password, "clientId": client_id},
        )
        if res.get("code") != 200 or not res.get("message"):
            raise SystemExit("OAuth 登录失败: " + json.dumps(res, ensure_ascii=False)[:400])
        _token["value"] = res["message"]
        _token["ts"] = now
        return _token["value"]


def fetch_page(page: int, page_size: int = 100) -> list[dict]:
    token = login()
    res = post_form(
        "/proxy_dms/content/selectContentList",
        {
            "columnId": COLUMN_ID,
            "modelId": MODEL_ID,
            "page": str(page),
            "pageSize": str(page_size),
            "states": "0,3",
        },
        token,
    )
    code = res.get("code")
    if code in (208, 212):
        token = login(True)
        res = post_form(
            "/proxy_dms/content/selectContentList",
            {
                "columnId": COLUMN_ID,
                "modelId": MODEL_ID,
                "page": str(page),
                "pageSize": str(page_size),
                "states": "0,3",
            },
            token,
        )
        code = res.get("code")
    if code == 202:
        return []
    if code != 200:
        raise RuntimeError(f"selectContentList page={page} {res}")
    return (res.get("content") or {}).get("data") or []


def load_personnel(workers: int = 8) -> list[dict]:
    first = fetch_page(0)
    rows = list(first)
    if len(first) < 100:
        return rows
    page = 1
    while True:
        batch_pages = list(range(page, page + workers))
        got_short = False
        with ThreadPoolExecutor(max_workers=workers) as ex:
            futs = {ex.submit(fetch_page, p): p for p in batch_pages}
            by_page = {}
            for fut in as_completed(futs):
                by_page[futs[fut]] = fut.result()
        for p in batch_pages:
            chunk = by_page[p]
            rows.extend(chunk)
            if len(chunk) < 100:
                got_short = True
                break
        if got_short:
            break
        page += workers
        if page > 800:
            break
        print(f"  已拉人员 {len(rows)} ...", flush=True)
    return rows


def load_excel(path: Path) -> list[dict]:
    wb = load_workbook(path, read_only=True, data_only=True)
    ws = wb[wb.sheetnames[0]]
    out = []
    for i, row in enumerate(ws.iter_rows(min_row=2, values_only=True), start=2):
        cells = list(row) + [None] * 9
        town, village, name, idn, ssc, bank, card, phone, amt = cells[:9]
        item = {
            "row": i,
            "town": "" if town is None else str(town).strip(),
            "village": "" if village is None else str(village).strip(),
            "name": "" if name is None else str(name).strip(),
            "id_number": norm_id(idn),
            "ssc": norm_ssc(ssc),
            "card": digits(card),
        }
        out.append(item)
    wb.close()
    return out


def index_people(people: list[dict]):
    by_id = defaultdict(list)
    by_card = defaultdict(list)
    for p in people:
        dms_id = str(p.get("id") or "").strip()
        if not dms_id:
            continue
        rec = {
            "dms_id": dms_id,
            "biz_id": str(p.get("c_biz_id") or "").strip(),
            "name": str(p.get("c_name") or "").strip(),
            "id_number": norm_id(p.get("c_id_number")),
            "card": digits(p.get("c_bank_card_number")),
            "ssc": norm_ssc(p.get("c_social_insurance_code")),
            "town": str(p.get("c_town_id") or "").strip(),
            "village": str(p.get("c_village_id") or "").strip(),
        }
        if rec["id_number"]:
            by_id[rec["id_number"]].append(rec)
        if rec["card"]:
            by_card[rec["card"]].append(rec)
    return by_id, by_card


def unique_or_none(items: list[dict]) -> tuple[dict | None, str]:
    if not items:
        return None, "none"
    ids = {x["dms_id"] for x in items}
    if len(ids) > 1:
        return None, "multi"
    return items[0], "one"


def match_one(row: dict, by_id, by_card):
    id_hit, id_st = unique_or_none(by_id.get(row["id_number"], []) if row["id_number"] else [])
    card_hit, card_st = unique_or_none(by_card.get(row["card"], []) if row["card"] else [])
    if id_st == "multi":
        return None, "id_multi"
    if card_st == "multi" and id_st != "one":
        return None, "card_multi"
    if id_st == "one" and card_st == "one" and id_hit["dms_id"] != card_hit["dms_id"]:
        return None, "id_card_mismatch"
    if id_st == "one":
        return id_hit, "id"
    if card_st == "one":
        return card_hit, "card"
    return None, "unmatched"


def update_ssc(dms_id: str, ssc: str) -> tuple[bool, str]:
    payload = {"id": dms_id, "c_social_insurance_code": ssc}
    last = ""
    for attempt in range(4):
        try:
            token = login(attempt > 0)
            res = post_form(
                "/proxy_dms/content/updateContent",
                {
                    "columnId": COLUMN_ID,
                    "modelId": MODEL_ID,
                    "content": json.dumps(payload, ensure_ascii=False),
                },
                token,
            )
            code = res.get("code")
            if code in (208, 212):
                login(True)
                continue
            if code == 200:
                return True, "ok"
            last = str(res.get("message") or code)
            if code in (500, 599) or "超时" in last:
                time.sleep(1.5 * (attempt + 1))
                continue
            return False, last
        except Exception as e:
            last = str(e)
            time.sleep(1.5 * (attempt + 1))
    return False, last


def write_csv(path: Path, rows: list[dict], fields: list[str]):
    with path.open("w", encoding="utf-8-sig", newline="") as f:
        w = csv.DictWriter(f, fieldnames=fields, extrasaction="ignore")
        w.writeheader()
        for r in rows:
            w.writerow(r)


def main():
    ap = argparse.ArgumentParser()
    ap.add_argument("--xlsx", default=str(DEFAULT_XLSX))
    ap.add_argument("--apply", action="store_true")
    ap.add_argument("--workers", type=int, default=6)
    args = ap.parse_args()
    xlsx = Path(args.xlsx)
    if not xlsx.exists():
        raise SystemExit("找不到 Excel: " + str(xlsx))

    print("解析 Excel ...", flush=True)
    excel_rows = load_excel(xlsx)
    with_ssc = [r for r in excel_rows if r["ssc"]]
    print(f"  明细 {len(excel_rows)} 行，有社保卡 {len(with_ssc)}", flush=True)

    print("登录并拉取人员底数 ...", flush=True)
    login(True)
    people = load_personnel()
    print(f"  库中人员 {len(people)}", flush=True)
    by_id, by_card = index_people(people)
    db_empty = sum(1 for p in people if not norm_ssc(p.get("c_social_insurance_code")))
    print(f"  库中社保卡为空 {db_empty}", flush=True)

    planned = {}
    details = []
    stats = defaultdict(int)
    for row in with_ssc:
        person, how = match_one(row, by_id, by_card)
        stats[how] += 1
        if person is None:
            details.append({**row, "result": how, "biz_id": "", "db_ssc": "", "action": "skip"})
            continue
        cur = person["ssc"]
        action = "skip"
        if not cur:
            action = "fill"
        elif cur == row["ssc"]:
            action = "same"
            stats["same"] += 1
        else:
            action = "conflict_diff"
            stats["conflict_diff"] += 1
        rec = {
            **row,
            "result": how,
            "biz_id": person["biz_id"],
            "dms_id": person["dms_id"],
            "db_name": person["name"],
            "db_ssc": cur,
            "action": action,
        }
        details.append(rec)
        if action == "fill":
            prev = planned.get(person["dms_id"])
            if prev and prev["ssc"] != row["ssc"]:
                rec["action"] = "conflict_multi_excel"
                stats["conflict_multi_excel"] += 1
                stats["fill"] -= 1
                planned.pop(person["dms_id"], None)
            else:
                planned[person["dms_id"]] = rec
                stats["fill"] += 1

    summary = {
        "excel_rows": len(excel_rows),
        "excel_with_ssc": len(with_ssc),
        "db_people": len(people),
        "db_ssc_empty": db_empty,
        "match": {k: stats[k] for k in sorted(stats)},
        "to_fill_unique_people": len(planned),
        "apply": bool(args.apply),
    }
    print(json.dumps(summary, ensure_ascii=False, indent=2), flush=True)

    applied = []
    failed = []
    if args.apply and planned:
        items = list(planned.values())
        print(f"开始回填 {len(items)} 人 ...", flush=True)
        done = 0
        with ThreadPoolExecutor(max_workers=max(1, args.workers)) as ex:
            futs = {ex.submit(update_ssc, rec["dms_id"], rec["ssc"]): rec for rec in items}
            for fut in as_completed(futs):
                rec = futs[fut]
                ok, msg = fut.result()
                done += 1
                if ok:
                    applied.append(rec)
                else:
                    rec = dict(rec)
                    rec["error"] = msg
                    failed.append(rec)
                if done % 200 == 0 or done == len(items):
                    print(f"  进度 {done}/{len(items)} 成功 {len(applied)} 失败 {len(failed)}", flush=True)
        summary["applied"] = len(applied)
        summary["failed"] = len(failed)

    stamp = time.strftime("%Y%m%d-%H%M%S")
    report_json = REPORT_DIR / f"社保卡号回填报告-{stamp}.json"
    report_csv = REPORT_DIR / f"社保卡号回填明细-{stamp}.csv"
    report_json.write_text(json.dumps(summary, ensure_ascii=False, indent=2), encoding="utf-8")
    write_csv(
        report_csv,
        details,
        [
            "row", "town", "village", "name", "id_number", "card", "ssc",
            "result", "biz_id", "db_name", "db_ssc", "action",
        ],
    )
    if failed:
        fail_csv = REPORT_DIR / f"社保卡号回填失败-{stamp}.csv"
        write_csv(
            fail_csv,
            failed,
            ["row", "town", "village", "name", "id_number", "ssc", "biz_id", "dms_id", "error"],
        )
        print("失败清单", fail_csv)
    print("报告", report_json)
    print("明细", report_csv)


if __name__ == "__main__":
    main()