#!/usr/bin/env python3 """ 排班数据解析器模块 支持月度表格和年度表格解析 """ import re from typing import Dict, List, Optional, Tuple import logging from src.logging_config import get_logger logger = get_logger(__name__) class ScheduleDataParser: """排班数据解析器（支持月度表格和年度表格）""" @staticmethod def _parse_chinese_date(date_str: str) -> Optional[str]: """ 解析中文日期格式 参数: date_str: 中文日期，如 "12月30日" 或 "12/30" 或 "12月1日" 或 "1月1日" 返回: 标准化日期字符串 "M月D日" (不补零) 异常: ValueError: 日期格式无效 """ if not date_str or not isinstance(date_str, str): return None date_str = date_str.strip() try: # 如果是 "12/30" 格式 if '/' in date_str: month, day = date_str.split('/') # 移除可能的空格和前导零 month = month.strip().lstrip('0') day = day.strip().lstrip('0') if not month.isdigit() or not day.isdigit(): raise ValueError(f"日期格式无效: {date_str}") return f"{int(month)}月{int(day)}日" # 如果是 "12月30日" 或 "1月1日" 格式 if '月' in date_str and '日' in date_str: # 移除前导零，如 "01月01日" -> "1月1日" parts = date_str.split('月') if len(parts) == 2: month_part = parts[0].lstrip('0') day_part = parts[1].rstrip('日').lstrip('0') if not month_part or not day_part: raise ValueError(f"日期格式无效: {date_str}") return f"{month_part}月{day_part}日" return date_str # 如果是 "12月1日" 格式（已经包含"日"字） if '月' in date_str: # 检查是否已经有"日"字 if '日' not in date_str: return f"{date_str}日" return date_str # 如果是纯数字，尝试解析 if date_str.isdigit() and len(date_str) == 4: # 假设是 "1230" 格式 month = date_str[:2].lstrip('0') day = date_str[2:].lstrip('0') return f"{month}月{day}日" return None except Exception as e: logger.warning(f"解析日期失败: {date_str}, 错误: {e}") return None @staticmethod def _find_date_column_index(headers: List[str], target_date: str) -> Optional[int]: """ 在表头中查找目标日期对应的列索引 参数: headers: 表头行 ["姓名", "12月1日", "12月2日", ...] target_date: 目标日期 "12月30日" 返回: 列索引（从0开始），未找到返回None """ if not headers or not target_date: return None # 标准化目标日期 target_std = ScheduleDataParser._parse_chinese_date(target_date) if not target_std: logger.warning(f"无法标准化目标日期: {target_date}") return None # 遍历表头查找匹配的日期 for i, header in enumerate(headers): if not header: continue header_std = ScheduleDataParser._parse_chinese_date(header) if header_std == target_std: logger.debug(f"找到日期列: {target_date} -> {header} (索引: {i})") return i logger.warning(f"未找到日期列: {target_date}, 表头: {headers}") return None def parse_monthly_sheet(self, values: List[List[str]], target_date: str) -> Dict[str, any]: """ 解析月度表格数据（如12月表格） 参数: values: 飞书表格返回的二维数组 target_date: 目标日期（格式: "12月30日" 或 "12/30"） 返回: 排班信息字典 """ if not values or len(values) < 2: logger.warning("表格数据为空或不足") return self._empty_result() # 第一行是表头 headers = values[0] date_column_index = self._find_date_column_index(headers, target_date) if date_column_index is None: logger.warning(f"未找到日期列: {target_date}") return self._empty_result() # 收集白班和夜班人员 day_shift_names = [] night_shift_names = [] # 从第二行开始是人员数据 for row_idx, row in enumerate(values[1:], start=2): if len(row) <= date_column_index: continue name = row[0] if row else '' shift = row[date_column_index] if date_column_index < len(row) else '' if not name or not shift: continue # 清理班次值 shift = shift.strip() if shift == '白': day_shift_names.append(name.strip()) elif shift == '夜': night_shift_names.append(name.strip()) elif shift: # 其他班次类型 logger.debug(f"忽略未知班次类型: {shift} (行: {row_idx})") return self._format_result(day_shift_names, night_shift_names) def parse_yearly_sheet(self, values: List[List[str]], target_date: str) -> Dict[str, any]: """ 解析年度表格数据（如2026年排班表） 参数: values: 飞书表格返回的二维数组 target_date: 目标日期（格式: "12月30日" 或 "12/30"） 返回: 排班信息字典 """ if not values: logger.warning("年度表格数据为空") return self._empty_result() # 查找目标月份的数据块 target_month = target_date.split('月')[0] if '月' in target_date else '' if not target_month: logger.warning(f"无法从 {target_date} 提取月份") return self._empty_result() # 在年度表格中查找对应的月份块 current_block_start = -1 current_month = '' for i, row in enumerate(values): if not row: continue first_cell = str(row[0]) if row else '' # 检查是否是月份标题行，如 "福州港1月排班表" if '排班表' in first_cell and '月' in first_cell: # 提取月份数字 month_match = re.search(r'(\d+)月', first_cell) if month_match: current_month = month_match.group(1).lstrip('0') current_block_start = i logger.debug(f"找到月份块: {current_month}月 (行: {i+1})") # 如果找到目标月份，检查下一行是否是表头行 if current_month == target_month and i == current_block_start + 1: # 当前行是表头行 headers = row date_column_index = self._find_date_column_index(headers, target_date) if date_column_index is None: logger.warning(f"在年度表格中未找到日期列: {target_date}") return self._empty_result() # 收集人员数据（从表头行的下一行开始） day_shift_names = [] night_shift_names = [] for j in range(i + 1, len(values)): person_row = values[j] if not person_row: # 遇到空行，继续检查下一行 continue # 检查是否是下一个月份块的开始 if person_row[0] and isinstance(person_row[0], str) and '排班表' in person_row[0] and '月' in person_row[0]: break # 跳过星期行（第一列为空的行） if not person_row[0]: continue if len(person_row) <= date_column_index: continue name = person_row[0] if person_row else '' shift = person_row[date_column_index] if date_column_index < len(person_row) else '' if not name or not shift: continue # 清理班次值 shift = shift.strip() if shift == '白': day_shift_names.append(name.strip()) elif shift == '夜': night_shift_names.append(name.strip()) return self._format_result(day_shift_names, night_shift_names) logger.warning(f"在年度表格中未找到 {target_month}月 的数据块") return self._empty_result() def parse(self, values: List[List[str]], target_date: str, sheet_title: str = '') -> Dict[str, any]: """ 解析排班数据，自动判断表格类型 参数: values: 飞书表格返回的二维数组 target_date: 目标日期（格式: "12月30日" 或 "12/30"） sheet_title: 表格标题，用于判断表格类型 返回: 排班信息字典 """ # 根据表格标题判断表格类型 if '年' in sheet_title and '排班表' in sheet_title: # 年度表格 logger.info(f"使用年度表格解析器: {sheet_title}") return self.parse_yearly_sheet(values, target_date) else: # 月度表格 logger.info(f"使用月度表格解析器: {sheet_title}") return self.parse_monthly_sheet(values, target_date) def _empty_result(self) -> Dict[str, any]: """返回空结果""" return { 'day_shift': '', 'night_shift': '', 'day_shift_list': [], 'night_shift_list': [] } def _format_result(self, day_shift_names: List[str], night_shift_names: List[str]) -> Dict[str, any]: """格式化结果""" # 去重并排序 day_shift_names = sorted(set(day_shift_names)) night_shift_names = sorted(set(night_shift_names)) # 格式化输出 day_shift_str = '、'.join(day_shift_names) if day_shift_names else '' night_shift_str = '、'.join(night_shift_names) if night_shift_names else '' return { 'day_shift': day_shift_str, 'night_shift': night_shift_str, 'day_shift_list': day_shift_names, 'night_shift_list': night_shift_names } if __name__ == '__main__': # 测试代码 import sys # 设置日志 logging.basicConfig(level=logging.DEBUG) parser = ScheduleDataParser() # 测试日期解析 test_dates = ["12/30", "12月30日", "1月1日", "01/01", "1230", "无效日期"] for date in test_dates: parsed = parser._parse_chinese_date(date) print(f"解析 '{date}' -> '{parsed}'") # 测试月度表格解析 monthly_values = [ ["姓名", "12月1日", "12月2日", "12月3日"], ["张三", "白", "夜", ""], ["李四", "夜", "白", "白"], ["王五", "", "白", "夜"] ] result = parser.parse_monthly_sheet(monthly_values, "12月2日") print(f"\n月度表格解析结果: {result}") # 测试年度表格解析 yearly_values = [ ["福州港2026年排班表"], ["姓名", "1月1日", "1月2日", "1月3日"], ["张三", "白", "夜", ""], ["李四", "夜", "白", "白"], ["福州港2月排班表"], ["姓名", "2月1日", "2月2日"], ["王五", "白", "夜"] ] result = parser.parse_yearly_sheet(yearly_values, "1月2日") print(f"年度表格解析结果: {result}")