main.py

"""FastAPI backend for Dialectica."""

import os
import time
from collections import defaultdict, deque

from dotenv import load_dotenv
from fastapi import FastAPI, File, Form, Request, UploadFile
from fastapi.responses import FileResponse, JSONResponse
from pydantic import BaseModel

import config
from scripts import parsing
from scripts.expert import Expert
from scripts.inference import ORDERED_LABELS, CognitiveClassifier, ConceptMatcher

load_dotenv()

app = FastAPI(title="Dialectica")

VALID_PROVIDERS = ("deepseek", "gemini")

# Basic limits.
DAILY_CALL_CAP = 200          # expert-call cap per day
PER_IP_PER_MINUTE = 6         # max requests per IP per minute
MAX_QUESTION_CHARS = 2000     # max question length

# In-memory rate-limit state.
_ip_hits = defaultdict(deque)   # ip -> recent timestamps
_daily = {"day": None, "count": 0}

SESSION = {
    "material": "",
    "concepts": [],
    "coverage": {},   # concept -> {Surface: n, Mechanistic: n, Critical: n}
    "history": [],
}

COMPONENTS = {"classifier": None, "matcher": None, "experts": {}}


def get_classifier_and_matcher():
    """Get classifier and matcher (lazy init)."""
    product_config = config.ProductConfig()
    if COMPONENTS["classifier"] is None:
        COMPONENTS["classifier"] = CognitiveClassifier(product_config)
    if COMPONENTS["matcher"] is None:
        COMPONENTS["matcher"] = ConceptMatcher(product_config)
    return COMPONENTS["classifier"], COMPONENTS["matcher"]


def get_expert(provider):
    """Get cached expert for provider."""
    if provider not in VALID_PROVIDERS:
        provider = "deepseek"
    if provider not in COMPONENTS["experts"]:
        product_config = config.ProductConfig()
        product_config.provider = provider
        try:
            COMPONENTS["experts"][provider] = Expert(product_config)
        except SystemExit as error:
            return None, str(error)
        except Exception as error:
            return None, f"Could not start the {provider} expert: {error}"
    return COMPONENTS["experts"][provider], None


def _blank_coverage(concepts):
    """Make empty coverage map."""
    return {c: {level: 0 for level in ORDERED_LABELS} for c in concepts}


def _check_rate_limit(request):
    """Check per-IP rate limit."""
    client_ip = request.client.host if request.client else "unknown"
    now = time.time()
    hits = _ip_hits[client_ip]
    while hits and now - hits[0] > 60:
        hits.popleft()
    if len(hits) >= PER_IP_PER_MINUTE:
        return "You are sending questions too quickly. Wait a moment and retry."
    hits.append(now)
    return None


def _check_daily_cap():
    """Check daily usage cap."""
    today = time.strftime("%Y-%m-%d", time.gmtime())
    if _daily["day"] != today:
        _daily["day"] = today
        _daily["count"] = 0
    if _daily["count"] >= DAILY_CALL_CAP:
        return ("The demo has reached its daily usage limit. "
                "Please try again tomorrow.")
    return None


class AskRequest(BaseModel):
    """Request body for /api/ask."""

    question: str
    provider: str = "deepseek"


@app.get("/")
def index():
    """Serve frontend page."""
    return FileResponse(os.path.join("frontend", "index.html"))


@app.post("/api/upload")
async def upload(
    request: Request,
    file: UploadFile = File(None),
    text: str = Form(None),
    provider: str = Form("deepseek"),
):
    """Load material and extract concepts."""
    rate_error = _check_rate_limit(request)
    if rate_error:
        return JSONResponse(status_code=429, content={"error": rate_error})
    cap_error = _check_daily_cap()
    if cap_error:
        return JSONResponse(status_code=429, content={"error": cap_error})

    if file is not None:
        data = await file.read()
        try:
            raw = parsing.parse_material(file.filename, data)
        except ValueError as error:
            return JSONResponse(status_code=400, content={"error": str(error)})
    elif text:
        raw = text
    else:
        return JSONResponse(
            status_code=400,
            content={"error": "Upload a file or paste some text to begin."},
        )

    material = parsing.clean_text(raw)
    if len(material) < 40:
        return JSONResponse(
            status_code=400,
            content={"error": "That material is too short to work with."},
        )

    _, matcher = get_classifier_and_matcher()
    expert, error = get_expert(provider)
    if expert is None:
        return JSONResponse(status_code=400, content={"error": error})

    concepts = expert.extract_concepts(material)
    matcher.set_concepts(concepts)
    _daily["count"] += 1  # concept extraction counts as one call

    SESSION["material"] = material
    SESSION["concepts"] = concepts
    SESSION["coverage"] = _blank_coverage(concepts)
    SESSION["history"] = []

    return {"concepts": concepts, "coverage": SESSION["coverage"],
            "char_count": len(material), "provider": provider}


@app.post("/api/ask")
def ask(request: Request, body: AskRequest):
    """Answer one question and update coverage."""
    rate_error = _check_rate_limit(request)
    if rate_error:
        return JSONResponse(status_code=429, content={"error": rate_error})

    question = body.question.strip()
    if not question:
        return JSONResponse(status_code=400,
                            content={"error": "Ask the expert something."})
    if len(question) > MAX_QUESTION_CHARS:
        return JSONResponse(
            status_code=400,
            content={"error": "That question is too long. Please shorten it."},
        )
    if not SESSION["material"]:
        return JSONResponse(
            status_code=400,
            content={"error": "Load some material before you start asking."},
        )

    cap_error = _check_daily_cap()
    if cap_error:
        return JSONResponse(status_code=429, content={"error": cap_error})

    classifier, matcher = get_classifier_and_matcher()
    expert, error = get_expert(body.provider)
    if expert is None:
        return JSONResponse(status_code=400, content={"error": error})

    classification = classifier.classify(question)
    level = classification["level"]
    concepts = matcher.match(question)  # may be empty

    answer_text = expert.answer(question, SESSION["material"], SESSION["history"])
    _daily["count"] += 1

    SESSION["history"].append({"role": "student", "text": question})
    SESSION["history"].append({"role": "expert", "text": answer_text})

    for concept in concepts:
        if concept in SESSION["coverage"]:
            SESSION["coverage"][concept][level] += 1

    return {
        "answer": answer_text,
        "level": level,
        "confidence": classification["confidence"],
        "concepts": concepts,
        "coverage": SESSION["coverage"],
        "provider": body.provider,
    }


@app.get("/api/coverage")
def coverage():
    """Return current coverage state."""
    return {"concepts": SESSION["concepts"], "coverage": SESSION["coverage"]}