wandering
237 lines
7.7 KiB
Python
237 lines
7.7 KiB
Python
"""发票数据模型与 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}")
|