qichi.liang
0cbc587bf3
1. 新增月底/月初智能数据调整功能 - 月底最后一天自动弹出剔除数据对话框 - 月初1号自动弹出添加数据对话框 - 普通日期不弹出对话框 2. 实现月底剔除数据自动转移到次月1号 - 月底剔除的数据自动添加到次月1号统计 - 支持跨月、跨年数据转移 - 数据备注自动记录转移信息 3. 修复自动获取数据后不弹出调整对话框的问题 - 修改auto_fetch_data()方法，成功获取数据后调用调整处理 - 确保第一次打开GUI也能弹出相应对话框 4. 修复月度统计不包含调整数据的问题 - 修改get_monthly_stats()方法包含手动调整数据 - 确保调整数据正确影响月度统计 5. 恢复日报原始模板格式 - 移除调整数据的详细说明 - 保持原始日报模板，只显示最终结果 6. 数据库增强 - 新增manual_adjustments表存储手动调整数据 - 实现调整数据的增删改查方法 - 实现包含调整数据的每日数据获取方法 测试通过：所有功能正常工作，数据计算准确。
447 lines
16 KiB
Python
447 lines
16 KiB
Python
#!/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:
|
||
# 使用数据库的新方法获取包含调整的数据
|
||
if hasattr(self.db, 'get_daily_data_with_adjustments'):
|
||
return self.db.get_daily_data_with_adjustments(date)
|
||
|
||
# 降级处理：如果没有新方法，使用原始逻辑
|
||
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),
|
||
'adjustments': [],
|
||
'total_adjustment_teu': 0
|
||
}
|
||
|
||
except Exception as e:
|
||
logger.error(f"获取每日数据失败: {date}, 错误: {e}")
|
||
return {
|
||
'date': date,
|
||
'ships': {},
|
||
'total_teu': 0,
|
||
'ship_count': 0,
|
||
'adjustments': [],
|
||
'total_adjustment_teu': 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']
|
||
|
||
# 获取当月所有日期的调整数据
|
||
total_adjustment_teu = 0
|
||
adjustment_details: Dict[str, Dict[str, int]] = {}
|
||
|
||
# 获取当月所有日期的调整数据
|
||
for day in range(1, target_date.day + 1):
|
||
day_str = f"{year_month}-{day:02d}"
|
||
if day_str <= date: # 只统计到指定日期
|
||
# 获取该日期的调整数据
|
||
if hasattr(self.db, 'get_daily_data_with_adjustments'):
|
||
daily_data = self.db.get_daily_data_with_adjustments(day_str)
|
||
adjustment_teu = daily_data.get('total_adjustment_teu', 0)
|
||
if adjustment_teu != 0:
|
||
total_adjustment_teu += adjustment_teu
|
||
adjustment_details[day_str] = {
|
||
'adjustment_teu': adjustment_teu,
|
||
'total_teu': daily_data.get('total_teu', 0)
|
||
}
|
||
|
||
# 计算当月天数（已过的天数）
|
||
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 + total_adjustment_teu
|
||
|
||
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,
|
||
'adjustment_total': total_adjustment_teu,
|
||
'completion': completion,
|
||
'daily_totals': daily_totals,
|
||
'adjustment_details': adjustment_details
|
||
}
|
||
|
||
except Exception as e:
|
||
logger.error(f"获取月度统计失败: {date}, 错误: {e}")
|
||
return {
|
||
'year_month': date[:7],
|
||
'days_passed': 0,
|
||
'planned': 0,
|
||
'actual': 0,
|
||
'unaccounted': 0,
|
||
'adjustment_total': 0,
|
||
'completion': 0,
|
||
'daily_totals': {},
|
||
'adjustment_details': {}
|
||
}
|
||
|
||
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()
|