import pandas as pd import requests import logging import time import json from sqlalchemy import create_engine from urllib.parse import quote_plus from datetime import datetime # 配置日志 logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(levelname)s - %(essage)s', datefmt='%Y-%m-%d %H:%M:%S' ) # ====================== 全局配置 ====================== # 数据库连接配置 DB_CONFIG = { 'host': 'eutsp-prod.mysql.germany.rds.aliyuncs.com', 'port': 3306, 'user': 'international_tsp_eu_r', 'password': 'ZXhBgo1TB2XbF3kP', 'database': 'chery_international_tsp_eu' } # 飞书多维表格配置 FEISHU_APP_TOKEN = "T3QLbmpqma014ussjy9cgitqnSh" # 应用Token FEISHU_TABLE_ID = "tbloPJHLc05MHIAY" # 表格ID VIN_FIELD_NAME = "车架号" # VIN字段名称 ACTIVE_FIELD_NAME = "激活状态" # 激活状态字段名称 # 飞书应用凭证 APP_ID = "cli_a8b50ec0eed1500d" APP_SECRET = "ccxkE7ZCFQZcwkkM1rLy0ccZRXYsT2xK" # 飞书机器人配置 WEBHOOK_URL = "https://open.feishu.cn/open-apis/bot/v2/hook/3bc45247-c469-47a0-bfe9-c62e46c5aca0" MULTI_TABLE_URL = "https://my-ichery.feishu.cn/base/T3QLbmpqma014ussjy9cgitqnSh?from=from_copylink" # ====================== 通用函数 ====================== def get_access_token(): """获取飞书访问令牌""" api_token_url = "https://open.feishu.cn/open-apis/auth/v3/tenant_access_token/internal/" api_post_data = {"app_id": APP_ID, "app_secret": APP_SECRET} try: response = requests.post(api_token_url, json=api_post_data, timeout=10) response.raise_for_status() resp_data = response.json() if resp_data.get("code") == 0: return resp_data["tenant_access_token"] else: raise Exception(f"获取Token失败: {resp_data.get('msg')}") except Exception as e: logging.error(f"获取飞书Token异常: {str(e)}") raise def send_feishu_message(message): """发送飞书通知[9,10](@ref)""" headers = {'Content-Type': 'application/json'} # 构建交互式卡片 payload = { "msg_type": "interactive", "card": { "config": {"wide_screen_mode": True}, "elements": [ { "tag": "div", "text": {"content": message, "tag": "lark_md"} }, { "tag": "action", "actions": [ { "tag": "button", "text": {"content": "查看多维表格", "tag": "plain_text"}, "type": "primary", "url": MULTI_TABLE_URL } ] } ], "header": { "title": {"content": "车辆状态同步报告", "tag": "plain_text"}, "template": "blue" } } } try: response = requests.post(WEBHOOK_URL, headers=headers, json=payload, timeout=10) response.raise_for_status() return response.json() except Exception as e: logging.error(f"飞书消息发送失败: {str(e)}") return None # ====================== 车辆激活状态同步 ====================== def read_feishu_records(app_token, table_id, retry=3): """从飞书多维表格读取车辆记录""" access_token = get_access_token() headers = { "Authorization": f"Bearer {access_token}", "Content-Type": "application/json; charset=utf-8", } url = f"https://open.feishu.cn/open-apis/bitable/v1/apps/{app_token}/tables/{table_id}/records" params = {"page_size": 500} # 单次请求最大记录数 all_records = [] page_token = None while True: if page_token: params["page_token"] = page_token records_fetched = False for attempt in range(retry): try: response = requests.get(url, headers=headers, params=params, timeout=30) response.raise_for_status() resp_data = response.json() if resp_data.get("code") != 0: raise Exception(f"API错误: {resp_data.get('msg')}") records = resp_data["data"]["items"] all_records.extend(records) # 检查是否有下一页 if "page_token" in resp_data["data"] and resp_data["data"]["has_more"]: page_token = resp_data["data"]["page_token"] else: page_token = None records_fetched = True break except Exception as e: logging.warning(f"读取多维表格失败(尝试 {attempt + 1}/{retry}): {str(e)}") if attempt < retry - 1: time.sleep(2 ** attempt) if not records_fetched: raise Exception("无法读取多维表格数据") if not page_token: break return all_records def fetch_activation_data(): """从数据库查询车辆激活状态""" try: # 创建SQLAlchemy引擎 safe_password = quote_plus(DB_CONFIG['password']) engine = create_engine( f"mysql+pymysql://{DB_CONFIG['user']}:{safe_password}@{DB_CONFIG['host']}:{DB_CONFIG['port']}/{DB_CONFIG['database']}", pool_recycle=3600 ) sql_query = """ SELECT v_vehicle_info.VIN AS car_vins, CASE v_sim_info.STATE WHEN 2 THEN 'active' ELSE 'no_active' END AS active_status FROM v_vehicle_info JOIN v_tbox_info USING (VIN) JOIN v_sim_info USING (SN); """ # 执行查询并转换为DataFrame df = pd.read_sql(sql_query, engine) return df.set_index('car_vins')['active_status'].to_dict() except Exception as e: logging.error(f"数据库查询失败: {str(e)}") raise def compare_and_update_records(app_token, table_id, activation_data): """比较并更新飞书多维表格记录""" access_token = get_access_token() headers = { "Authorization": f"Bearer {access_token}", "Content-Type": "application/json; charset=utf-8", } # 1. 读取多维表格所有记录 feishu_records = read_feishu_records(app_token, table_id) logging.info(f"成功读取{len(feishu_records)}条多维表格记录") # 统计变量 total_updated = 0 activated_count = 0 not_activated_count = 0 # 2. 准备需要更新的记录 records_to_update = [] for record in feishu_records: try: record_id = record["record_id"] fields = record.get("fields", {}) # 获取车架号(VIN) vin = fields.get(VIN_FIELD_NAME) if not vin: logging.warning(f"记录 {record_id} 缺少VIN字段") continue # 获取多维表格中的激活状态 current_status = fields.get(ACTIVE_FIELD_NAME, "").lower() # 获取数据库中的激活状态 db_status = activation_data.get(vin) if not db_status: logging.warning(f"VIN {vin} 在数据库中未找到") continue # 比较状态是否需要更新 if db_status != current_status: records_to_update.append({ "record_id": record_id, "fields": { ACTIVE_FIELD_NAME: db_status } }) logging.info(f"VIN {vin} 状态需更新: {current_status} → {db_status}") # 统计更新状态 if db_status == 'active': activated_count += 1 else: not_activated_count += 1 except Exception as e: logging.error(f"处理记录异常: {record} - {str(e)}") total_updated = len(records_to_update) logging.info(f"需要更新{total_updated}条记录") logging.info(f"更新统计: 激活车辆: {activated_count}台, 未激活车辆: {not_activated_count}台") # 3. 批量更新记录 (每批次50条) if not records_to_update: logging.info("没有需要更新的记录") return total_updated, activated_count, not_activated_count batch_size = 50 success_count = 0 for i in range(0, len(records_to_update), batch_size): batch = records_to_update[i:i + batch_size] batch_payload = {"records": batch} update_url = f"https://open.feishu.cn/open-apis/bitable/v1/apps/{app_token}/tables/{table_id}/records/batch_update" for attempt in range(3): # 最多重试3次 try: response = requests.post(update_url, headers=headers, json=batch_payload, timeout=30) resp_data = response.json() if response.status_code == 200 and resp_data.get("code") == 0: success_count += len(batch) logging.info(f"成功更新批次 {i // batch_size + 1}: {len(batch)}条记录") break else: error_msg = resp_data.get('msg', '未知错误') logging.warning(f"更新失败(尝试 {attempt + 1}/3): {error_msg}") time.sleep(2 ** attempt) except Exception as e: logging.warning(f"网络错误(尝试 {attempt + 1}/3): {str(e)}") time.sleep(2 ** attempt) logging.info(f"总计更新成功{success_count}条记录") return total_updated, activated_count, not_activated_count def sync_vehicle_activation(): """执行车辆激活状态同步流程""" try: start_time = time.time() logging.info("开始车辆激活状态同步流程...") # 1. 从数据库获取激活状态数据 activation_data = fetch_activation_data() logging.info(f"从数据库获取{len(activation_data)}条激活状态记录") # 统计数据库中的激活状态 db_activated = sum(1 for status in activation_data.values() if status == 'active') db_not_activated = sum(1 for status in activation_data.values() if status == 'no_active') logging.info(f"数据库统计: 激活车辆: {db_activated}台, 未激活车辆: {db_not_activated}台") # 2. 与飞书多维表格数据比对并更新 total_updated, activated_count, not_activated_count = compare_and_update_records( FEISHU_APP_TOKEN, FEISHU_TABLE_ID, activation_data ) # 3. 输出最终统计结果 logging.info("=" * 50) logging.info(f"本次更新统计结果:") logging.info(f" - 更新车辆总数: {total_updated}台") logging.info(f" - 激活车辆数量: {activated_count}台") logging.info(f" - 未激活车辆数量: {not_activated_count}台") logging.info("=" * 50) # 4. 完成日志 elapsed_time = time.time() - start_time logging.info(f"流程完成! 耗时: {elapsed_time:.2f}秒, 更新记录数: {total_updated}") return { "total_updated": total_updated, "activated_count": activated_count, "not_activated_count": not_activated_count, "elapsed_time": elapsed_time } except Exception as e: logging.error(f"流程执行异常: {str(e)}") logging.exception("详细错误信息") return None # ====================== 主执行流程 ====================== if __name__ == '__main__': # 执行车辆激活状态同步 sync_result = sync_vehicle_activation() if sync_result: # 构建飞书消息 current_time = datetime.now().strftime("%Y-%m-%d %H:%M") message = ( f"🚗 **车辆激活状态同步完成** ✅\n" f"⏰ 同步时间: {current_time}\n" f"⏱️ 处理耗时: {sync_result['elapsed_time']:.2f}秒\n\n" f"📊 **同步统计结果:**\n" f"• 更新车辆总数: {sync_result['total_updated']}台\n" f"• ✅ 激活车辆数量: {sync_result['activated_count']}台\n" f"• ❌ 未激活车辆数量: {sync_result['not_activated_count']}台\n\n" f"🔗 查看多维表格: [点击访问]({MULTI_TABLE_URL})" ) # 发送飞书通知 send_result = send_feishu_message(message) if send_result: logging.info(f"飞书消息发送成功: {send_result}") else: logging.error("飞书消息发送失败") else: error_message = "❌ 车辆激活状态同步失败,请检查日志" send_feishu_message(error_message) logging.error(error_message)