gmao/app_new/documents/service.py

"""Agrégation en lecture des documents des modules métier.

Ce service ne possède aucun fichier et ne crée aucune ligne documentaire : il
ne fait que présenter des adaptateurs homogènes aux vues du centre global.
"""
from dataclasses import dataclass
from datetime import date, datetime
from typing import Optional

from flask import url_for

from ..core.authorization import has_permission
from ..core.models.college import Building, Room, Zone
from ..core.models.cleaning import ProductDocument
from ..core.models.equipment import EquipmentDocument
from ..core.models.maintenance import InterventionDocument


DOCUMENT_TYPE_LABELS = {
    "notice": "Notice / manuel",
    "facture": "Facture",
    "devis": "Devis",
    "rapport": "Rapport / compte rendu",
    "certificat": "Certificat / contrôle",
    "photo": "Photo",
    "plan": "Plan",
    "fds": "FDS",
    "fiche_technique": "Fiche technique",
    "autre": "Autre",
    "other": "Autre",
}


@dataclass
class DocumentResult:
    source: str
    source_id: int
    parent_id: int
    title: str
    filename: str
    document_type: str
    document_type_label: str
    description: Optional[str]
    document_date: Optional[date]
    context_label: str
    context_url: Optional[str]
    download_url: str
    filepath: str
    location_labels: tuple[str, ...]
    location_summary: str
    version: Optional[str] = None
    issuer: Optional[str] = None


def _location_path(room):
    if not room:
        return None
    parts = []
    if room.building:
        parts.append(room.building.name)
    if room.zone:
        parts.append(room.zone.name)
    parts.append(room.name)
    return " > ".join(parts)


def _equipment_location(equipment):
    if not equipment:
        return ()
    room = equipment.effective_room or equipment.room
    path = _location_path(room)
    return (path,) if path else ()


def _intervention_location(intervention):
    return _equipment_location(intervention.equipment) or (
        (_location_path(intervention.room),) if _location_path(intervention.room) else ()
    )


def _product_locations(product):
    paths = set()
    for lot in product.lots:
        for balance in lot.balances:
            if not balance.quantity or balance.quantity <= 0 or not balance.location:
                continue
            location = balance.location
            parts = []
            building = Building.query.get(location.building_id) if location.building_id else None
            zone = Zone.query.get(location.zone_id) if location.zone_id else None
            room = Room.query.get(location.room_id) if location.room_id else None
            if building:
                parts.append(building.name)
            if zone:
                parts.append(zone.name)
            if room:
                parts.append(room.name)
            parts.append(location.name)
            paths.add(" > ".join(parts))
    return tuple(sorted(paths))


def _summary(locations):
    if not locations:
        return "Localisation non renseignée"
    if len(locations) == 1:
        return locations[0]
    return f"{len(locations)} emplacements"


def _result(source, document, parent, *, title, document_type, description,
            document_date, context_label, context_url, locations, version=None,
            issuer=None):
    return DocumentResult(
        source=source,
        source_id=document.id,
        parent_id=parent.id,
        title=title or document.filename,
        filename=document.filename,
        document_type=document_type or "autre",
        document_type_label=DOCUMENT_TYPE_LABELS.get(document_type or "autre", "Autre"),
        description=description,
        document_date=document_date,
        context_label=context_label,
        context_url=context_url,
        download_url=url_for("documents.download", source=source, document_id=document.id),
        filepath=document.filepath,
        location_labels=tuple(locations),
        location_summary=_summary(locations),
        version=version,
        issuer=issuer,
    )


def list_documents(*, query="", source=None, document_type=None,
                   date_from=None, date_to=None, user=None, fds_only=False):
    """Retourne les documents accessibles, filtrés sans stockage parallèle."""
    user = user
    results = []
    if has_permission("documents.view", user) and has_permission("intervention.view", user):
        for document in InterventionDocument.query.order_by(InterventionDocument.uploaded_at.desc()).all():
            parent = document.intervention
            if not parent:
                continue
            item = _result(
                "intervention", document, parent, title=document.filename,
                document_type=getattr(document, "document_type", "autre"),
                description=document.description, document_date=document.uploaded_at,
                context_label=f"Intervention #{parent.id} — {parent.title}",
                context_url=url_for("interventions.detail", id=parent.id),
                locations=_intervention_location(parent),
            )
            results.append(item)
    if has_permission("documents.view", user) and has_permission("patrimoine.view", user):
        for document in EquipmentDocument.query.order_by(EquipmentDocument.uploaded_at.desc()).all():
            parent = document.equipment
            if not parent:
                continue
            results.append(_result(
                "equipment", document, parent, title=document.filename,
                document_type=getattr(document, "document_type", "autre"),
                description=document.description, document_date=document.uploaded_at,
                context_label=f"Équipement — {parent.name}",
                context_url=url_for("equipments.detail", id=parent.id),
                locations=_equipment_location(parent),
            ))
    if has_permission("documents.view", user) and has_permission("stock.view", user):
        for document in ProductDocument.query.order_by(ProductDocument.uploaded_at.desc()).all():
            parent = document.commercial_product
            if not parent:
                continue
            generic = parent.generic_product
            label = parent.commercial_name
            results.append(_result(
                "product", document, parent, title=document.title or document.filename,
                document_type=document.document_type, description=None,
                document_date=document.published_at or document.uploaded_at,
                context_label=f"Produit — {label}",
                context_url=url_for("cleaning.references", highlight=parent.id),
                locations=_product_locations(parent),
                version=document.version,
                issuer=(parent.manufacturer.name if parent.manufacturer else None)
                or (parent.supplier.name if parent.supplier else None),
            ))
    needle = (query or "").strip().casefold()
    if needle:
        results = [item for item in results if needle in " ".join((
            item.title, item.filename, item.description or "", item.context_label,
            item.location_summary,
        )).casefold()]
    if source:
        results = [item for item in results if item.source == source]
    if document_type:
        results = [item for item in results if item.document_type == document_type]
    if fds_only:
        results = [item for item in results if item.document_type == "fds"]
    def document_day(item):
        if not item.document_date:
            return None
        return item.document_date.date() if isinstance(item.document_date, datetime) else item.document_date
    if date_from:
        results = [item for item in results if document_day(item) and document_day(item) >= date_from]
    if date_to:
        results = [item for item in results if document_day(item) and document_day(item) <= date_to]
    # Les sources historiques mélangent ``date`` (produits) et ``datetime``
    # (interventions/équipements).  Normaliser avant le tri évite une
    # TypeError Python et garde un ordre stable entre les trois adaptateurs.
    results.sort(
        key=lambda item: datetime.combine(document_day(item), datetime.min.time())
        if document_day(item) else datetime.min,
        reverse=True,
    )
    return results


def get_document(source, document_id):
    models = {
        "intervention": InterventionDocument,
        "equipment": EquipmentDocument,
        "product": ProductDocument,
    }
    model = models.get(source)
    return model.query.get_or_404(document_id) if model else None


def can_read_source(source, user=None):
    return has_permission("documents.view", user) and has_permission({
        "intervention": "intervention.view",
        "equipment": "patrimoine.view",
        "product": "stock.view",
    }.get(source, "system.admin"), user)