""" Web Dashboard Service Interactive UI for code review management and collaboration """ from fastapi import FastAPI, HTTPException, Request from fastapi.staticfiles import StaticFiles from fastapi.templating import Jinja2Templates from pydantic import BaseModel from typing import Dict, List, Optional, Any import json import datetime import hashlib from enum import Enum app = FastAPI(title="Dashboard Service", version="1.0.0") templates = Jinja2Templates(directory="templates") class DashboardService: def __init__(self): self.active_reviews = {} self.user_sessions = {} self.analytics_data = {} def get_dashboard_data(self, user_id: str) -> Dict[str, Any]: return { "user_id": user_id, "active_reviews": self.active_reviews.get(user_id, []), "recent_activity": self._get_recent_activity(user_id), "analytics": self._get_user_analytics(user_id), "notifications": self._get_notifications(user_id) } def _get_recent_activity(self, user_id: str) -> List[Dict]: return [ { "type": "review_completed", "message": "Review completed for PR #123", "timestamp": datetime.datetime.now().isoformat(), "repository": "example/repo" }, { "type": "new_comment", "message": "New AI comment on your pull request", "timestamp": (datetime.datetime.now() - datetime.timedelta(hours=2)).isoformat(), "repository": "example/another-repo" } ] def _get_user_analytics(self, user_id: str) -> Dict[str, Any]: return { "total_reviews": 42, "average_score": 85.5, "issues_fixed": 128, "productivity_gain": "23%" } def _get_notifications(self, user_id: str) -> List[Dict]: return [ { "type": "info", "title": "New feature available", "message": "AI-powered security scanning is now enabled", "timestamp": datetime.datetime.now().isoformat(), "read": False } ] dashboard_service = DashboardService() @app.get("/health") async def health(): return {"status": "healthy", "service": "dashboard"} @app.get("/dashboard/{user_id}") async def get_dashboard(request: Request, user_id: str): try: dashboard_data = dashboard_service.get_dashboard_data(user_id) return templates.TemplateResponse("dashboard.html", {"request": request, "data": dashboard_data}) except Exception as e: raise HTTPException(status_code=500, detail=f"Dashboard loading failed: {str(e)}") @app.get("/api/dashboard/{user_id}") async def get_dashboard_api(user_id: str): try: return dashboard_service.get_dashboard_data(user_id) except Exception as e: raise HTTPException(status_code=500, detail=f"Dashboard API failed: {str(e)}") if __name__ == "__main__": import uvicorn uvicorn.run(app, host="0.0.0.0", port=8005)