帮财务小姐姐写了几个 Python 自动化脚本，结果...

某个无聊的下午，财务小姐姐找到了我，跟我说她厌倦了每周重复那些无聊的点击操作，想要我帮忙开发一个工具，它可以：监控文件夹、从 PDF 中提取数据、丰富数据、推送报告。

我想，闲着也是闲着，就帮她这个忙吧，说不定还可以...咳咳咳

1. 我要解决的问题

大多数自动化项目失败是因为他们试图解决所有问题。相反，应该选择一个重复出现的痛点，并且投资回报率可衡量。我的做法是：

痛点：

• 客户每天都会以分散的 PDF 格式发送发票。
• 我手动打开它们，提取供应商、日期、金额，然后放入 excel。
• 每天浪费约 20 分钟。

目标：
将其减少到零人力分钟。

2. 快速 MVP — 构建文件监视器 + PDF 提取器

从小事做起：查看文件夹，检测新的 PDF，提取文本。使用watchdog+ PyMuPDF(fitz)。

# file_watcher.py
import time
from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler
import fitz  # pymupdf

classPDFHandler(FileSystemEventHandler):
defon_created(self, event):
if event.src_path.lower().endswith(".pdf"):
            print(f"[+] New PDF: {event.src_path}")
            text = extract_text(event.src_path)
            print(text[:200], "...\n")  # quick preview

defextract_text(path: str) -> str:
    doc = fitz.open(path)
    pages = []
for page in doc:
        pages.append(page.get_text())
    doc.close()
return"\n".join(pages)

if __name__ == "__main__":
    observer = Observer()
    handler = PDFHandler()
    observer.schedule(handler, path="./inbox", recursive=False)
    observer.start()
try:
whileTrue:
            time.sleep(1)
except KeyboardInterrupt:
        observer.stop()
    observer.join()

这个脚本已经将小姐姐每天的工作时间缩短至 5 分钟——主要用于审查。

3. 增强提取器的鲁棒性：OCR + 文本回退

部分 PDF 是扫描图像。请添加pytesseract后备功能。

pip install pytesseract pill 
# 还必须在系统上安装 tesseract (apt/brew/choco)
from PIL import Image
import pytesseract
import fitz

defextract_text_with_ocr(path: str) -> str:
    doc = fitz.open(path)
    aggregated = []
for page in doc:
        text = page.get_text()
if text.strip():
            aggregated.append(text)
else:
            pix = page.get_pixmap(dpi=200)
            img = Image.frombytes("RGB", [pix.width, pix.height], pix.samples)
            aggregated.append(pytesseract.image_to_string(img))
    doc.close()
return"\n".join(aggregated)

这种混合方法（文本层 -> OCR）使该工具对我所见的 95% 的发票都具有可靠性。

4. 使用 OOP 构建结构——构建插件友好的管道

如果想要产品化，请将你的流程模块化。每个步骤都是一个类：加载器 → 解析器 → 丰富器 → 接收器。这样你就可以在不重写代码的情况下更换存储（excel表格、数据库、Webhook）。

# pipeline.py
from abc import ABC, abstractmethod
from typing import Dict

classStep(ABC):
    @abstractmethod
defrun(self, data: Dict) -> Dict:
pass

classLoader(Step):
def__init__(self, path): self.path = path
defrun(self, data):
        data['text'] = extract_text_with_ocr(self.path)
return data

classParser(Step):
defrun(self, data):
# naive example; replace with regex or NLP later
        text = data['text']
        data['vendor'] = find_vendor(text)
        data['amount'] = find_amount(text)
return data

classSink(Step):
defrun(self, data):
        save_to_excel_sheet(data)
return data

classPipeline:
def__init__(self, steps):
        self.steps = steps
defexecute(self, initial):
        data = initial
for step in self.steps:
            data = step.run(data)
return data

此模式可扩展：添加ClassifierStep语言检测、TranslatorStep非英语文档等。

5. 信息丰富与提取——先用正则表达式，再用机器学习

从确定性解析（正则表达式）开始。如果发票内容混乱或包含多种布局，请添加机器学习模型（或使用layout-parser）。正则表达式代码片段示例：

import re

AMOUNT_RE = re.compile(r"(?<!\d)(?:USD|EUR|\$)?\s?([\d{1,3}(?:,\d{3})*(?:\.\d{2})?)\b")

deffind_amount(text: str) -> float | None:
    m = AMOUNT_RE.search(text.replace("\n", " "))
if m:
        s = m.group(1).replace(',', '')
return float(s)
returnNone

为了提高可靠性，请使用spacy+ 自定义 NER 或layout-parser在空间上检测发票字段。

6. Web 自动化和抓取——Playwright 用于下载和仪表盘

当发票位于网络仪表板后面时，使用 Playwright 自动下载。

pip install playwright
playwright install
deflogin_and_download(url, user, password, download_path):
with sync_playwright() as p:
        browser = p.chromium.launch(headless=True)
        page = browser.new_page()
        page.goto(url)
        page.fill('#username', user)
        page.fill('#password', password)
        page.click('#login')
        page.wait_for_selector('a.download')
with page.expect_download() as download_info:
            page.click('a.download')
        download = download_info.value
        download.save_as(download_path)
        browser.close()

这使得服务可以自动收集源 PDF——如果小姐姐想运行系统每天早上获取客户文档的订阅，这一点至关重要。

7. 打包工具 — CLI 使用Typer/Click

对于分发，将功能包装为 CLI，以便非开发客户可以在本地运行它，或者可以在服务器上运行它。

pip install typer
import typer
from pipeline import Pipeline, Loader, Parser, Sink

app = typer.Typer()

@app.command()
defprocess(path: str):
    steps = [Loader(path), Parser(), Sink()]
    p = Pipeline(steps)
    p.execute({})
    typer.echo("Processed!")

if __name__ == "__main__":
    app()

构建一个setup.py/pyproject.toml并发布到 PyPI，或者打包为 wheel / Docker 镜像。

8. 使用 worker 进行扩展：Celery + Redis（或 FastAPI + 后台任务）

如果你想让更多的小姐姐一起同时使用时，那么在工作队列中运行处理工作，而不是阻止所有内容。

pip install celery redis
#tasks.py
from celery import Celery
from pipeline import Pipeline, Loader, Parser, Sink

app = Celery('tasks', broker='redis://localhost:6379/0')

@app.task
defprocess_file(path):
    steps = [Loader(path), Parser(), Sink()]
    Pipeline(steps).execute({})

Web 前端/API 入队process_file.delay(path)并立即返回。工作线程负责处理并将结果推送至存储。

9. 可观察性和可靠性——日志、指标、可重试步骤

使用loguru+结构化日志，并导出正常运行时间、队列长度和故障率的指标（Prometheus）。

pip install loguru
from loguru import logger
logger.add("service.log", rotation="10 MB", level="INFO")

try:
    process_file("/tmp/a.pdf")
except Exception as e:
    logger.exception("Processing failed")

工具做完了，让财务小姐姐试用时，结果她投来了崇拜的眼神......