390 lines
13 KiB
Python
390 lines
13 KiB
Python
"""
|
||
财务报销自动化 — Web 界面
|
||
|
||
用户上传 PDF 发票和支付截图，配置账号信息，自动完成：
|
||
1. 发票提取 2. OCR 识别 3. 浏览器填报（可选）
|
||
|
||
启动: python web/app.py
|
||
访问: http://localhost:5000
|
||
"""
|
||
|
||
import io
|
||
import json
|
||
import logging
|
||
import sys
|
||
import threading
|
||
import time
|
||
import uuid
|
||
from pathlib import Path
|
||
|
||
from flask import Flask, Response, jsonify, render_template, request, stream_with_context
|
||
|
||
# 确保项目根目录在 sys.path
|
||
PROJECT_ROOT = Path(__file__).parent.parent.resolve()
|
||
sys.path.insert(0, str(PROJECT_ROOT))
|
||
|
||
from app.config import load_config as load_project_config
|
||
from app.extractor import extract_invoices, save_csv, save_markdown
|
||
from app.ocr import enrich_with_ocr, _save_csv as save_ocr_csv, save_markdown_from_csv, _load_csv as load_ocr_csv
|
||
|
||
app = Flask(__name__, template_folder="templates")
|
||
|
||
UPLOAD_BASE = PROJECT_ROOT / "web" / "uploads"
|
||
SESSION_LOG_FILE = "session.log"
|
||
SESSION_RESULT_FILE = "result.json"
|
||
|
||
|
||
# ================================================================
|
||
# 日志收集器 — 捕获管道日志到文件，SSE 端点通过 tail -f 读取
|
||
# ================================================================
|
||
|
||
class _SSELogHandler(logging.Handler):
|
||
"""将日志写入指定文件（线程安全）"""
|
||
|
||
def __init__(self, log_path: Path):
|
||
super().__init__()
|
||
self._lock = threading.Lock()
|
||
self._file = open(log_path, "w", encoding="utf-8")
|
||
|
||
def emit(self, record: logging.LogRecord):
|
||
try:
|
||
msg = self.format(record) + "\n"
|
||
with self._lock:
|
||
self._file.write(msg)
|
||
self._file.flush()
|
||
except Exception:
|
||
pass
|
||
|
||
def close_file(self):
|
||
try:
|
||
self._file.close()
|
||
except Exception:
|
||
pass
|
||
|
||
|
||
def _install_log_collector(session_dir: Path) -> _SSELogHandler:
|
||
"""安装日志收集器到 app.* 模块"""
|
||
log_path = session_dir / SESSION_LOG_FILE
|
||
fmt = logging.Formatter(
|
||
"%(asctime)s [%(levelname)-5s] %(name)s: %(message)s",
|
||
"%Y-%m-%d %H:%M:%S",
|
||
)
|
||
handler = _SSELogHandler(log_path)
|
||
handler.setFormatter(fmt)
|
||
handler.setLevel(logging.INFO)
|
||
|
||
for name in ["extractor", "ocr", "pipeline", "bot"]:
|
||
logger = logging.getLogger(name)
|
||
logger.setLevel(logging.INFO)
|
||
logger.addHandler(handler)
|
||
|
||
return handler
|
||
|
||
|
||
def _remove_log_collector(handler: _SSELogHandler):
|
||
for name in ["extractor", "ocr", "pipeline", "bot"]:
|
||
logging.getLogger(name).removeHandler(handler)
|
||
handler.close_file()
|
||
|
||
|
||
# ================================================================
|
||
# 管道入口
|
||
# ================================================================
|
||
|
||
# ================================================================
|
||
# 管道入口
|
||
# ================================================================
|
||
|
||
def run_pipeline_web(session_dir: Path, config: dict, run_bot: bool = False):
|
||
"""在 Web 会话目录中执行管道，结果写入 session 目录下的文件"""
|
||
start = time.time()
|
||
|
||
# ---- Step 1: 发票提取 ----
|
||
invoices = extract_invoices(str(session_dir))
|
||
if not invoices:
|
||
return {"ok": False, "error": "未提取到任何发票数据"}
|
||
|
||
save_csv(invoices, session_dir / "invoice_summary.csv")
|
||
save_markdown(invoices, session_dir / "invoice_summary.md")
|
||
|
||
# ---- Step 2: OCR 识别 ----
|
||
csv_path = session_dir / "invoice_summary.csv"
|
||
rows = load_ocr_csv(csv_path)
|
||
if rows is None:
|
||
return {"ok": False, "error": "CSV 读取失败"}
|
||
|
||
rows = enrich_with_ocr(rows, str(session_dir))
|
||
save_ocr_csv(csv_path, rows)
|
||
save_markdown_from_csv(csv_path, rows)
|
||
|
||
# ---- Step 3: 浏览器填报（可选）----
|
||
if run_bot:
|
||
from app.bot import load_invoice_data, run_bot_web
|
||
|
||
bot_invoices = load_invoice_data(str(csv_path), config)
|
||
run_bot_web(config, bot_invoices, session_dir)
|
||
|
||
elapsed = time.time() - start
|
||
return {
|
||
"ok": True,
|
||
"elapsed": f"{elapsed:.1f}s",
|
||
"invoice_count": len(rows),
|
||
"csv_url": f"/api/download/{session_dir.name}/invoice_summary.csv",
|
||
"md_url": f"/api/download/{session_dir.name}/invoice_summary.md",
|
||
}
|
||
|
||
|
||
def run_csv_pipeline_web(session_dir: Path, config: dict, csv_filename: str, run_bot: bool = False):
|
||
"""直接使用上传的 CSV 文件进行填报，跳过 PDF 提取和 OCR"""
|
||
start = time.time()
|
||
|
||
csv_path = session_dir / csv_filename
|
||
if not csv_path.exists():
|
||
return {"ok": False, "error": "CSV 文件不存在"}
|
||
|
||
# 读取 CSV 行数
|
||
rows = load_ocr_csv(csv_path)
|
||
if rows is None:
|
||
return {"ok": False, "error": "CSV 读取失败"}
|
||
|
||
# 浏览器填报
|
||
if run_bot:
|
||
from app.bot import load_invoice_data, run_bot_web
|
||
|
||
bot_invoices = load_invoice_data(str(csv_path), config)
|
||
run_bot_web(config, bot_invoices, session_dir)
|
||
|
||
elapsed = time.time() - start
|
||
return {
|
||
"ok": True,
|
||
"elapsed": f"{elapsed:.1f}s",
|
||
"invoice_count": len(rows),
|
||
"csv_url": f"/api/download/{session_dir.name}/{csv_filename}",
|
||
}
|
||
|
||
|
||
# ================================================================
|
||
# Flask 路由
|
||
# ================================================================
|
||
|
||
@app.route("/")
|
||
def index():
|
||
return render_template("index.html")
|
||
|
||
|
||
@app.route("/api/session", methods=["POST"])
|
||
def create_session():
|
||
"""创建上传会话，返回 session_id"""
|
||
sid = uuid.uuid4().hex[:12]
|
||
session_dir = UPLOAD_BASE / sid
|
||
session_dir.mkdir(parents=True, exist_ok=True)
|
||
return jsonify({"session_id": sid})
|
||
|
||
|
||
@app.route("/api/upload/<session_id>", methods=["POST"])
|
||
def upload_file(session_id: str):
|
||
"""上传 PDF 或图片"""
|
||
session_dir = _validate_session(session_id)
|
||
if isinstance(session_dir, tuple):
|
||
return session_dir
|
||
|
||
f = request.files.get("file")
|
||
if not f or not f.filename:
|
||
return jsonify({"error": "未选择文件"}), 400
|
||
|
||
safe_name = Path(f.filename).name
|
||
f.save(str(session_dir / safe_name))
|
||
return jsonify({"ok": True, "filename": safe_name})
|
||
|
||
|
||
@app.route("/api/upload-csv/<session_id>", methods=["POST"])
|
||
def upload_csv(session_id: str):
|
||
"""上传 CSV 发票数据文件（跳过 PDF 提取和 OCR）"""
|
||
session_dir = _validate_session(session_id)
|
||
if isinstance(session_dir, tuple):
|
||
return session_dir
|
||
|
||
f = request.files.get("file")
|
||
if not f or not f.filename:
|
||
return jsonify({"error": "未选择文件"}), 400
|
||
|
||
safe_name = Path(f.filename).name
|
||
f.save(str(session_dir / safe_name))
|
||
return jsonify({"ok": True, "filename": safe_name})
|
||
|
||
|
||
@app.route("/api/files/<session_id>", methods=["GET"])
|
||
def list_files(session_id: str):
|
||
"""列出会话目录中的文件"""
|
||
session_dir = _validate_session(session_id)
|
||
if isinstance(session_dir, tuple):
|
||
return session_dir
|
||
|
||
pdfs = sorted(f.name for f in session_dir.glob("*.pdf"))
|
||
imgs = sorted(
|
||
f.name for ext in {".png", ".jpg", ".jpeg", ".bmp", ".webp"}
|
||
for f in session_dir.glob(f"*{ext}")
|
||
)
|
||
return jsonify({"pdfs": pdfs, "images": imgs})
|
||
|
||
|
||
@app.route("/api/process/<session_id>", methods=["POST"])
|
||
def start_process(session_id: str):
|
||
"""启动管道处理"""
|
||
session_dir = _validate_session(session_id)
|
||
if isinstance(session_dir, tuple):
|
||
return session_dir
|
||
|
||
body = request.get_json(silent=True) or {}
|
||
run_bot_flag = body.get("submit", False)
|
||
mode = body.get("mode", "auto") # "pdf", "csv", or "auto"
|
||
|
||
# 读取配置
|
||
config = _build_web_config(body)
|
||
|
||
# 写入配置到会话目录
|
||
with open(session_dir / "config.json", "w", encoding="utf-8") as f:
|
||
json.dump(config, f, ensure_ascii=False, indent=2, default=str)
|
||
|
||
# 在后台线程执行
|
||
handler = _install_log_collector(session_dir)
|
||
|
||
def _run():
|
||
result = {"ok": False, "error": "未知错误"}
|
||
try:
|
||
if mode == "csv":
|
||
# CSV 模式：直接使用上传的 CSV，跳过 PDF 提取和 OCR
|
||
csv_files = list(session_dir.glob("*.csv"))
|
||
if not csv_files:
|
||
result = {"ok": False, "error": "未找到 CSV 文件"}
|
||
else:
|
||
result = run_csv_pipeline_web(session_dir, config, csv_files[0].name, run_bot_flag)
|
||
else:
|
||
# 自动检测：如果有 CSV 则走 csv 管道，否则走 pdf 管道
|
||
csv_files = list(session_dir.glob("*.csv"))
|
||
pdf_files = list(session_dir.glob("*.pdf"))
|
||
if csv_files and not pdf_files:
|
||
result = run_csv_pipeline_web(session_dir, config, csv_files[0].name, run_bot_flag)
|
||
else:
|
||
result = run_pipeline_web(session_dir, config, run_bot_flag)
|
||
except BaseException as e:
|
||
result = {"ok": False, "error": str(e)}
|
||
if isinstance(e, (KeyboardInterrupt, SystemExit)):
|
||
raise
|
||
finally:
|
||
try:
|
||
# 原子写入：先写临时文件，再重命名，避免 SSE 读到截断的空文件
|
||
tmp_path = session_dir / (SESSION_RESULT_FILE + ".tmp")
|
||
with open(tmp_path, "w", encoding="utf-8") as f:
|
||
json.dump(result, f, ensure_ascii=False)
|
||
tmp_path.replace(session_dir / SESSION_RESULT_FILE)
|
||
except Exception:
|
||
pass
|
||
_remove_log_collector(handler)
|
||
|
||
threading.Thread(target=_run, daemon=True).start()
|
||
|
||
return jsonify({"status": "started"})
|
||
|
||
|
||
@app.route("/api/logs/<session_id>")
|
||
def stream_logs(session_id: str):
|
||
"""SSE 日志流"""
|
||
session_dir = _validate_session(session_id)
|
||
if isinstance(session_dir, tuple):
|
||
return session_dir
|
||
|
||
def generate():
|
||
# 先发送已有日志
|
||
log_file = session_dir / SESSION_LOG_FILE
|
||
last_size = 0
|
||
start_time = time.time()
|
||
timeout = 600 # 10 分钟超时
|
||
|
||
while time.time() - start_time < timeout:
|
||
if log_file.exists():
|
||
current_size = log_file.stat().st_size
|
||
if current_size > last_size:
|
||
with open(log_file, encoding="utf-8", errors="replace") as f:
|
||
f.seek(last_size)
|
||
chunk = f.read()
|
||
if chunk:
|
||
yield f"data: {_escape_sse(chunk)}\n\n"
|
||
last_size = current_size
|
||
|
||
# 也通过队列发送实时日志
|
||
# 检查是否完成
|
||
result_file = session_dir / SESSION_RESULT_FILE
|
||
if result_file.exists():
|
||
with open(result_file, encoding="utf-8") as f:
|
||
result = json.load(f)
|
||
yield f"data: {_escape_sse(json.dumps({'type': 'done', 'result': result}, ensure_ascii=False))}\n\n"
|
||
break
|
||
|
||
time.sleep(0.5)
|
||
|
||
return Response(
|
||
stream_with_context(generate()),
|
||
mimetype="text/event-stream",
|
||
headers={"Cache-Control": "no-cache", "X-Accel-Buffering": "no"},
|
||
)
|
||
|
||
|
||
@app.route("/api/download/<session_id>/<filename>")
|
||
def download_file(session_id: str, filename: str):
|
||
"""下载生成的文件"""
|
||
session_dir = _validate_session(session_id)
|
||
if isinstance(session_dir, tuple):
|
||
return session_dir
|
||
|
||
filepath = session_dir / filename
|
||
if not filepath.exists():
|
||
return jsonify({"error": "文件不存在"}), 404
|
||
|
||
return Response(filepath.read_bytes(), mimetype="application/octet-stream")
|
||
|
||
|
||
# ================================================================
|
||
# 辅助函数
|
||
# ================================================================
|
||
|
||
def _validate_session(session_id: str):
|
||
session_dir = UPLOAD_BASE / session_id
|
||
if not session_dir.exists():
|
||
return jsonify({"error": "会话不存在"}), 404
|
||
return session_dir
|
||
|
||
|
||
def _build_web_config(body: dict) -> dict:
|
||
"""从请求体构建配置"""
|
||
config = load_project_config()
|
||
for key in ("username", "password", "default_name", "default_card_no", "default_person_id"):
|
||
if body.get(key):
|
||
config[key] = body[key]
|
||
return config
|
||
|
||
|
||
def _escape_sse(text: str) -> str:
|
||
"""SSE 数据转义，同时处理 Windows 行尾 \\r\\n"""
|
||
return text.replace("\r\n", "\n").replace("\r", "\n").replace("\n", "\ndata: ")
|
||
|
||
|
||
@app.route("/mobile/<session_id>")
|
||
def mobile_upload(session_id: str):
|
||
"""移动端上传页面"""
|
||
session_dir = UPLOAD_BASE / session_id
|
||
if not session_dir.exists():
|
||
return render_template("mobile_upload.html", error="会话不存在"), 404
|
||
return render_template("mobile_upload.html", session_id=session_id)
|
||
|
||
|
||
@app.route("/api/mobile-upload/<session_id>", methods=["POST"])
|
||
def mobile_upload_file(session_id: str):
|
||
"""移动端上传图片（复用 PC 上传逻辑）"""
|
||
return upload_file(session_id)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
UPLOAD_BASE.mkdir(parents=True, exist_ok=True)
|
||
print(f"启动 Web 服务: http://localhost:5000")
|
||
app.run(host="0.0.0.0", port=5000, debug=True, threaded=True, use_reloader=False) |