sync: pull latest backend from schemasense repo

- Activity tracking, chat caching, PDF export, noise reduction

Author: amitdevx

backend/main.py

@@ -4,8 +4,8 @@
 from fastapi.middleware.cors import CORSMiddleware
 from config import get_settings
 from utils.database import db
-from routes import tables, chat, chat_stream, export, auth, connection, dashboard, profile, settings as settings_routes, integrations, cache as cache_routes
-from utils.cache_db import init_cache
+from routes import tables, chat, chat_stream, export, auth, connection, dashboard, profile, settings as settings_routes, integrations, cache as cache_routes, activity
+from utils.cache_db import init_cache, init_chat_cache
 
 # Setup logging
 logging.basicConfig(
@@ -25,6 +25,7 @@ async def lifespan(app: FastAPI):
     logger.info("Starting SchemaSense Backend...")
     logger.info(f"Binding to {settings.API_HOST}:{settings.API_PORT}")
     await init_cache()
+    await init_chat_cache()
     logger.info("Backend ready! (Database connection will be established per user via /api/connect-db)")
 
     yield
@@ -67,6 +68,7 @@ async def lifespan(app: FastAPI):
 app.include_router(settings_routes.router)
 app.include_router(integrations.router)
 app.include_router(cache_routes.router)
+app.include_router(activity.router)
 
 # ===== Root Routes =====
 

backend/requirements.txt

@@ -10,3 +10,4 @@ bcrypt==4.1.1
 PyJWT==2.11.0
 python-dotenv==1.0.0
 pymssql>=2.2.8
+fpdf2>=2.7.0

backend/routes/activity.py

@@ -0,0 +1,15 @@
+from fastapi import APIRouter
+from typing import Optional
+from utils.activity import get_recent_activities
+import logging
+
+logger = logging.getLogger(__name__)
+
+router = APIRouter(prefix="/api", tags=["activity"])
+
+
+@router.get("/activity/recent")
+async def recent_activity(limit: Optional[int] = 20):
+    """Get recent activity feed"""
+    activities = get_recent_activities(limit=limit or 20)
+    return {"activities": activities, "total": len(activities)}

backend/routes/chat.py

@@ -4,6 +4,7 @@
 from schemas import ChatRequest, ChatResponse
 from routes.connection import get_user_db, get_db_pool, get_schema_filter_for
 from utils.deepseek_client import deepseek_client
+from utils.activity import log_activity, ActivityType
 import logging
 
 logger = logging.getLogger(__name__)
@@ -45,6 +46,13 @@ async def chat_with_schema(request: ChatRequestWithConnection):
             context=context
         )
 
+        log_activity(
+            ActivityType.CHAT_QUERY,
+            "Chat query",
+            f"Asked: {request.question[:80]}{'...' if len(request.question) > 80 else ''}",
+            {"question": request.question[:200]}
+        )
+        
         return ChatResponse(
             question=request.question,
             answer=answer,

backend/routes/chat_stream.py

@@ -1,9 +1,11 @@
 from fastapi import APIRouter, HTTPException, Request
 from fastapi.responses import StreamingResponse
 from schemas import ChatRequest
-from routes.connection import get_user_db, get_db_pool, get_schema_filter_for, get_db_type
+from routes.connection import get_user_db, get_db_pool, get_schema_filter_for, get_db_type, get_connection_details
 from utils.deepseek_client import deepseek_client
 from utils.schema_queries import get_count_query
+from utils.activity import log_activity, ActivityType
+from utils.cache_db import get_cached_chat_response, store_chat_response
 import logging
 import json
 from typing import Optional
@@ -26,6 +28,30 @@ async def chat_with_schema_streaming(request: ChatStreamRequest):
     try:
         db = get_db_pool(request.connection_id) if request.connection_id else get_user_db()
 
+        # Check cache first
+        conn_details = get_connection_details(request.connection_id)
+        if conn_details:
+            cached = await get_cached_chat_response(
+                host=conn_details["host"],
+                port=conn_details["port"],
+                database=conn_details["database"],
+                schema_filter=conn_details["schema_filter"],
+                user=conn_details["user"],
+                question=request.question,
+            )
+            if cached:
+                log_activity(
+                    ActivityType.CHAT_QUERY,
+                    "Chat query (cached)",
+                    f"Asked: {request.question[:80]}{'...' if len(request.question) > 80 else ''}",
+                    {"question": request.question[:200], "cached": True}
+                )
+                async def cached_response():
+                    yield json.dumps({"type": "status", "message": "⚡ Found cached response..."}) + "\n"
+                    yield json.dumps({"type": "content", "data": cached}) + "\n"
+                    yield json.dumps({"type": "done"}) + "\n"
+                return StreamingResponse(cached_response(), media_type="application/x-ndjson")
+        
         # Build comprehensive context from all tables
         async with db.acquire() as conn:
             tables_query = """
@@ -40,7 +66,6 @@ async def chat_with_schema_streaming(request: ChatStreamRequest):
             context = "DATABASE SCHEMA:\n"
             db_type = get_db_type(request.connection_id)
             for table_name in table_names:
-                # Get row count
                 count_query = get_count_query(table_name, db_type)
                 count_result = await conn.fetchval(count_query)
 
@@ -54,30 +79,49 @@ async def chat_with_schema_streaming(request: ChatStreamRequest):
 
         async def stream_response():
             """Generator that streams response chunks"""
-            # Send initial status that appears in blue box
+            log_activity(
+                ActivityType.CHAT_QUERY,
+                "Chat query (streaming)",
+                f"Asked: {request.question[:80]}{'...' if len(request.question) > 80 else ''}",
+                {"question": request.question[:200]}
+            )
             yield json.dumps({"type": "status", "message": "🤖 Connecting to AI..."}) + "\n"
             yield json.dumps({"type": "status", "message": "📊 Analyzing your database..."}) + "\n"
             yield json.dumps({"type": "status", "message": "🔍 Processing your question..."}) + "\n"
             yield json.dumps({"type": "status", "message": "⚡ Generating response..."}) + "\n"
 
             try:
-                # Stream the actual response content (without progress messages)
+                full_response = ""
                 async for chunk in deepseek_client.stream_chat_about_schema(
                     question=request.question,
                     context=context
                 ):
-                    # Send content chunks directly (no status messages mixed in)
+                    full_response += chunk
                     yield json.dumps({"type": "content", "data": chunk}) + "\n"
 
-                # Completion signal
+                # Store in cache for future requests
+                if conn_details and full_response:
+                    try:
+                        await store_chat_response(
+                            host=conn_details["host"],
+                            port=conn_details["port"],
+                            database=conn_details["database"],
+                            schema_filter=conn_details["schema_filter"],
+                            user=conn_details["user"],
+                            question=request.question,
+                            response=full_response,
+                        )
+                    except Exception as cache_err:
+                        logger.warning(f"Failed to cache chat response: {cache_err}")
+                
                 yield json.dumps({"type": "done"}) + "\n"
             except Exception as e:
                 logger.error(f"Streaming error: {e}")
                 yield json.dumps({"type": "error", "message": f"Error: {str(e)}"}) + "\n"
 
         return StreamingResponse(
             stream_response(),
-            media_type="application/x-ndjson"  # Newline-delimited JSON
+            media_type="application/x-ndjson"
         )
 
     except HTTPException:

backend/routes/connection.py

@@ -5,6 +5,7 @@
 import uuid
 from typing import Dict, Optional, List
 from datetime import datetime
+from utils.activity import log_activity, ActivityType
 
 logger = logging.getLogger(__name__)
 router = APIRouter(prefix="/api", tags=["connection"])
@@ -217,6 +218,13 @@ async def connect_database(request: ConnectionRequest, response: Response):
 
         logger.info(f"Connected [{conn_id}] ({request.database_type}): {request.host}:{request.port}/{request.database}")
 
+        log_activity(
+            ActivityType.DATABASE_CONNECTED,
+            f"Connected to {conn_name}",
+            f"Database {request.database} on {request.host}:{request.port} ({request.database_type})",
+            {"connection_id": conn_id, "database": request.database, "host": request.host}
+        )
+        
         # Set a cookie
         response.set_cookie(
             key="db_connected", value="true",
@@ -295,6 +303,12 @@ async def activate_connection(connection_id: str):
     active_connection_id = connection_id
     _sync_user_db_connection()
     logger.info(f"Activated connection: {connection_id}")
+    log_activity(
+        ActivityType.DATABASE_ACTIVATED,
+        f"Activated {connections[connection_id].get('name', connection_id)}",
+        f"Switched active database to {connections[connection_id]['database']}",
+        {"connection_id": connection_id}
+    )
     return {"success": True, "message": f"Connection '{connection_id}' is now active", "active_id": connection_id}
 
 @router.delete("/connections/{connection_id}")
@@ -322,6 +336,12 @@ async def remove_connection(connection_id: str):
         _sync_user_db_connection()
 
     logger.info(f"Removed connection: {connection_id}")
+    log_activity(
+        ActivityType.DATABASE_DISCONNECTED,
+        f"Disconnected database",
+        f"Removed database connection {connection_id}",
+        {"connection_id": connection_id}
+    )
     return {"success": True, "message": "Connection removed", "active_id": active_connection_id}
 
 @router.post("/disconnect-db")

backend/routes/dashboard.py

@@ -3,6 +3,7 @@
 from typing import List, Optional
 from routes import connection as conn_module
 from routes.connection import get_user_db, get_db_pool, get_schema_filter_for
+from utils.activity import get_analyses_count
 from pydantic import BaseModel
 import logging
 
@@ -116,7 +117,7 @@ async def get_statistics():
         return {
             "connectedDatabases": len(conn_module.connections),
             "totalTables": total_tables,
-            "analysesRun": len(conn_module.connections)
+            "analysesRun": get_analyses_count()
         }
     except HTTPException:
         raise

backend/routes/export.py

@@ -1,9 +1,11 @@
 from fastapi import APIRouter, HTTPException
+from fastapi.responses import Response
 from datetime import datetime
 from typing import Optional
 from schemas import ExportResponse
 from routes.connection import get_user_db, get_db_pool, get_schema_filter_for
 from routes.tables import get_schema, get_data_quality, explain_table
+from utils.pdf_generator import generate_table_pdf
 import logging
 
 logger = logging.getLogger(__name__)
@@ -68,3 +70,41 @@ async def export_documentation(table_name: str, connection_id: Optional[str] = N
     except Exception as e:
         logger.error(f"Error exporting {table_name}: {e}")
         raise HTTPException(status_code=500, detail=f"Export error: {str(e)}")
+
+
+@router.get("/export/{table_name}/pdf")
+async def export_pdf(table_name: str, connection_id: Optional[str] = None):
+    """Export table documentation as a PDF report (reuses cached AI analysis)"""
+    try:
+        db = get_db_pool(connection_id) if connection_id else get_user_db()
+        schema = get_schema_filter_for(connection_id)
+        async with db.acquire() as conn:
+            tables = await conn.fetch("""
+                SELECT table_name FROM information_schema.tables 
+                WHERE table_schema=$1 AND table_type='BASE TABLE'
+            """, schema)
+            table_names = [t['table_name'] for t in tables]
+            if table_name not in table_names:
+                raise HTTPException(status_code=404, detail=f"Table '{table_name}' not found")
+
+        schema_info = await get_schema(table_name, connection_id=connection_id)
+        quality = await get_data_quality(table_name, connection_id=connection_id)
+        explanation = await explain_table(table_name, connection_id=connection_id)
+
+        pdf_bytes = generate_table_pdf(
+            table_name=table_name,
+            schema_info=schema_info,
+            quality=quality,
+            business_context=explanation.business_explanation,
+        )
+
+        return Response(
+            content=pdf_bytes,
+            media_type="application/pdf",
+            headers={"Content-Disposition": f'attachment; filename="{table_name}_report.pdf"'},
+        )
+    except HTTPException:
+        raise
+    except Exception as e:
+        logger.error(f"Error generating PDF for {table_name}: {e}")
+        raise HTTPException(status_code=500, detail=f"PDF export error: {str(e)}")

backend/routes/tables.py

@@ -5,6 +5,7 @@
 from utils.deepseek_client import deepseek_client
 from utils.schema_queries import get_sample_query, get_count_query, get_count_alias_query, get_null_stats_query
 from utils.cache_db import get_cached_explanation, store_explanation
+from utils.activity import log_activity, ActivityType
 from datetime import datetime
 import logging
 
@@ -192,6 +193,13 @@ async def get_data_quality(table_name: str, connection_id: Optional[str] = None)
             overall_score=round(overall_score, 1)
         )
 
+        log_activity(
+            ActivityType.ANALYSIS_RUN,
+            f"Quality analysis: {table_name}",
+            f"Data quality score: {quality_grade} ({round(avg_completeness, 1)}% completeness)",
+            {"table_name": table_name, "grade": quality_grade}
+        )
+        
         return TableQuality(
             table_name=table_name,
             row_count=total_count,
@@ -271,6 +279,13 @@ async def explain_table(table_name: str, connection_id: Optional[str] = None):
                 row_count=schema_info.row_count,
             )
 
+        log_activity(
+            ActivityType.TABLE_EXPLAINED,
+            f"AI explained: {table_name}",
+            f"Generated AI business explanation for {table_name}",
+            {"table_name": table_name}
+        )
+        
         return TableExplanation(
             table_name=table_name,
             business_explanation=explanation,