13bf1a0146
仅保存说明、脚本和 Excel 模板，示例中的个人和公司信息已匿名化。 Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
329 lines
11 KiB
Python
329 lines
11 KiB
Python
#!/usr/bin/env python3
|
||
"""
|
||
差旅报销自动化脚本
|
||
|
||
Python 只做两类确定性工作：
|
||
1. extract：提取 PDF 原文到 TXT
|
||
2. fill：读取大模型生成的 JSON，填入 Excel 模板
|
||
|
||
中间的发票理解、金额分类、日期/事由/人员判断全部由大模型完成。
|
||
模板使用 .xlsx，完全依赖 Python 包实现跨平台，不依赖 LibreOffice。
|
||
"""
|
||
|
||
import argparse
|
||
import json
|
||
import re
|
||
import shutil
|
||
from pathlib import Path
|
||
|
||
import pdfplumber
|
||
|
||
|
||
OUTPUT_DIR = Path("报销输出")
|
||
RAW_TEXT_FILE = OUTPUT_DIR / "pdf_raw_text.txt"
|
||
TEMPLATE_FILE = Path("差旅费明细表.xlsx")
|
||
|
||
|
||
SCHEMA_EXAMPLE = {
|
||
"meta": {
|
||
"出差人": "从文本中提取或询问",
|
||
"出差事由": "从文本中提取或询问",
|
||
"出发日期": "从文本中提取或询问",
|
||
"结束日期": "从文本中提取或询问",
|
||
"出发地": "从文本中提取或询问",
|
||
"目的地": "从文本中提取或询问",
|
||
"出行方式": "从文本中提取或询问",
|
||
"出差天数": "从文本中提取或询问",
|
||
"事由简写": "从文本中提取或询问，适合文件名使用的简短事由描述"
|
||
},
|
||
"expenses": {
|
||
"住宿": {
|
||
"金额": 510.0,
|
||
"单据张数": 1,
|
||
"说明": "广州市番禺明乐宾馆住宿费"
|
||
},
|
||
"交通": {
|
||
"金额": 2013.77,
|
||
"单据张数": 5,
|
||
"说明": "机票1690.00 + 滴滴323.77"
|
||
},
|
||
"饮食": {
|
||
"金额": 0.0,
|
||
"单据张数": 0,
|
||
"说明": "未发现餐饮发票"
|
||
},
|
||
"其他": {
|
||
"金额": 0.0,
|
||
"单据张数": 0,
|
||
"说明": "无其他费用"
|
||
}
|
||
},
|
||
"summary": {
|
||
"附件总张数": 6,
|
||
"报销总额": 2523.77,
|
||
"预借差旅费": 0.0,
|
||
"补领金额": 2523.77,
|
||
"退还金额": 0.0
|
||
},
|
||
"evidence": [
|
||
{
|
||
"文件": "住宿.pdf",
|
||
"类别": "住宿",
|
||
"金额": 510.0,
|
||
"依据": "价税合计（小写）¥510.00"
|
||
},
|
||
{
|
||
"文件": "机票预订出行订单尾号87377629-电子普通发票.pdf",
|
||
"类别": "交通",
|
||
"金额": 1690.0,
|
||
"依据": "价税合计（小写）¥1690.00"
|
||
}
|
||
],
|
||
"questions": []
|
||
}
|
||
|
||
|
||
def ensure_output_dir():
|
||
OUTPUT_DIR.mkdir(exist_ok=True)
|
||
|
||
|
||
def command_extract(_args):
|
||
ensure_output_dir()
|
||
pdf_files = sorted(Path(".").glob("*.pdf"))
|
||
if not pdf_files:
|
||
raise SystemExit("当前目录未找到 PDF 文件。")
|
||
|
||
with open(RAW_TEXT_FILE, "w", encoding="utf-8") as output:
|
||
output.write("=" * 60 + "\n")
|
||
output.write("PDF 原始内容汇总\n")
|
||
output.write("=" * 60 + "\n\n")
|
||
|
||
for pdf_path in pdf_files:
|
||
output.write(f"\n{'=' * 60}\n")
|
||
output.write(f"文件: {pdf_path.name}\n")
|
||
output.write("=" * 60 + "\n")
|
||
|
||
try:
|
||
with pdfplumber.open(pdf_path) as pdf:
|
||
for page_index, page in enumerate(pdf.pages, start=1):
|
||
text = page.extract_text() or ""
|
||
output.write(f"\n--- 第 {page_index} 页 ---\n")
|
||
output.write(text)
|
||
output.write("\n")
|
||
except Exception as exc:
|
||
output.write(f"\n[读取失败: {exc}]\n")
|
||
|
||
print(f"已提取 {len(pdf_files)} 个 PDF 文件到 {RAW_TEXT_FILE}")
|
||
|
||
|
||
def command_schema(args):
|
||
ensure_output_dir()
|
||
text = json.dumps(SCHEMA_EXAMPLE, ensure_ascii=False, indent=2)
|
||
if args.output:
|
||
schema_path = OUTPUT_DIR / "schema_example.json"
|
||
schema_path.write_text(text + "\n", encoding="utf-8")
|
||
print(f"已写入 {schema_path}")
|
||
else:
|
||
print(text)
|
||
|
||
|
||
def command_fill(args):
|
||
result = load_json(args.json_path)
|
||
validate_result(result)
|
||
ensure_fill_dependencies()
|
||
|
||
output_file = fill_template(result)
|
||
verify_output(output_file, result)
|
||
print(f"完成：{output_file}")
|
||
|
||
|
||
def load_json(json_path):
|
||
path = Path(json_path)
|
||
if not path.exists():
|
||
raise SystemExit(f"找不到 JSON 文件：{json_path}")
|
||
|
||
with open(path, encoding="utf-8") as f:
|
||
try:
|
||
return json.load(f)
|
||
except json.JSONDecodeError as exc:
|
||
raise SystemExit(f"JSON 格式错误：{exc}")
|
||
|
||
|
||
def validate_result(result):
|
||
require_keys(result, ["meta", "expenses", "summary"], "根对象")
|
||
|
||
meta = result["meta"]
|
||
expenses = result["expenses"]
|
||
summary = result["summary"]
|
||
|
||
require_keys(meta, [
|
||
"出差人", "出差事由", "出发日期", "结束日期", "出发地", "目的地",
|
||
"出行方式", "出差天数", "事由简写"
|
||
], "meta")
|
||
|
||
require_keys(expenses, ["住宿", "交通", "饮食", "其他"], "expenses")
|
||
for category in ["住宿", "交通", "饮食", "其他"]:
|
||
require_keys(expenses[category], ["金额", "单据张数", "说明"], f"expenses.{category}")
|
||
require_number(expenses[category]["金额"], f"expenses.{category}.金额")
|
||
require_integer(expenses[category]["单据张数"], f"expenses.{category}.单据张数")
|
||
|
||
require_keys(summary, ["附件总张数", "报销总额", "预借差旅费", "补领金额", "退还金额"], "summary")
|
||
|
||
expense_total = sum(float(expenses[c]["金额"]) for c in ["住宿", "交通", "饮食", "其他"])
|
||
if abs(float(summary["报销总额"]) - expense_total) > 0.01:
|
||
raise SystemExit(
|
||
f"报销总额不一致：summary.报销总额={summary['报销总额']}，"
|
||
f"费用合计={expense_total:.2f}"
|
||
)
|
||
|
||
receipt_total = sum(int(expenses[c]["单据张数"]) for c in ["住宿", "交通", "饮食", "其他"])
|
||
if int(summary["附件总张数"]) != receipt_total:
|
||
raise SystemExit(
|
||
f"附件总张数不一致：summary.附件总张数={summary['附件总张数']}，"
|
||
f"分类单据合计={receipt_total}"
|
||
)
|
||
|
||
|
||
def require_keys(obj, keys, label):
|
||
missing = [key for key in keys if key not in obj]
|
||
if missing:
|
||
raise SystemExit(f"{label} 缺少字段：{', '.join(missing)}")
|
||
|
||
|
||
def require_number(value, label):
|
||
if not isinstance(value, (int, float)):
|
||
raise SystemExit(f"{label} 必须是数字，当前为：{value!r}")
|
||
|
||
|
||
def require_integer(value, label):
|
||
if not isinstance(value, int):
|
||
raise SystemExit(f"{label} 必须是整数，当前为：{value!r}")
|
||
|
||
|
||
def ensure_fill_dependencies():
|
||
if not TEMPLATE_FILE.exists():
|
||
raise SystemExit(
|
||
"找不到模板文件：差旅费明细表.xlsx\n"
|
||
"请将 Excel 模板另存为 .xlsx 后放到当前目录。"
|
||
)
|
||
|
||
try:
|
||
import openpyxl # noqa: F401
|
||
except ImportError:
|
||
raise SystemExit("缺少 openpyxl，请运行：python -m pip install openpyxl")
|
||
|
||
|
||
def fill_template(result):
|
||
ensure_output_dir()
|
||
import openpyxl
|
||
|
||
meta = result["meta"]
|
||
expenses = result["expenses"]
|
||
output_file = OUTPUT_DIR / build_output_filename(meta)
|
||
|
||
shutil.copy(TEMPLATE_FILE, output_file)
|
||
wb = openpyxl.load_workbook(output_file)
|
||
ws = wb.active
|
||
|
||
ws["E3"] = meta["出差事由"]
|
||
ws["A6"] = f'{meta["出发日期"]}-{meta["结束日期"]}'
|
||
ws["B6"] = f'{meta["出发地"]}-{meta["目的地"]}'
|
||
ws["C6"] = f'{meta["出差天数"]}天'
|
||
ws["D6"] = meta["出行方式"]
|
||
|
||
ws["H6"] = expenses["住宿"]["单据张数"]
|
||
ws["I6"] = expenses["住宿"]["金额"]
|
||
ws["H7"] = expenses["饮食"]["单据张数"]
|
||
ws["I7"] = expenses["饮食"]["金额"]
|
||
ws["H8"] = expenses["交通"]["单据张数"]
|
||
ws["I8"] = expenses["交通"]["金额"]
|
||
ws["H9"] = expenses["其他"]["单据张数"]
|
||
ws["I9"] = expenses["其他"]["金额"]
|
||
|
||
wb.save(output_file)
|
||
return output_file
|
||
|
||
|
||
def build_output_filename(meta):
|
||
date_str = normalize_date_for_filename(meta["出发日期"])
|
||
safe_parts = [
|
||
"差旅费明细表",
|
||
sanitize_filename(meta["出差人"]),
|
||
date_str,
|
||
sanitize_filename(meta["目的地"]),
|
||
sanitize_filename(meta["事由简写"]),
|
||
]
|
||
return "_".join(safe_parts) + ".xlsx"
|
||
|
||
|
||
def normalize_date_for_filename(date_text):
|
||
parts = re.split(r"[/-]", str(date_text))
|
||
if len(parts) != 3:
|
||
return sanitize_filename(str(date_text))
|
||
return f"{int(parts[0])}{int(parts[1]):02d}{int(parts[2]):02d}"
|
||
|
||
|
||
def sanitize_filename(text):
|
||
text = str(text).strip()
|
||
text = re.sub(r"[\\/:*?\"<>|\s]+", "", text)
|
||
return text or "未填写"
|
||
|
||
|
||
def verify_output(output_file, result):
|
||
import openpyxl
|
||
|
||
wb = openpyxl.load_workbook(output_file, data_only=False)
|
||
ws = wb.active
|
||
meta = result["meta"]
|
||
|
||
expected_route = f'{meta["出发地"]}-{meta["目的地"]}'
|
||
expected_date = f'{meta["出发日期"]}-{meta["结束日期"]}'
|
||
|
||
checks = [
|
||
(ws["E3"].value, meta["出差事由"], "出差事由"),
|
||
(ws["A6"].value, expected_date, "日期"),
|
||
(ws["B6"].value, expected_route, "起止地点"),
|
||
]
|
||
for actual, expected, label in checks:
|
||
if actual != expected:
|
||
raise SystemExit(f"输出校验失败：{label} 实际={actual!r} 预期={expected!r}")
|
||
|
||
formula_checks = {
|
||
"H2": "附件张数合计",
|
||
"F9": "天数合计",
|
||
"C10": "报销总额",
|
||
"H10": "补领金额",
|
||
}
|
||
for cell, label in formula_checks.items():
|
||
value = ws[cell].value
|
||
if not isinstance(value, str) or not value.startswith("="):
|
||
raise SystemExit(f"输出校验失败：{label} 单元格 {cell} 公式丢失，当前值={value!r}")
|
||
|
||
|
||
def build_parser():
|
||
parser = argparse.ArgumentParser(description="差旅报销自动化脚本")
|
||
subparsers = parser.add_subparsers(dest="command", required=True)
|
||
|
||
extract_parser = subparsers.add_parser("extract", help="提取当前目录所有 PDF 原文到 报销输出/pdf_raw_text.txt")
|
||
extract_parser.set_defaults(func=command_extract)
|
||
|
||
schema_parser = subparsers.add_parser("schema", help="输出大模型需要生成的 JSON schema 示例")
|
||
schema_parser.add_argument("--output", action="store_true", help="写入 报销输出/schema_example.json，而不是打印到终端")
|
||
schema_parser.set_defaults(func=command_schema)
|
||
|
||
fill_parser = subparsers.add_parser("fill", help="读取 JSON 并填入差旅费明细表模板")
|
||
fill_parser.add_argument("json_path", help="大模型生成的 analyzed_result.json 路径")
|
||
fill_parser.set_defaults(func=command_fill)
|
||
|
||
return parser
|
||
|
||
|
||
def main():
|
||
parser = build_parser()
|
||
args = parser.parse_args()
|
||
args.func(args)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|