59 lines
2.6 KiB
Python
59 lines
2.6 KiB
Python
|
"""Moteur unique de génération des échéances depuis les tâches de lot."""
|
||
|
|
from datetime import date, timedelta
|
||
|
|
|
||
|
|
from app_new.extensions import db
|
||
|
|
from app_new.core.models.planning import Meter, ScheduledTask
|
||
|
|
from app_new.core.models.maintenance import LotTask
|
||
|
|
from app_new.core.services.planning_service import PlanningService
|
||
|
|
|
||
|
|
|
||
|
|
def _targets(lot):
|
||
|
|
result = []
|
||
|
|
for root in lot.equipments.filter_by(is_deleted=False):
|
||
|
|
if root.tracked_individually:
|
||
|
|
result.extend([x for x in root.all_children if not x.is_group and not x.is_deleted])
|
||
|
|
elif root.is_group and root.children.count():
|
||
|
|
result.extend([x for x in root.children if not x.is_deleted])
|
||
|
|
else:
|
||
|
|
result.append(root)
|
||
|
|
return list({item.id: item for item in result}.values())
|
||
|
|
|
||
|
|
|
||
|
|
def _working_day(candidate):
|
||
|
|
for _ in range(370):
|
||
|
|
if PlanningService.get_working_hours(candidate):
|
||
|
|
return candidate
|
||
|
|
candidate += timedelta(days=1)
|
||
|
|
raise ValueError("Aucune journée travaillée disponible dans les 12 prochains mois.")
|
||
|
|
|
||
|
|
|
||
|
|
def generate_due_tasks(task: LotTask, event=None, today=None):
|
||
|
|
"""Génère les échéances sans doublon et retourne les objets créés."""
|
||
|
|
today = today or date.today()
|
||
|
|
if not task.is_active:
|
||
|
|
return []
|
||
|
|
if task.trigger_type in ("event", "weather") and task.trigger_event != event:
|
||
|
|
return []
|
||
|
|
if task.trigger_type == "season":
|
||
|
|
start, end = task.season_start_month or 1, task.season_end_month or 12
|
||
|
|
inside = start <= today.month <= end if start <= end else today.month >= start or today.month <= end
|
||
|
|
if not inside:
|
||
|
|
return []
|
||
|
|
created = []
|
||
|
|
for equipment in _targets(task.lot):
|
||
|
|
if task.trigger_type == "meter":
|
||
|
|
meter = Meter.query.filter_by(equipment_id=equipment.id, is_active=True).order_by(Meter.current_value.desc()).first()
|
||
|
|
if not meter or task.meter_threshold is None or meter.current_value < task.meter_threshold:
|
||
|
|
continue
|
||
|
|
existing = ScheduledTask.query.filter_by(lot_task_id=task.id, equipment_id=equipment.id, status="planned").first()
|
||
|
|
if existing:
|
||
|
|
continue
|
||
|
|
due = _working_day(today + timedelta(days=max(task.advance_days or 0, 0)))
|
||
|
|
item = ScheduledTask(lot_task_id=task.id, equipment_id=equipment.id,
|
||
|
|
room_id=equipment.effective_room.id if equipment.effective_room else None,
|
||
|
|
scheduled_date=due, estimated_duration=task.effective_duration(), status="planned")
|
||
|
|
db.session.add(item)
|
||
|
|
created.append(item)
|
||
|
|
db.session.commit()
|
||
|
|
return created
|