使用Python从零打造智能票据处理机器人

# 安装依赖
pip install paddlepaddle paddleocr openpyxl pillow opencv-python python-dotenv
# 邮件发送使用标准库，无需额外安装

from paddleocr import PaddleOCR

# 初始化OCR引擎（首次运行会自动下载模型）
ocr = PaddleOCR(
    use_angle_cls=True,   # 使用方向分类器，处理旋转文字
    lang='ch',            # 中文模型
    show_log=False        # 关闭冗余日志
)

import re

def extract_invoice_info(ocr_result):
    """
    从PaddleOCR识别结果中提取发票关键信息
    ocr_result格式: list of [box, (text, confidence)]
    """
    full_text = ""
    for line in ocr_result:
        text = line[1][0]
        confidence = line[1][1]
        if confidence > 0.7:  # 只保留高置信度结果
            full_text += text + " "
    
    # 提取金额（支持多种写法）
    amount_pattern = r'(?:金额|合计|总计)[:：\s]*([¥￥]?(\d+(?:\.\d{1,2})?))'
    amount_match = re.search(amount_pattern, full_text)
    amount = amount_match.group(1) if amount_match else None
    
    # 提取开票日期（常见格式：YYYY年MM月DD日 或 YYYY-MM-DD）
    date_pattern = r'(?:开票日期|发票日期)[:：\s]*(\d{4}[年-]\d{1,2}[月-]\d{1,2}日?)'
    date_match = re.search(date_pattern, full_text)
    invoice_date = date_match.group(1) if date_match else None
    
    # 提取供应商（销售方名称）
    seller_pattern = r'(?:销售方|出售方|供应商)[:：\s]*([\u4e00-\u9fa5a-zA-Z0-9（）()]+公司[\u4e00-\u9fa5]*)'
    seller_match = re.search(seller_pattern, full_text)
    seller = seller_match.group(1) if seller_match else None
    
    return {
        "amount": amount,
        "date": invoice_date,
        "supplier": seller,
        "raw_text": full_text
    }

def ocr_invoice_image(image_path):
    """对单张发票图片执行OCR并返回结构化信息"""
    try:
        result = ocr.ocr(image_path, cls=True)
        if not result or not result[0]:
            raise ValueError("OCR未识别到任何文本")
        # result[0] 是图片中所有文本行
        info = extract_invoice_info(result[0])
        return info
    except Exception as e:
        print(f"OCR识别失败: {image_path}, 错误: {e}")
        return None

pip install pdf2image
# 还需要安装poppler（Windows需下载，Linux apt install poppler-utils）

from pdf2image import convert_from_path

def ocr_invoice_pdf(pdf_path):
    images = convert_from_path(pdf_path, dpi=200, first_page=1, last_page=1)
    if not images:
        return None
    # 将PIL Image转换为临时文件或直接处理
    import tempfile
    with tempfile.NamedTemporaryFile(suffix=".png", delete=False) as tmp:
        images[0].save(tmp.name, "PNG")
        result = ocr.ocr(tmp.name, cls=True)
        info = extract_invoice_info(result[0]) if result and result[0] else None
        return info

from openpyxl import load_workbook
import os

def append_to_excel(excel_path, invoice_info):
    """
    将发票信息追加到Excel文件的报销明细表中
    假设工作表名为"报销明细"，表头为：日期, 供应商, 金额, 备注
    """
    if not os.path.exists(excel_path):
        # 如果文件不存在，创建新文件并写入表头
        from openpyxl import Workbook
        wb = Workbook()
        ws = wb.active
        ws.title = "报销明细"
        ws.append(["日期", "供应商", "金额", "备注"])
    else:
        wb = load_workbook(excel_path)
        ws = wb["报销明细"]
    
    # 追加一行数据
    row = [
        invoice_info.get("date", ""),
        invoice_info.get("supplier", ""),
        invoice_info.get("amount", ""),
        f"OCR自动识别 {invoice_info.get('raw_text', '')[:20]}..."
    ]
    ws.append(row)
    
    # 保存文件
    wb.save(excel_path)
    print(f"已追加记录到 {excel_path}")

import smtplib
from email.mime.multipart import MIMEMultipart
from email.mime.text import MIMEText
from email.mime.base import MIMEBase
from email import encoders
import os

def send_excel_by_email(receiver_email, excel_path, subject="报销单待审核", body="请查收本周报销明细"):
    """
    发送Excel文件作为附件到指定邮箱
    """
    # 邮箱配置（从环境变量读取）
    smtp_server = os.getenv("SMTP_SERVER", "smtp.qq.com")
    smtp_port = int(os.getenv("SMTP_PORT", "465"))
    sender_email = os.getenv("SENDER_EMAIL")
    sender_password = os.getenv("SENDER_PASSWORD")  # 授权码
    
    if not sender_email or not sender_password:
        raise ValueError("请在.env文件中配置发件邮箱和密码")
    
    # 构建邮件
    msg = MIMEMultipart()
    msg["From"] = sender_email
    msg["To"] = receiver_email
    msg["Subject"] = subject
    msg.attach(MIMEText(body, "plain", "utf-8"))
    
    # 添加附件
    with open(excel_path, "rb") as f:
        part = MIMEBase("application", "octet-stream")
        part.set_payload(f.read())
        encoders.encode_base64(part)
        filename = os.path.basename(excel_path)
        part.add_header("Content-Disposition", f"attachment; filename={filename}")
        msg.attach(part)
    
    # 发送
    with smtplib.SMTP_SSL(smtp_server, smtp_port) as server:
        server.login(sender_email, sender_password)
        server.sendmail(sender_email, receiver_email, msg.as_string())
    print(f"邮件已发送至 {receiver_email}")

SMTP_SERVER=smtp.qq.com
SMTP_PORT=465
SENDER_EMAIL=rpa@company.com
SENDER_PASSWORD=your_authorization_code
AUDITOR_EMAIL=auditor@company.com

from dotenv import load_dotenv
load_dotenv()

import os
import time
import shutil
from pathlib import Path
import logging
from dotenv import load_dotenv

from ocr_engine import ocr_invoice_image, ocr_invoice_pdf
from excel_handler import append_to_excel
from mail_sender import send_excel_by_email

# 加载配置
load_dotenv()
INPUT_DIR = "invoices"
PROCESSED_DIR = "processed"
OUTPUT_EXCEL = "output/报销明细.xlsx"
AUDITOR_EMAIL = os.getenv("AUDITOR_EMAIL")
SCAN_INTERVAL = 60  # 扫描间隔（秒）

# 配置日志
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s - %(levelname)s - %(message)s",
    handlers=[
        logging.FileHandler("invoice_robot.log"),
        logging.StreamHandler()
    ]
)
logger = logging.getLogger(__name__)

def process_single_file(file_path):
    """处理单个发票文件"""
    logger.info(f"开始处理文件: {file_path}")
    ext = file_path.suffix.lower()
    
    # 1. OCR识别
    if ext in ['.jpg', '.jpeg', '.png']:
        info = ocr_invoice_image(str(file_path))
    elif ext == '.pdf':
        info = ocr_invoice_pdf(str(file_path))
    else:
        logger.warning(f"不支持的文件类型: {ext}")
        return False
    
    if not info:
        logger.error(f"OCR识别失败: {file_path}")
        return False
    
    logger.info(f"识别结果: 金额={info['amount']}, 日期={info['date']}, 供应商={info['supplier']}")
    
    # 2. 写入Excel
    try:
        append_to_excel(OUTPUT_EXCEL, info)
    except Exception as e:
        logger.error(f"Excel写入失败: {e}")
        return False
    
    # 3. 移动文件到已处理文件夹
    processed_path = Path(PROCESSED_DIR) / file_path.name
    shutil.move(str(file_path), str(processed_path))
    logger.info(f"文件已归档: {processed_path}")
    
    return True

def scan_and_process():
    """扫描文件夹，处理所有待处理文件"""
    input_path = Path(INPUT_DIR)
    if not input_path.exists():
        input_path.mkdir()
    
    files = list(input_path.glob("*.*"))
    if not files:
        logger.info("暂无待处理文件")
        return
    
    for file_path in files:
        if file_path.suffix.lower() in ['.jpg', '.jpeg', '.png', '.pdf']:
            process_single_file(file_path)
        else:
            logger.info(f"跳过非图片/PDF文件: {file_path.name}")

def main():
    logger.info("智能票据处理机器人启动")
    logger.info(f"监控文件夹: {INPUT_DIR}, 扫描间隔: {SCAN_INTERVAL}秒")
    
    # 确保输出目录存在
    Path("output").mkdir(exist_ok=True)
    Path(PROCESSED_DIR).mkdir(exist_ok=True)
    
    # 持续运行模式（也可改为单次执行后退出）
    try:
        while True:
            scan_and_process()
            time.sleep(SCAN_INTERVAL)
    except KeyboardInterrupt:
        logger.info("机器人已停止")

if __name__ == "__main__":
    main()

import cv2

def preprocess_image(image_path):
    img = cv2.imread(image_path, cv2.IMREAD_GRAYSCALE)
    # 自适应阈值二值化
    img = cv2.adaptiveThreshold(img, 255, cv2.ADAPTIVE_THRESH_GAUSSIAN_C, cv2.THRESH_BINARY, 11, 2)
    # 去噪
    img = cv2.medianBlur(img, 3)
    return img

from tenacity import retry, stop_after_attempt, wait_exponential

@retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=4, max=10))
def send_excel_with_retry(*args, **kwargs):
    send_excel_by_email(*args, **kwargs)

FROM python:3.9-slim
RUN apt-get update && apt-get install -y poppler-utils
WORKDIR /app
COPY requirements.txt .
RUN pip install -r requirements.txt
COPY . .
CMD ["python", "main.py"]