""" 邮件处理管线 - 协调解析、分类、匹配、写入的完整流程 """ import asyncio import logging from dataclasses import dataclass, field from datetime import datetime from typing import Optional, Callable from .email_parser import ParsedEmail, parse_eml_file from .ai_classifier import ClassificationResult, classify_email_async from .sheets_updater import SheetsUpdater, MatchResult, WriteResult logger = logging.getLogger(__name__) # 处理状态常量 STATUS_PENDING = "待处理" STATUS_PARSING = "解析中" STATUS_CLASSIFYING = "AI分析中" STATUS_MATCHING = "匹配公司中" STATUS_WRITING = "回填中" STATUS_DONE = "✅ 已完成" STATUS_AMBIGUOUS = "⚠️ 需人工确认" STATUS_NOT_FOUND = "❌ 未找到公司" STATUS_PARSE_ERROR = "❌ 解析失败" STATUS_AI_ERROR = "❌ AI分析失败" STATUS_WRITE_ERROR = "❌ 写入失败" STATUS_SKIPPED = "⏭️ 已跳过" @dataclass class EmailProcessResult: """单封邮件的完整处理结果""" # 输入信息 filename: str file_path: str = "" # 解析结果 sender: str = "" recipient: str = "" date_str: str = "" subject: str = "" body_preview: str = "" body_text: str = "" # AI 分析结果 company_name: str = "" category: str = "" summary: str = "" ai_confidence: str = "" # 匹配结果 match_status: str = "" matched_row: int = -1 matched_name: str = "" matched_sheet: str = "" # 匹配到的工作表名称（West/South/Northeast/Midwest） match_score: float = 0.0 candidates: list = field(default_factory=list) # 写入结果 write_action: str = "" write_reason: str = "" # 最终状态 status: str = STATUS_PENDING error_msg: str = "" # 时间戳 processed_at: str = "" @property def needs_confirmation(self) -> bool: return self.match_status == "ambiguous" @property def to_table_row(self) -> list: """格式化为表格展示行""" return [ self.filename, self.sender, self.date_str, self.company_name, self.category, self.summary or self.body_preview, self.status, self.write_reason or self.error_msg, ] class EmailProcessor: """邮件处理管线""" def __init__(self, sheets_updater: Optional[SheetsUpdater] = None): self.sheets_updater = sheets_updater self._results: list[EmailProcessResult] = [] self._pending_confirmations: list[EmailProcessResult] = [] def _update_status(self, result: EmailProcessResult, status: str, callback: Optional[Callable] = None): """更新状态并触发回调""" result.status = status if callback: callback(result) async def process_single_email( self, file_path: str, progress_callback: Optional[Callable] = None, ) -> EmailProcessResult: """ 处理单封邮件的完整流程 Args: file_path: .eml 文件路径 progress_callback: 进度回调 (result) -> None Returns: EmailProcessResult """ import os filename = os.path.basename(file_path) result = EmailProcessResult(filename=filename, file_path=file_path) # Step 1: 解析 self._update_status(result, STATUS_PARSING, progress_callback) try: email = parse_eml_file(file_path) if not email.is_valid: result.status = STATUS_PARSE_ERROR result.error_msg = email.parse_error return result result.sender = email.sender result.recipient = email.recipient result.date_str = email.date_str result.subject = email.subject result.body_preview = email.body_preview result.body_text = email.body_text except Exception as e: result.status = STATUS_PARSE_ERROR result.error_msg = str(e) logger.error(f"解析失败 ({filename}): {e}") return result # Step 2: AI 分类 self._update_status(result, STATUS_CLASSIFYING, progress_callback) try: classification = await classify_email_async(email) if not classification.is_valid: result.status = STATUS_AI_ERROR result.error_msg = classification.error # AI 失败但解析成功，保留邮件信息仍可人工处理 result.company_name = classification.company_name or email.sender_domain.title() return result result.company_name = classification.company_name result.category = classification.category result.summary = classification.summary result.ai_confidence = classification.confidence except Exception as e: result.status = STATUS_AI_ERROR result.error_msg = str(e) logger.error(f"AI 分类失败 ({filename}): {e}") return result # Step 3: 模糊匹配 if self.sheets_updater and self.sheets_updater.is_connected(): self._update_status(result, STATUS_MATCHING, progress_callback) try: match = self.sheets_updater.fuzzy_match_company(result.company_name) result.match_status = match.status result.matched_row = match.row_index result.matched_name = match.matched_name result.matched_sheet = match.sheet_name result.match_score = match.score result.candidates = match.candidates if match.status == "not_found": result.status = STATUS_NOT_FOUND result.error_msg = match.message return result if match.status == "ambiguous": result.status = STATUS_AMBIGUOUS result.error_msg = match.message return result # Step 4: 写入（传入匹配到的工作表名） self._update_status(result, STATUS_WRITING, progress_callback) write_result = self.sheets_updater.write_email_data( row_index=result.matched_row, category=result.category, body_text=result.body_text, date_str=result.date_str, sheet_name=result.matched_sheet, ) result.write_action = write_result.action result.write_reason = write_result.reason if not write_result.success: result.status = STATUS_WRITE_ERROR result.error_msg = write_result.error return result if write_result.action == "skipped": result.status = STATUS_SKIPPED else: result.status = STATUS_DONE except Exception as e: result.status = STATUS_WRITE_ERROR result.error_msg = str(e) logger.error(f"Sheets 操作失败 ({filename}): {e}") return result else: # Sheets 未连接，仅展示分析结果 result.match_status = "sheets_not_configured" result.status = "✅ 分析完成（未配置Sheets）" result.processed_at = datetime.now().strftime("%Y-%m-%d %H:%M:%S") self._update_status(result, result.status, progress_callback) return result async def process_batch( self, file_paths: list[str], progress_callback: Optional[Callable] = None, ) -> tuple[list[EmailProcessResult], list[EmailProcessResult]]: """ 批量处理邮件 Args: file_paths: 文件路径列表 progress_callback: 进度回调 (current, total, result) -> None Returns: (completed_results, pending_confirmation_results) """ self._results = [] self._pending_confirmations = [] semaphore = asyncio.Semaphore(5) # 最大并发 5 completed_count = [0] async def process_one(file_path: str): async with semaphore: result = await self.process_single_email(file_path) completed_count[0] += 1 self._results.append(result) if result.needs_confirmation: self._pending_confirmations.append(result) if progress_callback: await asyncio.coroutine( lambda: progress_callback(completed_count[0], len(file_paths), result) )() if asyncio.iscoroutinefunction(progress_callback) else progress_callback( completed_count[0], len(file_paths), result ) return result tasks = [process_one(fp) for fp in file_paths] await asyncio.gather(*tasks, return_exceptions=True) return self._results, self._pending_confirmations def confirm_and_write( self, result: EmailProcessResult, confirmed_row: int, confirmed_name: str, confirmed_sheet: str = "", ) -> WriteResult: """ 人工确认后强制写入 Args: result: 待确认的处理结果 confirmed_row: 用户选择的行号 confirmed_name: 用户选择的公司名 confirmed_sheet: 用户选择的工作表名（West/South/Northeast/Midwest） Returns: WriteResult """ if not self.sheets_updater or not self.sheets_updater.is_connected(): from .sheets_updater import WriteResult return WriteResult(success=False, error="Google Sheets 未连接") # 优先使用传入的工作表名，否则从候选中查找 sheet_name = confirmed_sheet if not sheet_name: # 尝试从候选列表里根据行号找工作表名 for c in result.candidates: if c.get("row") == confirmed_row: sheet_name = c.get("sheet", "") break write_result = self.sheets_updater.write_email_data( row_index=confirmed_row, category=result.category, body_text=result.body_text, date_str=result.date_str, force_overwrite=True, sheet_name=sheet_name, ) if write_result.success: result.matched_row = confirmed_row result.matched_name = confirmed_name result.matched_sheet = sheet_name result.status = STATUS_DONE result.write_action = write_result.action sheet_info = f" [{sheet_name}]" if sheet_name else "" result.write_reason = f"人工确认写入（{confirmed_name}{sheet_info}）" return write_result