""" 订单服务模块 --------- 提供订单处理服务，协调Excel处理和订单合并流程。 """ import os import json import re import shutil from datetime import datetime from pathlib import Path from typing import Dict, List, Optional, Tuple, Union, Any, Callable from ..config.settings import ConfigManager from ..core.utils.log_utils import get_logger from ..core.excel.processor import ExcelProcessor from ..core.excel.merger import PurchaseOrderMerger from ..core.db.product_db import ProductDatabase from ..core.db.order_metadata_db import OrderMetadataDB from ..core.ocr.metadata_extractor import OrderMetadataExtractor, sanitize_for_filename logger = get_logger(__name__) class OrderService: """ 订单服务：协调Excel处理和订单合并流程 """ def __init__(self, config: Optional[ConfigManager] = None): """ 初始化订单服务 Args: config: 配置管理器，如果为None则创建新的 """ logger.info("初始化OrderService") self.config = config or ConfigManager() # 创建共享的商品数据库实例 db_path = self.config.get_path('Paths', 'product_db', fallback='data/product_cache.db') if hasattr(self.config, 'get_path') else 'data/product_cache.db' tpl_folder = self.config.get('Paths', 'template_folder', fallback='templates') item_data = self.config.get('Templates', 'item_data', fallback='商品资料.xlsx') tpl_path = os.path.join(tpl_folder, item_data) self.product_db = ProductDatabase(db_path, tpl_path) # 创建Excel处理器和采购单合并器 self.excel_processor = ExcelProcessor(self.config, product_db=self.product_db) self.order_merger = PurchaseOrderMerger(self.config) # 元信息识别器 + 单据元信息库 self.extractor = OrderMetadataExtractor() self.metadata_db = OrderMetadataDB(db_path) logger.info("OrderService初始化完成") def get_latest_excel(self) -> Optional[str]: """ 获取最新的Excel文件 Returns: 最新Excel文件路径，如果未找到则返回None """ return self.excel_processor.get_latest_excel() def process_excel(self, file_path: Optional[str] = None, progress_cb: Optional[Callable[[int], None]] = None) -> Optional[str]: """ 处理Excel订单文件，生成标准采购单 Args: file_path: Excel文件路径，如果为None则处理最新的文件 Returns: 输出采购单文件路径，如果处理失败则返回None """ if not file_path: file_path = self.excel_processor.get_latest_excel() if not file_path: logger.warning("未找到可处理的Excel文件") return None logger.info("OrderService开始处理最新Excel文件") else: logger.info(f"OrderService开始处理指定Excel文件: {file_path}") # 检查是否需要特殊的供应商预处理（如杨碧月） try: from .special_suppliers_service import SpecialSuppliersService special_service = SpecialSuppliersService(self.config) # 尝试识别并预处理（注意：这里不再传入 progress_cb 避免无限递归或重复进度条， # 或者我们在 special_service 内部逻辑中处理完后直接返回结果） # 为了避免循环调用，我们在 SpecialSuppliersService 内部不再调用 process_excel， # 而是让 process_excel 识别后自己决定是否处理预处理后的文件。 # 我们新增一个 check_and_preprocess 方法 preprocessed_path = self._check_special_preprocess(file_path) if preprocessed_path: logger.info(f"检测到特殊供应商，已生成预处理文件: {preprocessed_path}") file_path = preprocessed_path except Exception as e: logger.error(f"检查特殊预处理时出错: {e}") result_path = self.excel_processor.process_specific_file(file_path, progress_cb=progress_cb) if not result_path: return None # 应用单据元信息识别 + 重命名 result 与原图 try: meta = self._extract_and_save_metadata(file_path) if meta: result_path = self._apply_metadata_to_filenames( result_path, file_path, meta ) except Exception as e: logger.error(f"应用单据元信息失败（不影响 result 文件）: {e}") return result_path def _check_special_preprocess(self, file_path: str) -> Optional[str]: """检查并执行特殊的预处理（支持杨碧月、烟草公司、蓉城易购）""" try: from app.core.utils.file_utils import smart_read_excel import pandas as pd import re # 仅读取前 50 行进行智能识别 (header=None 确保能读到第一行内容) df_head = smart_read_excel(file_path, nrows=50, header=None) df_str = df_head.astype(str) # 1. 识别：烟草公司 (Tobacco) # 特征：内容中包含“专卖证号”或特定证号“510109104938” is_tobacco = df_str.apply(lambda x: x.str.contains('专卖证号|510109104938')).any().any() if is_tobacco: logger.info("识别到烟草公司订单，执行专用预处理...") from .tobacco_service import TobaccoService tobacco_svc = TobaccoService(self.config) return tobacco_svc.preprocess_tobacco_order(file_path) # 2. 识别：蓉城易购 (Rongcheng Yigou) # 特征：内容中包含单号标识“RCDH” is_rongcheng = df_str.apply(lambda x: x.str.contains('RCDH')).any().any() if is_rongcheng: logger.info("识别到蓉城易购订单，执行专用预处理...") from .special_suppliers_service import SpecialSuppliersService special_svc = SpecialSuppliersService(self.config) return special_svc.preprocess_rongcheng_yigou(file_path) # 3. 识别：杨碧月 (Yang Biyue) # 特征：经手人列包含“杨碧月” handler_col = None for col in df_head.columns: # 在前50行中搜索“经手人”关键字 if df_head[col].astype(str).str.contains('经手人').any(): handler_col = col break if handler_col is not None: # 检查该列是否有“杨碧月” if df_head[handler_col].astype(str).str.contains('杨碧月').any(): logger.info("识别到杨碧月订单，执行专用预处理...") from .special_suppliers_service import SpecialSuppliersService special_svc = SpecialSuppliersService(self.config) return special_svc.process_yang_biyue_only(file_path) except Exception as e: logger.warning(f"智能预处理识别失败: {e}") return None def get_purchase_orders(self) -> List[str]: """ 获取采购单文件列表 Returns: 采购单文件路径列表 """ return self.order_merger.get_purchase_orders() def merge_purchase_orders(self, file_paths: List[str], progress_cb: Optional[Callable[[int], None]] = None) -> Optional[str]: """ 合并指定的采购单文件 Args: file_paths: 采购单文件路径列表 Returns: 合并后的采购单文件路径，如果合并失败则返回None """ logger.info(f"OrderService开始合并指定采购单: {file_paths}") return self.merge_orders(file_paths, progress_cb) def merge_all_purchase_orders(self, progress_cb: Optional[Callable[[int], None]] = None) -> Optional[str]: """ 合并所有可用的采购单文件 Returns: 合并后的采购单文件路径，如果合并失败则返回None """ logger.info("OrderService开始合并所有采购单") return self.merge_orders(None, progress_cb) def merge_orders(self, file_paths: Optional[List[str]] = None, progress_cb: Optional[Callable[[int], None]] = None) -> Optional[str]: """ 合并采购单 Args: file_paths: 采购单文件路径列表，如果为None则处理所有采购单 Returns: 合并后的采购单文件路径，如果合并失败则返回None """ if file_paths: logger.info(f"OrderService开始合并指定采购单: {file_paths}") else: logger.info("OrderService开始合并所有采购单") return self.order_merger.process(file_paths, progress_cb) def validate_unit_price(self, result_path: str) -> List[str]: """ 校验采购单单价与商品资料进货价的差异 Args: result_path: 待校验的采购单路径 Returns: 差异信息列表，无差异返回空列表 """ try: import pandas as pd from app.core.utils.file_utils import smart_read_excel from app.core.handlers.column_mapper import ColumnMapper as CM # 使用共享的商品数据库实例 product_db = self.product_db # 读取待校验的采购单 df_res = smart_read_excel(result_path) res_barcode_col = CM.find_column(list(df_res.columns), 'barcode') res_price_col = CM.find_column(list(df_res.columns), 'unit_price') if not res_barcode_col or not res_price_col: logger.warning("未能在采购单中找到条码或单价列") return [] # 批量查询进货价 barcodes = df_res[res_barcode_col].astype(str).str.strip().tolist() item_prices = product_db.get_prices(barcodes) results = [] for _, row in df_res.iterrows(): bc = str(row[res_barcode_col]).strip() if bc not in item_prices: continue try: res_price = float(row[res_price_col]) except (ValueError, TypeError): continue item_price = item_prices[bc] diff = abs(res_price - item_price) if diff > 1.0: results.append(f"条码 {bc}: 采购单价={res_price} vs 进货价={item_price} 差异={diff:.2f}") return results except Exception as e: logger.error(f"单价校验过程中发生错误: {e}") return [] # ══════════════════════════════════════════════════════════════ # 单据元信息识别 + 文件重命名 # ══════════════════════════════════════════════════════════════ def _extract_and_save_metadata(self, ocr_excel_path: str) -> Optional[Any]: """从 OCR 输出的 xlsx 同目录的 .meta.json 提取元信息，写入 SQLite。 Returns: OrderMetadata 或 None（失败时） """ try: base = Path(ocr_excel_path) file_hash = base.stem meta_path = base.with_suffix('.meta.json') ocr_text = '' ocr_rows: List[List[str]] = [] source_image = '' if meta_path.exists(): try: payload = json.loads(meta_path.read_text(encoding='utf-8')) # 优先用通用识别文本（含手写抬头/日期） ocr_text = payload.get('general_text') or payload.get('ocr_text', '') or '' ocr_rows = payload.get('ocr_rows', []) or [] source_image = payload.get('image_path', '') or '' except Exception as e: logger.warning(f"读 meta.json 失败: {e}") # 兜底：从 xlsx 拼文本（与 OCRService._write_meta_json 的兜底一致） if not ocr_text: try: import xlrd rb = xlrd.open_workbook(str(ocr_excel_path)) ws = rb.sheet_by_index(0) lines = [] for r in range(ws.nrows): row_vals = [str(ws.cell_value(r, c)) for c in range(ws.ncols)] ocr_rows.append(row_vals) lines.append(' '.join(row_vals)) ocr_text = '\n'.join(lines) except Exception as e: logger.debug(f"从 xlsx 拼 OCR 文本失败: {e}") meta = self.extractor.extract(ocr_text, ocr_rows) self.metadata_db.save( file_hash=file_hash, supplier=meta.supplier, bill_date=meta.bill_date, total_amount=meta.total_amount, raw_supplier_text=meta.raw_supplier_text, source_image=source_image, ) logger.info( f"元信息识别: hash={file_hash} supplier={meta.supplier!r} " f"bill_date={meta.bill_date!r} total_amount={meta.total_amount:.2f}" ) return meta except Exception as e: logger.error(f"_extract_and_save_metadata 失败: {e}") return None def _apply_metadata_to_filenames(self, result_path: str, ocr_excel_path: str, meta) -> str: """应用新文件名规则： - result: 采购单_{YYYYMMDD}_{供应商}_{hash}.xls - 原图: {原stem}_{YYYYMMDD}_{供应商}_{hash}.{ext} 任一步骤失败不影响 result 文件本身。 Returns: 新 result 路径（无论重命名是否成功都返回；失败时返回原路径） """ try: file_hash = Path(ocr_excel_path).stem supplier_clean = sanitize_for_filename(meta.supplier) or '未知供应商' date_part = meta.bill_date or '未知日期' # ── 1. result 重命名 ── new_result_name = f"采购单_{date_part}_{supplier_clean}_{file_hash}.xls" result_dir = Path(result_path).parent new_result_path = result_dir / new_result_name try: # 冲突时加 _N if new_result_path.exists() and str(new_result_path) != str(result_path): new_result_path = self._dedup_path(new_result_path) if str(new_result_path) != str(result_path): os.rename(result_path, str(new_result_path)) logger.info(f"result 重命名: {result_path} -> {new_result_path}") result_path = str(new_result_path) except Exception as e: logger.warning(f"result 重命名失败: {e}") # ── 2. 原图重命名 ── try: meta_row = self.metadata_db.get(file_hash) src_image = (meta_row or {}).get('source_image', '') if src_image and os.path.exists(src_image): src_p = Path(src_image) stem = src_p.stem ext = src_p.suffix new_image_name = f"{stem}_{date_part}_{supplier_clean}_{file_hash}{ext}" new_image_path = src_p.parent / new_image_name # 不覆盖已重命名的图片 if str(new_image_path) != str(src_p) and not new_image_path.exists(): os.rename(src_p, new_image_path) logger.info(f"原图重命名: {src_p.name} -> {new_image_path.name}") # 更新 source_image 路径 self.metadata_db.save( file_hash=file_hash, supplier=meta.supplier, bill_date=meta.bill_date, total_amount=meta.total_amount, raw_supplier_text=meta.raw_supplier_text, source_image=str(new_image_path), ) except Exception as e: logger.warning(f"原图重命名失败: {e}") return result_path except Exception as e: logger.error(f"_apply_metadata_to_filenames 失败: {e}") return result_path @staticmethod def _dedup_path(p: Path) -> Path: """路径冲突时加 _N 后缀。""" stem, suffix = p.stem, p.suffix parent = p.parent n = 1 while True: cand = parent / f"{stem}_{n}{suffix}" if not cand.exists(): return cand n += 1