marvdeng/zero-gpt

1import os2import re3import sys4import logging5from fastapi import FastAPI, HTTPException6from pydantic import BaseModel7from transformers import pipeline8 9# 1. Configure the logging layout to stream cleanly to the console10logging.basicConfig(11    level=logging.INFO,12    format="%(asctime)s [%(levelname)s] %(name)s: %(message)s",13    handlers=[logging.StreamHandler(sys.stdout)]14)15logger = logging.getLogger("ai-detector-api")16 17# Force Hugging Face cache to live in writeable temporary directory18os.environ["HF_HOME"] = "/tmp/hf_cache"19 20logger.info("Initializing FastAPI Application Layout...")21app = FastAPI(title="Advanced AI Text Segment Classifier")22 23logger.info("Downloading/Loading Segmented RoBERTa AI Detection pipeline...")24try:25    detector = pipeline("text-classification", model="openai-community/roberta-base-openai-detector")26    logger.info("Model weights loaded into RAM successfully and ready for inference.")27except Exception as e:28    logger.critical(f"FAILED TO LOAD MODEL PIPELINE ON STARTUP: {str(e)}", exc_info=True)29    raise e30 31class TextPayload(BaseModel):32    text: str33 34def split_into_sentences(text: str):35    """36    Splits long body text into clean individual sentences 37    using regex to avoid breaking on abbreviations.38    """39    sentence_endings = re.compile(r'(?<!\w\.\w.)(?<![A-Z][a-z]\.)(?<=\.|\?)\s')40    sentences = sentence_endings.split(text.strip())41    return [s.strip() for s in sentences if s.strip()]42 43@app.get("/")44def home():45    logger.info("Health check ping received at root '/' endpoint.")46    return {"status": "healthy", "message": "Segmented AI Detector is operational."}47 48@app.post("/detect")49def detect_text(payload: TextPayload):50    raw_text = payload.text.strip()51    52    # Track metadata about the incoming payload size53    logger.info(f"Incoming payload received. Total raw characters: {len(raw_text)}")54 55    if not raw_text:56        logger.warning("Empty text payload submitted. Aborting with 400 Bad Request.")57        raise HTTPException(status_code=400, detail="Text cannot be empty.")58        59    try:60        # 1. Analyze the global block of text as a whole61        global_prediction = detector(raw_text)[0]62        global_is_ai = global_prediction["label"].lower() == "fake"63        64        # 2. Segment the text into sentences for fine-grained analysis65        sentences = split_into_sentences(raw_text)66        logger.info(f"Successfully segmented document into {len(sentences)} distinct sentences.")67        68        ai_segments = []69        ai_sentence_count = 070        71        # 3. Analyze each individual sentence segment72        for index, sentence in enumerate(sentences):73            sentence_pred = detector(sentence)[0]74            is_sentence_ai = sentence_pred["label"].lower() == "fake"75            confidence = round(sentence_pred["score"], 4)76            77            # Map "Fake" directly to an AI percentage representation (0.0 to 100.0)78            ai_percentage = round(confidence * 100, 2) if is_sentence_ai else round((1 - confidence) * 100, 2)79            80            # If the segment leans AI (over 50%), flag it81            if ai_percentage >= 50.0:82                ai_sentence_count += 183                ai_segments.append({84                    "sentence": sentence,85                    "ai_probability": ai_percentage86                })87                88        # 4. Calculate the percentage of the document composed of AI sentences89        total_sentences = len(sentences) if len(sentences) > 0 else 190        percentage_of_text_ai = round((ai_sentence_count / total_sentences) * 100, 2)91        92        # Fallback adjustment93        global_confidence = round(global_prediction["score"] * 100, 2) if global_is_ai else round((1 - global_prediction["score"]) * 100, 2)94 95        # Log flags for operational tracking96        if global_is_ai:97            logger.warning(f"AI Detected! Document Flagged as AI Generated. Score: {global_confidence}%")98        else:99            logger.info(f"Document labeled as Human Written. Score: {global_confidence}%")100 101        return {102            "document_assessment": {103                "verdict": "AI Generated" if global_is_ai else "Human Written",104                "global_ai_score_pct": global_confidence,105                "percentage_sentences_flagged_ai": percentage_of_text_ai106            },107            "total_sentences_analyzed": len(sentences),108            "identified_ai_segments": ai_segments109        }110        111    except Exception as e:112        # Logs the exact stack trace line number if things crash113        logger.error(f"Inference pipeline execution error occurred: {str(e)}", exc_info=True)114        raise HTTPException(status_code=500, detail="Internal analysis exception encountered.")