#!/usr/bin/env python3 """ 日报生成模块 更新依赖，使用新的配置和数据库模块 """ from datetime import datetime, timedelta from typing import Dict, List, Optional, Any import logging from src.config import config from src.logging_config import get_logger from src.database.daily_logs import DailyLogsDatabase from src.feishu.manager import FeishuScheduleManager logger = get_logger(__name__) class ReportGeneratorError(Exception): """日报生成错误""" pass class DailyReportGenerator: """每日作业报告生成器""" def __init__(self, db_path: Optional[str] = None): """ 初始化日报生成器 参数: db_path: 数据库文件路径，如果为None则使用配置 """ self.db = DailyLogsDatabase(db_path) logger.info("日报生成器初始化完成") def get_latest_date(self) -> str: """ 获取数据库中最新的日期 返回: 最新日期字符串，格式 "YYYY-MM-DD" """ try: logs = self.db.query_all(limit=1) if logs: return logs[0]['date'] return datetime.now().strftime('%Y-%m-%d') except Exception as e: logger.error(f"获取最新日期失败: {e}") return datetime.now().strftime('%Y-%m-%d') def get_daily_data(self, date: str) -> Dict[str, Any]: """ 获取指定日期的数据 参数: date: 日期字符串，格式 "YYYY-MM-DD" 返回: 每日数据字典 """ try: logs = self.db.query_by_date(date) # 按船名汇总TEU和尺寸箱量 ships: Dict[str, Dict[str, Any]] = {} for log in logs: ship = log['ship_name'] if ship not in ships: ships[ship] = { 'teu': 0, 'twenty_feet': 0, 'forty_feet': 0 } if log.get('teu'): ships[ship]['teu'] += log['teu'] if log.get('twenty_feet'): ships[ship]['twenty_feet'] += log['twenty_feet'] if log.get('forty_feet'): ships[ship]['forty_feet'] += log['forty_feet'] total_teu = sum(ship_data['teu'] for ship_data in ships.values()) return { 'date': date, 'ships': ships, 'total_teu': total_teu, 'ship_count': len(ships) } except Exception as e: logger.error(f"获取每日数据失败: {date}, 错误: {e}") return { 'date': date, 'ships': {}, 'total_teu': 0, 'ship_count': 0 } def get_monthly_stats(self, date: str) -> Dict[str, Any]: """ 获取月度统计（截止到指定日期） 参数: date: 日期字符串，格式 "YYYY-MM-DD" 返回: 月度统计字典 """ try: year_month = date[:7] # YYYY-MM target_date = datetime.strptime(date, '%Y-%m-%d').date() logs = self.db.query_all(limit=10000) # 只统计当月且在指定日期之前的数据 monthly_logs = [ log for log in logs if log['date'].startswith(year_month) and datetime.strptime(log['date'], '%Y-%m-%d').date() <= target_date ] # 按日期汇总 daily_totals: Dict[str, int] = {} for log in monthly_logs: d = log['date'] if d not in daily_totals: daily_totals[d] = 0 if log.get('teu'): daily_totals[d] += log['teu'] # 计算当月天数（已过的天数） current_date = datetime.strptime(date, '%Y-%m-%d') if current_date.day == config.FIRST_DAY_OF_MONTH_SPECIAL: days_passed = 1 else: days_passed = current_date.day # 获取未统计数据 unaccounted = self.db.get_unaccounted(year_month) planned = days_passed * config.DAILY_TARGET_TEU actual = sum(daily_totals.values()) + unaccounted completion = round(actual / planned * 100, 2) if planned > 0 else 0 return { 'year_month': year_month, 'days_passed': days_passed, 'planned': planned, 'actual': actual, 'unaccounted': unaccounted, 'completion': completion, 'daily_totals': daily_totals } except Exception as e: logger.error(f"获取月度统计失败: {date}, 错误: {e}") return { 'year_month': date[:7], 'days_passed': 0, 'planned': 0, 'actual': 0, 'unaccounted': 0, 'completion': 0, 'daily_totals': {} } def get_shift_personnel(self, date: str) -> Dict[str, str]: """ 获取班次人员（从飞书排班表获取） 注意：日报中显示的是次日的班次人员，所以需要获取 date+1 的排班 例如：生成 12/29 的日报，显示的是 12/30 的人员 参数: date: 日期字符串，格式 "YYYY-MM-DD" 返回: 班次人员字典 """ try: # 检查飞书配置（支持应用凭证和手动token两种方式） has_feishu_config = bool(config.FEISHU_SPREADSHEET_TOKEN) and ( bool(config.FEISHU_APP_ID and config.FEISHU_APP_SECRET) or bool(config.FEISHU_TOKEN) ) if not has_feishu_config: logger.warning("飞书配置不完整，跳过排班信息获取") logger.warning("需要配置 FEISHU_SPREADSHEET_TOKEN 和 (FEISHU_APP_ID+FEISHU_APP_SECRET 或 FEISHU_TOKEN)") return self._empty_personnel() # 初始化飞书排班管理器 manager = FeishuScheduleManager() # 计算次日日期（日报中显示的是次日班次） parsed_date = datetime.strptime(date, '%Y-%m-%d') tomorrow = (parsed_date + timedelta(days=1)).strftime('%Y-%m-%d') logger.info(f"获取 {date} 日报的班次人员，对应排班表日期: {tomorrow}") # 获取次日的排班信息 schedule = manager.get_schedule_for_date(tomorrow) # 如果从飞书获取到数据，使用飞书数据 if schedule.get('day_shift') or schedule.get('night_shift'): return { 'day_shift': schedule.get('day_shift', ''), 'night_shift': schedule.get('night_shift', ''), 'duty_phone': config.DUTY_PHONE } # 如果飞书数据为空，返回空值 logger.warning(f"无法从飞书获取 {tomorrow} 的排班信息") return self._empty_personnel() except Exception as e: logger.error(f"获取排班信息失败: {e}") # 降级处理：返回空值 return self._empty_personnel() def generate_report(self, date: Optional[str] = None) -> str: """ 生成日报 参数: date: 日期字符串，格式 "YYYY-MM-DD"，如果为None则使用最新日期 返回: 日报文本 异常: ReportGeneratorError: 生成失败 """ try: if not date: date = self.get_latest_date() # 验证日期格式 try: parsed = datetime.strptime(date, '%Y-%m-%d') display_date = parsed.strftime('%m/%d') query_date = parsed.strftime('%Y-%m-%d') except ValueError as e: error_msg = f"日期格式无效: {date}, 错误: {e}" logger.error(error_msg) raise ReportGeneratorError(error_msg) from e # 获取数据 daily_data = self.get_daily_data(query_date) monthly_data = self.get_monthly_stats(query_date) personnel = self.get_shift_personnel(query_date) # 生成日报 lines: List[str] = [] lines.append(f"日期：{display_date}") lines.append("") # 船次信息 if daily_data['ships']: ship_lines: List[str] = [] for ship, ship_data in sorted(daily_data['ships'].items(), key=lambda x: -x[1]['teu']): ship_lines.append(f"船名：{ship}") teu = ship_data['teu'] twenty_feet = ship_data.get('twenty_feet', 0) forty_feet = ship_data.get('forty_feet', 0) # 构建尺寸箱量字符串 size_parts = [] if twenty_feet > 0: size_parts.append(f"20尺*{twenty_feet}") if forty_feet > 0: size_parts.append(f"40尺*{forty_feet}") if size_parts: size_str = " ".join(size_parts) ship_lines.append(f"作业量：{teu}TEU（{size_str}）") else: ship_lines.append(f"作业量：{teu}TEU") lines.extend(ship_lines) lines.append("") # 当日实际作业量 lines.append(f"当日实际作业量：{daily_data['total_teu']}TEU") # 月度统计 lines.append(f"当月计划作业量：{monthly_data['planned']}TEU (用天数*{config.DAILY_TARGET_TEU}TEU)") lines.append(f"当月实际作业量：{monthly_data['actual']}TEU") lines.append(f"当月完成比例：{monthly_data['completion']}%") lines.append("") # 人员信息 day_personnel = personnel['day_shift'] night_personnel = personnel['night_shift'] duty_phone = personnel['duty_phone'] # 班次日期使用次日 next_day = (parsed + timedelta(days=1)).strftime('%m/%d') lines.append(f"{next_day} 白班人员：{day_personnel}") lines.append(f"{next_day} 夜班人员：{night_personnel}") lines.append(f"24小时值班手机：{duty_phone}") report = "\n".join(lines) logger.info(f"日报生成完成: {date}") return report except ReportGeneratorError: raise except Exception as e: error_msg = f"生成日报失败: {e}" logger.error(error_msg) raise ReportGeneratorError(error_msg) from e def print_report(self, date: Optional[str] = None) -> str: """ 打印日报 参数: date: 日期字符串，格式 "YYYY-MM-DD"，如果为None则使用最新日期 返回: 日报文本 """ try: report = self.generate_report(date) print(report) return report except ReportGeneratorError as e: print(f"生成日报失败: {e}") return "" def save_report_to_file(self, date: Optional[str] = None, filepath: Optional[str] = None) -> bool: """ 保存日报到文件 参数: date: 日期字符串，如果为None则使用最新日期 filepath: 文件路径，如果为None则使用默认路径 返回: 是否成功 """ try: report = self.generate_report(date) if filepath is None: # 使用默认路径 import os report_dir = "reports" os.makedirs(report_dir, exist_ok=True) if date is None: date = self.get_latest_date() filename = f"daily_report_{date}.txt" filepath = os.path.join(report_dir, filename) with open(filepath, 'w', encoding='utf-8') as f: f.write(report) logger.info(f"日报已保存到文件: {filepath}") return True except Exception as e: logger.error(f"保存日报到文件失败: {e}") return False def _empty_personnel(self) -> Dict[str, str]: """返回空的人员信息""" return { 'day_shift': '', 'night_shift': '', 'duty_phone': config.DUTY_PHONE } def close(self): """关闭数据库连接""" self.db.close() if __name__ == '__main__': # 测试代码 import sys # 设置日志 logging.basicConfig(level=logging.INFO) generator = DailyReportGenerator() try: # 测试获取最新日期 latest_date = generator.get_latest_date() print(f"最新日期: {latest_date}") # 测试生成日报 report = generator.generate_report(latest_date) print(f"\n日报内容:\n{report}") # 测试保存到文件 success = generator.save_report_to_file(latest_date) print(f"\n保存到文件: {'成功' if success else '失败'}") except ReportGeneratorError as e: print(f"日报生成错误: {e}") sys.exit(1) except Exception as e: print(f"未知错误: {e}") sys.exit(1) finally: generator.close()