import pandas as pd import requests import json import logging import time from datetime import datetime, timedelta, date from sqlalchemy import create_engine from urllib.parse import quote_plus # 配置日志 logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s', datefmt='%Y-%m-%d %H:%M:%S' ) def get_access_token(): """获取飞书访问令牌""" api_token_url = "https://open.feishu.cn/open-apis/auth/v3/tenant_access_token/internal/" api_post_data = {"app_id": "cli_a8b50ec0eed1500d", "app_secret": "ccxkE7ZCFQZcwkkM1rLy0ccZRXYsT2xK"} 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 read_daily_data(spreadsheet_token, sheet_id, retry=3): """从飞书电子表格读取日统计数据""" access_token = get_access_token() url = f"https://open.feishu.cn/open-apis/sheets/v2/spreadsheets/{spreadsheet_token}/values/{sheet_id}" headers = { "Authorization": f"Bearer {access_token}", "Content-Type": "application/json; charset=utf-8", } for attempt in range(retry): try: response = requests.get(url, headers=headers, timeout=30) response.raise_for_status() resp_data = response.json() if resp_data.get("code") != 0: raise Exception(f"API错误: {resp_data.get('msg')}") # 解析数据：跳过标题行，只取日期和激活数量 values = resp_data['data']['valueRange']['values'][1:] daily_data = [] for row in values: try: date_str = row[0] count = int(row[1]) # 转换日期格式 date_obj = datetime.strptime(date_str, "%Y-%m-%d").date() daily_data.append({"date": date_obj, "count": count}) except (IndexError, ValueError) as e: logging.warning(f"数据格式错误: {row} - {str(e)}") return daily_data except Exception as e: logging.warning(f"读取飞书表格失败(尝试 {attempt + 1}/{retry}): {str(e)}") if attempt < retry - 1: time.sleep(2 ** attempt) # 指数退避重试 raise Exception(f"读取飞书表格失败，重试{retry}次后仍不成功") def write_weekly_data(base_token, table_id, records, retry=3): """写入数据到飞书多维表格（周统计）""" access_token = get_access_token() url = f"https://open.feishu.cn/open-apis/bitable/v1/apps/{base_token}/tables/{table_id}/records/batch_create" headers = { "Authorization": f"Bearer {access_token}", "Content-Type": "application/json; charset=utf-8", } # 构建API请求体 payload = {"records": []} for row in records: try: # === 关键修复：使用毫秒时间戳格式（飞书官方要求）[3,8](@ref) === # 注意：需要将date对象转换为datetime对象（设置时间为00:00:00） start_dt = datetime.combine(row['week_start'], datetime.min.time()) end_dt = datetime.combine(row['week_end'], datetime.min.time()) start_timestamp = int(start_dt.timestamp() * 1000) end_timestamp = int(end_dt.timestamp() * 1000) payload["records"].append({ "fields": { "周数": str(row['week_number']), "周开始日期": start_timestamp, # 时间戳格式[3,8](@ref) "周结束日期": end_timestamp, # 时间戳格式[3,8](@ref) "激活数量": int(row['active_count']) } }) except Exception as e: logging.error(f"记录格式错误: {row} - {str(e)}") # 处理空记录情况 if not payload["records"]: logging.warning("没有有效记录可写入多维表格") return None # 分批写入（每次最多100条） batch_size = 100 success_count = 0 for i in range(0, len(payload["records"]), batch_size): batch = payload["records"][i:i + batch_size] batch_payload = {"records": batch} batch_success = False # 标记批次是否成功 for attempt in range(retry): try: response = requests.post(url, json=batch_payload, headers=headers, timeout=30) resp_data = response.json() if response.status_code == 200 and resp_data.get("code") == 0: if not batch_success: # 仅首次成功时计数 success_count += len(batch) batch_success = True logging.info(f"成功写入批次 {i // batch_size + 1}: {len(batch)}条记录") break # 成功则跳出重试循环 else: error_msg = resp_data.get('msg', '未知错误') error_code = resp_data.get("code", "") # 添加详细的错误诊断[3,8](@ref) if error_code == "DatetimeFieldConvFail": logging.error(f"日期字段转换失败: 请检查多维表格中日期字段类型配置[3,8](@ref)") # 记录导致错误的字段值 logging.error(f"问题记录示例: 周开始日期={start_timestamp}, 周结束日期={end_timestamp}") elif error_code == "TextFieldConvFail": logging.error(f"文本字段转换失败: 请检查数字字段是否包含非数字字符") elif error_code == "FieldNameNotFound": logging.error(f"字段未找到: 请检查字段名称是否与多维表格一致") # 添加详细错误日志 if 'data' in resp_data and 'errors' in resp_data['data']: logging.error(f"详细错误信息: {resp_data['data']['errors']}") logging.warning(f"飞书多维表格API错误(尝试 {attempt + 1}/{retry}): {error_msg}") time.sleep(2 ** attempt) except Exception as e: logging.warning(f"网络错误(尝试 {attempt + 1}/{retry}): {str(e)}") time.sleep(2 ** attempt) if not batch_success: logging.error(f"批次 {i // batch_size + 1} 写入失败") return success_count def calculate_iso_week(date): """计算ISO周数（处理跨年）""" # 获取ISO年份和周数 iso_year, iso_week, iso_day = date.isocalendar() # 处理跨年情况 if iso_week == 1 and date.month == 12: return f"{iso_year}-W{iso_week:02d}" elif iso_week >= 52 and date.month == 1: return f"{date.year - 1}-W{iso_week:02d}" else: return f"{date.year}-W{iso_week:02d}" def calculate_week_range(date): """计算所在周的周一和周日日期（ISO标准）""" # 计算周一 (0=Monday, 6=Sunday) start_date = date - timedelta(days=date.weekday()) end_date = start_date + timedelta(days=6) return start_date, end_date def send_feishu_message(webhook_url, message, daily_url=None, weekly_url=None, retry=2): """发送飞书通知""" headers = {'Content-Type': 'application/json'} # 构建交互式卡片 elements = [ { "tag": "div", "text": {"content": message, "tag": "lark_md"} } ] actions = [] # 添加日统计表格按钮 if daily_url: actions.append({ "tag": "button", "text": {"content": "查看日统计", "tag": "plain_text"}, "type": "default", "url": daily_url }) # 添加周统计表格按钮 if weekly_url: actions.append({ "tag": "button", "text": {"content": "查看周统计", "tag": "plain_text"}, "type": "primary", "url": weekly_url }) if actions: elements.append({ "tag": "action", "actions": actions }) payload = { "msg_type": "interactive", "card": { "config": {"wide_screen_mode": True}, "elements": elements, "header": { "title": {"content": "数据同步完成通知", "tag": "plain_text"}, "template": "green" } } } for attempt in range(retry): try: response = requests.post(webhook_url, headers=headers, json=payload, timeout=10) response.raise_for_status() return response.json() except Exception as e: logging.warning(f"飞书消息发送失败(尝试 {attempt + 1}/{retry}): {str(e)}") if attempt < retry - 1: time.sleep(2) # 等待后重试 raise Exception(f"飞书消息发送失败，重试{retry}次后仍不成功") if __name__ == '__main__': # 配置参数 DAILY_SPREADSHEET_ID = "MIqssvuB3h3XmgtkI26cBX7Angc" # 日统计电子表格ID DAILY_SHEET_ID = "46ea20" # 日统计工作表ID WEEKLY_BASE_TOKEN = "T3QLbmpqma014ussjy9cgitqnSh" # 周统计多维表格ID WEEKLY_TABLE_ID = "tbloB11YuVOSdi9e" # 周统计表ID WEBHOOK_URL = "https://open.feishu.cn/open-apis/bot/v2/hook/d82e4ada-8f78-40ae-b98e-75af0c448c95" # 表格链接 DAILY_URL = "https://my-ichery.feishu.cn/sheets/MIqssvuB3h3XmgtkI26cBX7Angc?sheet=46ea20&table=tblKYOLYYl1CFwn9&view=vews5Ash0A" WEEKLY_URL = "https://my-ichery.feishu.cn/base/T3QLbmpqma014ussjy9cgitqnSh?table=tbloB11YuVOSdi9e&view=vewFl6lbpB" try: logging.info("开始周统计计算流程...") current_time = datetime.now().strftime("%Y-%m-%d %H:%M") # 1. 从飞书日统计表读取数据 daily_data = read_daily_data(DAILY_SPREADSHEET_ID, DAILY_SHEET_ID) logging.info(f"成功读取{len(daily_data)}条日统计记录") # 2. 计算周统计数据（处理跨年） weekly_stats = {} for entry in daily_data: date = entry['date'] count = entry['count'] # 计算ISO周数（处理跨年） week_number = calculate_iso_week(date) # 计算周范围 week_start, week_end = calculate_week_range(date) # 初始化周统计 if week_number not in weekly_stats: weekly_stats[week_number] = { 'week_number': week_number, 'week_start': week_start, 'week_end': week_end, 'active_count': 0 } # 累加激活数量 weekly_stats[week_number]['active_count'] += count # 转换为列表并排序 weekly_list = sorted(weekly_stats.values(), key=lambda x: x['week_start']) logging.info(f"生成{len(weekly_list)}条周统计记录（含跨年处理）") # 3. 准备写入数据 weekly_values = [] for week in weekly_list: weekly_values.append({ 'week_number': week['week_number'], 'week_start': week['week_start'], 'week_end': week['week_end'], 'active_count': week['active_count'] }) # 4. 写入周统计表（分批处理） success_count = write_weekly_data(WEEKLY_BASE_TOKEN, WEEKLY_TABLE_ID, weekly_values) if success_count != len(weekly_values): raise Exception(f"周统计写入失败: 成功写入{success_count}条，总记录{len(weekly_values)}条") logging.info(f"成功写入{success_count}条周统计记录到飞书多维表格") # 5. 构建通知消息 today = datetime.now().date() current_week_number = calculate_iso_week(today) stats_summary = f"• 本周({current_week_number})激活车辆: {weekly_stats.get(current_week_number, {}).get('active_count', 0)}\n" stats_summary += f"• 累计激活车辆: {sum(week['active_count'] for week in weekly_list)}\n" # 添加最新周统计摘要 if weekly_list: latest_week = weekly_list[-1] stats_summary += f"• 最新一周({latest_week['week_number']}): {latest_week['week_start'].strftime('%Y-%m-%d')}至{latest_week['week_end'].strftime('%Y-%m-%d')}, 激活{latest_week['active_count']}辆" message = f"📅 周统计计算完成 ✅\n" message += f"⏰ 时间: {current_time}\n" message += f"📊 统计摘要:\n{stats_summary}\n" message += f"📝 处理结果: 成功处理{len(daily_data)}条日记录 → 生成{len(weekly_list)}条周记录" # 6. 发送通知 result = send_feishu_message( WEBHOOK_URL, message, daily_url=DAILY_URL, weekly_url=WEEKLY_URL ) logging.info(f"飞书消息发送成功: {result}") except Exception as e: error_message = f"❌ 周统计计算异常: {str(e)}" logging.exception("程序执行异常") send_feishu_message(WEBHOOK_URL, error_message)