qichi.liang
00d2218c6d
- 添加日报生成功能 (report_generator.py) - 添加 GUI 界面 (daily_report_gui.py) - 添加班次交接报告功能 (shift_report.py) - 集成飞书 API 获取排班信息 - 集成 Metabase 查询作业数据 - 生成 AGENTS.md 文档
364 lines
12 KiB
Python
364 lines
12 KiB
Python
#!/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)} 条")
|