# -*- coding: utf-8 -*- """ 工单同步服务 实现飞书多维表格与本地工单系统的双向同步 """ import json import logging from typing import Dict, List, Optional, Any from datetime import datetime from src.integrations.feishu_client import FeishuClient from src.integrations.ai_suggestion_service import AISuggestionService from src.integrations.flexible_field_mapper import FlexibleFieldMapper from src.core.database import db_manager from src.core.models import WorkOrder # 工单状态和优先级枚举 class WorkOrderStatus: PENDING = "pending" IN_PROGRESS = "in_progress" COMPLETED = "completed" CLOSED = "closed" class WorkOrderPriority: LOW = "low" MEDIUM = "medium" HIGH = "high" URGENT = "urgent" logger = logging.getLogger(__name__) class WorkOrderSyncService: """工单同步服务""" def __init__(self, feishu_client: FeishuClient, app_token: str, table_id: str): """ 初始化同步服务 Args: feishu_client: 飞书客户端 app_token: 多维表格应用token table_id: 表格ID """ self.feishu_client = feishu_client self.app_token = app_token self.table_id = table_id self.ai_service = AISuggestionService() # 初始化灵活字段映射器 self.field_mapper = FlexibleFieldMapper() # 保留原有的字段映射作为默认配置（向后兼容） self.field_mapping = { # 核心字段 "TR Number": "order_id", # TR编号映射到工单号 "TR Description": "description", # TR描述作为详细描述 "Type of problem": "category", # 问题类型作为分类 "TR Level": "priority", # TR Level作为优先级 "TR Status": "status", # TR Status作为状态 "Source": "source", # 来源信息（Mail, Telegram bot等） "Date creation": "created_at", # 创建日期 "处理过程": "solution", # 处理过程作为解决方案 "TR tracking": "resolution", # TR跟踪作为解决方案详情 # 扩展字段 "Created by": "created_by", # 创建人 "Module（模块）": "module", # 模块 "Wilfulness（责任人）": "wilfulness", # 责任人 "Date of close TR": "date_of_close", # 关闭日期 "Vehicle Type01": "vehicle_type", # 车型 "VIN|sim": "vin_sim", # 车架号/SIM "App remote control version": "app_remote_control_version", # 应用远程控制版本 "HMI SW": "hmi_sw", # HMI软件版本 "父记录": "parent_record", # 父记录 "Has it been updated on the same day": "has_updated_same_day", # 是否同日更新 "Operating time": "operating_time", # 操作时间 # AI建议字段 "AI建议": "ai_suggestion", # AI建议字段 "Issue Start Time": "updated_at" # 问题开始时间作为更新时间 } # 将原有映射添加到灵活映射器中 self._init_flexible_mapper() # 状态映射 - 根据飞书表格中的实际值 self.status_mapping = { "close": WorkOrderStatus.CLOSED, # 已关闭 "temporary close": WorkOrderStatus.IN_PROGRESS, # 临时关闭对应处理中 "OTA": WorkOrderStatus.IN_PROGRESS, # OTA状态对应处理中 "open": WorkOrderStatus.PENDING, # 开放状态对应待处理 "pending": WorkOrderStatus.PENDING, # 待处理 "completed": WorkOrderStatus.COMPLETED # 已完成 } # 优先级映射 - 根据飞书表格中的实际值 self.priority_mapping = { "Low": WorkOrderPriority.LOW, "Medium": WorkOrderPriority.MEDIUM, "High": WorkOrderPriority.HIGH, "Urgent": WorkOrderPriority.URGENT } def _init_flexible_mapper(self): """初始化灵活映射器，将原有映射添加到其中""" for feishu_field, local_field in self.field_mapping.items(): self.field_mapper.add_field_mapping(feishu_field, local_field) def get_field_discovery_report(self, feishu_fields: Dict[str, Any]) -> Dict[str, Any]: """ 获取字段发现报告 Args: feishu_fields: 飞书字段数据 Returns: 字段发现报告 """ return self.field_mapper.discover_fields(feishu_fields) def add_field_mapping(self, feishu_field: str, local_field: str, aliases: List[str] = None, patterns: List[str] = None, priority: int = 3) -> bool: """ 添加字段映射 Args: feishu_field: 飞书字段名 local_field: 本地字段名 aliases: 别名列表 patterns: 模式列表 priority: 优先级 Returns: 是否添加成功 """ return self.field_mapper.add_field_mapping(feishu_field, local_field, aliases, patterns, priority) def remove_field_mapping(self, feishu_field: str) -> bool: """ 移除字段映射 Args: feishu_field: 飞书字段名 Returns: 是否移除成功 """ return self.field_mapper.remove_field_mapping(feishu_field) def get_mapping_status(self) -> Dict[str, Any]: """ 获取映射状态 Returns: 映射状态信息 """ return self.field_mapper.get_mapping_status() def sync_from_feishu(self, generate_ai_suggestions: bool = True, limit: int = 10) -> Dict[str, Any]: """ 从飞书同步数据到本地系统 Args: generate_ai_suggestions: 是否生成AI建议 limit: 处理记录数量限制 Returns: 同步结果统计 """ try: logger.info("开始从飞书同步工单数据...") # 获取飞书表格记录（限制数量） records = self.feishu_client.get_table_records(self.app_token, self.table_id, page_size=limit) if records.get("code") != 0: raise Exception(f"获取飞书记录失败: {records.get('msg', '未知错误')}") items = records.get("data", {}).get("items", []) logger.info(f"从飞书获取到 {len(items)} 条记录") # 生成AI建议 if generate_ai_suggestions: logger.info("开始生成AI建议...") items = self.ai_service.batch_generate_suggestions(items, limit) # 将AI建议更新回飞书表格 for item in items: if "ai_suggestion" in item: try: self.feishu_client.update_table_record( self.app_token, self.table_id, item["record_id"], {"AI建议": item["ai_suggestion"]} ) logger.info(f"更新飞书记录 {item['record_id']} 的AI建议") except Exception as e: logger.error(f"更新飞书AI建议失败: {e}") synced_count = 0 updated_count = 0 created_count = 0 errors = [] with db_manager.get_session() as session: for record in items: try: # 解析飞书记录 parsed_fields = self.feishu_client.parse_record_fields(record) feishu_id = record.get("record_id") # 查找本地是否存在对应记录 existing_workorder = session.query(WorkOrder).filter( WorkOrder.feishu_record_id == feishu_id ).first() # 转换为本地工单格式 workorder_data = self._convert_feishu_to_local(parsed_fields) workorder_data["feishu_record_id"] = feishu_id if existing_workorder: # 更新现有记录 for key, value in workorder_data.items(): if key != "feishu_record_id": setattr(existing_workorder, key, value) existing_workorder.updated_at = datetime.now() updated_count += 1 else: # 创建新记录 workorder_data["created_at"] = datetime.now() workorder_data["updated_at"] = datetime.now() new_workorder = WorkOrder(**workorder_data) session.add(new_workorder) created_count += 1 synced_count += 1 except Exception as e: error_msg = f"处理记录 {record.get('record_id', 'unknown')} 失败: {str(e)}" logger.error(error_msg) errors.append(error_msg) session.commit() result = { "success": True, "total_records": len(items), "synced_count": synced_count, "created_count": created_count, "updated_count": updated_count, "ai_suggestions_generated": generate_ai_suggestions, "errors": errors } logger.info(f"飞书同步完成: {result}") return result except Exception as e: logger.error(f"飞书同步失败: {e}") return { "success": False, "error": str(e) } def sync_to_feishu(self, workorder_id: int) -> Dict[str, Any]: """ 将本地工单同步到飞书 Args: workorder_id: 工单ID Returns: 同步结果 """ try: with db_manager.get_session() as session: workorder = session.query(WorkOrder).filter(WorkOrder.id == workorder_id).first() if not workorder: return {"success": False, "error": "工单不存在"} # 转换为飞书格式 feishu_fields = self._convert_local_to_feishu(workorder) if workorder.feishu_record_id: # 更新飞书记录 result = self.feishu_client.update_table_record( self.app_token, self.table_id, workorder.feishu_record_id, feishu_fields ) else: # 创建新飞书记录 result = self.feishu_client.create_table_record( self.app_token, self.table_id, feishu_fields ) if result.get("code") == 0: # 保存飞书记录ID到本地 workorder.feishu_record_id = result["data"]["record"]["record_id"] session.commit() if result.get("code") == 0: return {"success": True, "message": "同步成功"} else: return {"success": False, "error": result.get("msg", "同步失败")} except Exception as e: logger.error(f"同步到飞书失败: {e}") return {"success": False, "error": str(e)} def create_workorder_from_feishu_record(self, record_id: str) -> Dict[str, Any]: """ 从飞书单条记录创建工单 Args: record_id: 飞书记录ID Returns: 创建结果 """ try: logger.info(f"从飞书记录 {record_id} 创建工单") # 获取单条飞书记录 feishu_data = self.feishu_client.get_table_record( self.app_token, self.table_id, record_id ) if feishu_data.get("code") != 0: return { "success": False, "message": f"获取飞书记录失败: {feishu_data.get('msg', '未知错误')}" } record = feishu_data.get("data", {}).get("record") if not record: return { "success": False, "message": "飞书记录不存在" } fields = record.get("fields", {}) # 转换为本地工单格式 local_data = self._convert_feishu_to_local(fields) local_data["feishu_record_id"] = record_id # 检查是否已存在 existing_workorder = self._find_existing_workorder(record_id) if existing_workorder: return { "success": False, "message": f"工单已存在: {existing_workorder.order_id}" } # 创建新工单 workorder = self._create_workorder(local_data) return { "success": True, "message": f"工单创建成功: {local_data.get('order_id')}", "workorder_id": workorder.id, "order_id": local_data.get('order_id') } except Exception as e: logger.error(f"从飞书记录创建工单失败: {e}") return { "success": False, "message": f"创建工单失败: {str(e)}" } def _find_existing_workorder(self, feishu_record_id: str) -> Optional[WorkOrder]: """查找已存在的工单""" try: with db_manager.get_session() as session: return session.query(WorkOrder).filter( WorkOrder.feishu_record_id == feishu_record_id ).first() except Exception as e: logger.error(f"查找现有工单失败: {e}") return None def _create_workorder(self, local_data: Dict[str, Any]) -> WorkOrder: """创建新工单""" try: with db_manager.get_session() as session: workorder = WorkOrder( order_id=local_data.get("order_id"), title=local_data.get("title"), description=local_data.get("description"), category=local_data.get("category"), priority=local_data.get("priority"), status=local_data.get("status"), created_at=local_data.get("created_at"), updated_at=local_data.get("updated_at"), resolution=local_data.get("solution"), feishu_record_id=local_data.get("feishu_record_id"), assignee=local_data.get("assignee"), solution=local_data.get("solution"), ai_suggestion=local_data.get("ai_suggestion") ) session.add(workorder) session.commit() session.refresh(workorder) logger.info(f"创建工单成功: {workorder.order_id}") return workorder except Exception as e: logger.error(f"创建工单失败: {e}") raise def _update_workorder(self, workorder: WorkOrder, local_data: Dict[str, Any]) -> WorkOrder: """更新现有工单""" try: with db_manager.get_session() as session: workorder.title = local_data.get("title", workorder.title) workorder.description = local_data.get("description", workorder.description) workorder.category = local_data.get("category", workorder.category) workorder.priority = local_data.get("priority", workorder.priority) workorder.status = local_data.get("status", workorder.status) workorder.updated_at = local_data.get("updated_at", workorder.updated_at) workorder.resolution = local_data.get("solution", workorder.resolution) workorder.assignee = local_data.get("assignee", workorder.assignee) workorder.solution = local_data.get("solution", workorder.solution) workorder.ai_suggestion = local_data.get("ai_suggestion", workorder.ai_suggestion) session.commit() session.refresh(workorder) logger.info(f"更新工单成功: {workorder.order_id}") return workorder except Exception as e: logger.error(f"更新工单失败: {e}") raise def _update_feishu_ai_suggestion(self, record_id: str, ai_suggestion: str) -> bool: """更新飞书表格中的AI建议""" try: result = self.feishu_client.update_record( self.app_token, self.table_id, record_id, {"AI建议": ai_suggestion} ) return result.get("code") == 0 except Exception as e: logger.error(f"更新飞书AI建议失败: {e}") return False def _convert_feishu_to_local(self, feishu_fields: Dict[str, Any]) -> Dict[str, Any]: """将飞书字段转换为本地工单字段""" logger.info(f"开始转换飞书字段: {feishu_fields}") # 使用灵活映射器进行字段转换 local_data, conversion_stats = self.field_mapper.convert_fields(feishu_fields) # 记录转换统计信息 logger.info(f"字段转换统计: 总字段 {conversion_stats['total_fields']}, " f"已映射 {conversion_stats['mapped_fields']}, " f"未映射 {len(conversion_stats['unmapped_fields'])}") # 如果有未映射的字段，记录详细信息 if conversion_stats['unmapped_fields']: logger.warning(f"未映射字段: {conversion_stats['unmapped_fields']}") for field in conversion_stats['unmapped_fields']: suggestions = conversion_stats['mapping_details'][field].get('suggestions', []) if suggestions: logger.info(f"字段 '{field}' 的建议映射: {suggestions[0] if suggestions else '无'}") # 特殊字段处理 for local_field, value in local_data.items(): if local_field == "status" and value in self.status_mapping: local_data[local_field] = self.status_mapping[value] elif local_field == "priority" and value in self.priority_mapping: local_data[local_field] = self.priority_mapping[value] elif local_field in ["created_at", "updated_at", "date_of_close"] and value: try: # 处理飞书时间戳（毫秒） if isinstance(value, (int, float)): # 飞书时间戳是毫秒，需要转换为秒 local_data[local_field] = datetime.fromtimestamp(value / 1000) else: # 处理ISO格式时间字符串 local_data[local_field] = datetime.fromisoformat(value.replace('Z', '+00:00')) except Exception as e: logger.warning(f"时间字段转换失败: {e}, 使用当前时间") local_data[local_field] = datetime.now() # 生成标题 - 使用TR Number和问题类型 tr_number = feishu_fields.get("TR Number", "") problem_type = feishu_fields.get("Type of problem", "") if tr_number and problem_type: local_data["title"] = f"{tr_number} - {problem_type}" elif tr_number: local_data["title"] = f"{tr_number} - TR工单" else: local_data["title"] = "TR工单" # 设置默认值 if "status" not in local_data: local_data["status"] = WorkOrderStatus.PENDING if "priority" not in local_data: local_data["priority"] = WorkOrderPriority.MEDIUM if "category" not in local_data: local_data["category"] = "Remote control" # 根据表格中最常见的问题类型 return local_data def _convert_local_to_feishu(self, workorder: WorkOrder) -> Dict[str, Any]: """将本地工单字段转换为飞书字段""" feishu_fields = {} # 反向映射 reverse_mapping = {v: k for k, v in self.field_mapping.items()} for local_field, feishu_field in reverse_mapping.items(): value = getattr(workorder, local_field, None) if value is not None: # 特殊字段处理 if local_field == "status": # 反向状态映射 reverse_status = {v: k for k, v in self.status_mapping.items()} value = reverse_status.get(value, str(value)) elif local_field == "priority": # 反向优先级映射 reverse_priority = {v: k for k, v in self.priority_mapping.items()} value = reverse_priority.get(value, str(value)) elif local_field in ["created_at", "updated_at"] and isinstance(value, datetime): value = value.isoformat() feishu_fields[feishu_field] = value return feishu_fields def get_sync_status(self) -> Dict[str, Any]: """获取同步状态统计""" try: with db_manager.get_session() as session: total_local = session.query(WorkOrder).count() synced_count = session.query(WorkOrder).filter( WorkOrder.feishu_record_id.isnot(None) ).count() return { "total_local_workorders": total_local, "synced_workorders": synced_count, "unsynced_workorders": total_local - synced_count } except Exception as e: logger.error(f"获取同步状态失败: {e}") return {"error": str(e)}