别再手动整理工作台账了，这套 AI 流水线让我每天多出 2 小时

# collect.py —— 把 raw 目录下的所有表读成一个 DataFrame
import pandas as pd
from pathlib import Path

SOURCE_DIR = Path("./raw")

def load_all() -> pd.DataFrame:
    frames = []
    for f in SOURCE_DIR.iterdir():
        try:
            if f.suffix.lower() in (".xlsx", ".xls"):
                df = pd.read_excel(f)
            elif f.suffix.lower() == ".csv":
                df = pd.read_csv(f, encoding="utf-8-sig")
            else:
                continue
        except Exception as e:
            print(f"[跳过] {f.name} 读不了：{e}")
            continue
        df["__来源"] = f.name   # 留个来源标记，后面排查问题靠它
        frames.append(df)
        print(f"[读取] {f.name}  {len(df)} 行")

    if not frames:
        raise SystemExit("raw 目录里没有可用文件")
    return pd.concat(frames, ignore_index=True)

if __name__ == "__main__":
    raw = load_all()
    print(f"合计 {len(raw)} 条原始记录")
    raw.to_pickle("./cache/raw.pkl")

# normalize.py —— 统一口径、去重、排序
import re
import pandas as pd

def normalize(df: pd.DataFrame) -> pd.DataFrame:
    # 1. 字段名先统一：按关键字模糊映射到一套标准列名
    df = df.rename(columns=lambda c: str(c).strip())
    col_map = {}
    for c in df.columns:
        key = re.sub(r"[\s（）()【】:：]", "", c)
        if any(k in key for k in ("时间", "日期")):
            col_map[c] = "时间"
        elif any(k in key for k in ("事项", "内容", "描述", "摘要")):
            col_map[c] = "事项"
        elif "负责人" in key or ("人" in key and "创建" not in key):
            col_map[c] = "负责人"
        elif any(k in key for k in ("状态", "进度", "结果")):
            col_map[c] = "状态"
    df = df.rename(columns=col_map)

    # 2. 缺的列补空，保证后面不会 KeyError
    for c in ["时间", "事项", "负责人", "状态"]:
        if c not in df.columns:
            df[c] = ""

    # 3. 时间统一成 datetime，解析不了的（比如"上周"）直接丢掉
    df["时间"] = pd.to_datetime(df["时间"], errors="coerce")
    before = len(df)
    df = df.dropna(subset=["时间"])
    print(f"时间解析失败丢掉 {before - len(df)} 条")

    # 4. 事项去空白、扔掉空行
    df["事项"] = df["事项"].astype(str).str.strip()
    df = df[df["事项"].str.len() > 1]

    # 5. 去重：同一时间同一件事同一个人，只留一条
    df = df.drop_duplicates(subset=["时间", "事项", "负责人"])

    return df[["时间", "事项", "负责人", "状态", "__来源"]] \
             .sort_values("时间").reset_index(drop=True)

# classify.py —— 调 AI 打分类和重要度
import os, json, time
from openai import OpenAI

client = OpenAI(
    api_key=os.getenv("AI_API_KEY"),
    base_url=os.getenv("AI_BASE_URL"),   # 换成你自己用的服务地址
)

CATEGORIES = ["商品运营", "供应链", "客户与售后", "数据与报表", "会议与协同", "临时事务"]

PROMPT = """你是公司运营岗的老同事，熟悉业务口径。
下面是一批一天内的工作流水，请逐条打上分类和重要度。

【分类】只能从这几个里选：{cats}
【重要度】只能是三个值之一：高 / 中 / 低

判断重要度的标准：
- 高：涉及钱、客户、交付时间，或者领导大概率会追问的事
- 中：常规推进中的事项，做完了但还没结果
- 低：行政、通知、搬运、例行动作，丢了也没人问

注意：
- 宁可标中，不要乱标高。全是高的台账没有意义。
- 看不懂在说什么的，分类给"临时事务"，重要度给"低"。

只输出 JSON 数组，不要任何解释文字，不要 markdown 代码块：
[{{"i": 0, "cat": "商品运营", "level": "高"}}]

流水：
{items}
"""

def classify_batch(items: list[str]) -> list[dict]:
    payload = "\n".join(f"{i}. {t}" for i, t in enumerate(items))
    resp = client.chat.completions.create(
        model=os.getenv("AI_MODEL", "gpt-4o-mini"),
        messages=[{"role": "user",
                   "content": PROMPT.format(cats="、".join(CATEGORIES), items=payload)}],
        temperature=0,
    )
    txt = resp.choices[0].message.content.strip()
    txt = txt.replace("```json", "").replace("```", "").strip()
    return json.loads(txt)

# render.py —— 生成最终台账 Excel
import pandas as pd

def render(df: pd.DataFrame, out_path: str):
    with pd.ExcelWriter(out_path, engine="openpyxl") as w:
        df.to_excel(w, sheet_name="明细", index=False)

        # 分类 × 重要度 的交叉汇总
        pivot = (df.pivot_table(index="分类", columns="重要度",
                                values="事项", aggfunc="count", fill_value=0)
                   .reset_index())
        pivot.to_excel(w, sheet_name="分类汇总", index=False)

        # 只留重点，按时间倒序，领导只看这页
        key = (df[df["重要度"] == "高"]
               .sort_values("时间", ascending=False))
        key.to_excel(w, sheet_name="重点事项", index=False)

    # 控制台也打一份摘要，方便直接复制到群里
    print(f"台账已生成：{out_path}（共 {len(df)} 条，重点 {len(key)} 条）")
    print("\n".join(f"  [{r.重要度}] {r.时间:%m-%d} {r.事项}"
                    for r in key.itertuples()))

# main.py
import time
from datetime import datetime
from collect import load_all
from normalize import normalize
from classify import classify_batch
from render import render

BATCH = 30   # 每批最多30条，太多 AI 容易串行号

def main():
    df = normalize(load_all())
    print(f"清洗后有效记录 {len(df)} 条")

    items = df["事项"].tolist()
    cats   = ["未分类"] * len(items)
    levels = ["中"]     * len(items)

    for s in range(0, len(items), BATCH):
        batch = items[s:s + BATCH]
        try:
            for r in classify_batch(batch):
                idx = s + int(r["i"])
                cats[idx]   = r.get("cat", "临时事务")
                levels[idx] = r.get("level", "中")
        except Exception as e:
            # 失败了不要让整条流程挂掉，标记出来人工补
            print(f"[警告] 第 {s // BATCH + 1} 批处理失败：{e}")
        time.sleep(1)   # 简单限流，别把接口打爆

    df["分类"], df["重要度"] = cats, levels
    out = f"./out/工作台账_{datetime.now():%Y%m%d}.xlsx"
    render(df, out)

if __name__ == "__main__":
    main()

@echo off
cd /d D:\workspace\ledger
python main.py >> run.log 2>&1

微信导出：7月18日 | 跟B供应商确认发货时间 | 我 | 已沟通
钉钉审批：2026-07-18 15:20 | 申请E款补货300件 | 我 | 审批通过
邮件摘要：7.18 | 回复客户张总关于交期的询问 | 我 | 已回复
微信导出：7月18日 | 在群里同步了本周数据 | 我 | 已发
备忘录：7-18 | 帮同事导了一份历史订单表 | 我 | 完成