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

完成差旅发票录入流程

Browse Source
This commit is contained in:
wandering
2026-06-11 19:22:34 +08:00
parent cf567c22f2
commit 10115214aa
50 changed files with 3033 additions and 2756 deletions
Download Patch File Download Diff File Expand all files Collapse all files

254
src/doc/invoice.py
View File

@@ -1,16 +1,12 @@
"""发票数据模型与 CSV 工具
定义发票类型常量、CSV 列结构，提供发票分类和 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
save_application_json(applications, path) 保存出差申请单 JSON
"""
import csv
@@ -21,76 +17,63 @@ from .. import get_logger
log = get_logger("invoice")
# ------------------------------------------------------------------
# CSV 列定义
# ------------------------------------------------------------------
# 发票级别 CSV 列（用于 invoice_summary.csv，每行一张发票）
# 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",
]
# 支付记录级别 CSV 列（用于 payment_records.csv，每行一笔支付）
PAYMENT_RECORD_COLUMNS = [
"序号",
# 支付信息
"刷卡日期",
"公务卡号",
"刷卡金额",
# 发票聚合信息
"关联发票数",
"发票详情", # 格式: 类型[号码]¥金额 | 类型[号码]¥金额
"备注",
# 内部字段（用于下游解析）
"_invoices_json", # JSON 序列化的发票列表，供 bot/fill_doc 使用
"工号",
"index",
"card_date",
"card_no",
"card_amount",
"relative_invoice_count",
"invoice_detail",
"remark",
"_matched_invoices",
"person_id",
]
# ------------------------------------------------------------------
# 发票类型常量
# ------------------------------------------------------------------
INVOICE_TYPE_TRAIN = "高铁票"
INVOICE_TYPE_HOTEL = "酒店住宿"
INVOICE_TYPE_GENERAL = "普通发票"
INVOICE_TYPE_TRAVEL = frozenset([INVOICE_TYPE_TRAIN, INVOICE_TYPE_HOTEL])
def _is_application_document(invoice_type: str) -> bool:
"""判断是否为出差事前申请单"""
return invoice_type == "application"
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 = []
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_travel_invoice(inv_type):
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}
return {"travel": travel, "general": general, "application": application}
# ------------------------------------------------------------------
@@ -108,69 +91,37 @@ def _clean_invoice_for_json(inv: dict[str, str]) -> dict[str, str]:
return clean
def load_csv(csv_path: Path) -> list[dict[str, str]] | None:
"""读取支付记录 CSV 为 dict 列表，失败返回 None"""
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", newline="") as f:
with open(csv_path, encoding="utf-8-sig", newline="") as f:
reader = csv.DictReader(f)
fieldnames = reader.fieldnames or []
missing = [c for c in PAYMENT_RECORD_COLUMNS if c not in fieldnames]
missing = [c for c in required_columns if c not in fieldnames]
if missing:
log.error(f"CSV 缺少必要列: {missing}")
log.error(f"{label} 缺少必要列: {missing}")
return None
return [row for row in reader]
except FileNotFoundError:
log.error(f"CSV 文件不存在: {csv_path.name}")
log.error(f"{label} 文件不存在: {csv_path.name}")
return None
except Exception as e:
log.error(f"CSV 读取失败: {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"""
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
return _load_csv(csv_path, INVOICE_LEVEL_COLUMNS, "发票 CSV")
def save_csv(
@@ -180,8 +131,8 @@ def save_csv(
"""将支付记录列表保存为 CSV（以支付记录为主键）
每条支付记录包含：
- 刷卡日期、公务卡号、刷卡金额（支付信息）
- 关联发票数、发票详情（发票聚合信息）
- card_date, card_no, card_amount（支付信息）
- relative_invoice_count, invoice_detail（发票聚合信息）
- _matched_invoices（内部字段，序列化为 JSON 存储在 CSV 中）
"""
csv_path = Path(output_path)
@@ -191,7 +142,6 @@ def save_csv(
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],
@@ -201,14 +151,14 @@ def save_csv(
writer.writerow(
[
idx,
record.get("刷卡日期", ""),
record.get("公务卡号", ""),
record.get("刷卡金额", ""),
record.get("关联发票数", str(len(matched_invoices))),
record.get("发票详情", ""),
record.get("备注", ""),
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("工号", ""),
record.get("person_id", ""),
]
)
@@ -223,6 +173,7 @@ def save_invoice_csv(
从 _matched_invoices 中还原每张发票，回填刷卡信息，
生成以发票为主键的 CSV，用于人工填写报销单参考。
出差事前申请单不会被写入此文件（它们有独立的 CSV）。
"""
csv_path = Path(output_path)
@@ -235,27 +186,29 @@ def save_invoice_csv(
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("发票类型", ""),
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("工号", ""),
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
@@ -263,11 +216,18 @@ def save_invoice_csv(
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)
def save_application_json(
applications: list[dict[str, str]],
output_path: str | Path = "travel_applications.json",
) -> None:
"""将出差事前申请单列表保存为独立 JSON 文件
log.info(f"CSV 已保存: {csv_path.name}")
使用 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}")