kg-agent/scripts/analyze_temporal.py

"""分析考勤数据中的人员-项目-时间分布，验证时间建模方案。"""

from __future__ import annotations

import re
from pathlib import Path

import pandas as pd


HR_ROOT = Path(r"E:\CODE\knowledge_agent\data\20260529_申勤提供数据\人事部")
MONTH_PAT = re.compile(r"(\d{1,2})\s*月")


def _month_from_sheet(sh: str, default: int | None = None) -> int | None:
    m = MONTH_PAT.search(sh)
    return int(m.group(1)) if m else default


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 main() -> None:
    # employee -> {project: set(months)}
    emp_proj: dict[str, dict[str, set[int]]] = {}
    # project -> {month: employee count}
    proj_month: dict[str, dict[int, set[str]]] = {}
    total_rows = 0

    for p in sorted(HR_ROOT.rglob("*.xls")):
        try:
            xls = pd.ExcelFile(p)
        except Exception:  # noqa: BLE001
            continue
        for sh in xls.sheet_names:
            try:
                df = pd.read_excel(xls, sheet_name=sh, header=None, dtype=str)
            except Exception:  # noqa: BLE001
                continue
            if df.shape[0] < 5 or df.shape[1] < 4:
                continue
            project, year = _header_info(df)
            if not project:
                project = p.name
            month = _month_from_sheet(sh)
            sub = df.iloc[4:].copy()
            sub.columns = range(sub.shape[1])
            sub = sub[sub[3].notna()]
            for _, row in sub.iterrows():
                emp_id = str(row[1]).strip() if pd.notna(row[1]) else ""
                name = str(row[3]).strip()
                key = emp_id or f"{project}-{name}"
                total_rows += 1
                emp_proj.setdefault(key, {}).setdefault(project, set()).add(month or -1)
                proj_month.setdefault(project, {}).setdefault(month or -1, set()).add(key)

    print(f"考勤记录总数: {total_rows}")
    print(f"考勤文件涉及的唯一员工(按工号/姓名+项目): {len(emp_proj)}")

    multi = {k: v for k, v in emp_proj.items() if len(v) > 1}
    print(f"\n== 跨多个项目的人员数: {len(multi)} ==")
    for k, v in sorted(multi.items())[:20]:
        detail = ", ".join(f"{pj}({sorted(ms)})" for pj, ms in v.items())
        print(f"  {k}: {detail}")

    print(f"\n== 项目 × 月份 覆盖（人数） ==")
    for pj, months in sorted(proj_month.items()):
        cells = ", ".join(f"{m}月:{len(s)}人" for m, s in sorted(months.items()))
        print(f"  {pj}: {cells}")

    all_months = {m for pj in proj_month.values() for m in pj}
    print(f"\n涉及月份: {sorted(all_months)}")


if __name__ == "__main__":
    main()