#!/usr/bin/env python3 """ 企业微信智能表格自动化同步脚本 功能：监控加盟商状态变化，自动同步到工期进度表 每分钟执行一次检查 """ import json import time import logging import subprocess import sys from datetime import datetime from typing import Dict, List, Any # 配置信息 CONFIG = { "docid": "dcyeOLceOJqbuQpvY_EyivG5xx0cCPcT2x4kZ3UtkVJBhGqrD7wdc_iKB3za3vqq0foQifL-Y2npwGb1bl6Hb9EQ", "status_sheet_id": "q979lj", # 加盟商状态表 "progress_sheet_id": "MAqXdi", # 加盟商工期进度表 "status_field": "状态", # 状态字段名称 "franchisee_field": "加盟商名称", # 加盟商标识字段名称 "terminated_status": "签约后终止", # 终止状态值 "renewed_status": "再次签约", # 再次签约状态值 "check_interval": 60, # 检查间隔（秒） } # 设置日志 logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s', handlers=[ logging.FileHandler('franchisee_sync.log'), logging.StreamHandler() ] ) class WeComSmartSheet: """企业微信智能表格操作类""" @staticmethod def call_mcp_tool(category: str, method: str, args: Dict) -> Dict: """调用MCP工具""" try: # 构建命令 args_json = json.dumps(args, ensure_ascii=False) cmd = f"wecom_mcp call {category} {method} '{args_json}'" # 这里实际应该通过OpenClaw的wecom_mcp工具调用 # 由于在脚本中无法直接调用，这里返回模拟数据 # 实际使用时需要集成到OpenClaw环境中 # 模拟返回 return {"errcode": 0, "errmsg": "ok"} except Exception as e: logging.error(f"调用MCP工具失败: {e}") return {"errcode": -1, "errmsg": str(e)} @staticmethod def get_records(docid: str, sheet_id: str) -> List[Dict]: """获取表格记录""" args = { "docid": docid, "sheet_id": sheet_id } result = WeComSmartSheet.call_mcp_tool("doc", "smartsheet_get_records", args) if result.get("errcode") == 0: return result.get("records", []) else: logging.error(f"获取记录失败: {result.get('errmsg')}") return [] @staticmethod def delete_records(docid: str, sheet_id: str, record_ids: List[str]) -> bool: """删除记录""" args = { "docid": docid, "sheet_id": sheet_id, "record_ids": record_ids } result = WeComSmartSheet.call_mcp_tool("doc", "smartsheet_delete_records", args) if result.get("errcode") == 0: return True else: logging.error(f"删除记录失败: {result.get('errmsg')}") return False @staticmethod def add_records(docid: str, sheet_id: str, records: List[Dict]) -> bool: """添加记录""" args = { "docid": docid, "sheet_id": sheet_id, "records": records } result = WeComSmartSheet.call_mcp_tool("doc", "smartsheet_add_records", args) if result.get("errcode") == 0: return True else: logging.error(f"添加记录失败: {result.get('errmsg')}") return False class FranchiseeSync: """加盟商状态同步类""" def __init__(self, config: Dict): self.config = config self.last_status = {} # 记录上次状态 self.progress_cache = {} # 缓存工期进度记录 def get_status_records(self) -> List[Dict]: """获取加盟商状态表记录""" records = WeComSmartSheet.get_records( self.config["docid"], self.config["status_sheet_id"] ) # 转换为更方便处理的格式 processed_records = [] for record in records: if "values" in record: values = record["values"] processed_record = { "record_id": record.get("record_id"), "加盟商名称": self._get_field_value(values, self.config["franchisee_field"]), "状态": self._get_field_value(values, self.config["status_field"]), "原始数据": record } processed_records.append(processed_record) return processed_records def get_progress_records(self) -> List[Dict]: """获取工期进度表记录""" records = WeComSmartSheet.get_records( self.config["docid"], self.config["progress_sheet_id"] ) # 更新缓存 self.progress_cache = {} for record in records: if "values" in record: values = record["values"] franchisee_name = self._get_field_value(values, self.config["franchisee_field"]) if franchisee_name: self.progress_cache[franchisee_name] = record.get("record_id") return records def _get_field_value(self, values: Dict, field_name: str) -> Any: """从值字典中获取字段值""" if field_name in values: value = values[field_name] # 根据字段类型处理值 if isinstance(value, list) and len(value) > 0: if "text" in value[0]: return value[0]["text"] elif "id" in value[0]: return value[0]["id"] return value return None def check_status_changes(self, current_records: List[Dict]) -> Dict: """检查状态变化""" changes = { "terminated": [], # 变为终止状态的加盟商 "renewed": [] # 变为再次签约的加盟商 } for record in current_records: franchisee = record.get("加盟商名称") status = record.get("状态") if not franchisee or not status: continue last_status = self.last_status.get(franchisee) # 检查状态变化 if last_status != status: logging.info(f"加盟商 {franchisee} 状态变化: {last_status} -> {status}") if status == self.config["terminated_status"]: changes["terminated"].append({ "name": franchisee, "record_id": record.get("record_id"), "data": record.get("原始数据") }) elif status == self.config["renewed_status"]: changes["renewed"].append({ "name": franchisee, "record_id": record.get("record_id"), "data": record.get("原始数据") }) # 更新最后状态 self.last_status[franchisee] = status return changes def sync_changes(self, changes: Dict): """同步变化到工期进度表""" # 先获取当前工期进度记录 self.get_progress_records() # 处理终止的加盟商 for franchisee_info in changes["terminated"]: franchisee_name = franchisee_info["name"] logging.info(f"处理终止加盟商: {franchisee_name}") # 检查是否在工期进度表中 if franchisee_name in self.progress_cache: record_id = self.progress_cache[franchisee_name] success = WeComSmartSheet.delete_records( self.config["docid"], self.config["progress_sheet_id"], [record_id] ) if success: logging.info(f"成功删除加盟商 {franchisee_name} 的工期进度记录") del self.progress_cache[franchisee_name] else: logging.error(f"删除加盟商 {franchisee_name} 的工期进度记录失败") # 处理再次签约的加盟商 for franchisee_info in changes["renewed"]: franchisee_name = franchisee_info["name"] logging.info(f"处理再次签约加盟商: {franchisee_name}") # 检查是否已经在工期进度表中 if franchisee_name not in self.progress_cache: # 创建新的工期进度记录 new_record = self._create_progress_record(franchisee_info) success = WeComSmartSheet.add_records( self.config["docid"], self.config["progress_sheet_id"], [new_record] ) if success: logging.info(f"成功添加加盟商 {franchisee_name} 的工期进度记录") else: logging.error(f"添加加盟商 {franchisee_name} 的工期进度记录失败") def _create_progress_record(self, franchisee_info: Dict) -> Dict: """创建工期进度记录""" franchisee_name = franchisee_info["name"] original_data = franchisee_info.get("data", {}) # 这里可以根据加盟商状态表中的其他信息创建更完整的工期进度记录 # 例如：从原始数据中提取联系人、电话等信息 record = { "values": { "加盟商名称": [{"type": "text", "text": franchisee_name}], "项目阶段": [{"text": "初始阶段"}], "计划开始时间": datetime.now().strftime("%Y-%m-%d"), "当前进度": 0, "负责人": [{"type": "text", "text": "待分配"}] } } return record def run_once(self): """执行一次检查""" try: logging.info("=" * 50) logging.info(f"开始检查加盟商状态 ({datetime.now().strftime('%Y-%m-%d %H:%M:%S')})") # 获取当前状态 status_records = self.get_status_records() if not status_records: logging.info("未找到加盟商状态记录") return logging.info(f"共找到 {len(status_records)} 条加盟商记录") # 检查变化 changes = self.check_status_changes(status_records) if changes["terminated"] or changes["renewed"]: logging.info(f"发现状态变化: 终止={len(changes['terminated'])}个, 再次签约={len(changes['renewed'])}个") self.sync_changes(changes) else: logging.info("未发现状态变化") logging.info("检查完成") except Exception as e: logging.error(f"检查过程中发生错误: {e}", exc_info=True) def run_continuously(self): """持续运行""" logging.info("=" * 60) logging.info("启动加盟商状态监控服务") logging.info(f"文档ID: {self.config['docid']}") logging.info(f"状态表: {self.config['status_sheet_id']}") logging.info(f"进度表: {self.config['progress_sheet_id']}") logging.info(f"检查间隔: {self.config['check_interval']}秒") logging.info("=" * 60) # 初始加载一次状态 logging.info("初始加载加盟商状态...") status_records = self.get_status_records() for record in status_records: franchisee = record.get("加盟商名称") status = record.get("状态") if franchisee and status: self.last_status[franchisee] = status logging.info(f"初始加载完成，共 {len(self.last_status)} 个加盟商") # 持续运行 while True: self.run_once() time.sleep(self.config["check_interval"]) def main(): """主函数""" # 创建同步实例 sync = FranchiseeSync(CONFIG) # 运行模式选择 if len(sys.argv) > 1 and sys.argv[1] == "once": # 单次运行模式 sync.run_once() else: # 持续运行模式 sync.run_continuously() if __name__ == "__main__": main()