"""配送时效预警定时任务 依据订单分配时写入的配送截止时间(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) if remaining_minutes < 0: if not self._get_existing_unresolved_warning(task.id, "timeout"): w = self._create_warning( task, "timeout", "已超时", remaining_minutes, f"已超过配送时效 {abs(remaining_minutes)} 分钟,仍未完成配送", ) if w: warnings.append(w) return warnings if remaining_minutes <= advance: if not self._get_existing_unresolved_warning(task.id, "advance_deadline"): w = self._create_warning( task, "advance_deadline", "时效预警", remaining_minutes, f"距配送截止不足 {advance} 分钟(剩余约 {remaining_minutes} 分钟),请督促配送", ) if w: warnings.append(w) return warnings def _create_warning( self, task: BizDeliveryTask, warning_type: str, urgency_level: str, remaining_minutes: int, message: str ) -> Optional[BizDeliveryWarning]: """创建预警记录""" try: 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_existing_unresolved_warning(self, task_id: str, warning_type: str) -> Optional[BizDeliveryWarning]: return self.db.query(BizDeliveryWarning).filter( BizDeliveryWarning.task_id == task_id, BizDeliveryWarning.warning_type == warning_type, BizDeliveryWarning.is_del == 0, BizDeliveryWarning.is_resolved == 0, ).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