"""发票数据模型与 CSV 工具 定义 CSV 列结构，提供发票分类和 CSV 读写功能。 对外接口: load_csv(path) 读取支付记录 CSV save_csv(payment_records, path) 保存支付记录 CSV save_invoice_csv(payment_records, path) 保存发票级别 CSV save_application_json(applications, path) 保存出差申请单 JSON """ import csv import json from pathlib import Path from .. import get_logger log = get_logger("invoice") # 缓存目录名（相对于源文件目录） CACHE_DIR_NAME = ".invoice_cache" # CSV 列名 INVOICE_LEVEL_COLUMNS = [ "index", "invoice_type", "invoice_number", "invoice_date", "item_name", "spec_model", "total_amount", "seller_name", "departure", "arrival", "train_no", "ride_date", "seat_class", "person_name", "card_date", "card_no", "card_amount", "remark", "person_id", ] PAYMENT_RECORD_COLUMNS = [ "index", "card_date", "card_no", "card_amount", "relative_invoice_count", "invoice_detail", "remark", "_matched_invoices", "person_id", ] def _is_application_document(invoice_type: str) -> bool: """判断是否为出差事前申请单""" return invoice_type == "application" def classify_invoice_batch( invoices: list[dict[str, str]], ) -> dict[str, list[dict[str, str]]]: """按发票类型分组""" travel: list[dict[str, str]] = [] general: list[dict[str, str]] = [] application: list[dict[str, str]] = [] for inv in invoices: inv_type = inv.get("invoice_type", "general") if _is_application_document(inv_type): application.append(inv) elif inv_type in ("train", "hotel"): travel.append(inv) else: general.append(inv) return {"travel": travel, "general": general, "application": application} # ------------------------------------------------------------------ # 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, required_columns: list[str], label: str = "CSV", ) -> list[dict[str, str]] | None: """通用 CSV 读取器：按 required_columns 校验列，失败返回 None""" try: with open(csv_path, encoding="utf-8-sig", newline="") as f: reader = csv.DictReader(f) fieldnames = reader.fieldnames or [] missing = [c for c in required_columns if c not in fieldnames] if missing: log.error(f"{label} 缺少必要列: {missing}") return None return [row for row in reader] except FileNotFoundError: log.error(f"{label} 文件不存在: {csv_path.name}") return None except Exception as e: log.error(f"{label} 读取失败: {e}") return None def load_csv(csv_path: Path) -> list[dict[str, str]] | None: """读取支付记录 CSV 为 dict 列表，失败返回 None""" return _load_csv(csv_path, PAYMENT_RECORD_COLUMNS, "CSV") def load_invoice_csv(csv_path: Path) -> list[dict[str, str]] | None: """读取发票级别 CSV 为 dict 列表（每行一张发票），失败返回 None""" return _load_csv(csv_path, INVOICE_LEVEL_COLUMNS, "发票 CSV") def save_csv( payment_records: list[dict[str, str]], output_path: str | Path = "payment_records.csv", ) -> None: """将支付记录列表保存为 CSV（以支付记录为主键） 每条支付记录包含： - card_date, card_no, card_amount（支付信息） - relative_invoice_count, invoice_detail（发票聚合信息） - _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): 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("card_date", ""), record.get("card_no", ""), record.get("card_amount", ""), record.get("relative_invoice_count", str(len(matched_invoices))), record.get("invoice_detail", ""), record.get("remark", ""), invoices_json, record.get("person_id", ""), ] ) 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）。 """ 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) if _is_application_document(clean_inv.get("invoice_type", "")): continue writer.writerow( [ idx, clean_inv.get("invoice_type", ""), clean_inv.get("invoice_number", ""), clean_inv.get("invoice_date", ""), clean_inv.get("item_name", ""), clean_inv.get("spec_model", ""), clean_inv.get("total_amount", ""), clean_inv.get("seller_name", ""), clean_inv.get("departure", ""), clean_inv.get("arrival", ""), clean_inv.get("train_no", ""), clean_inv.get("ride_date", ""), clean_inv.get("seat_class", ""), clean_inv.get("person_name", ""), record.get("card_date", ""), record.get("card_no", ""), record.get("card_amount", ""), record.get("remark", ""), record.get("person_id", ""), ] ) idx += 1 log.info(f"发票级别 CSV 已保存: {csv_path.name}") def save_application_json( applications: list[dict[str, str]], output_path: str | Path = "travel_applications.json", ) -> None: """将出差事前申请单列表保存为独立 JSON 文件 使用 JSON 保留完整嵌套结构（如出差人员信息的列表形式）， 避免 CSV 扁平化导致的字段丢失。 """ json_path = Path(output_path) with open(json_path, "w", encoding="utf-8") as f: json.dump(applications, f, ensure_ascii=False, indent=2) log.info(f"出差申请单 JSON 已保存: {json_path.name}")