使用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"]