"""GameGrip Safety API — FastAPI Application (Phases 1-3) Run: pip install -r requirements.txt python main.py Test: curl -X POST "http://localhost:8000/v1/moderate" \ -H "Content-Type: application/json" \ -d '{"text":"free skins click here noob","age_band":"under_13","platform":"trade","user_id":"u1","conversation_id":"c1","policy_pack":"child_strict"}' """ from fastapi import FastAPI, HTTPException, Query from typing import Optional, List from config import settings, AgeBand, Platform, Decision from models import ( ModerateRequest, ModerateResponse, ConversationHistoryResponse, AdminOverrideRequest, AdminOverrideResponse, ReviewQueueResponse, ReviewItem, PolicyPackListResponse, PolicyPackDetailResponse, ) from decision_engine import compute_moderation from policy_engine import policy_engine from conversation_store import conversation_store from review_queue import review_queue from datetime import datetime app = FastAPI( title=settings.app_name, description=""" Age-aware real-time content moderation for games and interactive platforms. **Phases Implemented:** - **Phase 1 (MVP):** Core moderation endpoint with text/image scoring - **Phase 2 (Gaming Context):** Conversation history, risk trends, platform context - **Phase 3 (Policy Engine):** Configurable policy packs, admin overrides, review queue """, version=settings.version, docs_url="/docs", redoc_url="/redoc", ) # ═══════════════════════════════════════════════ # Phase 1 — Core Moderation # ═══════════════════════════════════════════════ @app.post("/v1/moderate", response_model=ModerateResponse, tags=["Moderation"]) async def moderate(request: ModerateRequest): """ Primary moderation endpoint. Accepts text and/or image content, evaluates against the selected policy pack, applies age-aware scoring, and returns a decision. """ if not request.text and not request.image_url and not request.image_base64: raise HTTPException(status_code=400, detail="Provide at least one of: text, image_url, image_base64") return compute_moderation(request) @app.get("/health", tags=["System"]) async def health(): """Health check endpoint.""" return { "status": "ok", "version": settings.version, "phases": ["phase_1_mvp", "phase_2_gaming_context", "phase_3_policy_engine"], } # ═══════════════════════════════════════════════ # Phase 2 — Gaming Context & Conversation History # ═══════════════════════════════════════════════ @app.get("/v1/conversations/{conversation_id}", response_model=ConversationHistoryResponse, tags=["Gaming Context"]) async def get_conversation(conversation_id: str): """ Retrieve conversation history with risk trend analysis. Shows last N messages, average risk, and trend direction. """ summary = conversation_store.get_conversation_summary(conversation_id) if not summary: raise HTTPException(status_code=404, detail="Conversation not found") return summary @app.get("/v1/users/{user_id}/risk", tags=["Gaming Context"]) async def get_user_risk(user_id: str): """ Get aggregate risk score for a user across all conversations. Useful for account-level risk profiling. """ risk = conversation_store.get_user_risk_score(user_id) conv_ids = conversation_store.get_user_conversations(user_id) return { "user_id": user_id, "aggregate_risk": round(risk, 2), "conversation_count": len(conv_ids), "conversation_ids": conv_ids, } # ═══════════════════════════════════════════════ # Phase 3 — Policy Engine # ═══════════════════════════════════════════════ @app.get("/v1/policies", response_model=PolicyPackListResponse, tags=["Policy Engine"]) async def list_policies(): """List all available policy packs with summaries.""" packs = policy_engine.list_policies() return PolicyPackListResponse( packs=[{"name": k, **v} for k, v in packs.items()], default_pack=settings.default_policy, ) @app.get("/v1/policies/{policy_name}", response_model=PolicyPackDetailResponse, tags=["Policy Engine"]) async def get_policy(policy_name: str): """Get detailed configuration for a specific policy pack.""" policy = policy_engine.get_policy(policy_name) if policy.name != policy_name and policy_name not in policy_engine._packs: raise HTTPException(status_code=404, detail="Policy pack not found") return PolicyPackDetailResponse( name=policy.name, description=policy.description, config=policy.model_dump(), ) # ═══════════════════════════════════════════════ # Phase 3 — Review Queue # ═══════════════════════════════════════════════ @app.get("/v1/review-queue", response_model=ReviewQueueResponse, tags=["Review Queue"]) async def get_review_queue( status: Optional[str] = Query(None, description="Filter by status: pending, reviewed, all"), limit: int = Query(50, ge=1, le=200), ): """ Get items in the manual review queue. - `status=pending` — items awaiting admin review - `status=reviewed` — already resolved items - `status=all` or omit — everything """ if status == "pending": items = review_queue.list_pending() elif status == "reviewed": all_items = review_queue.list_all() items = [i for i in all_items if i.status == "reviewed"] else: items = review_queue.list_all() items = items[:limit] pending = len(review_queue.list_pending()) reviewed = len([i for i in review_queue.list_all() if i.status == "reviewed"]) return ReviewQueueResponse( items=items, total_pending=pending, total_reviewed=reviewed, ) @app.get("/v1/review-queue/{review_id}", response_model=ReviewItem, tags=["Review Queue"]) async def get_review_item(review_id: str): """Get a specific review queue item by ID.""" item = review_queue.get(review_id) if not item: raise HTTPException(status_code=404, detail="Review item not found") return item @app.post("/v1/review-queue/{review_id}/resolve", response_model=AdminOverrideResponse, tags=["Review Queue"]) async def resolve_review_item(review_id: str, override: AdminOverrideRequest): """ Resolve a review queue item with an admin decision override. This creates an audit trail of the original vs. new decision. """ if not settings.enable_admin_override: raise HTTPException(status_code=403, detail="Admin override is disabled") item = review_queue.get(review_id) if not item: raise HTTPException(status_code=404, detail="Review item not found") resolved = review_queue.resolve( review_id=review_id, new_decision=override.new_decision, admin_id=override.admin_id, ) if not resolved: raise HTTPException(status_code=400, detail="Could not resolve review item") return AdminOverrideResponse( review_id=review_id, original_decision=item.original_decision, new_decision=override.new_decision, admin_id=override.admin_id, timestamp=datetime.utcnow(), applied=True, ) @app.get("/v1/review-queue/stats", tags=["Review Queue"]) async def review_queue_stats(): """Get review queue statistics.""" return review_queue.stats() # ═══════════════════════════════════════════════ # Run # ═══════════════════════════════════════════════ if __name__ == "__main__": import uvicorn uvicorn.run(app, host="0.0.0.0", port=8000)