app.py

"""
uvicorn app:app --reload --port 8000
"""

import os
from pathlib import Path

import torch
from fastapi import Depends, FastAPI, HTTPException, Security
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import HTMLResponse
from fastapi.security.api_key import APIKeyHeader
from huggingface_hub import snapshot_download
from pydantic import BaseModel
from transformers import AutoTokenizer, AutoModelForSequenceClassification

from formality_scorer import score_sentence, flag_soft_refusals

# Model location:
# - Production (HF Space): set MODEL_REPO_ID=YOUR_HF_USERNAME/keigo-radar-classifier
#   The weights are downloaded from HF Model Hub at container startup.
_MODEL_REPO_ID = os.getenv("MODEL_REPO_ID", "")
if _MODEL_REPO_ID:
    print(f"Downloading classifier from HF Hub: {_MODEL_REPO_ID}")
    CLASSIFIER_DIR = Path(snapshot_download(_MODEL_REPO_ID))
else:
    CLASSIFIER_DIR = Path("keigo_radar_classifier")
    print(f"Using local classifier: {CLASSIFIER_DIR.resolve()}")
INTENSITY_LABELS = ["none", "mild", "moderate", "strong"]

app = FastAPI(title="Keigo Radar API")


_raw_origins = os.getenv("ALLOWED_ORIGIN")

if not _raw_origins:
    raise RuntimeError("SECURITY ERROR: ALLOWED_ORIGIN environment variable is not set.")
origins = [origin.strip() for origin in _raw_origins.split(",")]

app.add_middleware(
    CORSMiddleware,
    allow_origins=origins,
    allow_methods=["GET", "POST"],
    allow_headers=["Content-Type", "X-API-Key"],
)

# ── API key guard ─────────────────────────────────────────────────────────────
_api_key_header = APIKeyHeader(name="X-API-Key", auto_error=False)

def verify_api_key(key: str | None = Security(_api_key_header)) -> str:
    """Validate the X-API-Key header against INTERNAL_API_KEY env var."""
    expected = os.getenv("INTERNAL_API_KEY")
    if not expected:
        raise HTTPException(status_code=500, detail="INTERNAL_API_KEY not configured on server.")
    if not key:
        raise HTTPException(status_code=401, detail="X-API-Key header missing.")
    if key != expected:
        raise HTTPException(status_code=403, detail="Forbidden")
    return key

class AnalyzeRequest(BaseModel):
    text: str

print("Loading classifier from", CLASSIFIER_DIR.resolve())
_tokenizer = AutoTokenizer.from_pretrained(CLASSIFIER_DIR)
_model = AutoModelForSequenceClassification.from_pretrained(CLASSIFIER_DIR)
_model.eval()
print("Classifier loaded.")


def predict_negative_intensity(text: str) -> dict:
    inputs = _tokenizer(text, return_tensors="pt", truncation=True, max_length=128)
    with torch.no_grad():
        logits = _model(**inputs).logits
    probs = torch.softmax(logits, dim=-1)[0].tolist()
    level = int(torch.argmax(logits, dim=-1).item())
    return {
        "level": level,
        "label": INTENSITY_LABELS[level],
        "probabilities": {INTENSITY_LABELS[i]: round(p, 3) for i, p in enumerate(probs)},
    }

    
@app.get("/", response_class=HTMLResponse)
def root():
    return """
    <!DOCTYPE html>
    <html>
    <head>
        <title>Keigo Radar API</title>
        <style>
            body {
                background-color: #0b0f19;
                color: #f3f4f6;
                font-family: -apple-system, BlinkMacSystemFont, "Segoe UI", Roboto, sans-serif;
                display: flex;
                justify-content: center;
                align-items: center;
                height: 100vh;
                margin: 0;
            }
            .card {
                background-color: #111827;
                border: 1px solid #1f2937;
                padding: 2.5rem;
                border-radius: 12px;
                max-width: 500px;
                text-align: center;
                box-shadow: 0 10px 25px -5px rgba(0, 0, 0, 0.3);
            }
            h1 { color: #f59e0b; font-size: 1.8rem; margin-top: 0; }
            p { color: #9ca3af; line-height: 1.6; font-size: 0.95rem; }
            .badge {
                display: inline-block;
                background-color: #065f46;
                color: #34d399;
                padding: 0.25rem 0.75rem;
                border-radius: 9999px;
                font-size: 0.8rem;
                font-weight: 600;
                margin-bottom: 1rem;
            }
            .link-btn {
                display: inline-block;
                margin-top: 1.5rem;
                background-color: #2563eb;
                color: white;
                text-decoration: none;
                padding: 0.6rem 1.3rem;
                border-radius: 6px;
                font-weight: 500;
                font-size: 0.9rem;
                transition: background-color 0.2s;
            }
            .link-btn:hover { background-color: #1d4ed8; }
        </style>
    </head>
    <body>
        <div class="card">
            <div class="badge">ONLINE</div>
            <h1>📡 Keigo Radar API</h1>
            <p>
                This is the high-performance NLP inference microservice container for Keigo Radar. 
                It processes Japanese morphological tokens via Fugashi and computes structural 
                frustration intensity log-odds using a custom-refined BERT model.
            </p>
            <a href="https://www.keigo-radar.xyz" target="_blank" class="link-btn">Go to Frontend Interface</a>
        </div>
    </body>
    </html>
    """

@app.post("/analyze")
def analyze(req: AnalyzeRequest, _: str = Depends(verify_api_key)) -> dict:
    text = req.text

    formality = score_sentence(text)
    refusals = flag_soft_refusals(text)
    intensity = predict_negative_intensity(text)

    # The core feature: surface politeness vs. underlying negative
    # intensity. 
    # High formality (sonkeigo/kenjougo) + high negative
    # intensity (level >= 2) = the text reads polite but the model
    # detects real frustration underneath.
    is_high_formality = formality["predicted_type"] in ("sonkeigo", "kenjougo")
    is_high_intensity = intensity["level"] >= 2
    masking_gap_detected = is_high_formality and is_high_intensity

    # Radar chart axes plot

    radar_axes = {
        "teineigo": 1.0 if formality["predicted_type"] in ("teineigo", "sonkeigo", "kenjougo") else 0.0,
        "sonkeigo": 1.0 if formality["predicted_type"] == "sonkeigo" else 0.0,
        "kenjougo": 1.0 if formality["predicted_type"] == "kenjougo" else 0.0,
        "negative_intensity": round(intensity["level"] / 3.0, 3),
        "soft_refusal_signal": round(min(len(refusals) / 2.0, 1.0), 3),
    }

    return {
        "text": text,
        "formality": formality,
        "negative_intensity": intensity,
        "soft_refusals": refusals,
        "masking_gap_detected": masking_gap_detected,
        "radar_axes": radar_axes,
    }


@app.get("/health")
def health():
    return {"status": "ok"}