9e7205eca2
- CSVBuilder.build_csv() e build_csv_content() accettano image_url_map opzionale - _resolve_image() risolve keyword->URL Unsplash con fallback keyword originale - _build_rows() chiama _resolve_image per cover, slides e cta image keywords - JobStatus ha campo image_url_map con persistenza su disco JSON - GenerationPipeline._resolve_unsplash_keywords() chiamato dopo batch LLM - Carica unsplash_api_key da settings.json, crea UnsplashService, chiama resolve_keywords - image_url_map salvato nel job JSON per riuso in export con edits - Export router recupera image_url_map dal job JSON e passa a build_csv_content - generate_single NON risolve Unsplash (velocità e riuso map job originale)
215 lines
7.9 KiB
Python
215 lines
7.9 KiB
Python
"""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
|