app.py

"""
IG Comment Gen - HuggingFace Space

Genera un commento breve, naturale, in italiano per un post Instagram,
usando la HF Inference Providers API (OpenAI-compatible).

Backend cascade:
    1. openai/gpt-oss-120b:fastest      (Groq/Cerebras, ~500ms)
    2. meta-llama/Llama-3.3-70B-Instruct:fastest
    3. mistralai/Mistral-Small-3.1-24B-Instruct-2503:fastest

Se tutti falliscono, torna comment=None -> il client (bot) ricade sul
suo file txt di fallback.

Auth verso HF: HF_TOKEN nei Secrets dello Space.
Auth verso questo Space: opzionale SPACE_API_KEY (bearer). Se non settato,
l'endpoint /api/generate e' aperto (utile in fase di test).
"""
from __future__ import annotations

import logging
import os
import re
import time
from typing import Optional

import gradio as gr
from fastapi import FastAPI, Header, HTTPException
from huggingface_hub import InferenceClient
from pydantic import BaseModel, Field

# ---------------------------------------------------------------------------
# Setup
# ---------------------------------------------------------------------------
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s [%(levelname)s] %(name)s: %(message)s",
)
logger = logging.getLogger("ig-comment-gen")

HF_TOKEN = os.environ.get("HF_TOKEN", "").strip()
SPACE_API_KEY = os.environ.get("SPACE_API_KEY", "").strip()

if not HF_TOKEN:
    logger.warning(
        "HF_TOKEN non trovato negli env: le chiamate falliranno finche' "
        "non lo setti nei Secrets dello Space."
    )

# Cascade di modelli. NON reasoning-style: quelli tipo gpt-oss-120b, DeepSeek-R1
# ecc. bruciano tutto max_tokens in "thinking" senza generare content.
# Per un commento Instagram serve un plain instruct model.
MODEL_CASCADE = [
    "meta-llama/Llama-3.3-70B-Instruct:fastest",
    "Qwen/Qwen2.5-72B-Instruct:fastest",
    "mistralai/Mistral-Small-3.1-24B-Instruct-2503:fastest",
]

MAX_CAPTION_CHARS = 800
REQUEST_TIMEOUT_S = 12
MAX_OUTPUT_TOKENS = 120
DEFAULT_LANGUAGE = "Italian"


# ---------------------------------------------------------------------------
# Prompting
# ---------------------------------------------------------------------------
def _build_prompt(
    caption: str,
    target_username: Optional[str],
    media_type: str,
    hint: Optional[str],
    language: str,
) -> str:
    """Prompt con 14 regole rigide, copiato/adattato dal repo di riferimento
    (dzeveckij/instagram-ai-commenter-bot) che sono empiricamente quelle
    che rendono i commenti indistinguibili da un umano."""
    safe_caption = (caption or "").strip()
    if len(safe_caption) > MAX_CAPTION_CHARS:
        safe_caption = safe_caption[:MAX_CAPTION_CHARS] + "…"

    target_clause = (
        f"by @{target_username}" if target_username else "by an Instagram user"
    )
    caption_clause = (
        f'Post caption: "{safe_caption}"'
        if safe_caption
        else "The post has no caption; comment on the photo/video itself."
    )
    hint_clause = f"Context for tone: {hint}\n\n" if hint else ""

    rules = f"""Rules (follow ALL):
1. Write ONE short, relevant comment. 1 sentence ideal, max 2 short.
2. Sound 100% authentic, like a real follower, NEVER like a bot.
3. If possible, refer to something specific from the caption.
4. ABSOLUTELY NO emojis.
5. ABSOLUTELY NO hashtags.
6. ABSOLUTELY NO exclamation marks.
7. NO generic compliments ("Great post", "Love this", "So inspiring", "Amazing", "Bravo").
8. NO questions, do not try to start a conversation.
9. NO first-person pronouns ("I", "me", "my", "io", "mi", "mio").
10. NO opinions, NO personal experiences.
11. Vary tone and phrasing; do not reuse common patterns.
12. Media type is: {media_type}. If video/reel, comment on motion/action.
13. Output language: {language}.
14. Output ONLY the final comment text. No quotes, no prefix, no explanation."""

    return (
        f"You write engaging human-like Instagram comments.\n"
        f"Write a comment for a post {target_clause}.\n\n"
        f"{caption_clause}\n\n"
        f"{hint_clause}"
        f"{rules}\n"
    )


# ---------------------------------------------------------------------------
# Output validation
# ---------------------------------------------------------------------------
def _sanitize_output(text: str) -> str:
    """Rimuove rumore tipico: virgolette wrap, prefissi 'Comment:', HTML,
    newline interni, spazi doppi. Taglia a 220 char."""
    if not text:
        return ""
    t = text.strip()
    t = re.sub(r"<[^>]+>", "", t).strip()
    for q in ('"', "'", "`"):
        if len(t) >= 2 and t.startswith(q) and t.endswith(q):
            t = t[1:-1].strip()
    for prefix in ("Comment:", "comment:", "Output:", "Reply:", "Commento:"):
        if t.lower().startswith(prefix.lower()):
            t = t[len(prefix):].strip()
    t = " ".join(t.split())
    if len(t) > 220:
        t = t[:217].rstrip() + "..."
    return t


def _passes_guardrails(text: str) -> bool:
    """Hard guardrail: no #, no !, no emoji (heuristic su codepoint > U+27BF)."""
    if not text:
        return False
    if any(ch in text for ch in "#!"):
        return False
    if any(ord(ch) > 0x27BF for ch in text):
        return False
    return True


# ---------------------------------------------------------------------------
# Inference
# ---------------------------------------------------------------------------
def _call_model(model: str, prompt: str) -> tuple[Optional[str], bool]:
    """
    Chiama HF Inference Providers (OpenAI-compat via InferenceClient).

    Returns:
        (text, retryable):
          - text: commento raw o None su errore
          - retryable: True se ha senso provare il prossimo modello
    """
    if not HF_TOKEN:
        return None, False
    try:
        client = InferenceClient(token=HF_TOKEN, timeout=REQUEST_TIMEOUT_S)
        completion = client.chat.completions.create(
            model=model,
            messages=[{"role": "user", "content": prompt}],
            max_tokens=MAX_OUTPUT_TOKENS,
            temperature=0.9,
            top_p=1.0,
        )
    except Exception as e:
        err_repr = repr(e).lower()
        # 401 / 403 / bad token -> fatal
        if "401" in err_repr or "unauthorized" in err_repr or "403" in err_repr:
            logger.warning("[%s] auth error: %s", model, e)
            return None, False
        # 404 (model not on providers) / 429 (rate-limit) / 5xx / timeout -> retryable
        logger.warning("[%s] error (retryable): %s", model, e)
        return None, True

    try:
        msg = completion.choices[0].message
        text = getattr(msg, "content", None) or ""
        # Reasoning models (gpt-oss, DeepSeek-R1, ...) mettono la risposta in
        # `reasoning` invece che in `content` e finiscono spesso senza generare
        # il content finale. Se text e' vuoto, trattalo come retryable: sposta
        # al prossimo modello (non-reasoning) invece di considerarlo fatal.
        if not text.strip():
            reasoning = getattr(msg, "reasoning", None) or getattr(
                msg, "reasoning_content", None
            )
            logger.warning(
                "[%s] empty content (finish=%s, has_reasoning=%s)",
                model,
                completion.choices[0].finish_reason,
                bool(reasoning),
            )
            return None, True  # retryable: prova il prossimo modello
    except Exception as e:
        logger.warning("[%s] parse error: %s", model, e)
        return None, True

    return text, False


def generate_comment_core(
    caption: str,
    media_type: str,
    target_username: Optional[str] = None,
    hint: Optional[str] = None,
    language: str = DEFAULT_LANGUAGE,
) -> dict:
    """
    Wrapper riusabile (chiamato sia da Gradio UI che da FastAPI endpoint).
    Ritorna dict compatibile con la response API.
    """
    if not HF_TOKEN:
        return {
            "comment": None,
            "model_used": None,
            "latency_ms": 0,
            "attempts": 0,
            "error": "HF_TOKEN non configurato nei Secrets dello Space.",
        }

    prompt = _build_prompt(
        caption=caption or "",
        target_username=target_username or None,
        media_type=str(media_type or "photo").lower(),
        hint=hint or None,
        language=language or DEFAULT_LANGUAGE,
    )

    started = time.time()
    attempts = 0
    for idx, model in enumerate(MODEL_CASCADE):
        attempts += 1
        text, retryable = _call_model(model, prompt)
        if text:
            cleaned = _sanitize_output(text)
            if _passes_guardrails(cleaned):
                latency_ms = int((time.time() - started) * 1000)
                if idx > 0:
                    logger.info(
                        "cascade recovered on model %s after %d failure(s)",
                        model, idx,
                    )
                return {
                    "comment": cleaned,
                    "model_used": model,
                    "latency_ms": latency_ms,
                    "attempts": attempts,
                    "error": None,
                }
            else:
                logger.info(
                    "[%s] output failed guardrails: %r",
                    model, cleaned[:60],
                )
                continue

        if not retryable:
            logger.warning("cascade stopped: fatal error on %s", model)
            break

    latency_ms = int((time.time() - started) * 1000)
    return {
        "comment": None,
        "model_used": None,
        "latency_ms": latency_ms,
        "attempts": attempts,
        "error": "Tutti i modelli hanno fallito o output non conforme.",
    }


# ---------------------------------------------------------------------------
# DM generation (feature esclusiva del branch personalcoaching lato bot)
# ---------------------------------------------------------------------------
_MAX_BIO_CHARS = 400
_MAX_CAPTION_CHARS_DM = 600
_MAX_FULLNAME_CHARS = 80
_MAX_DM_CHARS = 280


def _truncate_dm(text: Optional[str], limit: int) -> str:
    if not text:
        return ""
    t = text.strip()
    if len(t) > limit:
        return t[:limit].rstrip() + "…"
    return t


def _build_dm_prompt(
    target_username: Optional[str],
    full_name: Optional[str],
    bio: Optional[str],
    last_post_caption: Optional[str],
    hint: Optional[str],
    language: str,
    allow_emoji: bool,
) -> str:
    """Prompt per DM di apertura, assertivo per evitare che Llama/Qwen infili
    'Mi chiamo X, sono un coach...', hashtag, CTA hard-sell o emoji a raffica.
    Replica lo stile del vecchio ai_dm.py (Gemini era) ma adattato alla
    catena non-reasoning corrente."""
    name_part = _truncate_dm(full_name, _MAX_FULLNAME_CHARS)
    bio_part = _truncate_dm(bio, _MAX_BIO_CHARS)
    caption_part = _truncate_dm(last_post_caption, _MAX_CAPTION_CHARS_DM)

    context_lines = []
    if target_username:
        context_lines.append(f"- username: @{target_username}")
    if name_part:
        context_lines.append(f"- nome visualizzato: {name_part}")
    if bio_part:
        context_lines.append(f"- bio: {bio_part}")
    if caption_part:
        context_lines.append(f"- caption del suo ultimo post: {caption_part}")
    context_block = (
        "Profilo del destinatario (USA questi indizi per personalizzare,\n"
        "ma NON fare il pappagallo: non ripetere parole esatte della bio):\n"
        + "\n".join(context_lines)
        if context_lines
        else "Nessuna informazione sul profilo disponibile: scrivi un DM\n"
        "amichevole generico che ringrazia per il follow e fa una sola\n"
        "domanda aperta sul suo percorso fitness."
    )

    hint_clause = (
        f"Contesto su CHI scrive (autore del DM): {hint}\n\n" if hint else ""
    )

    emoji_rule = (
        "5. Puoi usare AL MASSIMO 1 emoji leggera (👋 🙌 💪) e SOLO se calza "
        "naturalmente. Nessuna emoji e' la scelta default migliore."
        if allow_emoji
        else "5. ASSOLUTAMENTE NESSUNA emoji."
    )

    rules = f"""Regole (rispettarle TUTTE):
1. Scrivi un DM Instagram di apertura, 2-3 frasi brevi totali, max {_MAX_DM_CHARS} caratteri.
2. Tono: amichevole, peer-to-peer, mai venditore, mai formale.
3. Ringrazia in modo naturale per il follow (lui ti ha appena seguito).
4. Fai UNA SOLA domanda aperta sul suo percorso fitness/sport, basandoti sugli
   indizi del profilo se ci sono. La domanda deve essere semplice e specifica,
   non generica ("come stai?" e' VIETATO).
{emoji_rule}
6. ASSOLUTAMENTE NIENTE hashtag.
7. ASSOLUTAMENTE NIENTE link, niente "DM me", niente "scrivimi per coaching",
   niente menzioni di servizi o prezzi.
8. NIENTE punti esclamativi multipli ("!!"). Massimo 1 "!" in tutto il DM.
9. NIENTE template robotici tipo "Ciao [nome], ho visto il tuo profilo e...".
10. NON menzionare la parola "bot", "automatico", "AI" o simili.
11. NON ripetere parole della bio in modo letterale.
12. Lingua di output: {language}.
13. Output SOLO il testo finale del DM. Niente virgolette, niente prefissi,
    niente spiegazioni, niente firma."""

    return (
        f"Sei un personal trainer / coach che ha appena ricevuto un nuovo\n"
        f"follower su Instagram e vuole aprire una conversazione sincera in DM.\n\n"
        f"{hint_clause}"
        f"{context_block}\n\n"
        f"{rules}\n"
    )


_DM_LINK_MARKERS = (
    "http://", "https://", "www.", "wa.me", "bit.ly", "linktr.ee", "t.me/",
)
_DM_SELL_MARKERS = (
    "coaching online",
    "scrivimi per",
    "contattami per",
    "pacchetto",
    "prezzo",
    "prenota",
)


def _sanitize_dm(text: str) -> str:
    """Rimuove tag HTML, wrap quotes, prefissi 'DM:' e collassa whitespace."""
    if not text:
        return ""
    t = text.strip()
    t = re.sub(r"<[^>]+>", "", t).strip()
    for q in ('"', "'", "`"):
        if len(t) >= 2 and t.startswith(q) and t.endswith(q):
            t = t[1:-1].strip()
    for prefix in ("DM:", "dm:", "Message:", "Messaggio:", "Reply:"):
        if t.lower().startswith(prefix.lower()):
            t = t[len(prefix):].strip()
    # normalizza spazi ma preserva newline singolo
    t = re.sub(r"[ \t]+", " ", t).strip()
    if len(t) > _MAX_DM_CHARS:
        t = t[: _MAX_DM_CHARS - 3].rstrip() + "..."
    return t


def _passes_dm_guardrails(text: str, allow_emoji: bool) -> bool:
    if not text:
        return False
    if "#" in text:
        return False
    lower = text.lower()
    if any(m in lower for m in _DM_LINK_MARKERS):
        return False
    if any(m in lower for m in _DM_SELL_MARKERS):
        return False
    if text.count("!") > 1:
        return False
    if not allow_emoji and any(ord(ch) > 0x27BF for ch in text):
        return False
    return True


def generate_dm_core(
    target_username: Optional[str] = None,
    full_name: Optional[str] = None,
    bio: Optional[str] = None,
    last_post_caption: Optional[str] = None,
    hint: Optional[str] = None,
    language: str = DEFAULT_LANGUAGE,
    allow_emoji: bool = True,
) -> dict:
    """Wrapper riusabile per DM. Stessa cascata di modelli di generate_comment_core."""
    if not HF_TOKEN:
        return {
            "dm": None,
            "model_used": None,
            "latency_ms": 0,
            "attempts": 0,
            "error": "HF_TOKEN non configurato nei Secrets dello Space.",
        }

    prompt = _build_dm_prompt(
        target_username=target_username,
        full_name=full_name,
        bio=bio,
        last_post_caption=last_post_caption,
        hint=hint,
        language=language or DEFAULT_LANGUAGE,
        allow_emoji=allow_emoji,
    )

    started = time.time()
    attempts = 0
    for idx, model in enumerate(MODEL_CASCADE):
        attempts += 1
        text, retryable = _call_model(model, prompt)
        if text:
            cleaned = _sanitize_dm(text)
            if _passes_dm_guardrails(cleaned, allow_emoji=allow_emoji):
                latency_ms = int((time.time() - started) * 1000)
                if idx > 0:
                    logger.info(
                        "DM cascade recovered on model %s after %d failure(s)",
                        model, idx,
                    )
                return {
                    "dm": cleaned,
                    "model_used": model,
                    "latency_ms": latency_ms,
                    "attempts": attempts,
                    "error": None,
                }
            else:
                logger.info(
                    "[%s] DM output failed guardrails: %r", model, cleaned[:60]
                )
                continue

        if not retryable:
            logger.warning("DM cascade stopped: fatal error on %s", model)
            break

    latency_ms = int((time.time() - started) * 1000)
    return {
        "dm": None,
        "model_used": None,
        "latency_ms": latency_ms,
        "attempts": attempts,
        "error": "Tutti i modelli hanno fallito o output non conforme.",
    }


# ---------------------------------------------------------------------------
# FastAPI
# ---------------------------------------------------------------------------
class GenerateRequest(BaseModel):
    caption: str = Field(..., description="Caption del post")
    media_type: str = Field("photo", description="photo|video|reel|carousel|igtv")
    target_username: Optional[str] = Field(None, description="Autore del post")
    hint: Optional[str] = Field(None, description="Contesto di tono / nicchia")
    language: str = Field(DEFAULT_LANGUAGE, description="Lingua output")


class GenerateResponse(BaseModel):
    comment: Optional[str]
    model_used: Optional[str]
    latency_ms: int
    attempts: int
    error: Optional[str] = None


class GenerateDmRequest(BaseModel):
    target_username: Optional[str] = Field(None, description="Destinatario del DM")
    full_name: Optional[str] = Field(None, description="Nome display sul profilo")
    bio: Optional[str] = Field(None, description="Biografia del destinatario")
    last_post_caption: Optional[str] = Field(None, description="Caption ultimo post")
    hint: Optional[str] = Field(None, description="Contesto su chi scrive")
    language: str = Field(DEFAULT_LANGUAGE, description="Lingua output")
    allow_emoji: bool = Field(True, description="Permette al massimo 1 emoji leggera")


class GenerateDmResponse(BaseModel):
    dm: Optional[str]
    model_used: Optional[str]
    latency_ms: int
    attempts: int
    error: Optional[str] = None


app = FastAPI(title="IG Comment Gen", version="1.0.0")


def _check_api_key(authorization: Optional[str]) -> None:
    if not SPACE_API_KEY:
        return  # endpoint aperto
    if not authorization or not authorization.startswith("Bearer "):
        raise HTTPException(status_code=401, detail="Missing bearer token")
    token = authorization.removeprefix("Bearer ").strip()
    if token != SPACE_API_KEY:
        raise HTTPException(status_code=401, detail="Invalid token")


@app.post("/api/generate", response_model=GenerateResponse)
def api_generate(
    req: GenerateRequest,
    authorization: Optional[str] = Header(None),
) -> GenerateResponse:
    _check_api_key(authorization)
    result = generate_comment_core(
        caption=req.caption,
        media_type=req.media_type,
        target_username=req.target_username,
        hint=req.hint,
        language=req.language,
    )
    return GenerateResponse(**result)


@app.post("/api/generate_dm", response_model=GenerateDmResponse)
def api_generate_dm(
    req: GenerateDmRequest,
    authorization: Optional[str] = Header(None),
) -> GenerateDmResponse:
    _check_api_key(authorization)
    result = generate_dm_core(
        target_username=req.target_username,
        full_name=req.full_name,
        bio=req.bio,
        last_post_caption=req.last_post_caption,
        hint=req.hint,
        language=req.language,
        allow_emoji=req.allow_emoji,
    )
    return GenerateDmResponse(**result)


@app.get("/api/health")
def health() -> dict:
    return {
        "status": "ok",
        "hf_token_configured": bool(HF_TOKEN),
        "space_api_key_enforced": bool(SPACE_API_KEY),
        "models": MODEL_CASCADE,
        "endpoints": ["/api/generate", "/api/generate_dm", "/api/health"],
    }


# ---------------------------------------------------------------------------
# Gradio UI
# ---------------------------------------------------------------------------
def _ui_generate(caption, media_type, target_username, hint, language):
    result = generate_comment_core(
        caption=caption,
        media_type=media_type,
        target_username=target_username,
        hint=hint,
        language=language,
    )
    comment = result.get("comment") or "(nessun commento generato)"
    meta = (
        f"Model: {result.get('model_used') or 'n/a'}\n"
        f"Latency: {result.get('latency_ms')}ms\n"
        f"Attempts: {result.get('attempts')}\n"
        f"Error: {result.get('error') or 'none'}"
    )
    return comment, meta


with gr.Blocks(title="IG Comment Gen", theme=gr.themes.Soft()) as demo:
    gr.Markdown(
        "# 💬 IG Comment Gen\n"
        "Genera un commento breve, naturale, in italiano per un post Instagram. "
        "Usato dal bot GramAddict come sostituto della vecchia integrazione Gemini."
    )
    with gr.Row():
        with gr.Column():
            caption_in = gr.Textbox(
                label="Caption del post",
                lines=5,
                placeholder="3 anni di allenamento continuo, ecco i risultati…",
            )
            media_in = gr.Dropdown(
                choices=["photo", "video", "reel", "carousel", "igtv"],
                value="reel",
                label="Media type",
            )
            username_in = gr.Textbox(
                label="Target username (opzionale)",
                placeholder="il.pump.quotidiano",
            )
            hint_in = gr.Textbox(
                label="Hint di contesto (opzionale)",
                placeholder="coach fitness, tono cameratesco",
            )
            lang_in = gr.Textbox(
                label="Lingua output", value=DEFAULT_LANGUAGE
            )
            btn = gr.Button("Genera commento", variant="primary")
        with gr.Column():
            out_comment = gr.Textbox(label="Commento generato", lines=3)
            out_meta = gr.Textbox(label="Meta", lines=4)

    btn.click(
        fn=_ui_generate,
        inputs=[caption_in, media_in, username_in, hint_in, lang_in],
        outputs=[out_comment, out_meta],
    )

    gr.Markdown(
        "### API REST\n"
        "`POST /api/generate` con body JSON `{ caption, media_type, "
        "target_username?, hint?, language? }`.\n"
        "Se `SPACE_API_KEY` e' settato nei Secrets, richiede header "
        "`Authorization: Bearer <SPACE_API_KEY>`.\n\n"
        "`GET /api/health` per verificare che HF_TOKEN sia configurato."
    )


# Mount Gradio dentro FastAPI sulla root: cosi' visiti "/" e vedi la UI,
# mentre "/api/*" resta libero per gli endpoint REST.
app = gr.mount_gradio_app(app, demo, path="/")


# Local dev only. Su HF Space (Docker SDK) uvicorn e' avviato dal Dockerfile
# come `uvicorn app:app --host 0.0.0.0 --port 7860`. Non serve il blocco
# __main__: se lo eseguissi qui e HF Docker facesse lo stesso ci sarebbero
# due processi in conflitto.
if __name__ == "__main__":
    import uvicorn
    port = int(os.environ.get("PORT", 7860))
    uvicorn.run(app, host="0.0.0.0", port=port)