"""CSVBuilder — produce CSV compatibile con Canva Bulk Create. Caratteristiche: - Encoding utf-8-sig (BOM) — critico per Excel + caratteri italiani (Pitfall 3) - Header CANVA_FIELDS locked — 33 colonne esatte - Mappa GeneratedPost + CalendarSlot -> riga CSV - Filtra solo PostResult con status="success" - Scrive su disco in OUTPUTS_PATH/{job_id}.csv - Supporta image_url_map opzionale: risolve keyword -> URL Unsplash nelle colonne _image_keyword """ from __future__ import annotations import csv import io import logging from pathlib import Path from typing import TYPE_CHECKING, Optional from backend.constants import CANVA_FIELDS from backend.schemas.generate import PostResult if TYPE_CHECKING: from backend.schemas.calendar import CalendarResponse logger = logging.getLogger(__name__) class CSVBuilder: """Costruisce file CSV Canva-compatibili dai risultati di generazione.""" def build_csv( self, posts: list[PostResult], calendar: "CalendarResponse", job_id: str, output_dir: Path, image_url_map: Optional[dict[str, str]] = None, ) -> Path: """Genera e scrive il CSV su disco. Filtra solo i PostResult con status="success", mappa i dati GeneratedPost + CalendarSlot alle 33 colonne CANVA_FIELDS, e scrive con encoding utf-8-sig per compatibilità Excel. Se image_url_map è fornita, le colonne _image_keyword contengono URL Unsplash reali invece delle keyword testuali originali. Args: posts: Lista di PostResult (include success e failed). calendar: CalendarResponse con i metadati degli slot. job_id: Identificatore univoco del job (usato come nome file). output_dir: Directory dove scrivere il file CSV. image_url_map: Mappa opzionale {keyword: url_unsplash}. Se None, usa le keyword testuali originali. Returns: Path del file CSV scritto su disco. """ output_dir.mkdir(parents=True, exist_ok=True) output_path = output_dir / f"{job_id}.csv" rows = self._build_rows(posts, calendar, image_url_map) # CRITICO: encoding utf-8-sig (BOM) per Excel + caratteri italiani with open(output_path, "w", newline="", encoding="utf-8-sig") as f: writer = csv.DictWriter(f, fieldnames=CANVA_FIELDS, extrasaction="ignore") writer.writeheader() writer.writerows(rows) url_count = len(image_url_map) if image_url_map else 0 logger.info( "CSV scritto | job_id=%s | righe_success=%d | url_unsplash=%d | path=%s", job_id, len(rows), url_count, output_path, ) return output_path def build_csv_content( self, posts: list[PostResult], calendar: "CalendarResponse", job_id: str, image_url_map: Optional[dict[str, str]] = None, ) -> str: """Genera il CSV come stringa (senza scrivere su disco). Usato per preview e per la route POST /export/{job_id}/csv con dati modificati inline dall'utente. Se image_url_map è fornita, le colonne _image_keyword contengono URL Unsplash reali invece delle keyword testuali originali. Args: posts: Lista di PostResult (include success e failed). calendar: CalendarResponse con i metadati degli slot. job_id: Identificatore univoco del job. image_url_map: Mappa opzionale {keyword: url_unsplash}. Returns: Stringa CSV con encoding utf-8-sig (BOM). """ rows = self._build_rows(posts, calendar, image_url_map) output = io.StringIO() # Aggiungi BOM manualmente per compatibilità Excel output.write("\ufeff") writer = csv.DictWriter(output, fieldnames=CANVA_FIELDS, extrasaction="ignore") writer.writeheader() writer.writerows(rows) return output.getvalue() # --------------------------------------------------------------------------- # Metodi privati # --------------------------------------------------------------------------- def _resolve_image(self, keyword: str, image_url_map: Optional[dict[str, str]]) -> str: """Risolve una keyword immagine in URL Unsplash se disponibile. Args: keyword: Keyword immagine originale. image_url_map: Mappa {keyword: url} o None. Returns: URL Unsplash se disponibile nella mappa, altrimenti la keyword originale. """ if image_url_map and keyword in image_url_map: return image_url_map[keyword] return keyword def _build_rows( self, posts: list[PostResult], calendar: "CalendarResponse", image_url_map: Optional[dict[str, str]] = None, ) -> list[dict[str, str]]: """Costruisce la lista di righe CSV dai risultati. Filtra solo i post con status="success" e mappa i dati GeneratedPost + CalendarSlot alle colonne CANVA_FIELDS. Se image_url_map è fornita, le colonne _image_keyword vengono risolte in URL Unsplash quando disponibili. Args: posts: Lista completa di PostResult. calendar: CalendarResponse con i metadati degli slot. image_url_map: Mappa opzionale {keyword: url_unsplash}. Returns: Lista di dict con chiavi = CANVA_FIELDS. """ # Crea un dizionario slot_index -> CalendarSlot per lookup veloce slot_map = {slot.indice: slot for slot in calendar.slots} rows: list[dict[str, str]] = [] for post_result in posts: if post_result.status != "success" or post_result.post is None: continue slot_index = post_result.slot_index slot = slot_map.get(slot_index) if slot is None: logger.warning( "Slot non trovato per slot_index=%d, skip", slot_index ) continue post = post_result.post row: dict[str, str] = {} # --- Metadati slot (8 colonne) --- row["campagna"] = calendar.campagna row["fase_campagna"] = slot.fase_campagna row["tipo_contenuto"] = slot.tipo_contenuto row["formato_narrativo"] = slot.formato_narrativo row["funzione"] = slot.funzione row["livello_schwartz"] = slot.livello_schwartz row["target_nicchia"] = slot.target_nicchia row["data_pub_suggerita"] = slot.data_pub_suggerita # --- Cover slide (3 colonne) --- row["cover_title"] = post.cover_title row["cover_subtitle"] = post.cover_subtitle row["cover_image_keyword"] = self._resolve_image(post.cover_image_keyword, image_url_map) # --- Slide centrali s2-s7 (6 slide x 3 colonne = 18 colonne) --- slide_labels = ["s2", "s3", "s4", "s5", "s6", "s7"] for idx, label in enumerate(slide_labels): if idx < len(post.slides): slide = post.slides[idx] row[f"{label}_headline"] = slide.headline row[f"{label}_body"] = slide.body row[f"{label}_image_keyword"] = self._resolve_image(slide.image_keyword, image_url_map) else: # Fallback se slides ha meno di 6 elementi (non dovrebbe accadere) row[f"{label}_headline"] = "" row[f"{label}_body"] = "" row[f"{label}_image_keyword"] = "" # --- CTA slide (3 colonne) --- row["cta_text"] = post.cta_text row["cta_subtext"] = post.cta_subtext row["cta_image_keyword"] = self._resolve_image(post.cta_image_keyword, image_url_map) # --- Caption Instagram (1 colonna) --- row["caption_instagram"] = post.caption_instagram rows.append(row) return rows