#!/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")