app.py

#!/usr/bin/env python
"""
RM-DETECT — AI-vs-human text detection API.

A self-hostable AI-text detector. Wraps the `aidetect`
engine (perplexity + stylometry, Korean + English) behind a FastAPI service and
serves a single-page web UI.

Endpoints
---------
  GET  /            -> the web UI (static/index.html)
  GET  /health      -> {"status":"ok", "model_loaded":bool, ...}
  POST /detect      -> full detection result with per-paragraph char spans
"""
from __future__ import annotations

import os
import sys
import time
from typing import List, Optional

from fastapi import FastAPI, HTTPException
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import FileResponse
from fastapi.staticfiles import StaticFiles
from pydantic import BaseModel, Field

# make the app dir importable (aidetect package + detector live alongside)
_HERE = os.path.dirname(os.path.abspath(__file__))
if _HERE not in sys.path:
    sys.path.insert(0, _HERE)

from aidetect import features as _F        # noqa: E402
from aidetect import langid as _L          # noqa: E402

import numpy as np                         # noqa: E402
import joblib                              # noqa: E402

MAX_CHARS = 20000
MODEL_PATH = os.environ.get("AIDETECT_MODEL", os.path.join(_HERE, "ai_detector_model.joblib"))

_VERDICT_BANDS = [
    (0.85, "AI-generated"),
    (0.60, "Likely AI"),
    (0.40, "Mixed"),
    (0.15, "Likely human"),
    (0.00, "Human-written"),
]


def verdict_for(p: float) -> str:
    for t, label in _VERDICT_BANDS:
        if p >= t:
            return label
    return "Human-written"


# ---------------------------------------------------------------------------
# Detector wrapper that also returns character offsets for each paragraph so
# the frontend can highlight the exact source regions ("의심 영역").
# ---------------------------------------------------------------------------
class RMDetector:
    def __init__(self, model_path: str = MODEL_PATH):
        blob = joblib.load(model_path)
        self.pipe = blob["pipeline"]
        self.feature_order = blob["feature_order"]
        self.scaler = self.pipe.named_steps["scale"]
        self.clf = self.pipe.named_steps["clf"]

    def _score(self, feats):
        vec = np.array([[float(feats.get(k, 0.0)) for k in self.feature_order]])
        vec = np.nan_to_num(vec, nan=0.0, posinf=1e6, neginf=-1e6)
        prob = float(self.pipe.predict_proba(vec)[0, 1])
        z = self.scaler.transform(vec)[0]
        contrib = z * self.clf.coef_[0]
        idx = np.argsort(-np.abs(contrib))[:5]
        top = [{"feature": self.feature_order[i],
                "contribution": round(float(contrib[i]), 3),
                "direction": "AI" if contrib[i] > 0 else "human"} for i in idx]
        return prob, top

    @staticmethod
    def _paragraph_spans(text: str):
        """Yield (start, end, paragraph_text) preserving original char offsets.
        Splits on blank lines; falls back to single newlines."""
        # try blank-line blocks first
        spans = []
        import re
        # split keeping offsets: iterate over blocks separated by >=1 blank line
        pattern = re.compile(r"\n\s*\n")
        if pattern.search(text):
            pos = 0
            for m in pattern.finditer(text):
                block = text[pos:m.start()]
                if block.strip():
                    s = pos + (len(block) - len(block.lstrip()))
                    e = pos + len(block.rstrip())
                    spans.append((s, e, text[s:e]))
                pos = m.end()
            block = text[pos:]
            if block.strip():
                s = pos + (len(block) - len(block.lstrip()))
                e = pos + len(block.rstrip())
                spans.append((s, e, text[s:e]))
        else:
            # single-newline paragraphs
            pos = 0
            for line in text.split("\n"):
                if line.strip():
                    s = pos + (len(line) - len(line.lstrip()))
                    e = pos + len(line.rstrip())
                    spans.append((s, e, text[s:e]))
                pos += len(line) + 1
        if not spans and text.strip():
            s = len(text) - len(text.lstrip())
            e = len(text.rstrip())
            spans.append((s, e, text[s:e]))
        return spans

    def detect(self, text: str, lang: Optional[str] = None,
               para_min_chars: int = 30) -> dict:
        text = text or ""
        if not text.strip():
            raise ValueError("empty input")
        if lang is None:
            lang = _L.detect_lang(text)

        doc_feats = _F.full_features(text, lang=lang)
        doc_prob, doc_top = self._score(doc_feats)

        paragraphs = []
        flagged = 0
        for i, (s, e, ptext) in enumerate(self._paragraph_spans(text)):
            if len(ptext) < para_min_chars:
                plang = lang
                low_conf = True
            else:
                plang = _L.detect_lang(ptext)
                low_conf = False
            pf = _F.full_features(ptext, lang=plang)
            pprob, ptop = self._score(pf)
            is_flagged = (pprob >= 0.60) and not low_conf
            if is_flagged:
                flagged += 1
            paragraphs.append({
                "index": i,
                "start": s,
                "end": e,
                "text": ptext,
                "language": plang,
                "ai_probability": round(pprob, 4),
                "verdict": verdict_for(pprob),
                "low_confidence": low_conf,
                "flagged": is_flagged,
                "top_features": ptop,
            })

        return {
            "language": lang,
            "overall_ai_probability": round(doc_prob, 4),
            "verdict": verdict_for(doc_prob),
            "n_paragraphs": len(paragraphs),
            "n_flagged": flagged,
            "top_features": doc_top,
            "paragraphs": paragraphs,
        }


# ---------------------------------------------------------------------------
# FastAPI app
# ---------------------------------------------------------------------------
app = FastAPI(title="RM-DETECT", version="1.0.0",
              description="AI-vs-human text detector (perplexity + stylometry)")
app.add_middleware(
    CORSMiddleware, allow_origins=["*"], allow_methods=["*"], allow_headers=["*"],
)

_detector: Optional[RMDetector] = None
_load_error: Optional[str] = None
_started = time.time()


@app.on_event("startup")
def _load():
    global _detector, _load_error
    try:
        _detector = RMDetector()
    except Exception as exc:  # noqa: BLE001
        _load_error = f"{type(exc).__name__}: {exc}"


class DetectRequest(BaseModel):
    text: str = Field(..., description="text to analyze")
    lang: Optional[str] = Field(None, description="'ko' or 'en'; auto-detected if omitted")


@app.get("/health")
def health():
    return {
        "status": "ok" if _detector is not None else "error",
        "model_loaded": _detector is not None,
        "load_error": _load_error,
        "uptime_s": round(time.time() - _started, 1),
        "max_chars": MAX_CHARS,
    }


@app.post("/detect")
def detect(req: DetectRequest):
    if _detector is None:
        raise HTTPException(status_code=503,
                            detail=f"model not loaded: {_load_error}")
    text = req.text or ""
    if not text.strip():
        raise HTTPException(status_code=400, detail="text is empty")
    if len(text) > MAX_CHARS:
        raise HTTPException(status_code=413,
                            detail=f"text too long (>{MAX_CHARS} chars)")
    if req.lang not in (None, "ko", "en"):
        raise HTTPException(status_code=400, detail="lang must be 'ko' or 'en'")
    t0 = time.time()
    try:
        result = _detector.detect(text, lang=req.lang)
    except ValueError as exc:
        raise HTTPException(status_code=400, detail=str(exc))
    result["elapsed_ms"] = round((time.time() - t0) * 1000, 1)
    return result


# static UI (mounted last so /health and /detect take precedence)
_STATIC = os.path.join(_HERE, "static")


@app.get("/")
def index():
    idx = os.path.join(_STATIC, "index.html")
    if os.path.isfile(idx):
        return FileResponse(idx)
    raise HTTPException(status_code=404, detail="UI not built")


if os.path.isdir(_STATIC):
    app.mount("/static", StaticFiles(directory=_STATIC), name="static")