"""配送时效预警定时任务 依据订单分配时写入的配送截止时间(biz_delivery_task.effective_time)及区域表 sys_region.delivery_warn_advance_minutes(缺省 30 分钟)生成运营端预警: - 进入「截止时间前 N 分钟」窗口且仍未完成配送:生成一条 advance_deadline 预警; - 已超过截止时间仍未完成:生成 timeout 预警; - 配送员确认送达(任务状态 delivering)或任务取消/完成时,由业务侧自动结案预警。 建议:每 1~5 分钟执行一次 check_delivery_warnings(db)。 """ import logging from datetime import datetime from typing import Optional from sqlalchemy.orm import Session from app.models.models import BizDeliveryTask, BizDeliveryWarning, SysRegion from app.core.utils import generate_id from app.crud.delivery_warning_resolve import auto_resolve_open_warnings_for_task logger = logging.getLogger(__name__) DEFAULT_ADVANCE_MINUTES = 30 class DeliveryWarningTask: """配送时效预警任务""" def __init__(self, db: Session): self.db = db def check_delivery_warnings(self) -> dict: """检查配送任务并生成预警""" result = { "checked": 0, "warnings_created": 0, "notifications_sent": 0, "errors": [] } try: tasks = ( self.db.query(BizDeliveryTask) .filter(BizDeliveryTask.is_del == 0) .all() ) result["checked"] = len(tasks) for task in tasks: try: if task.status in ("delivering", "completed", "cancelled"): auto_resolve_open_warnings_for_task(self.db, task.id) continue if task.status not in ("pending", "assigned", "picking"): continue warnings = self._check_task_warnings(task) result["warnings_created"] += len(warnings) for warning in warnings: self._send_warning_notification(warning) result["notifications_sent"] += 1 except Exception as e: logger.error(f"处理任务 {task.id} 时出错: {str(e)}") result["errors"].append({ "task_id": task.id, "error": str(e) }) self.db.commit() except Exception as e: logger.error(f"配送时效预警任务执行失败: {str(e)}") self.db.rollback() result["errors"].append({"task_id": "all", "error": str(e)}) return result def _region_advance_minutes(self, region_id: str) -> int: reg = self.db.query(SysRegion).filter(SysRegion.id == region_id, SysRegion.is_del == 0).first() if not reg: return DEFAULT_ADVANCE_MINUTES m = getattr(reg, "delivery_warn_advance_minutes", None) if m is None or int(m) < 1: return DEFAULT_ADVANCE_MINUTES return int(m) def _check_task_warnings(self, task: BizDeliveryTask) -> list: warnings = [] if not task.effective_time: return warnings now = datetime.now() remaining_minutes = int((task.effective_time - now).total_seconds() // 60) advance = self._region_advance_minutes(task.region_id) warning_type = None urgency_level = None message = None if remaining_minutes < 0: warning_type = "timeout" urgency_level = "已超时" message = f"已超过配送时效 {abs(remaining_minutes)} 分钟,仍未完成配送" elif remaining_minutes <= advance: warning_type = "advance_deadline" urgency_level = "时效预警" message = f"距配送截止不足 {advance} 分钟(剩余约 {remaining_minutes} 分钟),请督促配送" else: return warnings w = self._upsert_task_warning( task=task, warning_type=warning_type, urgency_level=urgency_level, remaining_minutes=remaining_minutes, message=message, ) if w: warnings.append(w) return warnings def _upsert_task_warning( self, task: BizDeliveryTask, warning_type: str, urgency_level: str, remaining_minutes: int, message: str ) -> Optional[BizDeliveryWarning]: """单任务单预警:存在则更新,不存在则创建""" try: warning = self._get_latest_warning_by_task(task.id) now = datetime.now() if warning: warning.warning_type = warning_type warning.urgency_level = urgency_level warning.remaining_minutes = remaining_minutes warning.message = message warning.order_id = task.order_id warning.is_sent = 0 # 未完成前统一保持未结案(若历史上被手工结案,则重新激活) warning.is_resolved = 0 warning.resolved_time = None warning.resolved_by = None warning.update_at = now warning.update_by = "system" self.db.flush() return warning warning = BizDeliveryWarning( id=generate_id(), task_id=task.id, order_id=task.order_id, warning_type=warning_type, urgency_level=urgency_level, remaining_minutes=remaining_minutes, message=message, is_sent=0, is_resolved=0, create_by="system", create_name="系统", ) self.db.add(warning) self.db.flush() return warning except Exception as e: logger.error(f"创建预警记录失败: {str(e)}") return None def _get_latest_warning_by_task(self, task_id: str) -> Optional[BizDeliveryWarning]: return self.db.query(BizDeliveryWarning).filter( BizDeliveryWarning.task_id == task_id, BizDeliveryWarning.is_del == 0, ).order_by(BizDeliveryWarning.update_at.desc(), BizDeliveryWarning.create_at.desc()).first() def _send_warning_notification(self, warning: BizDeliveryWarning) -> bool: """发送预警通知(可通过微信模板消息等)""" try: # 获取配送员信息 task = self.db.query(BizDeliveryTask).filter( BizDeliveryTask.id == warning.task_id ).first() if not task or not task.deliver_id: logger.warning(f"任务 {warning.task_id} 未分配配送员,跳过通知") return False # TODO: 调用微信模板消息接口发送通知 # from app.core.wechat_pay import send_delivery_warning_message # send_delivery_warning_message( # deliver_id=task.deliver_id, # task_no=task.task_no, # message=warning.message # ) logger.info(f"已发送预警通知: {task.task_no} - {warning.message}") return True except Exception as e: logger.error(f"发送预警通知失败: {str(e)}") return False def check_delivery_warnings(db: Session) -> dict: """执行配送时效预警检查(供外部调用)""" task = DeliveryWarningTask(db) return task.check_delivery_warnings() def cleanup_expired_warnings(db: Session, days: int = 7) -> int: """清理过期的预警记录 Args: db: 数据库会话 days: 保留天数,默认7天 Returns: int: 删除的记录数 """ from datetime import timedelta cutoff_date = datetime.now() - timedelta(days=days) count = db.query(BizDeliveryWarning).filter( BizDeliveryWarning.create_at < cutoff_date, BizDeliveryWarning.is_resolved == 1, BizDeliveryWarning.is_del == 0 ).update({"is_del": 1}) db.commit() logger.info(f"已清理 {count} 条过期预警记录") return count