""" Bilan final d'une réclamation. Déclenché en tâche de fond par /finalize-claim, une fois que les médias (photos + vidéos) ont été récupérés dans Medias Cameras par l'enrichissement Drive. Étapes : 1. Lire la ligne Reclamations et les lignes Medias Cameras associées via l'API AppSheet (Find). 2. Télécharger les images et vidéos depuis Drive dans un dossier temporaire. 3. Lancer l'inspection des pièces (Inspect_car_parts) sur les images. 4. Lancer la vérification du dommage (Check_claim_gemini.process_multi_claim : découpe la réclamation en dommages distincts si besoin, chacun est vérifié un par un sur les photos, puis en vidéo en complément si la zone concerne l'avant et qu'aucune photo ne confirme le dommage). 5. Créer Bilan Final/ sous le dossier de la plaque, y copier NATIVEMENT le template Google Sheet (jamais d'export xlsx -> zéro perte de mise en forme) et le remplir avec gspread (même service account que Drive, aucun compte Google personnel requis). 6. Créer Preuves/ (si verdict OUI) et Parties Visibles/ (toujours), y uploader les images/vidéos concernées. 7. Écrire le verdict et les liens dans Reclamations. ⚠️ Hypothèses à confirmer avec vous : - Enum "SR Responsable" (anciennement "Resultat Valide") : Oui / Non / Non_Concluant (confirmé). Le pipeline n'écrit jamais "Oui" — seule la supervision le fait à la main après vérification des caméras. "Non" = dommage confirmé déjà présent à l'entrée (réclamation rejetée). "Non_Concluant" = le pipeline n'a pas pu trancher (à vérifier manuellement). - le nom exact de l'onglet du template (SHEET_TITLE) ⚠️ Sécurisation du retraitement par mail : si le mail client ne précise pas explicitement sa demande, "Demande Reclamation" reçoit une valeur par défaut (extract_claim_from_email.DEFAULT_CLAIM_DEMAND). Si aucune pièce endommagée n'est identifiable dans le mail, "Localisation Carrosserie" reçoit "Non Spécifié" et Check_claim_gemini.process_claim() élargit la vérification du dommage à TOUTES les photos d'entrée disponibles (au lieu d'abandonner). ⚠️ NOTE ARCHITECTURE (2026-07-08) : les opérations qui CRÉENT un fichier (dossier Bilan Final, copie du template, uploads) utilisent le compte Google PERSONNEL via oauth_drive.py, pas le compte de service — un compte de service n'a aucun quota de stockage propre et se fait rejeter (403 storageQuotaExceeded) dès qu'il essaie de créer un fichier hors d'un Drive partagé. Les opérations de lecture seule (téléchargement des médias, recherche du dossier plaque) continuent d'utiliser le compte de service via drive_enrichment.get_drive_service(), qui fonctionne très bien pour ça. Si ce projet passe un jour sur Google Workspace, voir la note détaillée dans oauth_drive.py pour l'alternative "Drive partagé". """ import logging import os import shutil import tempfile from datetime import datetime from pathlib import Path import gspread from googleapiclient.http import MediaFileUpload, MediaIoBaseDownload from appsheet_client import edit_reclamation, edit_media, find_rows, lire_url, TABLE_RECLAMATIONS, TABLE_MEDIAS from drive_enrichment import get_drive_service, trouver_dossier_plaque from oauth_drive import get_user_drive_service, get_user_credentials from Inspect_car_parts import run_parts_inspection from Check_claim_gemini import process_multi_claim, build_media_role_resolver logging.basicConfig(level=logging.INFO) logger = logging.getLogger("finalize_claim") # ID du Google Sheet template (dans son URL : # https://docs.google.com/spreadsheets/d//edit). # Le service account doit avoir accès à ce fichier (partagez-le avec son # client_email, au moins en lecteur — il fera lui-même une COPIE, jamais # une modification du template original). TEMPLATE_SPREADSHEET_ID = os.environ.get("TEMPLATE_SPREADSHEET_ID") SHEET_TITLE = "CC Report Template EN" # nom exact de l'onglet dans le template # Mêmes cellules que Fill_report_gsheet.py (même mise en page, juste un # moteur d'écriture différent : openpyxl au lieu de gspread) CELL_CUSTOMER_NAME, CELL_COMPLAINT_DATE = "C5", "E5" CELL_LICENSE_PLATE = "C6" CELL_ENTRY_DATE, CELL_SITE_PARKING = "C7", "E7" CELL_EXIT_DATE, CELL_CLAIM_INTERNAL_ID = "C8", "E8" CELL_CLAIM_MESSAGE, CELL_CLAIM_DEMAND = "B11", "B17" CELL_CONCLUSION = "B28" CELL_REPORT_DATE, CELL_REPORT_AUTHOR = "C39", "E39" # Enum "SR Responsable" (anciennement "Resultat Valide") côté AppSheet : # Oui / Non / Non_Concluant. # Le pipeline automatique n'écrit JAMAIS "Oui" : seule la supervision peut # valider une réclamation à la main, après vérification des caméras. # - process_claim() renvoie "OUI" quand le dommage est CONFIRMÉ déjà # présent sur une photo d'entrée -> réclamation rejetée -> "Non". # - "NON", ou verdict indéterminé (aucune photo d'entrée disponible, # erreur de lecture du VLM) -> impossible de conclure automatiquement # -> "Non_Concluant", à vérifier manuellement. Une zone non spécifiée # ou non reconnue n'entraîne PLUS de verdict indéterminé : la # vérification est élargie à toutes les photos d'entrée disponibles # (voir Check_claim_gemini.process_claim). VERDICT_TO_RESULTAT = { "OUI": "Non", "NON": "Non_Concluant", } RESULTAT_PAR_DEFAUT = "Non_Concluant" def _copier_template(service, nom: str, parent_id: str) -> dict: """Copie NATIVEMENT le Google Sheet template (fileId -> fileId), sans jamais l'exporter/télécharger en xlsx : aucune conversion, donc aucune perte de mise en forme (checkboxes, cellules fusionnées, styles).""" if not TEMPLATE_SPREADSHEET_ID: raise EnvironmentError("TEMPLATE_SPREADSHEET_ID non défini.") return service.files().copy( fileId=TEMPLATE_SPREADSHEET_ID, body={"name": nom, "parents": [parent_id]}, fields="id, webViewLink", ).execute() def _remplir_rapport(spreadsheet_id: str, reclamation: dict, claim_message: str, claim_result: dict): """Remplit la copie du template avec gspread, authentifié avec le compte Google personnel (voir oauth_drive.py) — nécessaire car le fichier copié appartient à ce compte, pas au compte de service.""" gc = gspread.authorize(get_user_credentials()) ws = gc.open_by_key(spreadsheet_id).worksheet(SHEET_TITLE) text_values = { CELL_CUSTOMER_NAME: reclamation.get("Client Nom Prenom", ""), CELL_COMPLAINT_DATE: reclamation.get("Complaint Date", ""), CELL_LICENSE_PLATE: reclamation.get("Vehicule Plaque", ""), CELL_ENTRY_DATE: reclamation.get("Stay Entry Date", ""), CELL_SITE_PARKING: reclamation.get("Site Parking", ""), CELL_EXIT_DATE: reclamation.get("Stay Exit Date", ""), CELL_CLAIM_INTERNAL_ID: reclamation.get("Claim Internal ID", ""), CELL_CLAIM_MESSAGE: claim_message, CELL_CLAIM_DEMAND: reclamation.get("Demande Reclamation", ""), CELL_CONCLUSION: claim_result.get("message", ""), CELL_REPORT_AUTHOR: "Pipeline Automatique", CELL_REPORT_DATE: datetime.now().strftime("%d-%b-%Y"), } batch = [{"range": ref, "values": [[val]]} for ref, val in text_values.items()] ws.batch_update(batch, value_input_option="USER_ENTERED") def _telecharger_vers(service, file_id: str, dest_path: Path): request = service.files().get_media(fileId=file_id) with open(dest_path, "wb") as f: downloader = MediaIoBaseDownload(f, request) done = False while not done: _, done = downloader.next_chunk() def _creer_dossier(service, nom: str, parent_id: str) -> str: metadata = { "name": nom, "mimeType": "application/vnd.google-apps.folder", "parents": [parent_id], } return service.files().create(body=metadata, fields="id").execute()["id"] def _uploader(service, chemin: Path, parent_id: str) -> dict: media = MediaFileUpload(str(chemin), resumable=False) return service.files().create( body={"name": chemin.name, "parents": [parent_id]}, media_body=media, fields="id, webViewLink", ).execute() def _set_statut(id_reclamation: str, statut: str): """Met à jour la colonne Statut sans jamais lever d'exception (appelée depuis un except, elle ne doit pas masquer l'erreur d'origine).""" try: edit_reclamation(id_reclamation, {"Statut WorkFlow": statut}) except Exception: logger.exception( "Impossible de mettre à jour le Statut de %s en %r", id_reclamation, statut ) def executer_bilan_final(id_reclamation: str): """Orchestration complète. Ne relève jamais d'exception : tourne en tâche de fond, personne n'attend sa réponse HTTP.""" # Garde-fou anti double-traitement : si le bilan a déjà été généré pour # cette réclamation (ex: webhook rejoué, relance manuelle), on ne # relance pas tout le pipeline (upload Drive en double, nouvel appel # coûteux au VLM, etc.). reclamations = find_rows( TABLE_RECLAMATIONS, f'Filter(Reclamations, [ID Reclamation] = "{id_reclamation}")', ) if reclamations and lire_url(reclamations[0].get("Fichier Excel Rempli")): logger.info( "Bilan déjà généré pour %s (Fichier Excel Rempli déjà renseigné) — traitement ignoré.", id_reclamation, ) return _set_statut(id_reclamation, "Traitement de la réclamation") tmp = Path(tempfile.mkdtemp(prefix=f"bilan_{id_reclamation}_")) try: logger.info("Bilan final démarré pour %s", id_reclamation) # 1. Lecture des données AppSheet (déjà lue ci-dessus) -------------- if not reclamations: logger.warning("Réclamation %s introuvable dans AppSheet", id_reclamation) _set_statut(id_reclamation, "Erreur lors du traitement") return reclamation = reclamations[0] # Le Ref de Medias Cameras vers Reclamations stocke la vraie clé de # Reclamations, à savoir "ID Reclamation" (ex: "Ticket 1234587") — # PAS "Row ID". medias = find_rows( TABLE_MEDIAS, f'Filter("Medias Cameras", [ID Reclamation] = "{id_reclamation}")', ) # 2. Téléchargement des médias depuis Drive (lecture -> compte de service) -- service = get_drive_service() user_service = get_user_drive_service() # créations -> compte personnel (quota réel) images_dir, videos_dir = tmp / "images", tmp / "videos" images_dir.mkdir() videos_dir.mkdir() for m in medias: file_id = m.get("ID Media") if not file_id: continue is_image = m.get("Type Media") == "Image" dest_dir = images_dir if is_image else videos_dir ext = ".png" if is_image else ".mp4" _telecharger_vers(service, file_id, dest_dir / f"{file_id}{ext}") # 3. Inspection des pièces (images uniquement) ----------------------- parts_json = tmp / "vehicle_parts_detection.json" run_parts_inspection(images_dir, parts_json) # 4. Vérification du dommage (multi-dégâts, un par un, + fallback # vidéo pour les dommages concernant l'avant) --------------------- claim_message = reclamation.get("Dommage Detaille", "") claim_result = process_multi_claim(claim_message, str(parts_json), videos_dir=videos_dir) verdict = claim_result.get("verdict") # "OUI" / "NON" / None (cas d'erreur) # 4bis. Rôle de chaque média (colonne "Role Media" de Medias Cameras) : # "Dommages" si le dommage réclamé y est confirmé, "Parties Visibles" # si le média montre la pièce réclamée sans dommage confirmé, # "Non Utile" sinon (cf. Check_claim_gemini.build_media_role_resolver). resolver_role_media = build_media_role_resolver(claim_result) for m in medias: file_id = m.get("ID Media") if not file_id: continue role = resolver_role_media(file_id, m.get("Type Media") == "Image") try: edit_media(m.get("Row ID"), m.get("ID Media"), {"Role Media": role}) except Exception: logger.exception( "Échec de la mise à jour de 'Role Media' pour le média %s (réclamation %s)", file_id, id_reclamation, ) # 5. Dossiers Drive sous BDD Claim/Plaque//Bilan Final/ ------- plaque = reclamation.get("Vehicule Plaque", "") dossier_plaque = trouver_dossier_plaque(service, plaque) if not dossier_plaque: logger.warning("Dossier Drive introuvable pour la plaque %s", plaque) _set_statut(id_reclamation, "Erreur lors du traitement") return bilan_folder_id = _creer_dossier(user_service, "Bilan Final", dossier_plaque["id"]) # Copie native du template (jamais d'export xlsx) + remplissage rapport = _copier_template(user_service, f"Rapport_{id_reclamation}", bilan_folder_id) _remplir_rapport(rapport["id"], reclamation, claim_message, claim_result) # "Preuves" : créé dans tous les cas (même vide, pour que le lien # soit toujours disponible) ; rempli uniquement si le dommage est # confirmé (OUI) — images annotées + toutes les vidéos disponibles # (non analysées automatiquement, à visionner manuellement en # complément). preuves_folder_id = _creer_dossier(user_service, "Preuves", bilan_folder_id) if verdict == "OUI": for img_path in claim_result.get("annotated_images", []): if Path(img_path).exists(): _uploader(user_service, Path(img_path), preuves_folder_id) for video_path in videos_dir.iterdir(): _uploader(user_service, video_path, preuves_folder_id) lien_preuves = f"https://drive.google.com/drive/folders/{preuves_folder_id}" # "Parties Visibles" : dans tous les cas, toutes les photos qui # montrent la pièce concernée par la réclamation (ex: toutes les # photos où la portière droite est visible, si c'est elle qui est # signalée) — que le dommage y soit confirmé ou non. Si la zone n'a # pas pu être déterminée (mapped_part=None), `candidate_images` # couvre déjà TOUTES les photos d'entrée (cf. # Check_claim_gemini.process_claim) ; on ajoute aussi toutes les # vidéos, qui ne peuvent pas non plus être filtrées par pièce, pour # une vérification manuelle complète. parties_visibles_folder_id = _creer_dossier(user_service, "Parties Visibles", bilan_folder_id) for img_path in claim_result.get("candidate_images", []): if Path(img_path).exists(): _uploader(user_service, Path(img_path), parties_visibles_folder_id) if claim_result.get("mapped_part") is None: for video_path in videos_dir.iterdir(): _uploader(user_service, video_path, parties_visibles_folder_id) lien_parties_visibles = f"https://drive.google.com/drive/folders/{parties_visibles_folder_id}" # 6. Écriture des résultats dans Reclamations ------------------------ # Note : "Statut" ne passe PAS à "Fini !" ici — ce statut final est # positionné par le Bot 4 (Slack) lui-même, dans AppSheet, en dernière # étape de son process (après l'envoi de la notification), pas par ce # pipeline. Ici on reste sur "Traitement de la réclamation" en cas de # succès. champs = { "Lien Dossier Preuves": lien_preuves, "Lien Dossier Parties Visibles": lien_parties_visibles, "Fichier Excel Rempli": rapport.get("webViewLink", ""), } # "SR Responsable" est TOUJOURS renseigné (jamais laissé vide) — # c'est ce qui permet au Bot 4 (Slack) de se déclencher dans tous les # cas, y compris quand le pipeline n'a pas pu trancher automatiquement. resultat = VERDICT_TO_RESULTAT.get(verdict, RESULTAT_PAR_DEFAUT) champs["SR Responsable"] = resultat if resultat == RESULTAT_PAR_DEFAUT: logger.info( "Verdict non tranché pour %s (status=%s) — 'SR Responsable' = %r, " "à vérifier manuellement (caméras).", id_reclamation, claim_result.get("status"), RESULTAT_PAR_DEFAUT, ) edit_reclamation(id_reclamation, champs) logger.info("Bilan final terminé pour %s : verdict=%s", id_reclamation, verdict) except Exception: logger.exception("Échec du bilan final pour %s", id_reclamation) _set_statut(id_reclamation, "Erreur lors du traitement") finally: shutil.rmtree(tmp, ignore_errors=True)