import sys import os sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), '..'))) import uvicorn from fastapi import FastAPI, HTTPException, Request from fastapi.responses import HTMLResponse from pydantic import BaseModel from typing import Optional from env.core import GDPRAuditorEnvironment from models import Action as ActionModel app = FastAPI(title="GDPR Auditor") env = GDPRAuditorEnvironment() HTML_CONTENT = """ GDPR Compliance Auditor

🔒 GDPR Compliance Auditor

OpenEnv environment — Audit privacy policies for GDPR/CCPA compliance violations

Reset Environment

Submit Finding

Output

Select a task and click Reset to begin...
""" @app.get("/", response_class=HTMLResponse) async def root(): """Serve the interactive web UI.""" return HTML_CONTENT @app.get("/json") async def api_info(): """API information and available endpoints.""" return { "name": "GDPR Compliance Auditor", "version": "1.0.0", "description": "OpenEnv environment for AI-powered GDPR/CCPA compliance auditing", "tasks": ["easy_clause_existence", "medium_purpose_mapping", "hard_dark_patterns", "elite_multi_doc_reasoning"], "endpoints": { "health": "GET /health", "tasks": "GET /tasks", "reset": "GET /reset?task= or POST /reset {task: }", "step": "POST /step {message: string}", "state": "GET /state", }, } @app.get("/tasks") async def list_tasks(): """Enumerate all available tasks (as declared in openenv.yaml).""" return [ {"id": "easy_clause_existence", "name": "Clause Existence Check", "difficulty": "easy", "max_steps": 8, "reward_range": [0.001, 0.999]}, {"id": "medium_purpose_mapping", "name": "Purpose Mapping", "difficulty": "medium", "max_steps": 8, "reward_range": [0.001, 0.999]}, {"id": "hard_dark_patterns", "name": "Dark Pattern Detection", "difficulty": "hard", "max_steps": 8, "reward_range": [0.001, 0.999]}, {"id": "elite_multi_doc_reasoning","name": "Multi-Document Reasoning", "difficulty": "elite", "max_steps": 8, "reward_range": [0.001, 0.999]}, ] class ActionRequest(BaseModel): message: str class ResetRequest(BaseModel): task: Optional[str] = "easy" @app.get("/health") async def health(): """Health check endpoint.""" return {"status": "ok"} @app.get("/reset") async def reset_get(task: str = "easy"): """Reset the environment for a given task (GET).""" try: obs = env.reset(task_id=task) return {"observation": obs.model_dump(), "done": False} except Exception as e: raise HTTPException(status_code=500, detail=str(e)) @app.post("/reset") async def reset_post(request: Request): """Reset the environment for a given task (POST).""" try: body = await request.json() task = body.get("task", "easy") if body else "easy" except Exception: task = "easy" try: obs = env.reset(task_id=task) return {"observation": obs.model_dump(), "done": False} except Exception as e: raise HTTPException(status_code=500, detail=str(e)) @app.post("/step") async def step(action: ActionRequest): """Submit an action (compliance finding) to the environment.""" try: action_obj = ActionModel(message=action.message) obs, reward, done, info = env.step(action_obj) return { "observation": obs.model_dump(), "reward": reward.model_dump(), "done": done, "info": info, } except Exception as e: raise HTTPException(status_code=500, detail=str(e)) @app.get("/state") async def state(): """Get the current environment state.""" return env.state() def main(): uvicorn.run("server.app:app", host="0.0.0.0", port=7860) if __name__ == "__main__": main()