alban/gmao
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions 1

fix(c4): complete contract lifecycle and meter roles

Browse source
This commit is contained in:
root 2026-08-24 18:17:17 +00:00
parent 099db73d92
commit f0901f5061
4 changed files with 160 additions and 34 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

6
app_new/core/models/c4.py
View file

@ -85,10 +85,14 @@ class ConsumableOrder(db.Model):
received_at = db.Column(db.DateTime, nullable=True)
comment = db.Column(db.Text, nullable=True)
created_by_id = db.Column(db.Integer, db.ForeignKey("users.id", ondelete="SET NULL"), nullable=True)
cancelled_at = db.Column(db.DateTime, nullable=True)
cancelled_by_id = db.Column(db.Integer, db.ForeignKey("users.id", ondelete="SET NULL"), nullable=True)
cancellation_reason = db.Column(db.Text, nullable=True)
consumable = db.relationship("Consumable")
supplier = db.relationship("Company")
created_by = db.relationship("User")
created_by = db.relationship("User", foreign_keys=[created_by_id])
cancelled_by = db.relationship("User", foreign_keys=[cancelled_by_id])
receipts = db.relationship("ConsumableReceipt", back_populates="order", cascade="all, delete-orphan")
@property

4
app_new/core/models/planning.py
View file

@ -274,6 +274,10 @@ class Meter(db.Model):
meter_type = db.Column(db.String(50), default="hours")
unit = db.Column(db.String(20), default="h")
usage = db.Column(db.String(255), nullable=True)
# Rôle explicite uniquement pour les calculs de contrats photocopieurs.
# NONE conserve les compteurs historiques/non contractuels sans les
# classer silencieusement en N&B.
contract_role = db.Column(db.String(20), nullable=False, default="NONE", server_default="NONE")
status = db.Column(db.String(30), nullable=False, default="active", server_default="active")
remainder_label = db.Column(db.String(255), nullable=True)

146
app_new/core/services/c4_service.py
View file

@ -105,49 +105,52 @@ def replace_contract_equipment(*, old_assignment, new_equipment, replacement_dat
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"
return {"COPIER_BW": "bw", "COPIER_COLOR": "color"}.get(meter.contract_role)
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)
return _meter_consumption_from_rows(readings, start_date, end_date, meter=meter)
def _meter_consumption_from_rows(rows, start_date, end_date):
def _meter_consumption_from_rows(rows, start_date, end_date, meter=None):
"""Même calcul que ``_meter_consumption`` à partir d'un lot déjà chargé."""
start_dt = datetime.combine(start_date, datetime.min.time())
end_dt = datetime.combine(end_date, datetime.max.time())
relevant = [row for row in rows if row.reading_date <= end_dt]
before = next((row for row in reversed(relevant) if row.reading_date < start_dt), None)
inside = [row for row in relevant if start_dt <= row.reading_date <= end_dt]
ordered = ([before] if before else []) + inside
exact_start = bool(inside and inside[0].reading_date.date() == start_date)
ordered = inside
total = 0.0
previous = None
for reading in ordered:
if previous and not reading.is_reset and not previous.is_reset and reading.value >= previous.value:
total += reading.value - previous.value
previous = reading
quality = "EXACT" if before else ("ESTIMATED" if len(ordered) >= 2 else "PARTIAL")
if exact_start:
quality = "EXACT"
else:
first = inside[0] if inside else None
prior = [row for row in rows if row.reading_date < (first.reading_date if first else start_dt)]
prior.sort(key=lambda row: (row.reading_date, row.id))
prior_rates = []
for left, right in zip(prior, prior[1:]):
if not left.is_reset and not right.is_reset and right.value >= left.value:
days = max((right.reading_date.date() - left.reading_date.date()).days, 1)
prior_rates.append((right.value - left.value) / days)
if first and prior_rates:
missing_days = max((first.reading_date.date() - start_date).days, 0)
if meter is not None and meter.equipment is not None and getattr(meter.equipment, "device_type", "") in {"photocopier", "imprimante", "printer"}:
comparable_days = _school_days(start_date, first.reading_date.date() - timedelta(days=1))
missing_days = comparable_days or missing_days
total += mean(prior_rates) * missing_days
quality = "ESTIMATED"
else:
quality = "PARTIAL"
return total, quality, len(ordered)
@ -174,10 +177,14 @@ def calculate_contract_usage(period):
start = max(period.start_date, assignment.entry_date)
end = min(period.end_date, assignment.exit_date or period.end_date)
for meter in meter_by_equipment.get(assignment.equipment_id, []):
amount, quality, _ = _meter_consumption_from_rows(readings_by_meter.get(meter.id, []), start, end)
result[_meter_role(meter)] += amount
if quality != "EXACT":
result["quality"] = "PARTIAL" if result["quality"] == "EXACT" else result["quality"]
role = _meter_role(meter)
if role is None:
continue
amount, quality, _ = _meter_consumption_from_rows(readings_by_meter.get(meter.id, []), start, end, meter=meter)
result[role] += amount
rank = {"EXACT": 0, "ESTIMATED": 1, "PARTIAL": 2}
if rank[quality] > rank[result["quality"]]:
result["quality"] = quality
return result
@ -218,10 +225,14 @@ def evaluate_quota_alert(period, *, level="WARNING", commit=True):
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()
if quota is None or projected is None or projected <= quota:
if alert:
alert.is_resolved = True
alert.resolved_at = datetime.now(timezone.utc)
alert.message = "Projection redevenue compatible avec le quota."
continue
message = f"Projection {role} : {projected:.0f} copies pour un quota de {quota}."
if alert:
alert.message = message
@ -236,6 +247,28 @@ def evaluate_quota_alert(period, *, level="WARNING", commit=True):
return projection
def close_photocopier_contract(*, contract, closed_at=None, commit=True):
closed_at = closed_at or date.today()
if closed_at < contract.start_date:
raise C4DomainError("La date de clôture ne peut pas précéder le début du contrat.")
closed_at = min(closed_at, contract.end_date)
contract.status = "CLOSED"
contract.closed_at = closed_at
contract.end_date = closed_at
for period in contract.periods:
if period.start_date > closed_at:
period.status = "CANCELLED"
elif period.start_date <= closed_at <= period.end_date:
period.end_date = closed_at
period.status = "CLOSED"
for assignment in contract.assignments:
if assignment.entry_date <= closed_at and (assignment.exit_date is None or assignment.exit_date > closed_at):
assignment.exit_date = closed_at
if commit:
db.session.commit()
return contract
def create_consumable_order(*, consumable, quantity, supplier_id=None, user_id=None, commit=True):
quantity = int(quantity)
if quantity <= 0:
@ -253,18 +286,33 @@ def receive_consumable_order(*, order, quantity, user_id=None, comment=None, com
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)
db.session.add(receipt)
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 cancel_consumable_order(*, order, user_id=None, reason=None, commit=True):
if order.status == "RECEIVED":
raise C4DomainError("Une commande entièrement reçue ne peut pas être annulée.")
if order.status not in {"ORDERED", "PARTIALLY_RECEIVED"}:
raise C4DomainError("Cette commande ne peut pas être annulée dans son état actuel.")
order.status = "CANCELLED"
order.cancelled_at = datetime.now(timezone.utc)
order.cancelled_by_id = user_id
order.cancellation_reason = reason
refresh_consumable_stock_alert(order.consumable, commit=False)
if commit:
db.session.commit()
return order
def record_consumable_issue(*, consumable, equipment, quantity=1, user_id=None, intervention_id=None, notes=None, commit=True):
quantity = int(quantity)
if quantity <= 0:
@ -284,15 +332,16 @@ def record_consumable_issue(*, consumable, equipment, quantity=1, user_id=None,
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(
pending_orders = ConsumableOrder.query.filter(
ConsumableOrder.consumable_id == consumable.id,
ConsumableOrder.status.in_(("ORDERED", "PARTIALLY_RECEIVED")),
).first()
).all()
pending_quantity = sum(order.remaining_quantity for order in pending_orders)
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 pending_quantity:
message += f" Commande en cours — {pending_quantity} à recevoir."
if alert:
alert.message = message
else:
@ -305,6 +354,37 @@ def refresh_consumable_stock_alert(consumable, *, commit=True):
return alert
def estimate_consumable_stock_coverage(*, consumable):
links = EquipmentConsumable.query.filter_by(consumable_id=consumable.id).all()
active = [link.equipment for link in links if link.equipment and not link.equipment.is_deleted and
(link.equipment.lifecycle_status or "").lower() not in {"retired", "retire", "removed", "hors_service"}]
rates = []
for equipment in active:
lifetime = estimate_consumable_lifetime(consumable=consumable, equipment=equipment)
if lifetime["average_days"] and lifetime["average_days"] > 0:
rates.append(1.0 / lifetime["average_days"])
total_rate = sum(rates)
if not active or not rates:
quality = "NON_DISPONIBLE"
coverage = None
else:
quality = "ESTIMATED" if len(rates) == len(active) else "PARTIAL"
coverage = (consumable.quantity or 0) / total_rate if total_rate else None
return {"coverage_days": coverage, "quality": quality, "machines_total": len(active),
"machines_with_history": len(rates), "daily_usage_estimate": total_rate}
def visible_open_c4_alerts(user):
"""Alertes C4 informatives autorisées, sans les transformer en tâches."""
from ..authorization import has_permission
rows = []
if has_permission("contract.view", user):
rows.extend(Alert.query.filter_by(alert_type="QUOTA_OVERAGE_PROJECTED", is_resolved=False).all())
if has_permission("stock.view", user):
rows.extend(Alert.query.filter_by(alert_type="CONSUMABLE_STOCK_LOW", is_resolved=False).all())
return sorted(rows, key=lambda row: row.created_at or datetime.min.replace(tzinfo=timezone.utc), reverse=True)
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())

38
migrations/versions/t5h6i7j8k9l0_c4_completion.py Normal file
View file

@ -0,0 +1,38 @@
"""C4 completion: explicit copier meter roles and order cancellation audit.
Revision ID: t5h6i7j8k9l0
Revises: s4g5h6i7j8k9
"""
from alembic import op
import sqlalchemy as sa
revision = "t5h6i7j8k9l0"
down_revision = "s4g5h6i7j8k9"
branch_labels = None
depends_on = None
def upgrade():
op.add_column("meters", sa.Column("contract_role", sa.String(length=20), nullable=False, server_default="NONE"))
op.create_check_constraint(
"ck_meter_contract_role",
"meters",
"contract_role in ('COPIER_BW', 'COPIER_COLOR', 'NONE')",
)
op.add_column("consumable_orders", sa.Column("cancelled_at", sa.DateTime(), nullable=True))
op.add_column("consumable_orders", sa.Column("cancelled_by_id", sa.Integer(), nullable=True))
op.add_column("consumable_orders", sa.Column("cancellation_reason", sa.Text(), nullable=True))
op.create_foreign_key(
"fk_consumable_orders_cancelled_by",
"consumable_orders", "users", ["cancelled_by_id"], ["id"], ondelete="SET NULL",
)
def downgrade():
op.drop_constraint("fk_consumable_orders_cancelled_by", "consumable_orders", type_="foreignkey")
op.drop_column("consumable_orders", "cancellation_reason")
op.drop_column("consumable_orders", "cancelled_by_id")
op.drop_column("consumable_orders", "cancelled_at")
op.drop_constraint("ck_meter_contract_role", "meters", type_="check")
op.drop_column("meters", "contract_role")