#!/usr/bin/env python3 """Web demo: paste/upload a text, get human-vs-AI verdict and model attribution.""" import os import argparse import threading import time from pathlib import Path import joblib import numpy as np import pandas as pd import requests from flask import Flask, jsonify, request from scipy.sparse import hstack from huggingface_hub import hf_hub_download BASE = Path(__file__).parent UDPIPE_URL = "https://lindat.mff.cuni.cz/services/udpipe/api/process" REPO_ID = "milicka/ai-text-detector-models" # Updated to use just the filenames since they are at the root of your model repo LANGS = { "en": {"detector": "detector_brown.joblib", "attribution": "attribution_brown.joblib", "detector_local": "detector_brown_wordskip.joblib", "attribution_local": "attribution_brown_wordskip.joblib", "udpipe": "english-ewt-ud-2.17-251125"}, "cs": {"detector": "detector_koditex.joblib", "attribution": "attribution_koditex.joblib", "detector_local": "detector_koditex_wordskip.joblib", "attribution_local": "attribution_koditex_wordskip.joblib", "udpipe": "czech-pdtc-ud-2.17-251125"}, } MIN_TOKENS = 30 # refuse shorter inputs CHUNK = 200 CONLLU_COLS = {"words": 1, "lemmata": 2, "pos": 3, "TAG": 4, "FUN": 7} app = Flask(__name__) _bundles = {} _lock = threading.Lock() def get_bundles(lang, mode="udpipe"): """Lazy-load detector+attribution bundles for a language and mode from HF Hub.""" key = (lang, mode) suffix = "_local" if mode == "local" else "" with _lock: if key not in _bundles: cfg = LANGS[lang] print(f"loading {key} bundles from Hugging Face Hub ...", flush=True) # Načtení tokenu z tajných proměnných Space token = os.environ.get("for_models") # Přidání parametru token=token det_path = hf_hub_download(repo_id=REPO_ID, filename=cfg["detector" + suffix], token=token) attr_path = hf_hub_download(repo_id=REPO_ID, filename=cfg["attribution" + suffix], token=token) _bundles[key] = { "det": joblib.load(det_path), "attr": joblib.load(attr_path), } print(f"{key} bundles ready", flush=True) return _bundles[key] def detect_language(text): """Crude Czech/English heuristic based on characters and stopwords.""" czech_chars = sum(text.count(c) for c in "ěščřžýáíéůúďťň") if czech_chars / max(len(text), 1) > 0.005: return "cs" words = set(text.lower().split()) cs_hits = len(words & {"je", "se", "že", "na", "ale", "jako", "podle", "byl", "byla", "být", "jsou", "však"}) en_hits = len(words & {"the", "of", "and", "to", "is", "was", "that", "with", "for", "have"}) return "cs" if cs_hits > en_hits else "en" def udpipe_parse(text, model): resp = requests.post(UDPIPE_URL, data={ "model": model, "tokenizer": "", "tagger": "", "parser": "", "data": text}, timeout=300) resp.raise_for_status() return resp.json()["result"] def conllu_to_columns(conllu): cols = {c: [] for c in CONLLU_COLS} for line in conllu.splitlines(): if not line.strip() or line.startswith("#"): continue f = line.split("\t") if "-" in f[0] or "." in f[0]: continue for c, i in CONLLU_COLS.items(): cols[c].append(f[i]) return cols def make_chunks(cols, chunk_size): """Non-overlapping full chunks; the remainder is replaced by one full chunk anchored at the END of the text (overlapping the previous chunk), so every scored chunk has the full length. Texts shorter than one chunk yield a single short chunk.""" n = len(cols["words"]) if n < chunk_size: bounds, overlaps = [(0, n)], [False] else: bounds = [(s, s + chunk_size) for s in range(0, n - chunk_size + 1, chunk_size)] rem = n - len(bounds) * chunk_size overlaps = [False] * len(bounds) if rem >= MIN_TOKENS: bounds.append((n - chunk_size, n)) overlaps.append(True) chunks = [{c: " ".join(v[a:b]) for c, v in cols.items()} for a, b in bounds] lengths = [b - a for a, b in bounds] return chunks, lengths, overlaps def featurize(bundle, chunk_df): mats = [vec.transform(chunk_df[col]) for col, vec in bundle["vectorizers"]] return mats[0] if len(mats) == 1 else hstack(mats).tocsr() def anchored_probs(det, scores): """P(AI) re-anchored so that p=0.5 at the tuned decision threshold.""" A = float(det["platt"].coef_[0][0]) z = A * (scores - det["threshold"]) return 1.0 / (1.0 + np.exp(-z)) def attribute_nonhuman(attr, X): """Most likely non-human model line per chunk.""" clf = attr["classifier"] dec = clf.decision_function(X) if dec.ndim == 1: dec = np.stack([-dec, dec], axis=1) classes = np.asarray(clf.classes_) dec[:, classes == "human"] = -np.inf return classes[dec.argmax(axis=1)] @app.route("/") def index(): return _HTML, 200, {"Content-Type": "text/html; charset=utf-8"} @app.route("/info") def info(): return jsonify({"languages": {k: v["udpipe"] for k, v in LANGS.items()}, "loaded": list(_bundles), "chunk_size": CHUNK}) @app.route("/classify", methods=["POST"]) def classify(): data = request.get_json(silent=True) or {} text = str(data.get("text", "")).strip() lang = data.get("lang", "auto") mode = data.get("mode", "udpipe") if mode not in ("udpipe", "local"): mode = "udpipe" if not text: return jsonify({"error": "No text provided."}), 400 if lang not in ("en", "cs"): lang = detect_language(text) try: t0 = time.time() if mode == "udpipe": conllu = udpipe_parse(text, LANGS[lang]["udpipe"]) t_udpipe = time.time() - t0 cols = conllu_to_columns(conllu) else: # local mode: approximate tokenizer, surface features only from ud_tokenize import tokenize cols = {"words": tokenize(text, lang)} t_udpipe = 0.0 n_tokens = len(cols["words"]) if n_tokens < MIN_TOKENS: return jsonify({"error": f"Text too short: {n_tokens} tokens (need >= {MIN_TOKENS})."}), 400 chunks, lengths, overlaps = make_chunks(cols, CHUNK) chunk_df = pd.DataFrame(chunks) b = get_bundles(lang, mode) det, attr = b["det"], b["attr"] X = featurize(det, chunk_df) scores = det["classifier"].decision_function(X) probs = anchored_probs(det, scores) attributed = attribute_nonhuman(attr, X) chunk_out = [{ "idx": i, "n_tokens": lengths[i], "p_ai": round(float(probs[i]), 4), "attributed": str(attributed[i]), "text": chunks[i]["words"], "short": lengths[i] < CHUNK, "overlap": overlaps[i], } for i in range(len(chunks))] return jsonify({ "lang": lang, "mode": mode, "n_tokens": n_tokens, "n_chunks": len(chunks), "chunks": chunk_out, "seconds": {"udpipe": round(t_udpipe, 1), "total": round(time.time() - t0, 1)}, }) except requests.RequestException as exc: return jsonify({"error": f"UDPipe service error: {exc}"}), 502 except Exception as exc: return jsonify({"error": str(exc)}), 500 _HTML = r""" Human or LLM? — corpus text classifier

Human or LLM?

AI Brown / AI Koditex classifier · UDPipe 2 + sparse linear models

chunk: 200 tokens
Input text
0 characters
0.50
0.50 — neutral (the tuned operating point)
The text is sent to the LINDAT UDPipe service for annotation, cut into 200-token chunks and scored by classifiers trained on the AI Brown / AI Koditex corpora (19 chat models, 2024–2026). P(AI) is anchored so that 0.5 = the tuned decision point; the prior slider shifts it for contexts where false positives are costlier than false negatives (or vice versa). Verdicts for models newer than the training sample are less reliable; base-model (non-chat) text is out of scope.
🤔
Result will appear here
Verdict
?
mean P(AI)
0%
chunks flagged AI
0%
—language
—tokens
—chunks
—prior used
—UDPipe time
—total time
""" if __name__ == "__main__": ap = argparse.ArgumentParser() ap.add_argument("--port", type=int, default=8123) ap.add_argument("--host", default="127.0.0.1") ap.add_argument("--preload", action="store_true", help="load both language bundles at startup") args = ap.parse_args() if args.preload: for lang in LANGS: get_bundles(lang) print(f"Serving at http://{args.host}:{args.port}", flush=True) app.run(host=args.host, port=args.port, threaded=True)