File size: 6,947 Bytes
c2e02ba e106d7d 0f9f853 72f4cb5 e106d7d c2e02ba 0f9f853 1753da3 e106d7d 1753da3 e106d7d 1753da3 e106d7d 1753da3 e106d7d 1753da3 e106d7d 1753da3 e106d7d 1753da3 e106d7d 1753da3 e106d7d 1753da3 e106d7d 1753da3 e106d7d 0f9f853 c2e02ba 0f9f853 c2e02ba 0f9f853 c2e02ba 0f9f853 c2e02ba 0f9f853 c2e02ba 0f9f853 e106d7d 0e34a58 e106d7d 0e34a58 e106d7d 0e34a58 e106d7d 0e34a58 e106d7d |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 |
"""This module defines the /summary route for the Flask application."""
import importlib
from venv import logger
import re
from fastapi import APIRouter, HTTPException, Path
from fastapi.responses import JSONResponse
from controllers.summary import get_summary_data # pylint: disable=import-error
from models.database import knowledge_base # pylint: disable=import-error
router = APIRouter(prefix="/summary", tags=["summary"])
# @router.get("/{email}")
# async def get_user_document_stats(
# email: str = Path(..., description="User's email address")
# ) -> JSONResponse:
# """
# Get document statistics for a specific user.
# This endpoint counts the number of unique emails and files uploaded by the user.
# It groups documents by metadata.id to ensure unique document counting (not chunks).
# Args:
# email (str): The user's email address
# Returns:
# JSONResponse: A JSON response containing document counts:
# {
# "email": "[email protected]",
# "emails": 5,
# "files": 12,
# "total_documents": 17
# }
# Raises:
# HTTPException: 400 for invalid email, 500 for database errors
# """
# try:
# email_pattern = r'^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$'
# if not email or not re.match(email_pattern, email):
# raise HTTPException(
# status_code=400,
# detail="Invalid email format. Must be a valid email address."
# )
# # Get document counts from database
# result = knowledge_base.get_doc_count(email)
# # Add email to response
# result["email"] = email
# return JSONResponse(content=result, status_code=200)
# except ValueError as e:
# logger.error("Validation error for user %s: %s", email, str(e))
# raise HTTPException(status_code=400, detail=str(e)) # pylint: disable=raise-missing-from
# except Exception as e: # pylint: disable=broad-exception-caught
# logger.error("Database error for user %s: %s", email, str(e))
# raise HTTPException( # pylint: disable=raise-missing-from
# status_code=500,
# detail="Internal server error while retrieving document statistics."
# )
@router.get('')
async def get_summary() -> JSONResponse:
"""
Generate a summary dashboard with content flow and entity analysis data.
This endpoint provides a complete summary overview, including:
- Content Flow Tracker: Article counts by source and category
- Entity Analysis: Top entities by type with mentions
All data is returned together, divided into three time periods: today, week, and month.
Returns:
JSONResponse: A JSON response containing the complete summary dashboard data:
{
"content": {"today": {...}, "week": {...}, "month": {...}},
"entity": {"today": {...}, "week": {...}, "month": {...}}
}
"""
try:
summary_data = get_summary_data()
return JSONResponse(content=summary_data)
except Exception as e: #pylint: disable=broad-except
return JSONResponse(content={"error": str(e)}, status_code=500)
@router.get("/{module}/{chart_id}")
def get_summary_chart(module: str, chart_id: str) -> JSONResponse:
"""
Handles GET requests to the summary route with a specific module and chart ID.
Args:
module (str): The module identifier (content or entity).
chart_id (str): The chart identifier (today, weekly, monthly).
Returns:
tuple: The result of the chart's process function and HTTP status code 200.
Raises:
ImportError: If the specified chart module cannot be imported.
AttributeError: If the imported module does not have a 'process' function.
Endpoint:
GET /<module>/<chart_id>
"""
try:
result = importlib.import_module(f"controllers.summary.{module}.{chart_id}").process()
return JSONResponse(content=result)
except ImportError as e:
return JSONResponse(content={"error": str(e)}, status_code=404)
except AttributeError as e:
return JSONResponse(content={"error": str(e)}, status_code=500)
@router.get("/{module}")
async def get_summary_module(module: str) -> JSONResponse:
"""
Handles GET requests to the summary route for a specific module.
Triggers the process for each chart under this module concurrently.
Args:
module (str): The module identifier (content or entity).
Returns:
dict: {"module": module, "charts": [chart1, chart2, ...]}
"""
try:
result = importlib.import_module("controllers.summary").process(module)
return JSONResponse(content=result)
except ImportError as e:
return JSONResponse(content={"error": str(e)}, status_code=404)
except Exception as e: #pylint: disable=broad-except
return JSONResponse(content={"error": str(e)}, status_code=500)
# @router.get("/{email}")
# async def get_user_document_stats(
# email: str = Path(..., description="User's email address")
# ) -> JSONResponse:
# """
# Get document statistics for a specific user.
# This endpoint counts the number of unique emails and files uploaded by the user.
# It groups documents by metadata.id to ensure unique document counting (not chunks).
# Args:
# email (str): The user's email address
# Returns:
# JSONResponse: A JSON response containing document counts:
# {
# "email": "[email protected]",
# "emails": 5,
# "files": 12,
# "total_documents": 17
# }
# Raises:
# HTTPException: 400 for invalid email, 500 for database errors
# """
# try:
# email_pattern = r'^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$'
# if not email or not re.match(email_pattern, email):
# raise HTTPException(
# status_code=400,
# detail="Invalid email format. Must be a valid email address."
# )
# # Get document counts from database
# result = knowledge_base.get_doc_count(email)
# # Add email to response
# result["email"] = email
# return JSONResponse(content=result, status_code=200)
# except ValueError as e:
# logger.error("Validation error for user %s: %s", email, str(e))
# raise HTTPException(status_code=400, detail=str(e)) # pylint: disable=raise-missing-from
# except Exception as e: # pylint: disable=broad-exception-caught
# logger.error("Database error for user %s: %s", email, str(e))
# raise HTTPException( # pylint: disable=raise-missing-from
# status_code=500,
# detail="Internal server error while retrieving document statistics."
# )
|