Michele/postgenerator
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(01-02): CalendarService, PromptService, 7 prompt .txt in italiano

Browse Source 
- backend/services/calendar_service.py: genera 13 slot con distribuzione PN (4v+2s+2n+3r+1c+1p) e Schwartz (L5=3,L4=3,L3=4,L2=2,L1=1), ordina per funnel, ruota nicchie, calcola date
- backend/services/prompt_service.py: carica/compila/elenca prompt {{variabile}}, ValueError per variabili mancanti
- backend/data/prompts/system_prompt.txt: sistema prompt esperto content marketing B2B italiano
- backend/data/prompts/topic_generator.txt: generazione topic per slot calendario
- backend/data/prompts/pas_valore.txt: formato PAS per post valore educativo
- backend/data/prompts/listicle_valore.txt: formato Listicle per post valore
- backend/data/prompts/bab_storytelling.txt: formato BAB per post storytelling
- backend/data/prompts/aida_promozione.txt: formato AIDA per post promozionale
- backend/data/prompts/dato_news.txt: formato Dato+Implicazione per post news
This commit is contained in:
Michele
2026-03-08 02:00:00 +01:00
parent f6d9215b52
commit ef9b9471fc
9 changed files with 917 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

296
backend/services/calendar_service.py Normal file
View File

@@ -0,0 +1,296 @@
"""CalendarService — genera il calendario editoriale di 13 slot PN + Schwartz.
Costruisce un piano di pubblicazione strategico con:
- Distribuzione Persuasion Nurturing corretta (4 valore, 2 storytelling, etc.)
- Livelli Schwartz assegnati in base al tipo contenuto
- Fasi campagna ordinate (Attira → Cattura → Coinvolgi → Converti)
- Date di pubblicazione suggerite con frequenza configurabile
- Rotazione nicchie (50% generico, 50% verticali in rotazione)
- Formato narrativo selezionato via FormatSelector
"""
from __future__ import annotations
from datetime import date, timedelta
from itertools import cycle
from typing import Optional
from backend.constants import (
FASI_CAMPAGNA,
FUNZIONI_CONTENUTO,
NICCHIE_DEFAULT,
PERSUASION_DISTRIBUTION,
POST_PER_CICLO,
SCHWARTZ_DISTRIBUTION,
)
from backend.schemas.calendar import CalendarRequest, CalendarResponse, CalendarSlot
from backend.services.format_selector import FormatSelector
# ---------------------------------------------------------------------------
# Mapping tipo_contenuto -> funzione editoriale
# ---------------------------------------------------------------------------
_TIPO_TO_FUNZIONE: dict[str, str] = {
"valore": "Educare",
"storytelling": "Intrattenere",
"news": "Intrattenere",
"riprova_sociale": "Persuadere",
"coinvolgimento": "Intrattenere",
"promozione": "Convertire",
}
# ---------------------------------------------------------------------------
# Mapping tipo_contenuto -> livello_schwartz con distribuzione corretta
# Ogni tipo ha una lista ordinata dei livelli che usa (in sequenza)
# ---------------------------------------------------------------------------
# La distribuzione deve sommare a SCHWARTZ_DISTRIBUTION totale
# L5=3, L4=3, L3=4, L2=2, L1=1
_TIPO_TO_LIVELLI: dict[str, list[str]] = {
"valore": ["L4", "L4", "L3", "L3"], # 4 slot: 2xL4, 2xL3
"storytelling": ["L5", "L5"], # 2 slot: 2xL5
"news": ["L5", "L4"], # 2 slot: 1xL5, 1xL4
"riprova_sociale": ["L3", "L3", "L2"], # 3 slot: 2xL3, 1xL2
"coinvolgimento": ["L2"], # 1 slot: 1xL2
"promozione": ["L1"], # 1 slot: 1xL1
}
# Verifica distribuzioni a load-time
_livelli_totali: dict[str, int] = {}
for _tipo, _livelli in _TIPO_TO_LIVELLI.items():
assert len(_livelli) == PERSUASION_DISTRIBUTION[_tipo], (
f"Tipo '{_tipo}': attesi {PERSUASION_DISTRIBUTION[_tipo]} livelli, "
f"trovati {len(_livelli)}"
)
for _l in _livelli:
_livelli_totali[_l] = _livelli_totali.get(_l, 0) + 1
for _livello, _count in _livelli_totali.items():
assert _count == SCHWARTZ_DISTRIBUTION[_livello], (
f"Livello '{_livello}': attesi {SCHWARTZ_DISTRIBUTION[_livello]}, "
f"trovati {_count}"
)
# ---------------------------------------------------------------------------
# Mapping livello_schwartz -> fase_campagna
# ---------------------------------------------------------------------------
_LIVELLO_TO_FASE: dict[str, str] = {
"L5": "Attira",
"L4": "Cattura",
"L3": "Cattura", # L3 va in Cattura (upper-middle) ma alcuni in Coinvolgi
"L2": "Coinvolgi",
"L1": "Converti",
}
# Affinamento: L3 riprova_sociale va in "Coinvolgi" (conosce la soluzione/prodotto)
_TIPO_LIVELLO_TO_FASE: dict[tuple[str, str], str] = {
("riprova_sociale", "L3"): "Coinvolgi",
("valore", "L3"): "Coinvolgi",
}
# ---------------------------------------------------------------------------
# Giorni della settimana per pubblicazione (default: lun, mer, ven)
# ---------------------------------------------------------------------------
_PUBLISH_WEEKDAYS = [0, 2, 4] # 0=Lunedì, 2=Mercoledì, 4=Venerdì
class CalendarService:
"""Genera il calendario editoriale di 13 slot con distribuzione PN e Schwartz."""
def __init__(self, format_selector: Optional[FormatSelector] = None) -> None:
"""Inizializza il servizio con un FormatSelector (iniettabile per test).
Args:
format_selector: Istanza di FormatSelector. Se None, ne crea una di default.
"""
self._format_selector = format_selector or FormatSelector()
def generate_calendar(self, request: CalendarRequest) -> CalendarResponse:
"""Genera un calendario editoriale di 13 slot.
Processo:
1. Espande la distribuzione PN in 13 slot ordinati per funnel
2. Assegna livelli Schwartz per tipo
3. Seleziona formato narrativo via FormatSelector
4. Assegna funzione editoriale e fase campagna
5. Distribu nicchie (50% generico, 50% verticali in rotazione)
6. Calcola date di pubblicazione
Args:
request: Parametri di configurazione del calendario
Returns:
CalendarResponse con 13 slot ordinati per fase campagna
"""
nicchie = self._prepare_niches(request.nicchie)
data_inizio = self._parse_start_date(request.data_inizio)
date_pubblicazione = self._generate_dates(
data_inizio, POST_PER_CICLO, request.frequenza_post
)
# Crea gli slot base (tipo + livello)
raw_slots = self._build_raw_slots()
# Ordina per fase campagna (Attira → Cattura → Coinvolgi → Converti)
ordered_slots = self._sort_by_funnel(raw_slots)
# Assegna nicchie con rotazione
niched_slots = self._distribute_niches(ordered_slots, nicchie)
# Costruisci gli oggetti CalendarSlot finali
calendar_slots: list[CalendarSlot] = []
for indice, (tipo, livello, nicchia) in enumerate(niched_slots):
formato = self._format_selector.select_format(tipo, livello)
funzione = _TIPO_TO_FUNZIONE[tipo]
fase = _TIPO_LIVELLO_TO_FASE.get((tipo, livello), _LIVELLO_TO_FASE[livello])
slot = CalendarSlot(
indice=indice,
tipo_contenuto=tipo,
livello_schwartz=livello,
formato_narrativo=formato,
funzione=funzione,
fase_campagna=fase,
target_nicchia=nicchia,
data_pub_suggerita=date_pubblicazione[indice].isoformat(),
topic=None,
)
calendar_slots.append(slot)
return CalendarResponse(
campagna=request.obiettivo_campagna,
slots=calendar_slots,
totale_post=len(calendar_slots),
)
# ---------------------------------------------------------------------------
# Metodi privati
# ---------------------------------------------------------------------------
def _build_raw_slots(self) -> list[tuple[str, str]]:
"""Crea la lista di (tipo_contenuto, livello_schwartz) per tutti i 13 slot."""
slots: list[tuple[str, str]] = []
for tipo, livelli in _TIPO_TO_LIVELLI.items():
for livello in livelli:
slots.append((tipo, livello))
return slots
def _sort_by_funnel(
self, slots: list[tuple[str, str]]
) -> list[tuple[str, str]]:
"""Ordina gli slot per fase campagna (Attira → Cattura → Coinvolgi → Converti)."""
phase_order = {fase: i for i, fase in enumerate(FASI_CAMPAGNA)}
def slot_phase_key(slot: tuple[str, str]) -> int:
tipo, livello = slot
fase = _TIPO_LIVELLO_TO_FASE.get((tipo, livello), _LIVELLO_TO_FASE[livello])
return phase_order.get(fase, 99)
return sorted(slots, key=slot_phase_key)
@staticmethod
def _distribute_niches(
slots: list[tuple[str, str]],
nicchie: list[str],
) -> list[tuple[str, str, str]]:
"""Assegna nicchie agli slot con distribuzione 50% generico, 50% verticali.
Args:
slots: Lista di (tipo, livello)
nicchie: Lista nicchie disponibili (include "generico")
Returns:
Lista di (tipo, livello, nicchia)
"""
verticali = [n for n in nicchie if n != "generico"]
if not verticali:
# Se non ci sono verticali, tutto generico
return [(t, l, "generico") for t, l in slots]
verticali_cycle = cycle(verticali)
result: list[tuple[str, str, str]] = []
for i, (tipo, livello) in enumerate(slots):
if i % 2 == 0:
# Slot pari -> generico
nicchia = "generico"
else:
# Slot dispari -> verticale in rotazione
nicchia = next(verticali_cycle)
result.append((tipo, livello, nicchia))
return result
@staticmethod
def _prepare_niches(nicchie_input: list[str] | None) -> list[str]:
"""Prepara la lista nicchie assicurando che 'generico' sia sempre incluso."""
if not nicchie_input:
return list(NICCHIE_DEFAULT)
if "generico" not in nicchie_input:
return ["generico"] + list(nicchie_input)
return list(nicchie_input)
@staticmethod
def _parse_start_date(data_inizio: str | None) -> date:
"""Converte stringa YYYY-MM-DD in date, default a oggi."""
if data_inizio:
try:
return date.fromisoformat(data_inizio)
except ValueError:
pass
return date.today()
@staticmethod
def _generate_dates(
start: date,
count: int,
frequenza: int,
) -> list[date]:
"""Genera una lista di date di pubblicazione.
Con frequenza=3 usa lun/mer/ven. Con altre frequenze distribuisce
uniformemente nella settimana.
Args:
start: Data di inizio
count: Numero di date da generare
frequenza: Post per settimana
Returns:
Lista di 'count' date ordinate
"""
dates: list[date] = []
if frequenza == 3:
# Standard: lun, mer, ven
publish_days = _PUBLISH_WEEKDAYS
else:
# Distribuzione uniforme: calcola i giorni della settimana
step = max(1, 7 // frequenza)
publish_days = [i * step for i in range(frequenza)]
# Trova il primo giorno di pubblicazione >= start
current = start
day_cycle = cycle(sorted(publish_days))
next_day = next(day_cycle)
# Avanza fino al primo giorno valido
days_checked = 0
while current.weekday() != next_day and days_checked < 7:
current += timedelta(days=1)
days_checked += 1
# Genera le date
for _ in range(count):
dates.append(current)
# Passa al prossimo giorno di pubblicazione
next_day = next(day_cycle)
days_ahead = (next_day - current.weekday()) % 7
if days_ahead == 0:
days_ahead = 7
current = current + timedelta(days=days_ahead)
return dates

169
backend/services/prompt_service.py Normal file
View File

@@ -0,0 +1,169 @@
"""PromptService — carica, lista e compila prompt .txt con variabili.
Gestisce i file .txt dei prompt LLM nella directory PROMPTS_PATH.
Usa la sintassi {{variabile}} per i placeholder (doppia graffa).
"""
from __future__ import annotations
import re
from pathlib import Path
from typing import Optional
# Pattern per trovare le variabili {{nome}} nei template
_VARIABLE_PATTERN = re.compile(r"\{\{(\w+)\}\}")
class PromptService:
"""Servizio per gestire i prompt .txt del sistema di generazione.
Fornisce metodi per:
- Elencare i prompt disponibili
- Caricare il contenuto di un prompt
- Compilare un prompt sostituendo le variabili {{...}}
- Salvare un prompt (per l'editor di Phase 2)
- Estrarre la lista di variabili richieste da un template
"""
def __init__(self, prompts_dir: Path) -> None:
"""Inizializza il servizio con la directory dei prompt.
Args:
prompts_dir: Path alla directory contenente i file .txt dei prompt.
Tipicamente PROMPTS_PATH da backend.config.
Raises:
FileNotFoundError: Se la directory non esiste.
"""
if not prompts_dir.exists():
raise FileNotFoundError(
f"Directory prompt non trovata: {prompts_dir}. "
"Verifica che PROMPTS_PATH sia configurato correttamente."
)
if not prompts_dir.is_dir():
raise NotADirectoryError(
f"Il percorso non è una directory: {prompts_dir}"
)
self._prompts_dir = prompts_dir
def list_prompts(self) -> list[str]:
"""Elenca tutti i prompt .txt disponibili nella directory.
Returns:
Lista di nomi file senza estensione, ordinata alfabeticamente.
Es: ['aida_promozione', 'bab_storytelling', 'system_prompt', ...]
"""
return sorted(
p.stem for p in self._prompts_dir.glob("*.txt") if p.is_file()
)
def load_prompt(self, name: str) -> str:
"""Carica il contenuto grezzo di un prompt .txt.
Args:
name: Nome del prompt senza estensione (es. "pas_valore")
Returns:
Contenuto testuale del file prompt
Raises:
FileNotFoundError: Se il file non esiste
"""
path = self._get_path(name)
if not path.exists():
available = self.list_prompts()
raise FileNotFoundError(
f"Prompt '{name}' non trovato in {self._prompts_dir}. "
f"Prompt disponibili: {available}"
)
return path.read_text(encoding="utf-8")
def compile_prompt(self, name: str, variables: dict[str, str]) -> str:
"""Carica un prompt e sostituisce tutte le variabili {{nome}} con i valori forniti.
Args:
name: Nome del prompt senza estensione
variables: Dizionario { nome_variabile: valore }
Returns:
Testo del prompt con tutte le variabili sostituite
Raises:
FileNotFoundError: Se il prompt non esiste
ValueError: Se una variabile nel template non ha corrispondenza nel dict
"""
template = self.load_prompt(name)
# Verifica che tutte le variabili del template siano nel dict
required = set(_VARIABLE_PATTERN.findall(template))
provided = set(variables.keys())
missing = required - provided
if missing:
raise ValueError(
f"Variabili mancanti per il prompt '{name}': {sorted(missing)}. "
f"Fornire: {sorted(required)}"
)
def replace_var(match: re.Match) -> str:
var_name = match.group(1)
return variables[var_name]
return _VARIABLE_PATTERN.sub(replace_var, template)
def save_prompt(self, name: str, content: str) -> None:
"""Salva il contenuto di un prompt nel file .txt.
Usato dall'editor di prompt in Phase 2.
Args:
name: Nome del prompt senza estensione
content: Contenuto testuale da salvare
Raises:
ValueError: Se il nome contiene caratteri non sicuri
"""
# Sicurezza: validazione nome file (solo lettere, cifre, underscore, trattino)
if not re.match(r"^[\w\-]+$", name):
raise ValueError(
f"Nome prompt non valido: '{name}'. "
"Usa solo lettere, cifre, underscore e trattino."
)
path = self._get_path(name)
path.write_text(content, encoding="utf-8")
def get_required_variables(self, name: str) -> list[str]:
"""Analizza il template e ritorna la lista delle variabili richieste.
Args:
name: Nome del prompt senza estensione
Returns:
Lista ordinata di nomi variabile (senza doppie graffe)
Es: ['brand_name', 'livello_schwartz', 'obiettivo_campagna', 'target_nicchia', 'topic']
Raises:
FileNotFoundError: Se il prompt non esiste
"""
template = self.load_prompt(name)
variables = sorted(set(_VARIABLE_PATTERN.findall(template)))
return variables
def prompt_exists(self, name: str) -> bool:
"""Verifica se un prompt esiste.
Args:
name: Nome del prompt senza estensione
Returns:
True se il file esiste
"""
return self._get_path(name).exists()
# ---------------------------------------------------------------------------
# Metodi privati
# ---------------------------------------------------------------------------
def _get_path(self, name: str) -> Path:
"""Costruisce il percorso completo per un file prompt."""
return self._prompts_dir / f"{name}.txt"