kg-agent/scripts/sample_snapshot.py

"""示例：计算 青浦-北斗园区 2026-03 的月度快照字段。"""

from __future__ import annotations

import re
from pathlib import Path

import pandas as pd


HR_ROOT = Path(r"E:\CODE\knowledge_agent\data\20260529_申勤提供数据\人事部")
FIN = Path(r"E:\CODE\knowledge_agent\data\20260529_申勤提供数据\财务部\2026按合同物业收费进程26.5.17(1).xlsx")
TARGET_PROJECT = "青浦-北斗园区"
TARGET_MONTH = 3
YEAR = 2026


def _header_info(df: pd.DataFrame) -> tuple[str, int]:
    project, year = "", YEAR
    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_from_sheet(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:
    emp_ids: set[str] = set()
    att_rows = 0
    sums = {"平时加班小时": 0.0, "国定加班小时": 0.0, "餐费补助金额": 0.0,
            "值班费金额": 0.0, "加班超时费金额": 0.0, "国定加班费金额": 0.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:
            month = _month_from_sheet(sh)
            if month != TARGET_MONTH:
                continue
            try:
                df = pd.read_excel(xls, sheet_name=sh, header=None, dtype=str)
            except Exception:  # noqa: BLE001
                continue
            project, year = _header_info(df)
            if project != TARGET_PROJECT or year != YEAR:
                continue
            sub = df.iloc[4:].copy()
            sub.columns = range(sub.shape[1])
            sub = sub[sub[3].notna()]
            att_rows += len(sub)
            emp_ids.update(str(v).strip() for v in sub[1].dropna())
            for col, key in [(36, "平时加班小时"), (37, "国定加班小时"),
                             (38, "餐费补助金额"), (39, "加班超时费金额"),
                             (40, "国定加班费金额"), (41, "值班费金额")]:
                if sub.shape[1] > col:
                    sums[key] += pd.to_numeric(sub[col], errors="coerce").sum()

    print(f"== {TARGET_PROJECT} {YEAR}年{TARGET_MONTH}月 考勤快照字段 ==")
    print(f"在岗人数: {len(emp_ids)}（当月有考勤记录的不同工号）")
    print(f"考勤人次: {att_rows}")
    for k, v in sums.items():
        print(f"{k}: {v:.2f}")

    # 财务：该项目 3 月物业费
    df = pd.read_excel(FIN, header=None, dtype=str)
    data = df.iloc[4:].copy()
    data.columns = range(data.shape[1])
    fin = data[data[1].astype(str).str.contains(TARGET_PROJECT, na=False)]
    print(f"\n== 财务（项目名含 '{TARGET_PROJECT}' 的合同行）==")
    if len(fin):
        amt = pd.to_numeric(fin[24], errors="coerce").sum()  # 24 = 3月物业费列
        print(f"3月物业费(开票口径)合计: {amt:.2f}（{len(fin)} 个合同行）")
        print("涉及合同:", fin[3].dropna().tolist())
    else:
        print("财务表中未找到该项目（需核对项目简称）")


if __name__ == "__main__":
    main()