82 lines
2.5 KiB
Python
82 lines
2.5 KiB
Python
|
|
import os
|
|
import sys
|
|
import time
|
|
import logging
|
|
from pathlib import Path
|
|
|
|
# 添加项目根目录到路径
|
|
sys.path.append(os.path.abspath(os.path.dirname(__file__)))
|
|
|
|
from app.config.settings import ConfigManager
|
|
from app.services.ocr_service import OCRService
|
|
from app.services.order_service import OrderService
|
|
from app.services.batch_service import BatchService
|
|
|
|
def setup_test_logging():
|
|
logging.basicConfig(
|
|
level=logging.INFO,
|
|
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
|
|
)
|
|
|
|
def test_single_process():
|
|
print("\n--- 测试单文件处理流程 ---")
|
|
config = ConfigManager()
|
|
ocr_service = OCRService(config)
|
|
order_service = OrderService(config)
|
|
|
|
# 获取一张图片
|
|
input_dir = Path(config.get_path('Paths', 'input_folder'))
|
|
images = list(input_dir.glob("*.jpg")) + list(input_dir.glob("*.png"))
|
|
|
|
if not images:
|
|
print("跳过: 没有找到测试图片")
|
|
return
|
|
|
|
test_img = str(images[0])
|
|
print(f"处理图片: {test_img}")
|
|
|
|
# 1. OCR 处理 (包含双 OCR 逻辑)
|
|
excel_path = ocr_service.process_image(test_img)
|
|
if excel_path:
|
|
print(f"OCR 成功: {excel_path}")
|
|
|
|
# 2. 业务处理 (识别元信息 + 重命名)
|
|
result_path = order_service.process_excel(excel_path)
|
|
if result_path:
|
|
print(f"业务处理成功: {result_path}")
|
|
else:
|
|
print("业务处理失败")
|
|
else:
|
|
print("OCR 失败")
|
|
|
|
def test_batch_process():
|
|
print("\n--- 测试批量处理流程 ---")
|
|
config = ConfigManager()
|
|
batch_service = BatchService(config)
|
|
|
|
def progress(done, total, entry):
|
|
print(f"进度: {done}/{total} - {entry.get('status')} - {entry.get('image')}")
|
|
|
|
summary = batch_service.process_all_inputs(progress_cb=progress)
|
|
print(f"批量处理汇总: 总数={summary['total']}, 成功={summary['success']}, 失败={summary['failed']}")
|
|
|
|
if __name__ == "__main__":
|
|
setup_test_logging()
|
|
|
|
# 测试前先清理一下记录，确保会重新处理
|
|
config = ConfigManager()
|
|
pjson = config.get_path('Paths', 'processed_record')
|
|
if os.path.exists(pjson):
|
|
# os.remove(pjson) # 不真正删除，避免影响用户数据
|
|
pass
|
|
|
|
try:
|
|
test_single_process()
|
|
time.sleep(1) # 间隔一下
|
|
test_batch_process()
|
|
except Exception as e:
|
|
print(f"测试过程中出现异常: {e}")
|
|
import traceback
|
|
traceback.print_exc()
|