wandering1/Auto-Finance
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

重构项目为LLM 驱动

Browse Source
This commit is contained in:
wandering
2026-06-09 16:25:20 +08:00
parent 0074975591
commit 98e3d21c83
52 changed files with 7405 additions and 1485 deletions
Download Patch File Download Diff File Expand all files Collapse all files

273
src/doc/invoice.py Normal file
View File

@@ -0,0 +1,273 @@
"""发票数据模型与 CSV 工具
定义发票类型常量、CSV 列结构，提供发票分类和 CSV 读写功能。
对外接口:
INVOICE_LEVEL_COLUMNS 发票级别 CSV 列定义
PAYMENT_RECORD_COLUMNS 支付记录级别 CSV 列定义
INVOICE_TYPE_* 发票类型常量
is_travel_invoice(type) 判断是否为差旅发票
classify_invoice_batch(invoices) 按类型分组
load_csv(path) 读取支付记录 CSV
save_csv(payment_records, path) 保存支付记录 CSV
save_invoice_csv(payment_records, path) 保存发票级别 CSV
"""
import csv
import json
from pathlib import Path
from .. import get_logger
log = get_logger("invoice")
# ------------------------------------------------------------------
# CSV 列定义
# ------------------------------------------------------------------
# 发票级别 CSV 列（用于 invoice_summary.csv，每行一张发票）
INVOICE_LEVEL_COLUMNS = [
"序号",
"发票类型",
"发票号码",
"开票日期",
"项目名称",
"规格型号",
"价税合计",
"销售方名称",
"出发站",
"到达站",
"车次",
"乘车日期",
"座位等级",
"人员姓名",
"刷卡日期",
"公务卡号",
"刷卡金额",
"备注",
"工号",
]
# 支付记录级别 CSV 列（用于 payment_records.csv，每行一笔支付）
PAYMENT_RECORD_COLUMNS = [
"序号",
# 支付信息
"刷卡日期",
"公务卡号",
"刷卡金额",
# 发票聚合信息
"关联发票数",
"发票详情", # 格式: 类型[号码]¥金额 | 类型[号码]¥金额
"备注",
# 内部字段（用于下游解析）
"_invoices_json", # JSON 序列化的发票列表，供 bot/fill_doc 使用
"工号",
]
# ------------------------------------------------------------------
# 发票类型常量
# ------------------------------------------------------------------
INVOICE_TYPE_TRAIN = "高铁票"
INVOICE_TYPE_HOTEL = "酒店住宿"
INVOICE_TYPE_GENERAL = "普通发票"
INVOICE_TYPE_TRAVEL = frozenset([INVOICE_TYPE_TRAIN, INVOICE_TYPE_HOTEL])
def is_travel_invoice(invoice_type: str) -> bool:
"""判断是否为差旅发票（高铁票/酒店住宿）"""
return invoice_type in INVOICE_TYPE_TRAVEL
def classify_invoice_batch(invoices: list[dict[str, str]]) -> dict[str, list[dict[str, str]]]:
"""将发票列表按类型分组：{'travel': [...], 'general': [...]}"""
travel = []
general = []
for inv in invoices:
inv_type = inv.get("发票类型", INVOICE_TYPE_GENERAL)
if is_travel_invoice(inv_type):
travel.append(inv)
else:
general.append(inv)
return {"travel": travel, "general": general}
# ------------------------------------------------------------------
# CSV 读写工具
# ------------------------------------------------------------------
def _clean_invoice_for_json(inv: dict[str, str]) -> dict[str, str]:
"""清理发票字典中的内部字段，保留可序列化的字段"""
clean = {}
for k, v in inv.items():
if k.startswith("_"):
continue
clean[k] = v
return clean
def load_csv(csv_path: Path) -> list[dict[str, str]] | None:
"""读取支付记录 CSV 为 dict 列表，失败返回 None"""
try:
with open(csv_path, encoding="utf-8", newline="") as f:
reader = csv.DictReader(f)
fieldnames = reader.fieldnames or []
missing = [c for c in PAYMENT_RECORD_COLUMNS if c not in fieldnames]
if missing:
log.error(f"CSV 缺少必要列: {missing}")
return None
return [row for row in reader]
except FileNotFoundError:
log.error(f"CSV 文件不存在: {csv_path.name}")
return None
except Exception as e:
log.error(f"CSV 读取失败: {e}")
return None
def load_invoice_csv(csv_path: Path) -> list[dict[str, str]] | None:
"""读取发票级别 CSV 为 dict 列表（每行一张发票），失败返回 None"""
try:
with open(csv_path, encoding="utf-8", newline="") as f:
reader = csv.DictReader(f)
fieldnames = reader.fieldnames or []
missing = [c for c in INVOICE_LEVEL_COLUMNS if c not in fieldnames]
if missing:
log.error(f"发票 CSV 缺少必要列: {missing}")
return None
return [row for row in reader]
except FileNotFoundError:
log.error(f"发票 CSV 文件不存在: {csv_path.name}")
return None
except Exception as e:
log.error(f"发票 CSV 读取失败: {e}")
return None
def load_invoices_from_csv(csv_path: Path) -> list[dict[str, str]] | None:
"""从支付记录 CSV 中还原发票级别的数据（供 bot/fill_doc 使用）
读取 _invoices_json 字段，反序列化后展平为发票列表。
"""
rows = load_csv(csv_path)
if rows is None:
return None
invoices = []
for row in rows:
invoices_json = row.get("_invoices_json", "")
if not invoices_json:
continue
try:
inv_list = json.loads(invoices_json)
for inv in inv_list:
# 从支付记录回填刷卡信息
inv["刷卡日期"] = row.get("刷卡日期", inv.get("刷卡日期", ""))
inv["公务卡号"] = row.get("公务卡号", inv.get("公务卡号", ""))
inv["刷卡金额"] = row.get("刷卡金额", inv.get("刷卡金额", ""))
invoices.append(inv)
except json.JSONDecodeError:
log.warning(f"无法解析发票 JSON: {invoices_json[:50]}...")
return invoices
def save_csv(
payment_records: list[dict[str, str]],
output_path: str | Path = "payment_records.csv",
) -> None:
"""将支付记录列表保存为 CSV（以支付记录为主键）
每条支付记录包含：
- 刷卡日期、公务卡号、刷卡金额（支付信息）
- 关联发票数、发票详情（发票聚合信息）
- _matched_invoices（内部字段，序列化为 JSON 存储在 CSV 中）
"""
csv_path = Path(output_path)
with open(csv_path, "w", encoding="utf-8", newline="") as f:
writer = csv.writer(f)
writer.writerow(PAYMENT_RECORD_COLUMNS)
for idx, record in enumerate(payment_records, 1):
# 序列化关联发票为 JSON
matched_invoices: list[dict[str, str]] = record.get("_matched_invoices", []) # type: ignore[assignment]
invoices_json = json.dumps(
[_clean_invoice_for_json(inv) for inv in matched_invoices],
ensure_ascii=False,
)
writer.writerow(
[
idx,
record.get("刷卡日期", ""),
record.get("公务卡号", ""),
record.get("刷卡金额", ""),
record.get("关联发票数", str(len(matched_invoices))),
record.get("发票详情", ""),
record.get("备注", ""),
invoices_json,
record.get("工号", ""),
]
)
log.info(f"支付记录 CSV 已保存: {csv_path.name}")
def save_invoice_csv(
payment_records: list[dict[str, str]],
output_path: str | Path = "invoice_summary.csv",
) -> None:
"""将支付记录展平为发票级别 CSV（每行一张发票）
从 _matched_invoices 中还原每张发票，回填刷卡信息，
生成以发票为主键的 CSV，用于人工填写报销单参考。
"""
csv_path = Path(output_path)
with open(csv_path, "w", encoding="utf-8", newline="") as f:
writer = csv.writer(f)
writer.writerow(INVOICE_LEVEL_COLUMNS)
idx = 1
for record in payment_records:
matched_invoices: list[dict[str, str]] = record.get("_matched_invoices", []) # type: ignore[assignment]
for inv in matched_invoices:
clean_inv = _clean_invoice_for_json(inv)
writer.writerow(
[
idx,
clean_inv.get("发票类型", ""),
clean_inv.get("发票号码", ""),
clean_inv.get("开票日期", ""),
clean_inv.get("项目名称", ""),
clean_inv.get("规格型号", ""),
clean_inv.get("价税合计", ""),
clean_inv.get("销售方名称", ""),
clean_inv.get("出发站", ""),
clean_inv.get("到达站", ""),
clean_inv.get("车次", ""),
clean_inv.get("乘车日期", ""),
clean_inv.get("座位等级", ""),
clean_inv.get("人员姓名", ""),
record.get("刷卡日期", ""),
record.get("公务卡号", ""),
record.get("刷卡金额", ""),
record.get("备注", ""),
record.get("工号", ""),
]
)
idx += 1
log.info(f"发票级别 CSV 已保存: {csv_path.name}")
def save_csv_rows(csv_path: Path, rows: list[dict[str, str]]) -> None:
"""将 dict 列表保存为支付记录 CSV（用于更新已有 CSV）"""
with open(csv_path, "w", encoding="utf-8", newline="") as f:
writer = csv.DictWriter(f, fieldnames=PAYMENT_RECORD_COLUMNS)
writer.writeheader()
writer.writerows(rows)
log.info(f"CSV 已保存: {csv_path.name}")