"""Editorial Calendar router. Espone endpoint per il calendario editoriale con awareness levels e formati narrativi. Integra LLM per generare idee di contenuto reali basate su profilo personaggio. """ import csv import io from typing import Optional from fastapi import APIRouter, Depends, HTTPException from fastapi.responses import StreamingResponse from pydantic import BaseModel from sqlalchemy.orm import Session from ..auth import get_current_user from ..database import get_db from ..models import Character, SystemSetting, User from ..services.calendar_service import CalendarService, FORMATI_NARRATIVI, AWARENESS_LEVELS from ..services.llm import get_llm_provider router = APIRouter( prefix="/api/editorial", tags=["editorial"], dependencies=[Depends(get_current_user)], ) _calendar_service = CalendarService() def _get_setting(db: Session, key: str, user_id: int = None) -> str | None: if user_id is not None: setting = db.query(SystemSetting).filter( SystemSetting.key == key, SystemSetting.user_id == user_id ).first() if setting is not None: return setting.value setting = db.query(SystemSetting).filter( SystemSetting.key == key, SystemSetting.user_id == None ).first() return setting.value if setting else None # === Schemas locali === class CalendarGenerateRequest(BaseModel): topics: list[str] format_narrativo: Optional[str] = None awareness_level: Optional[int] = None num_posts: int = 7 start_date: Optional[str] = None character_id: Optional[int] = None brief: Optional[str] = None class ExportCsvRequest(BaseModel): slots: list[dict] # === Endpoints === @router.get("/formats") def get_formats(): """Restituisce i format narrativi disponibili con i relativi awareness levels.""" return { "formats": _calendar_service.get_formats(), "awareness_levels": [ {"value": k, "label": v} for k, v in AWARENESS_LEVELS.items() ], } @router.post("/generate-calendar") def generate_calendar( request: CalendarGenerateRequest, db: Session = Depends(get_db), current_user: User = Depends(get_current_user), ): """Genera un calendario editoriale con idee di contenuto AI.""" if not request.topics: return {"slots": [], "totale_post": 0} # Generate structural slots slots = _calendar_service.generate_calendar( topics=request.topics, num_posts=request.num_posts, format_narrativo=request.format_narrativo, awareness_level=request.awareness_level, start_date=request.start_date, ) # Try to enrich with LLM-generated ideas provider_name = _get_setting(db, "llm_provider", current_user.id) api_key = _get_setting(db, "llm_api_key", current_user.id) model = _get_setting(db, "llm_model", current_user.id) if provider_name and api_key and slots: try: base_url = _get_setting(db, "llm_base_url", current_user.id) llm = get_llm_provider(provider_name, api_key, model, base_url=base_url) # Build context from character if available char_context = "" if request.character_id: character = db.query(Character).filter( Character.id == request.character_id, Character.user_id == current_user.id, ).first() if character: char_context = ( f"Personaggio: {character.name}, nicchia: {character.niche}\n" f"Target: {character.target_audience or 'generico'}\n" f"Voce: {character.brand_voice or character.tone or 'professionale'}\n" ) brief_context = f"\nBrief strategico: {request.brief}" if request.brief else "" slot_descriptions = "\n".join( f"- Slot {s['indice']+1}: topic={s['topic']}, formato={s['formato_narrativo']}, " f"awareness={s['awareness_label']}, data={s['data_pubblicazione']}" for s in slots ) system_prompt = ( "Sei un social media strategist esperto. " "Per ogni slot del calendario editoriale, genera un'idea di post concreta: " "un titolo/hook (max 10 parole) e un mini-brief (max 25 parole) che descrive il contenuto. " "Rispondi SOLO con il formato:\n" "1. HOOK: ... | BRIEF: ...\n" "2. HOOK: ... | BRIEF: ...\n" "Una riga per slot, nient'altro." ) prompt = ( f"{char_context}{brief_context}\n\n" f"Calendario ({len(slots)} slot):\n{slot_descriptions}\n\n" f"Genera hook e brief per ogni slot:" ) result = llm.generate(prompt, system=system_prompt) # Parse result and enrich slots lines = [l.strip() for l in result.strip().splitlines() if l.strip()] for i, line in enumerate(lines): if i >= len(slots): break # Parse "N. HOOK: ... | BRIEF: ..." hook = "" brief = "" if "HOOK:" in line and "BRIEF:" in line: parts = line.split("BRIEF:") hook_part = parts[0] brief = parts[1].strip() if len(parts) > 1 else "" hook = hook_part.split("HOOK:")[-1].strip().rstrip("|").strip() elif "|" in line: parts = line.split("|", 1) hook = parts[0].strip().lstrip("0123456789. ") brief = parts[1].strip() if len(parts) > 1 else "" else: hook = line.lstrip("0123456789. ").strip() slots[i]["hook"] = hook slots[i]["idea_brief"] = brief except Exception: pass # Fall back to structural-only slots return { "slots": slots, "totale_post": len(slots), } @router.post("/export-csv") def export_csv(request: ExportCsvRequest): """Esporta il calendario editoriale come CSV per Canva.""" output = io.StringIO() fieldnames = [ "indice", "data_pubblicazione", "topic", "formato_narrativo", "awareness_level", "awareness_label", "note", ] writer = csv.DictWriter(output, fieldnames=fieldnames, extrasaction="ignore") writer.writeheader() for slot in request.slots: writer.writerow({ "indice": slot.get("indice", ""), "data_pubblicazione": slot.get("data_pubblicazione", ""), "topic": slot.get("topic", ""), "formato_narrativo": slot.get("formato_narrativo", ""), "awareness_level": slot.get("awareness_level", ""), "awareness_label": slot.get("awareness_label", ""), "note": slot.get("note", ""), }) output.seek(0) return StreamingResponse( iter([output.getvalue()]), media_type="text/csv", headers={ "Content-Disposition": "attachment; filename=calendario_editoriale.csv" }, )