kg-agent/scripts/analyze_renewal_folders.py

"""扫描项目档案目录，识别合同期文件夹并生成续签链清单（CSV）。"""

from __future__ import annotations

import csv
import re
import sys
from datetime import date, timedelta
from pathlib import Path


sys.stdout.reconfigure(encoding="utf-8")
ROOT = Path(r"E:\申勤物业项目资料")
OUT = Path(r"E:\CODE\knowledge_agent\output\续签链-档案目录.csv")
PAT = re.compile(r"^(\d{4})\.(\d{1,2})\.(\d{1,2})-(\d{4})\.(\d{1,2})\.(\d{1,2})(.*)$")


def _d(y: str, m: str, d: str) -> date:
    return date(int(y), int(m), int(d))


def _norm_subject(name: str) -> str:
    """归一化合同主体：去掉扩展名、点招/补充标记与空白。"""
    s = name
    for suf in (".pdf", ".doc", ".docx", ".xls", ".xlsx"):
        s = s.replace(suf, "")
    s = re.sub(r"[（(]?(点招|补充)[）)]?", "", s)
    s = re.sub(r"[—-].*(点招|补充)$", "", s)
    return re.sub(r"\s+", "", s)


def main() -> None:
    rows = []
    chains = 0
    period_total = 0
    for region in sorted(p for p in ROOT.iterdir() if p.is_dir() and not p.name.startswith(("6-", "7-"))):
        for proj in sorted(p for p in region.iterdir() if p.is_dir()):
            periods = []
            # 项目目录下可能直接是合同期文件夹，也可能再套一层项目名
            for cand in sorted(p for p in proj.iterdir() if p.is_dir()):
                m = PAT.match(cand.name)
                if m:
                    periods.append((cand.name, m))
                else:
                    for sub in sorted(p for p in cand.iterdir() if p.is_dir()):
                        m2 = PAT.match(sub.name)
                        if m2:
                            periods.append((sub.name, m2))
            if not periods:
                continue
            periods.sort(key=lambda x: (_d(x[1].group(1), x[1].group(2), x[1].group(3)), x[0]))
            period_total += len(periods)
            num = proj.name.split("-")[0] if "-" in proj.name else ""
            prev: tuple[date, str, str] | None = None
            for name, m in periods:
                start = _d(m.group(1), m.group(2), m.group(3))
                end = _d(m.group(4), m.group(5), m.group(6))
                subject = _norm_subject(m.group(7))
                renewal = ""
                if prev is not None:
                    contig = start <= prev[0] + timedelta(days=3)
                    same_subj = prev[2] == subject or not prev[2] or not subject
                    if contig and same_subj:
                        renewal = f"是（上一期: {prev[1]}）"
                        chains += 1
                rows.append([
                    region.name, proj.name, num, name,
                    start.isoformat(), end.isoformat(), subject, renewal,
                ])
                prev = (end, name, subject)

    with OUT.open("w", newline="", encoding="utf-8-sig") as f:
        w = csv.writer(f)
        w.writerow(["区域", "项目目录", "目录编号", "合同期文件夹", "开始日期", "结束日期", "合同主体(归一)", "续签自上一期"])
        w.writerows(rows)

    print(f"扫描根目录: {ROOT}")
    print(f"合同期文件夹总数: {period_total}")
    print(f"识别到续签衔接: {chains} 条")
    print(f"输出: {OUT}")
    # 示例
    print("\n示例（青浦区图书馆）:")
    for r in rows:
        if "102-青浦区图书馆" in r[1]:
            print("  ", " | ".join(r[2:]))


if __name__ == "__main__":
    main()