083621afd3
- LLMService: retry 3x, RateLimitError legge retry-after header, 5xx backoff esponenziale, ValidationError riprova con istruzione correttiva, inter_request_delay 2s - LLMService.generate_topic(): usa TopicResult come response_schema (passa per loop retry/validation) - CSVBuilder: encoding utf-8-sig, header CANVA_FIELDS locked, mappa GeneratedPost+CalendarSlot -> 33 colonne - GenerationPipeline: background task asyncio.create_task, _jobs dict progresso real-time, per-item try/except individuale, persistenza JSON su disco
182 lines
6.2 KiB
Python
182 lines
6.2 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
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import csv
|
|
import io
|
|
import logging
|
|
from pathlib import Path
|
|
from typing import TYPE_CHECKING
|
|
|
|
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,
|
|
) -> 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.
|
|
|
|
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.
|
|
|
|
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)
|
|
|
|
# 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)
|
|
|
|
logger.info(
|
|
"CSV scritto | job_id=%s | righe_success=%d | path=%s",
|
|
job_id,
|
|
len(rows),
|
|
output_path,
|
|
)
|
|
return output_path
|
|
|
|
def build_csv_content(
|
|
self,
|
|
posts: list[PostResult],
|
|
calendar: "CalendarResponse",
|
|
job_id: str,
|
|
) -> 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.
|
|
|
|
Args:
|
|
posts: Lista di PostResult (include success e failed).
|
|
calendar: CalendarResponse con i metadati degli slot.
|
|
job_id: Identificatore univoco del job.
|
|
|
|
Returns:
|
|
Stringa CSV con encoding utf-8-sig (BOM).
|
|
"""
|
|
rows = self._build_rows(posts, calendar)
|
|
|
|
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 _build_rows(
|
|
self,
|
|
posts: list[PostResult],
|
|
calendar: "CalendarResponse",
|
|
) -> 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.
|
|
|
|
Args:
|
|
posts: Lista completa di PostResult.
|
|
calendar: CalendarResponse con i metadati degli slot.
|
|
|
|
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"] = post.cover_image_keyword
|
|
|
|
# --- 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"] = slide.image_keyword
|
|
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"] = post.cta_image_keyword
|
|
|
|
# --- Caption Instagram (1 colonna) ---
|
|
row["caption_instagram"] = post.caption_instagram
|
|
|
|
rows.append(row)
|
|
|
|
return rows
|