kg-agent/scripts/analyze_shift_mapping.py

"""分析排班表的组织代码 → 项目 映射（用考勤表交叉比对推断）。"""

from __future__ import annotations

import re
import sys
from pathlib import Path

import pandas as pd


sys.stdout.reconfigure(encoding="utf-8")
HR_ROOT = Path(r"E:\CODE\knowledge_agent\reference\20260529_申勤提供数据\人事部")


def _norm_project(name: str) -> str:
    return name.removeprefix("青浦-").strip()


def _header_info(df: pd.DataFrame) -> tuple[str, int]:
    project, year = "", 2026
    for r in range(min(3, len(df))):
        for c in range(min(12, df.shape[1])):
            v = df.iat[r, c]
            if pd.isna(v):
                continue
            s = str(v)
            m = re.search(r"项目[:：]\s*(.+)", s)
            if m and not project:
                project = m.group(1).strip()
            ym = re.search(r"20(\d{2})\s*年", s)
            if ym:
                year = 2000 + int(ym.group(1))
    return project, year


def _month(sh: str) -> int | None:
    m = re.search(r"(\d{1,2})\s*月", sh)
    return int(m.group(1)) if m else None


def main() -> None:
    # 1) 考勤：工号 -> {(项目, 月)}
    att: dict[str, set[tuple[str, int]]] = {}
    for p in sorted(HR_ROOT.rglob("*.xls")):
        try:
            xls = pd.ExcelFile(p)
        except Exception:  # noqa: BLE001
            continue
        for sh in xls.sheet_names:
            month = _month(sh)
            if month is None:
                continue
            try:
                df = pd.read_excel(xls, sheet_name=sh, header=None, dtype=str)
            except Exception:  # noqa: BLE001
                continue
            project, _ = _header_info(df)
            project = _norm_project(project)
            sub = df.iloc[4:].copy()
            sub.columns = range(sub.shape[1])
            for emp in sub[1].dropna():
                e = str(emp).strip()
                if e:
                    att.setdefault(e, set()).add((project, month))

    # 2) 排班：组织代码 -> {工号集合}，按文件分组
    org_by_file: dict[str, dict[str, set[str]]] = {}
    for p in sorted(HR_ROOT.rglob("排班*.xlsx")):
        try:
            xls = pd.ExcelFile(p)
        except Exception:  # noqa: BLE001
            continue
        orgs: dict[str, set[str]] = {}
        for sh in xls.sheet_names:
            if sh == "班次列表":
                continue
            try:
                df = pd.read_excel(xls, sheet_name=sh, header=None, dtype=str)
            except Exception:  # noqa: BLE001
                continue
            if df.shape[1] < 3:
                continue
            sub = df.iloc[2:].copy()
            sub.columns = range(sub.shape[1])
            for _, row in sub.iterrows():
                oc = str(row[0]).strip() if pd.notna(row[0]) else ""
                emp = str(row[2]).strip() if pd.notna(row[2]) else ""
                if oc and emp:
                    orgs.setdefault(oc, set()).add(emp)
        org_by_file[p.name] = orgs

    # 3) 交叉比对：组织代码的员工在考勤里属于哪些项目
    print("== 组织代码 → 项目 推断（考勤交叉比对）==")
    total_orgs = 0
    resolved = 0
    for fname, orgs in org_by_file.items():
        print(f"\n### {fname[:30]}…  组织代码数: {len(orgs)}")
        for oc, emps in sorted(orgs.items()):
            total_orgs += 1
            projs = set()
            for e in emps:
                for (pj, _m) in att.get(e, set()):
                    projs.add(pj)
            if len(projs) == 1:
                resolved += 1
                status = "✓ 唯一"
            elif len(projs) > 1:
                status = f"? 多项目 {sorted(projs)}"
            else:
                status = "✗ 无法推断（考勤中无此人）"
            print(f"  {oc}: {status}  员工数: {len(emps)}")
    print(f"\n组织代码总数: {total_orgs}，唯一推断成功: {resolved}")


if __name__ == "__main__":
    main()