#!/usr/bin/env python3 """ 码头作业日志管理工具 从 Confluence 获取交接班日志并保存到数据库 更新依赖，使用新的模块结构 """ import argparse import sys import os from datetime import datetime from typing import Optional, List from src.config import config from src.logging_config import setup_logging, get_logger from src.confluence import ConfluenceClient, ConfluenceClientError, HTMLTextExtractor, HTMLTextExtractorError, HandoverLogParser, ShipLog, LogParserError from src.database.daily_logs import DailyLogsDatabase from src.report import DailyReportGenerator, ReportGeneratorError # 初始化日志 logger = get_logger(__name__) def ensure_debug_dir(): """确保debug目录存在""" if not os.path.exists(config.DEBUG_DIR): os.makedirs(config.DEBUG_DIR) logger.info(f"创建调试目录: {config.DEBUG_DIR}") def get_timestamp() -> str: """获取时间戳用于文件名""" return datetime.now().strftime('%Y%m%d_%H%M%S') def fetch_html() -> str: """ 获取HTML内容 返回: HTML字符串 异常: SystemExit: 配置错误或获取失败 """ # 验证配置 if not config.validate(): logger.error("配置验证失败，请检查 .env 文件") sys.exit(1) try: logger.info("正在从 Confluence 获取 HTML 内容...") client = ConfluenceClient() html = client.get_html(config.CONFLUENCE_CONTENT_ID) logger.info(f"获取成功，共 {len(html)} 字符") return html except ConfluenceClientError as e: logger.error(f"获取HTML失败: {e}") sys.exit(1) except Exception as e: logger.error(f"未知错误: {e}") sys.exit(1) def extract_text(html: str) -> str: """ 提取布局文本 参数: html: HTML字符串 返回: 提取的文本 """ try: logger.info("正在提取布局文本...") extractor = HTMLTextExtractor() layout_text = extractor.extract(html) logger.info(f"提取完成，共 {len(layout_text)} 字符") return layout_text except HTMLTextExtractorError as e: logger.error(f"提取文本失败: {e}") raise except Exception as e: logger.error(f"未知错误: {e}") raise def save_debug_file(content: str, suffix: str = '') -> str: """ 保存调试文件到debug目录 参数: content: 要保存的内容 suffix: 文件名后缀 返回: 保存的文件路径 """ ensure_debug_dir() filename = f'layout_output{suffix}.txt' if suffix else 'layout_output.txt' filepath = os.path.join(config.DEBUG_DIR, filename) try: with open(filepath, 'w', encoding='utf-8') as f: f.write(content) logger.info(f"已保存到 {filepath}") return filepath except Exception as e: logger.error(f"保存调试文件失败: {e}") raise def parse_logs(text: str) -> List[ShipLog]: """ 解析日志数据 参数: text: 日志文本 返回: 船次日志列表 """ try: logger.info("正在解析日志数据...") parser = HandoverLogParser() logs = parser.parse(text) logger.info(f"解析到 {len(logs)} 条记录") return logs except LogParserError as e: logger.error(f"解析日志失败: {e}") raise except Exception as e: logger.error(f"未知错误: {e}") raise def save_to_db(logs: List[ShipLog]) -> int: """ 保存到数据库 参数: logs: 船次日志列表 返回: 保存的记录数 """ if not logs: logger.warning("没有记录可保存") return 0 try: db = DailyLogsDatabase() count = db.insert_many([log.to_dict() for log in logs]) logger.info(f"已保存 {count} 条记录到数据库") stats = db.get_stats() logger.info(f"数据库统计: 总记录={stats['total']}, 船次={len(stats['ships'])}, " f"日期范围={stats['date_range']['start']}~{stats['date_range']['end']}") return count except Exception as e: logger.error(f"保存到数据库失败: {e}") raise def add_unaccounted(year_month: str, teu: int, note: str = ''): """ 添加未统计数据 参数: year_month: 年月字符串，格式 "2025-12" teu: 未统计TEU数量 note: 备注 """ try: db = DailyLogsDatabase() result = db.insert_unaccounted(year_month, teu, note) if result: logger.info(f"已添加 {year_month} 月未统计数据: {teu}TEU") else: logger.error("添加失败") except Exception as e: logger.error(f"添加未统计数据失败: {e}") raise def remove_unaccounted(year_month: str, teu_to_reduce: int = None): """ 去除未统计数据 参数: year_month: 年月字符串，格式 "2025-12" teu_to_reduce: 要减少的TEU数量，如果为None则删除整个记录 """ try: db = DailyLogsDatabase() if teu_to_reduce is None: # 如果没有指定减少数量，则删除整个记录 result = db.delete_unaccounted(year_month) if result: logger.info(f"已删除 {year_month} 月未统计数据") else: logger.error("删除失败") else: # 减少指定数量的TEU result = db.reduce_unaccounted(year_month, teu_to_reduce) if result: logger.info(f"已减少 {year_month} 月未统计数据: {teu_to_reduce}TEU") else: logger.error("减少失败") except Exception as e: logger.error(f"去除未统计数据失败: {e}") raise def add_cross_month_exclusion(source_date: str, target_date: str, ship_name: str, teu: int, twenty_feet: int = 0, forty_feet: int = 0, reason: str = ''): """ 添加跨月剔除调整（手动剔除次月多统计的船） 参数: source_date: 源日期（上月底日期） target_date: 目标日期（次月日期） ship_name: 船名 teu: TEU数量 twenty_feet: 20尺箱量 forty_feet: 40尺箱量 reason: 调整原因 """ try: db = DailyLogsDatabase() success = db.insert_cross_month_exclusion( source_date=source_date, target_date=target_date, ship_name=ship_name, teu=teu, twenty_feet=twenty_feet, forty_feet=forty_feet, reason=reason ) if success: logger.info(f"已添加跨月剔除调整: {source_date} -> {target_date} {ship_name} {teu}TEU") else: logger.error("添加跨月剔除调整失败") except Exception as e: logger.error(f"添加跨月剔除调整失败: {e}") raise def show_stats(date: str): """ 显示指定日期的统计 参数: date: 日期字符串，格式 "YYYY-MM-DD" """ try: generator = DailyReportGenerator() generator.print_report(date) except ReportGeneratorError as e: logger.error(f"生成统计失败: {e}") except Exception as e: logger.error(f"未知错误: {e}") def run_fetch() -> str: """执行：获取HTML并提取文本""" html = fetch_html() text = extract_text(html) save_debug_file(text) return text def run_fetch_and_save(): """执行：获取、提取、解析、保存到数据库""" text = run_fetch() logs = parse_logs(text) save_to_db(logs) def run_fetch_save_debug() -> str: """执行：获取、提取、保存到debug目录""" html = fetch_html() text = extract_text(html) suffix = f'_{get_timestamp()}' save_debug_file(text, suffix) return text def run_report(date: Optional[str] = None): """执行：生成日报""" if not date: date = datetime.now().strftime('%Y-%m-%d') show_stats(date) def run_config_test(): """执行：配置测试""" logger.info("配置测试:") config.print_summary() # 测试Confluence连接 try: client = ConfluenceClient() if client.test_connection(): logger.info("Confluence连接测试: 成功") else: logger.warning("Confluence连接测试: 失败") except Exception as e: logger.error(f"Confluence连接测试失败: {e}") # 测试数据库连接 try: db = DailyLogsDatabase() stats = db.get_stats() logger.info(f"数据库连接测试: 成功，总记录: {stats['total']}") except Exception as e: logger.error(f"数据库连接测试失败: {e}") # 功能映射 FUNCTIONS = { 'fetch': run_fetch, 'fetch-save': run_fetch_and_save, 'fetch-debug': run_fetch_save_debug, 'report': lambda: run_report(), 'report-today': lambda: run_report(datetime.now().strftime('%Y-%m-%d')), 'config-test': run_config_test, 'stats': lambda: show_stats(datetime.now().strftime('%Y-%m-%d')), } def main(): parser = argparse.ArgumentParser( description='码头作业日志管理工具', formatter_class=argparse.RawDescriptionHelpFormatter, epilog=''' 可选功能: fetch 获取HTML并提取文本（保存到debug目录） fetch-save 获取、提取、解析并保存到数据库 fetch-debug 获取、提取并保存带时间戳的debug文件 report 生成日报（默认今天） report-today 生成今日日报 config-test 配置测试 stats 显示今日统计 参数: --unaccounted, -u TEU 添加未统计数据（需同时指定月份） --remove-unaccounted, -r [TEU] 去除未统计数据（需同时指定月份）。如果指定TEU值，则减少该数量；如果不指定，则删除整个记录 --month, -m YEAR-MONTH 指定月份（与 -u 或 -r 配合使用） --cross-exclude, -c 手动剔除次月多统计的船（需指定源日期、目标日期、船名和TEU） 跨月剔除参数: --source-date DATE 源日期（上月底日期），格式: YYYY-MM-DD --target-date DATE 目标日期（次月日期），格式: YYYY-MM-DD --ship-name NAME 船名 --teu TEU TEU数量 --twenty-feet COUNT 20尺箱量（可选，默认0） --forty-feet COUNT 40尺箱量（可选，默认0） --reason REASON 调整原因（可选） 示例: python3 main.py fetch python3 main.py fetch-save python3 main.py report 2025-12-28 python3 main.py config-test python3 main.py --unaccounted 118 --month 2025-12 python3 main.py --remove-unaccounted --month 2025-12 # 删除整个记录 python3 main.py --remove-unaccounted 118 --month 2025-12 # 减少118TEU python3 main.py --cross-exclude --source-date 2025-12-31 --target-date 2026-01-01 --ship-name "学友洋山" --teu 100 ''' ) parser.add_argument( 'function', nargs='?', default='fetch-save', choices=['fetch', 'fetch-save', 'fetch-debug', 'report', 'report-today', 'config-test', 'stats'], help='要执行的功能 (默认: fetch-save)' ) parser.add_argument( 'date', nargs='?', help='日期 (格式: YYYY-MM-DD)，用于 report 功能' ) parser.add_argument( '--unaccounted', '-u', metavar='TEU', type=int, help='添加未统计数据（需同时指定月份，如 -u 118 2025-12）' ) parser.add_argument( '--remove-unaccounted', '-r', metavar='TEU', nargs='?', const=None, type=int, help='去除未统计数据（需同时指定月份，如 -r 118 2025-12）。如果指定TEU值，则减少该数量；如果不指定，则删除整个记录' ) parser.add_argument( '--month', '-m', metavar='YEAR-MONTH', help='指定月份（与 --unaccounted 或 --remove-unaccounted 配合使用）' ) parser.add_argument( '--cross-exclude', '-c', action='store_true', help='手动剔除次月多统计的船' ) parser.add_argument( '--source-date', metavar='DATE', help='源日期（上月底日期），格式: YYYY-MM-DD' ) parser.add_argument( '--target-date', metavar='DATE', help='目标日期（次月日期），格式: YYYY-MM-DD' ) parser.add_argument( '--ship-name', metavar='NAME', help='船名' ) parser.add_argument( '--teu', metavar='TEU', type=int, help='TEU数量' ) parser.add_argument( '--twenty-feet', metavar='COUNT', type=int, default=0, help='20尺箱量（可选，默认0）' ) parser.add_argument( '--forty-feet', metavar='COUNT', type=int, default=0, help='40尺箱量（可选，默认0）' ) parser.add_argument( '--reason', metavar='REASON', default='手动剔除次月多统计的船', help='调整原因（可选，默认: "手动剔除次月多统计的船"）' ) args = parser.parse_args() # 添加未统计数据 if args.unaccounted: year_month = args.month or datetime.now().strftime('%Y-%m') try: add_unaccounted(year_month, args.unaccounted) except Exception as e: logger.error(f"添加未统计数据失败: {e}") sys.exit(1) return # 去除未统计数据 # 检查是否提供了 --remove-unaccounted 或 -r 参数 has_remove_arg = any(arg in sys.argv for arg in ['--remove-unaccounted', '-r']) if has_remove_arg: year_month = args.month or datetime.now().strftime('%Y-%m') try: # args.remove_unaccounted 可能是整数（指定TEU）或 None（未指定） if isinstance(args.remove_unaccounted, int): # 指定了TEU值，减少指定数量 remove_unaccounted(year_month, args.remove_unaccounted) else: # 未指定TEU值，删除整个记录 remove_unaccounted(year_month) except Exception as e: logger.error(f"去除未统计数据失败: {e}") sys.exit(1) return # 跨月剔除功能 if args.cross_exclude: if not all([args.source_date, args.target_date, args.ship_name, args.teu]): logger.error("跨月剔除功能需要指定以下参数: --source-date, --target-date, --ship-name, --teu") sys.exit(1) try: add_cross_month_exclusion( source_date=args.source_date, target_date=args.target_date, ship_name=args.ship_name, teu=args.teu, twenty_feet=args.twenty_feet, forty_feet=args.forty_feet, reason=args.reason ) except Exception as e: logger.error(f"跨月剔除失败: {e}") sys.exit(1) return # 执行功能 try: if args.function == 'report' and args.date: run_report(args.date) else: FUNCTIONS[args.function]() except KeyboardInterrupt: logger.info("用户中断操作") sys.exit(0) except Exception as e: logger.error(f"执行功能失败: {e}") sys.exit(1) if __name__ == '__main__': # 初始化日志系统 setup_logging() # 打印启动信息 logger.info("=" * 50) logger.info("码头作业日志管理工具 - OrbitIn") logger.info(f"启动时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}") logger.info("=" * 50) # 运行主程序 main()