#!/usr/bin/env python3 """ 飞书排班管理器模块 统一入口，使用数据库存储和缓存 """ from datetime import datetime, timedelta from typing import Dict, List, Optional, Tuple import logging import os import sys # 添加项目根目录到路径 project_root = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) if project_root not in sys.path: sys.path.insert(0, project_root) # 简单实现，不依赖src模块 config = None get_logger = logging.getLogger class ScheduleDatabase: """简单的排班数据库占位实现""" def __init__(self, db_path=None): self.db_path = db_path def save_schedule(self, date_str, result, sheet_id, sheet_title): pass def get_schedule_by_range(self, start_date, end_date): return [] def get_stats(self): return {} from .client import FeishuSheetsClient from .parser import ScheduleDataParser logger = get_logger(__name__) class FeishuScheduleManager: """飞书排班管理器（统一入口）""" def __init__( self, base_url: Optional[str] = None, token: Optional[str] = None, spreadsheet_token: Optional[str] = None, db_path: Optional[str] = None, ): """ 初始化管理器 参数: base_url: 飞书API基础URL，如果为None则使用配置 token: 飞书API令牌，如果为None则使用配置 spreadsheet_token: 表格token，如果为None则使用配置 db_path: 数据库路径，如果为None则使用配置 """ # 检查配置是否完整 self._check_config(token, spreadsheet_token) # 初始化组件 self.client = FeishuSheetsClient(base_url, token, spreadsheet_token) self.parser = ScheduleDataParser() self.db = ScheduleDatabase(db_path) logger.info("飞书排班管理器初始化完成") def _check_config( self, token: Optional[str], spreadsheet_token: Optional[str] ) -> None: """检查必要配置""" # 检查是否有任何可用的认证方式 has_token = bool(token or os.getenv("FEISHU_TOKEN")) has_app_credentials = bool(os.getenv("FEISHU_APP_ID") and os.getenv("FEISHU_APP_SECRET")) if not has_token and not has_app_credentials: logger.warning("飞书认证未配置，排班功能将不可用") logger.warning("请配置 FEISHU_TOKEN 或 FEISHU_APP_ID + FEISHU_APP_SECRET") elif has_app_credentials: logger.info("使用飞书应用凭证自动获取token") elif has_token: logger.info("使用手动配置的FEISHU_TOKEN") if not spreadsheet_token and not os.getenv("FEISHU_SPREADSHEET_TOKEN"): logger.warning("飞书表格令牌未配置，排班功能将不可用") def _select_sheet_for_date( self, sheets: List[Dict[str, str]], target_year_month: str ) -> Optional[Dict[str, str]]: """ 为指定日期选择最合适的表格 参数: sheets: 表格列表 target_year_month: 目标年月，格式 "2025-12" 返回: 选中的表格信息，未找到返回None """ if not sheets: logger.error("表格列表为空") return None # 提取年份和月份 try: year = target_year_month[:4] month = target_year_month[5:7].lstrip("0") except (IndexError, ValueError) as e: logger.error(f"解析年月失败: {target_year_month}, 错误: {e}") return None # 对于2026年，优先使用年度表格 if year == "2026": # 查找年度表格，如 "2026年排班表" year_name = f"{year}年" for sheet in sheets: title = sheet.get("title", "") if year_name in title and "排班表" in title: logger.info(f"找到2026年年度表格: {title}") return sheet # 优先查找月份表格，如 "12月" month_name = f"{int(month)}月" for sheet in sheets: title = sheet.get("title", "") if month_name in title: logger.info(f"找到月份表格: {title}") return sheet # 查找年度表格，如 "2026年排班表" year_name = f"{year}年" for sheet in sheets: title = sheet.get("title", "") if year_name in title and "排班表" in title: logger.info(f"找到年度表格: {title}") return sheet # 如果没有找到匹配的表格，使用第一个表格 logger.warning( f"未找到 {target_year_month} 的匹配表格，使用第一个表格: {sheets[0]['title']}" ) return sheets[0] def get_schedule_for_date(self, date_str: str) -> Dict[str, any]: """ 获取指定日期的排班信息 修复：每次都从飞书获取最新数据并覆盖数据库，确保日报中显示最新排班信息 参数: date_str: 日期字符串，格式 "2025-12-30" 返回: 排班信息字典 异常: ValueError: 日期格式无效 Exception: 其他错误 """ try: # 解析日期 dt = datetime.strptime(date_str, "%Y-%m-%d") # 生成两种格式的日期字符串，用于匹配不同表格 target_date_mm_dd = dt.strftime("%m/%d") # "01/01" 用于月度表格 target_date_chinese = f"{dt.month}月{dt.day}日" # "1月1日" 用于年度表格 target_year_month = dt.strftime("%Y-%m") # "2025-12" logger.info( f"获取 {date_str} 的排班信息 (格式: {target_date_mm_dd}/{target_date_chinese})" ) # 1. 获取表格信息 sheets = self.client.get_sheets_info() if not sheets: logger.error("未获取到表格信息") return self._empty_result() # 2. 选择最合适的表格 selected_sheet = self._select_sheet_for_date(sheets, target_year_month) if not selected_sheet: logger.error("未找到合适的表格") return self._empty_result() sheet_id = selected_sheet["sheet_id"] sheet_title = selected_sheet["title"] # 3. 获取表格数据 sheet_data = self.client.get_sheet_data(sheet_id) if not sheet_data: logger.error("未获取到表格数据") return self._empty_result() values = sheet_data.get("valueRange", {}).get("values", []) if not values: logger.error("表格数据为空") return self._empty_result() # 4. 解析数据 - 根据表格类型选择合适的日期格式 # 如果是年度表格，使用中文日期格式；否则使用mm/dd格式 if "年" in sheet_title and "排班表" in sheet_title: target_date = target_date_chinese # "1月1日" else: target_date = target_date_mm_dd # "01/01" logger.info(f"使用日期格式: {target_date} 解析表格: {sheet_title}") result = self.parser.parse(values, target_date, sheet_title) # 5. 每次都保存到数据库，覆盖旧数据，确保人员变动能及时更新 if result["day_shift"] or result["night_shift"]: self.db.save_schedule(date_str, result, sheet_id, sheet_title) logger.info( f"已更新 {date_str} 的排班信息到数据库: 白班={result['day_shift']}, 夜班={result['night_shift']}" ) else: logger.warning(f"解析结果为空，{date_str} 未保存到数据库") return result except ValueError as e: logger.error(f"日期格式无效: {date_str}, 错误: {e}") raise except Exception as e: logger.error(f"获取排班信息失败: {e}") # 降级处理：返回空值 return self._empty_result() def get_schedule_for_today(self) -> Dict[str, any]: """获取今天的排班信息""" today = datetime.now().strftime("%Y-%m-%d") return self.get_schedule_for_date(today) def get_schedule_for_tomorrow(self) -> Dict[str, any]: """获取明天的排班信息""" tomorrow = (datetime.now() + timedelta(days=1)).strftime("%Y-%m-%d") return self.get_schedule_for_date(tomorrow) def refresh_all_schedules(self, days: Optional[int] = None): """ 刷新未来指定天数的排班信息 参数: days: 刷新未来多少天的排班信息，如果为None则使用配置 """ if days is None: days = int(os.getenv("SCHEDULE_REFRESH_DAYS", 7)) logger.info(f"开始刷新未来 {days} 天的排班信息") today = datetime.now() success_count = 0 error_count = 0 for i in range(days): date = (today + timedelta(days=i)).strftime("%Y-%m-%d") try: logger.debug(f"刷新 {date} 的排班信息...") self.get_schedule_for_date(date) success_count += 1 except Exception as e: logger.error(f"刷新 {date} 的排班信息失败: {e}") error_count += 1 logger.info(f"排班信息刷新完成，成功: {success_count}, 失败: {error_count}") def get_schedule_by_range( self, start_date: str, end_date: str ) -> List[Dict[str, any]]: """ 获取日期范围内的排班信息 参数: start_date: 开始日期 (YYYY-MM-DD) end_date: 结束日期 (YYYY-MM-DD) 返回: 排班信息列表 """ try: # 验证日期格式 datetime.strptime(start_date, "%Y-%m-%d") datetime.strptime(end_date, "%Y-%m-%d") return self.db.get_schedule_by_range(start_date, end_date) except ValueError as e: logger.error(f"日期格式无效: {e}") return [] except Exception as e: logger.error(f"获取排班范围失败: {e}") return [] def test_connection(self) -> bool: """测试飞书连接是否正常""" return self.client.test_connection() def get_stats(self) -> Dict[str, any]: """获取排班数据库统计信息""" return self.db.get_stats() def _empty_result(self) -> Dict[str, any]: """返回空结果""" return { "day_shift": "", "night_shift": "", "day_shift_list": [], "night_shift_list": [], } def _format_db_result(self, db_result: Dict[str, any]) -> Dict[str, any]: """格式化数据库结果""" return { "day_shift": db_result["day_shift"], "night_shift": db_result["night_shift"], "day_shift_list": db_result["day_shift_list"], "night_shift_list": db_result["night_shift_list"], } if __name__ == "__main__": # 测试代码 import sys # 设置日志 logging.basicConfig(level=logging.INFO) # 初始化管理器 manager = FeishuScheduleManager() # 测试连接 if not manager.test_connection(): print("飞书连接测试失败") sys.exit(1) print("飞书连接测试成功") # 测试获取今天和明天的排班 today_schedule = manager.get_schedule_for_today() print( f"今天排班: 白班={today_schedule['day_shift']}, 夜班={today_schedule['night_shift']}" ) tomorrow_schedule = manager.get_schedule_for_tomorrow() print( f"明天排班: 白班={tomorrow_schedule['day_shift']}, 夜班={tomorrow_schedule['night_shift']}" ) # 测试统计 stats = manager.get_stats() print(f"排班统计: {stats}") # 测试范围查询（最近7天） end_date = datetime.now().strftime("%Y-%m-%d") start_date = (datetime.now() - timedelta(days=7)).strftime("%Y-%m-%d") schedules = manager.get_schedule_by_range(start_date, end_date) print(f"最近7天排班记录: {len(schedules)} 条")