monitoring/feedback.py

"""Boucle de feedback : collecte des corrections pompiste et conversion en
données d'entraînement (voir docs/MONITORING.md).

Principe : chaque photo traitée par `/analyze` est déjà sauvegardée dans
`photos/<transaction_id>.jpg` (voir `app/main.py`). Une correction envoyée
via `POST /feedback` référence cette photo par son `photo_reference` et
fournit les valeurs réellement correctes. `convert_feedback_to_annotations`
transforme ensuite ces corrections en entrées au même format que
`annotator/annotations/annotations.json`, réutilisables par le pipeline
d'entraînement existant (`split_lines.py` -> `prepare_doctr_dataset.py` ->
`finetune_doctr.py`).
"""
import json
import os
import uuid
from datetime import datetime, timezone
from pathlib import Path
from typing import Any, Dict, List, Optional

from app.preprocessing import detect_screen_region

MODULE_ROOT = Path(__file__).resolve().parent.parent

# Même variable YELY_DATA_DIR que `app/main.py`/`monitoring/metrics.py` : si
# un stockage persistant est monté (ex. /data sur un Hugging Face Space),
# feedback.jsonl et les crops doivent survivre aux redémarrages du Space
# comme les photos et les logs, sinon la boucle de feedback perd tout à
# chaque redéploiement. `DEFAULT_ANNOTATIONS_PATH` reste dans le dépôt
# (`annotator/`) volontairement : la conversion en données d'entraînement
# est un script lancé en local par un développeur, pas sur le Space (voir
# docs/MONITORING.md — l'annotateur n'est de toute façon pas poussé sur HF).
DATA_DIR = Path(os.environ.get("YELY_DATA_DIR", str(MODULE_ROOT)))
DEFAULT_FEEDBACK_PATH = DATA_DIR / "monitoring" / "feedback.jsonl"
DEFAULT_PHOTOS_DIR = DATA_DIR / "photos"
DEFAULT_ANNOTATIONS_PATH = MODULE_ROOT.parent / "annotator" / "annotations" / "annotations.json"
DEFAULT_CROPS_DIR = DATA_DIR / "monitoring" / "feedback_crops"

CORRECTABLE_FIELDS = ("prix", "volume", "prix_litre")


def record_feedback(photo_reference: str,
                    corrected_fields: Dict[str, Any],
                    corrected_by: Optional[str] = None,
                    feedback_path: Optional[Path] = None) -> Dict[str, Any]:
    """Ajoute une correction pompiste au journal `feedback.jsonl` (append-only).

    `corrected_fields` : sous-ensemble de {"prix", "volume", "prix_litre"} ->
    valeur correcte (les champs non fournis restent inconnus, pas déduits).
    """
    unknown = set(corrected_fields) - set(CORRECTABLE_FIELDS)
    if unknown:
        raise ValueError(f"Champs de correction inconnus : {sorted(unknown)}")
    if not corrected_fields:
        raise ValueError("Aucune correction fournie.")

    feedback_path = feedback_path or DEFAULT_FEEDBACK_PATH
    feedback_path.parent.mkdir(parents=True, exist_ok=True)

    entry = {
        "feedback_id": str(uuid.uuid4()),
        "photo_reference": photo_reference,
        "corrected_fields": corrected_fields,
        "corrected_by": corrected_by,
        "created_at": datetime.now(timezone.utc).isoformat(),
        "converted": False,
    }
    with open(feedback_path, "a", encoding="utf-8") as f:
        f.write(json.dumps(entry, ensure_ascii=False) + "\n")
    return entry


def load_feedback_entries(feedback_path: Optional[Path] = None) -> List[Dict[str, Any]]:
    feedback_path = feedback_path or DEFAULT_FEEDBACK_PATH
    if not feedback_path.exists():
        return []
    entries = []
    with open(feedback_path, "r", encoding="utf-8") as f:
        for line in f:
            line = line.strip()
            if not line:
                continue
            try:
                entries.append(json.loads(line))
            except json.JSONDecodeError:
                continue
    return entries


def convert_feedback_to_annotations(photos_dir: Optional[Path] = None,
                                    feedback_path: Optional[Path] = None,
                                    annotations_path: Optional[Path] = None,
                                    crops_dir: Optional[Path] = None) -> int:
    """Convertit les corrections non encore traitées en entrées `annotations.json`.

    La détection d'écran (`detect_screen_region`) est automatique, donc pas
    fiable à 100% (voir docs/LIMITATIONS.md, point 2) : les entrées générées
    sont marquées `status="pending_review"` plutôt que `"annotated"`, pour
    qu'une relecture humaine (même rapide, via le visualiseur de l'annotateur)
    précède leur utilisation dans un ré-entraînement.

    Retourne le nombre d'entrées converties.
    """
    import cv2

    photos_dir = photos_dir or DEFAULT_PHOTOS_DIR
    feedback_path = feedback_path or DEFAULT_FEEDBACK_PATH
    annotations_path = annotations_path or DEFAULT_ANNOTATIONS_PATH
    crops_dir = crops_dir or DEFAULT_CROPS_DIR

    entries = load_feedback_entries(feedback_path)
    pending = [e for e in entries if not e.get("converted")]
    if not pending:
        return 0

    annotations_path.parent.mkdir(parents=True, exist_ok=True)
    if annotations_path.exists():
        with open(annotations_path, "r", encoding="utf-8") as f:
            annotations = json.load(f)
    else:
        annotations = {}

    crops_dir.mkdir(parents=True, exist_ok=True)

    converted_count = 0
    for entry in pending:
        photo_path = photos_dir / entry["photo_reference"]
        if not photo_path.exists():
            continue

        img = cv2.imread(str(photo_path))
        if img is None:
            continue
        crop, bbox = detect_screen_region(img)
        crop_name = f"feedback_{entry['feedback_id']}_lcd.jpg"
        cv2.imwrite(str(crops_dir / crop_name), crop)

        fields = {name: "" for name in CORRECTABLE_FIELDS}
        fields.update({"unite_prix": "", "unite_vol": "", "notes": "issu du feedback pompiste"})
        for name, value in entry["corrected_fields"].items():
            fields[name] = str(value)

        annotations[entry["photo_reference"]] = {
            "status": "pending_review",
            "annotated_at": entry["created_at"],
            "image_path": str(photo_path),
            "lcd_bbox": list(bbox),
            "lcd_crop": str(crops_dir / crop_name),
            "fields": fields,
        }
        entry["converted"] = True
        converted_count += 1

    with open(annotations_path, "w", encoding="utf-8") as f:
        json.dump(annotations, f, ensure_ascii=False, indent=2)

    with open(feedback_path, "w", encoding="utf-8") as f:
        for entry in entries:
            f.write(json.dumps(entry, ensure_ascii=False) + "\n")

    return converted_count


if __name__ == "__main__":
    n = convert_feedback_to_annotations()
    print(f"{n} correction(s) converties en entrées à relire dans annotations.json")