290 lines
14 KiB
Python
290 lines
14 KiB
Python
|
"""Services métier C4 : contrats photocopieurs et consommables partagés."""
|
||
|
|
from calendar import monthrange
|
||
|
|
from datetime import date, datetime, timedelta, timezone
|
||
|
|
from statistics import mean
|
||
|
|
|
||
|
|
from sqlalchemy import and_, or_
|
||
|
|
|
||
|
|
from ..models import (
|
||
|
|
Alert, Company, Consumable, ConsumableOrder, ConsumableReceipt,
|
||
|
|
ConsumableUsage, ContractEquipmentAssignment, Equipment, EquipmentConsumable,
|
||
|
|
Meter, MeterReading, PhotocopierContract, PhotocopierContractPeriod,
|
||
|
|
)
|
||
|
|
from ...extensions import db
|
||
|
|
|
||
|
|
|
||
|
|
class C4DomainError(ValueError):
|
||
|
|
pass
|
||
|
|
|
||
|
|
|
||
|
|
def _anniversary(year, month, day):
|
||
|
|
return date(year, month, min(day, monthrange(year, month)[1]))
|
||
|
|
|
||
|
|
|
||
|
|
def create_contract(*, supplier_id, name, start_date, end_date, anniversary_month,
|
||
|
|
anniversary_day, quota_start_date=None, reference=None, notes=None, commit=True):
|
||
|
|
if end_date < start_date:
|
||
|
|
raise C4DomainError("La fin du contrat doit être postérieure à son début.")
|
||
|
|
if not 1 <= int(anniversary_month) <= 12:
|
||
|
|
raise C4DomainError("Le mois anniversaire est invalide.")
|
||
|
|
quota_start_date = quota_start_date or start_date
|
||
|
|
if quota_start_date < start_date or quota_start_date > end_date:
|
||
|
|
raise C4DomainError("La date de début des quotas doit être comprise dans le contrat.")
|
||
|
|
contract = PhotocopierContract(
|
||
|
|
supplier_id=supplier_id, name=name, reference=reference,
|
||
|
|
start_date=start_date, end_date=end_date,
|
||
|
|
quota_start_date=quota_start_date,
|
||
|
|
anniversary_month=int(anniversary_month), anniversary_day=int(anniversary_day),
|
||
|
|
notes=notes, status="ACTIVE",
|
||
|
|
)
|
||
|
|
db.session.add(contract)
|
||
|
|
db.session.flush()
|
||
|
|
generate_contract_periods(contract, commit=False)
|
||
|
|
if commit:
|
||
|
|
db.session.commit()
|
||
|
|
return contract
|
||
|
|
|
||
|
|
|
||
|
|
def generate_contract_periods(contract, *, commit=True):
|
||
|
|
"""Crée les périodes annuelles sans doublon, bornées au contrat."""
|
||
|
|
cursor = contract.quota_start_date or contract.start_date
|
||
|
|
while cursor <= contract.end_date:
|
||
|
|
next_anniversary = _anniversary(cursor.year + 1, contract.anniversary_month, contract.anniversary_day)
|
||
|
|
period_end = min(next_anniversary - timedelta(days=1), contract.end_date)
|
||
|
|
if period_end < cursor:
|
||
|
|
break
|
||
|
|
existing = PhotocopierContractPeriod.query.filter_by(contract_id=contract.id, start_date=cursor).first()
|
||
|
|
if not existing:
|
||
|
|
db.session.add(PhotocopierContractPeriod(contract=contract, start_date=cursor, end_date=period_end))
|
||
|
|
cursor = period_end + timedelta(days=1)
|
||
|
|
if commit:
|
||
|
|
db.session.commit()
|
||
|
|
return contract.periods
|
||
|
|
|
||
|
|
|
||
|
|
def _overlaps(start, end, other_start, other_end):
|
||
|
|
return start <= (other_end or date.max) and (end or date.max) >= other_start
|
||
|
|
|
||
|
|
|
||
|
|
def assign_equipment_to_contract(*, contract, equipment, entry_date, exit_date=None,
|
||
|
|
replacement_reason=None, replacement_comment=None, commit=True):
|
||
|
|
if contract.status != "ACTIVE":
|
||
|
|
raise C4DomainError("Un contrat clos ne peut plus recevoir de machine.")
|
||
|
|
if entry_date < contract.start_date or entry_date > contract.end_date:
|
||
|
|
raise C4DomainError("La date d'entrée doit être comprise dans le contrat.")
|
||
|
|
if exit_date and exit_date > contract.end_date:
|
||
|
|
raise C4DomainError("La date de sortie dépasse la fin du contrat.")
|
||
|
|
for row in ContractEquipmentAssignment.query.filter_by(equipment_id=equipment.id).all():
|
||
|
|
if _overlaps(entry_date, exit_date, row.entry_date, row.exit_date):
|
||
|
|
raise C4DomainError("Cette machine est déjà rattachée à un contrat sur cette période.")
|
||
|
|
assignment = ContractEquipmentAssignment(
|
||
|
|
contract=contract, equipment=equipment, entry_date=entry_date, exit_date=exit_date,
|
||
|
|
replacement_reason=replacement_reason, replacement_comment=replacement_comment,
|
||
|
|
)
|
||
|
|
db.session.add(assignment)
|
||
|
|
if commit:
|
||
|
|
db.session.commit()
|
||
|
|
return assignment
|
||
|
|
|
||
|
|
|
||
|
|
def replace_contract_equipment(*, old_assignment, new_equipment, replacement_date,
|
||
|
|
reason="AUTRE", comment=None, commit=True):
|
||
|
|
if reason not in {"PANNE", "RENOUVELLEMENT_FOURNISSEUR", "CHANGEMENT_MODELE", "FIN_DE_VIE", "AUTRE"}:
|
||
|
|
raise C4DomainError("Motif de remplacement invalide.")
|
||
|
|
if replacement_date < old_assignment.entry_date:
|
||
|
|
raise C4DomainError("La date de remplacement est invalide.")
|
||
|
|
old_assignment.exit_date = replacement_date - timedelta(days=1)
|
||
|
|
new_assignment = assign_equipment_to_contract(
|
||
|
|
contract=old_assignment.contract, equipment=new_equipment,
|
||
|
|
entry_date=replacement_date, replacement_reason=reason,
|
||
|
|
replacement_comment=comment, commit=False,
|
||
|
|
)
|
||
|
|
if commit:
|
||
|
|
db.session.commit()
|
||
|
|
return new_assignment
|
||
|
|
|
||
|
|
|
||
|
|
def _meter_role(meter):
|
||
|
|
text = f"{meter.name} {meter.usage or ''}".lower()
|
||
|
|
return "color" if any(word in text for word in ("couleur", "color", "coul")) else "bw"
|
||
|
|
|
||
|
|
|
||
|
|
def _meter_consumption(meter, start_date, end_date):
|
||
|
|
readings = (MeterReading.query.filter(
|
||
|
|
MeterReading.meter_id == meter.id,
|
||
|
|
MeterReading.reading_date >= datetime.combine(start_date, datetime.min.time()),
|
||
|
|
MeterReading.reading_date <= datetime.combine(end_date, datetime.max.time()),
|
||
|
|
).order_by(MeterReading.reading_date, MeterReading.id).all())
|
||
|
|
before = (MeterReading.query.filter(
|
||
|
|
MeterReading.meter_id == meter.id,
|
||
|
|
MeterReading.reading_date < datetime.combine(start_date, datetime.min.time()),
|
||
|
|
).order_by(MeterReading.reading_date.desc(), MeterReading.id.desc()).first())
|
||
|
|
if before:
|
||
|
|
readings.insert(0, before)
|
||
|
|
total = 0.0
|
||
|
|
quality = "EXACT" if before else "PARTIAL"
|
||
|
|
previous = None
|
||
|
|
for reading in readings:
|
||
|
|
if previous and not reading.is_reset and not previous.is_reset and reading.value >= previous.value:
|
||
|
|
total += reading.value - previous.value
|
||
|
|
previous = reading
|
||
|
|
if not before and len(readings) >= 2:
|
||
|
|
quality = "ESTIMATED"
|
||
|
|
return total, quality, len(readings)
|
||
|
|
|
||
|
|
|
||
|
|
def calculate_contract_usage(period):
|
||
|
|
result = {"bw": 0.0, "color": 0.0, "quality": "EXACT", "machines": 0}
|
||
|
|
assignments = ContractEquipmentAssignment.query.filter(
|
||
|
|
ContractEquipmentAssignment.contract_id == period.contract_id,
|
||
|
|
ContractEquipmentAssignment.entry_date <= period.end_date,
|
||
|
|
or_(ContractEquipmentAssignment.exit_date.is_(None), ContractEquipmentAssignment.exit_date >= period.start_date),
|
||
|
|
).all()
|
||
|
|
for assignment in assignments:
|
||
|
|
result["machines"] += 1
|
||
|
|
start = max(period.start_date, assignment.entry_date)
|
||
|
|
end = min(period.end_date, assignment.exit_date or period.end_date)
|
||
|
|
for meter in Meter.query.filter_by(equipment_id=assignment.equipment_id).all():
|
||
|
|
amount, quality, _ = _meter_consumption(meter, start, end)
|
||
|
|
result[_meter_role(meter)] += amount
|
||
|
|
if quality != "EXACT":
|
||
|
|
result["quality"] = "PARTIAL" if result["quality"] == "EXACT" else result["quality"]
|
||
|
|
return result
|
||
|
|
|
||
|
|
|
||
|
|
def _school_days(start, end):
|
||
|
|
"""Approximation explicite : jours ouvrés, hors fermetures complètes."""
|
||
|
|
from ..models.planning import CollegeClosure, ClosureWorkDay
|
||
|
|
closures = CollegeClosure.query.filter(CollegeClosure.start_date <= end, CollegeClosure.end_date >= start).all()
|
||
|
|
work_days = {row.work_date for closure in closures for row in ClosureWorkDay.query.filter_by(closure_id=closure.id).all()}
|
||
|
|
closed = {day for closure in closures if closure.work_hours_type == "none"
|
||
|
|
for day in (start + timedelta(days=i) for i in range((end - start).days + 1))
|
||
|
|
if closure.start_date <= day <= closure.end_date}
|
||
|
|
return sum(1 for i in range((end - start).days + 1)
|
||
|
|
if (day := start + timedelta(days=i)).weekday() < 5 and day not in closed or day in work_days)
|
||
|
|
|
||
|
|
|
||
|
|
def project_contract_usage(period):
|
||
|
|
usage = calculate_contract_usage(period)
|
||
|
|
elapsed = max((date.today() - period.start_date).days + 1, 0)
|
||
|
|
remaining = max((period.end_date - max(date.today(), period.start_date)).days, 0)
|
||
|
|
school_elapsed = _school_days(period.start_date, min(date.today(), period.end_date)) if elapsed else 0
|
||
|
|
school_remaining = _school_days(max(date.today() + timedelta(days=1), period.start_date), period.end_date) if remaining else 0
|
||
|
|
result = {"usage": usage, "projection": {}, "remaining": {}, "confidence": "LOW"}
|
||
|
|
for role, quota_key in (("bw", "quota_bw"), ("color", "quota_color")):
|
||
|
|
quota = getattr(period, quota_key)
|
||
|
|
result["remaining"][role] = max(quota - usage[role], 0) if quota is not None else None
|
||
|
|
if quota is None or school_elapsed <= 0:
|
||
|
|
result["projection"][role] = None
|
||
|
|
else:
|
||
|
|
daily = usage[role] / school_elapsed
|
||
|
|
result["projection"][role] = usage[role] + daily * school_remaining
|
||
|
|
result["confidence"] = "NORMAL" if school_elapsed >= 10 else "LOW"
|
||
|
|
return result
|
||
|
|
|
||
|
|
|
||
|
|
def evaluate_quota_alert(period, *, level="WARNING", commit=True):
|
||
|
|
from ..models.company import Alert
|
||
|
|
projection = project_contract_usage(period)
|
||
|
|
for role, quota_key in (("bw", "quota_bw"), ("color", "quota_color")):
|
||
|
|
quota = getattr(period, quota_key)
|
||
|
|
projected = projection["projection"].get(role)
|
||
|
|
if quota is None or projected is None or projected <= quota:
|
||
|
|
continue
|
||
|
|
related_key = f"{period.id}:{role}"
|
||
|
|
alert = Alert.query.filter_by(alert_type="QUOTA_OVERAGE_PROJECTED", related_type=related_key, is_resolved=False).first()
|
||
|
|
message = f"Projection {role} : {projected:.0f} copies pour un quota de {quota}."
|
||
|
|
if alert:
|
||
|
|
alert.message = message
|
||
|
|
else:
|
||
|
|
db.session.add(Alert(
|
||
|
|
alert_type="QUOTA_OVERAGE_PROJECTED", title="Dépassement quota photocopieur",
|
||
|
|
message=message, priority="haute" if level == "CRITICAL" else "normale",
|
||
|
|
related_id=period.id, related_type=related_key,
|
||
|
|
))
|
||
|
|
if commit:
|
||
|
|
db.session.commit()
|
||
|
|
return projection
|
||
|
|
|
||
|
|
|
||
|
|
def create_consumable_order(*, consumable, quantity, supplier_id=None, user_id=None, commit=True):
|
||
|
|
quantity = int(quantity)
|
||
|
|
if quantity <= 0:
|
||
|
|
raise C4DomainError("La quantité commandée doit être positive.")
|
||
|
|
order = ConsumableOrder(consumable=consumable, quantity_ordered=quantity, supplier_id=supplier_id, created_by_id=user_id, status="ORDERED", ordered_at=datetime.now(timezone.utc))
|
||
|
|
db.session.add(order)
|
||
|
|
refresh_consumable_stock_alert(consumable, commit=False)
|
||
|
|
if commit:
|
||
|
|
db.session.commit()
|
||
|
|
return order
|
||
|
|
|
||
|
|
|
||
|
|
def receive_consumable_order(*, order, quantity, user_id=None, comment=None, commit=True):
|
||
|
|
quantity = int(quantity)
|
||
|
|
if quantity <= 0 or quantity > order.remaining_quantity:
|
||
|
|
raise C4DomainError("La quantité reçue dépasse le restant à recevoir.")
|
||
|
|
receipt = ConsumableReceipt(order=order, quantity=quantity, received_by_id=user_id, comment=comment)
|
||
|
|
order.quantity_received += quantity
|
||
|
|
order.consumable.quantity = (order.consumable.quantity or 0) + quantity
|
||
|
|
order.status = "RECEIVED" if order.remaining_quantity == 0 else "PARTIALLY_RECEIVED"
|
||
|
|
if order.status == "RECEIVED":
|
||
|
|
order.received_at = receipt.received_at
|
||
|
|
db.session.add(receipt)
|
||
|
|
refresh_consumable_stock_alert(order.consumable, commit=False)
|
||
|
|
if commit:
|
||
|
|
db.session.commit()
|
||
|
|
return receipt
|
||
|
|
|
||
|
|
|
||
|
|
def record_consumable_issue(*, consumable, equipment, quantity=1, user_id=None, intervention_id=None, notes=None, commit=True):
|
||
|
|
quantity = int(quantity)
|
||
|
|
if quantity <= 0:
|
||
|
|
raise C4DomainError("La sortie doit être positive.")
|
||
|
|
if (consumable.quantity or 0) < quantity:
|
||
|
|
raise C4DomainError("Stock insuffisant.")
|
||
|
|
if not EquipmentConsumable.query.filter_by(equipment_id=equipment.id, consumable_id=consumable.id).first():
|
||
|
|
raise C4DomainError("Cette référence n'est pas compatible avec la machine.")
|
||
|
|
usage = ConsumableUsage(consumable=consumable, equipment=equipment, quantity_used=quantity, used_by_id=user_id, intervention_id=intervention_id, notes=notes)
|
||
|
|
consumable.quantity -= quantity
|
||
|
|
db.session.add(usage)
|
||
|
|
refresh_consumable_stock_alert(consumable, commit=False)
|
||
|
|
if commit:
|
||
|
|
db.session.commit()
|
||
|
|
return usage
|
||
|
|
|
||
|
|
|
||
|
|
def refresh_consumable_stock_alert(consumable, *, commit=True):
|
||
|
|
"""Maintient une seule alerte de stock bas, sans convertir une commande en stock."""
|
||
|
|
pending = ConsumableOrder.query.filter(
|
||
|
|
ConsumableOrder.consumable_id == consumable.id,
|
||
|
|
ConsumableOrder.status.in_(("ORDERED", "PARTIALLY_RECEIVED")),
|
||
|
|
).first()
|
||
|
|
alert = Alert.query.filter_by(alert_type="CONSUMABLE_STOCK_LOW", related_type="consumable", related_id=consumable.id, is_resolved=False).first()
|
||
|
|
if (consumable.quantity or 0) <= (consumable.min_quantity or 0):
|
||
|
|
message = f"Stock bas : {consumable.quantity or 0} / minimum {consumable.min_quantity or 0}."
|
||
|
|
if pending:
|
||
|
|
message += f" Commande en cours : {pending.remaining_quantity} à recevoir."
|
||
|
|
if alert:
|
||
|
|
alert.message = message
|
||
|
|
else:
|
||
|
|
db.session.add(Alert(alert_type="CONSUMABLE_STOCK_LOW", title="Stock consommable bas", message=message, related_type="consumable", related_id=consumable.id))
|
||
|
|
elif alert:
|
||
|
|
alert.is_resolved = True
|
||
|
|
alert.resolved_at = datetime.now(timezone.utc)
|
||
|
|
if commit:
|
||
|
|
db.session.commit()
|
||
|
|
return alert
|
||
|
|
|
||
|
|
|
||
|
|
def estimate_consumable_lifetime(*, consumable, equipment):
|
||
|
|
usages = (ConsumableUsage.query.filter_by(consumable_id=consumable.id, equipment_id=equipment.id)
|
||
|
|
.order_by(ConsumableUsage.used_at).all())
|
||
|
|
durations = [(b.used_at - a.used_at).days for a, b in zip(usages, usages[1:]) if (b.used_at - a.used_at).days > 0]
|
||
|
|
current_days = None
|
||
|
|
if usages:
|
||
|
|
last = usages[-1].used_at
|
||
|
|
if last.tzinfo is None:
|
||
|
|
last = last.replace(tzinfo=timezone.utc)
|
||
|
|
current_days = (datetime.now(timezone.utc) - last).days
|
||
|
|
return {"average_days": mean(durations) if durations else None, "observations": len(durations), "current_days": current_days}
|