#!/usr/bin/env python3 """ FastAPI REST API for Aumentum Browser Access Exposes endpoints for third-party applications and browser extensions to: - Query property documents - Convert documents to PDF for browser viewing - Retrieve document metadata """ # Load .env from project root before any config is read (fixes "using password: NO" when DB_PASSWORD is in .env) import os from pathlib import Path try: from dotenv import load_dotenv _root = Path(__file__).resolve().parents[2] if (_root / ".env").exists(): load_dotenv(_root / ".env") if (Path.cwd() / ".env").exists(): load_dotenv(Path.cwd() / ".env") except Exception: pass from fastapi import FastAPI, Query, HTTPException, File, UploadFile, Depends, status, Form, Request from fastapi import Path as FastAPIPath from fastapi.responses import FileResponse, StreamingResponse from fastapi.middleware.cors import CORSMiddleware from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm from pydantic import BaseModel from typing import Optional, List, Dict, Any from contextlib import asynccontextmanager from datetime import timedelta import logging import os import tempfile # Auth-related request logging (can be enabled to debug login/logout) logger = logging.getLogger("aumentum_api.auth") from backend.services.browser_service import AumentumBrowserService, DEFAULT_DB_CONFIG, DEFAULT_CONTENTSTORE_BASE # Backend auth (industry standard: short-lived access, refresh token, rate limiting) try: from backend.auth.service import AuthService from backend.auth.routes import router as auth_router from backend.auth.dependencies import get_current_user, require_auth, require_admin from backend.auth.schemas import Token, UserInfo, CreateUserRequest, CreateRoleRequest, UpdateUserRequest from backend.app.config import ACCESS_TOKEN_EXPIRE_MINUTES AUTH_AVAILABLE = True _USE_BACKEND_AUTH = True except ImportError: try: from backend.legacy_auth import AuthService, ACCESS_TOKEN_EXPIRE_MINUTES AUTH_AVAILABLE = True except ImportError: AuthService = None ACCESS_TOKEN_EXPIRE_MINUTES = 480 AUTH_AVAILABLE = False _USE_BACKEND_AUTH = False auth_router = None # TEMP_PDF_DIR = os.path.join(tempfile.gettempdir(), "aumentum_pdfs") # os.makedirs(TEMP_PDF_DIR, exist_ok=True) # Temporary directory for PDF cache TEMP_PDF_DIR = os.getenv( "TEMP_PDF_DIR", os.path.join(tempfile.gettempdir(), "aumentum_pdfs") ) os.makedirs(TEMP_PDF_DIR, exist_ok=True) def _configure_auth_logging(): """Ensure auth activity is logged for debugging login/logout. Logs to stdout and to logs/auth.log.""" if logger.handlers: return fmt = logging.Formatter("%(asctime)s [%(levelname)s] [AUTH] %(message)s") # Stdout (so it appears in api.log when process is started with redirect) stream = logging.StreamHandler() stream.setFormatter(fmt) logger.addHandler(stream) # Dedicated auth log file (so auth is always in one place) try: log_dir = os.path.join(os.path.dirname(os.path.abspath(__file__)), "logs") os.makedirs(log_dir, exist_ok=True) auth_log_path = os.path.join(log_dir, "auth.log") fh = logging.FileHandler(auth_log_path, encoding="utf-8") fh.setFormatter(fmt) logger.addHandler(fh) except Exception: pass # ignore if we can't write (e.g. read-only filesystem) logger.setLevel(logging.INFO) @asynccontextmanager async def _lifespan(app): _configure_auth_logging() yield app = FastAPI( title="Boundary Commission Administrative System API", description="API for Boundary Commission administrative software - managing interstate boundaries and land disputes", version="2.0.0", lifespan=_lifespan, ) # Include boundary commission router try: from backend.api.boundary import router as boundary_router app.include_router(boundary_router) except Exception as e: print(f"⚠️ Boundary commission API not available: {e}") # Include landing page content router (public GET + admin CRUD) try: from backend.api.landing import router as landing_router app.include_router(landing_router) except Exception as e: print(f"⚠️ Landing API not available: {e}") # Enable CORS (explicit origins when using credentials; * is invalid with allow_credentials=True) _cors_origins = [ "http://localhost:3000", "http://127.0.0.1:3000", "http://localhost:7000", "http://10.10.10.127:3000", "http://10.10.10.127:7000", ] app.add_middleware( CORSMiddleware, allow_origins=_cors_origins, allow_credentials=True, allow_methods=["GET", "POST", "PUT", "PATCH", "DELETE", "OPTIONS", "HEAD"], allow_headers=["*"], expose_headers=["*"], ) # Initialize services service = AumentumBrowserService( db_config=DEFAULT_DB_CONFIG, contentstore_base=DEFAULT_CONTENTSTORE_BASE ) # Initialize authentication service if AUTH_AVAILABLE: auth_service = AuthService(db_config=DEFAULT_DB_CONFIG) oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/auth/login") if _USE_BACKEND_AUTH: app.state.auth_service = auth_service app.include_router(auth_router) else: auth_service = None oauth2_scheme = None # =========================== # PYDANTIC MODELS (only when not using backend auth schemas) # =========================== class DocumentInfo(BaseModel): content_path: str file_exists: bool file_path: Optional[str] = None pdf_path: Optional[str] = None is_jpeg: bool error: Optional[str] = None class SchemaInfo(BaseModel): tables: List[str] property_tables: List[str] document_tables: List[str] doc_storage_schema: Optional[List] = None errors: List[str] class DocumentMetadata(BaseModel): document_number: str document_type: Optional[int] = None page_count: Optional[int] = None acceptance: Optional[str] = None recordation: Optional[str] = None submission: Optional[str] = None issued_by: Optional[str] = None comments: Optional[str] = None class PDFGenerationResult(BaseModel): document_number: str success: bool pdf_path: Optional[str] = None expected_pages: int actual_pages: int errors: List[str] class ContentURLInfo(BaseModel): node_id: int node_uuid: str content_url: str content_size: Optional[int] = None mime_type_str: Optional[str] = None if not _USE_BACKEND_AUTH: class Token(BaseModel): access_token: str token_type: str user: Optional[dict] = None class CreateUserRequest(BaseModel): """Request body for creating a new user (admin only).""" username: str password: str roles: List[str] = [] class CreateRoleRequest(BaseModel): """Request body for creating a new role (admin only).""" role_name: str class UserInfo(BaseModel): id: Optional[int] = None username: str roles: List[str] = [] groups: List[str] = [] is_admin: bool = False primary_role: Optional[str] = None first_name: Optional[str] = None last_name: Optional[str] = None full_name: Optional[str] = None email: Optional[str] = None job_title: Optional[str] = None organization: Optional[str] = None phone: Optional[str] = None mobile_phone: Optional[str] = None enabled: Optional[bool] = None # =========================== # API ENDPOINTS # =========================== @app.get("/") async def root(): """Root endpoint - API information.""" return { "service": "Aumentum Browser API", "version": "1.0.0", "endpoints": { "/schema": "Discover database schema", "/documents/lookup": "Lookup specific document", "/documents/pdf": "Get PDF for browser viewing", "/documents/by-document-number": "Resolve content by document_number", "/documents/pdf-by-document-number": "Stream PDF by document_number", "/documents/by-property": "Get documents by property", "/lrs/source-documents": "Recent LRS source documents (MSSQL)", "/lrs/content/recent": "Recent Alfresco content URLs (MSSQL)", "/documents/pdf-by-store-url": "Convert and stream PDF by store:// URL", "/health": "Health check" } } @app.get("/health") async def health_check(): """Health check endpoint.""" return {"status": "healthy", "service": "aumentum-api"} # =========================== # AUTHENTICATION HELPERS (used only when not using backend auth) # =========================== if not _USE_BACKEND_AUTH: async def get_current_user(token: Optional[str] = Depends(oauth2_scheme) if oauth2_scheme else None) -> Optional[UserInfo]: if not token: return None payload = auth_service.verify_token(token) if not payload: return None username = payload.get("sub") if not username: return None user = auth_service.authenticate_user(username, "", include_profile=False) if user: return UserInfo(**user) return None get_current_user_optional = get_current_user async def require_auth(current_user: UserInfo = Depends(get_current_user)) -> UserInfo: if not current_user: raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Not authenticated", headers={"WWW-Authenticate": "Bearer"}) return current_user async def require_admin(current_user: UserInfo = Depends(require_auth)) -> UserInfo: if not current_user.is_admin: raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Admin access required") return current_user if _USE_BACKEND_AUTH: get_current_user_optional = get_current_user # =========================== # AUTHENTICATION ENDPOINTS (login/me/logout served by backend auth router when _USE_BACKEND_AUTH) # =========================== if not _USE_BACKEND_AUTH: @app.post("/auth/login", response_model=Token) async def login(form_data: OAuth2PasswordRequestForm = Depends()): if not AUTH_AVAILABLE or not auth_service: raise HTTPException(status_code=status.HTTP_503_SERVICE_UNAVAILABLE, detail="Authentication service not available") logger.info("POST /auth/login: attempt username=%s", form_data.username) user = auth_service.authenticate_user(form_data.username, form_data.password, include_profile=False) if not user: logger.info("POST /auth/login: failed username=%s", form_data.username) raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Incorrect username or password", headers={"WWW-Authenticate": "Bearer"}) logger.info("POST /auth/login: success username=%s", form_data.username) access_token_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES) access_token = auth_service.create_access_token(data={"sub": user["username"]}, expires_delta=access_token_expires) return {"access_token": access_token, "token_type": "bearer", "user": user} @app.get("/auth/me", response_model=UserInfo) async def get_current_user_info(current_user: UserInfo = Depends(require_auth)): logger.info("GET /auth/me: returning user=%s", current_user.username) return current_user @app.post("/auth/logout") async def logout(current_user: UserInfo = Depends(require_auth), token: Optional[str] = Depends(oauth2_scheme)): logger.info("POST /auth/logout: username=%s", current_user.username) if token: auth_service.logout(token) return {"message": "Logged out successfully", "username": current_user.username, "session_ended": True} @app.get("/auth/users", response_model=List[UserInfo]) async def get_all_users(current_user: UserInfo = Depends(require_admin)): """ Get all users (admin only) Args: current_user: Current authenticated admin user Returns: List of all users """ users = auth_service.get_all_users() return [UserInfo(**user) for user in users] @app.post("/auth/users", status_code=status.HTTP_201_CREATED) async def create_user( body: CreateUserRequest, current_user: UserInfo = Depends(require_admin), ): """ Create a new user with the given username, password, and roles (admin only). """ if not AUTH_AVAILABLE or not auth_service: raise HTTPException( status_code=status.HTTP_503_SERVICE_UNAVAILABLE, detail="Authentication service not available", ) username = (body.username or "").strip() if not username: raise HTTPException(status_code=400, detail="Username is required") if not body.password: raise HTTPException(status_code=400, detail="Password is required") if len(body.password) < 6: raise HTTPException(status_code=400, detail="Password must be at least 6 characters") # Password length guardrail: bcrypt_sha256 supports arbitrary length, but we still cap to prevent abuse. if len(body.password.encode("utf-8")) > 4096: raise HTTPException(status_code=400, detail="Password is too long (max 4096 bytes)") conn = None cursor = None try: import zlib def calculate_crc(authority: str) -> int: return zlib.crc32(authority.encode("utf-8")) & 0xFFFFFFFF # Use auth_service DB config so /auth/users and /auth/login share the same DB in production. conn = auth_service._get_db_connection() cursor = conn.cursor() db_type = auth_service.db_config.get("type", "mssql").lower() # Ensure credentials table exists if db_type == "mysql": cursor.execute(""" CREATE TABLE IF NOT EXISTS plg_user_credentials ( username VARCHAR(255) PRIMARY KEY, password_hash VARCHAR(255) NOT NULL ) """) else: # T-SQL requires BEGIN/END for conditional CREATE TABLE. # Also scope the existence check to the intended schema. cursor.execute(""" IF NOT EXISTS ( SELECT 1 FROM sys.tables t JOIN sys.schemas s ON s.schema_id = t.schema_id WHERE t.name = 'plg_user_credentials' AND s.name = 'LRSAdmin' ) BEGIN CREATE TABLE LRSAdmin.plg_user_credentials ( username NVARCHAR(255) PRIMARY KEY, password_hash NVARCHAR(255) NOT NULL ) END """) # Check if user already exists if db_type == "mysql": cursor.execute( "SELECT id FROM alf_authority WHERE authority = %s AND authority NOT LIKE 'ROLE_%%' AND authority NOT LIKE 'GROUP_%%'", (username,), ) else: cursor.execute( "SELECT id FROM LRSAdmin.alf_authority WHERE authority = ? AND authority NOT LIKE 'ROLE_%' AND authority NOT LIKE 'GROUP_%'", (username,), ) if cursor.fetchone(): cursor.close() conn.close() raise HTTPException(status_code=409, detail=f"User '{username}' already exists") # Next ID for alf_authority if db_type == "mysql": cursor.execute("SELECT COALESCE(MAX(id), 0) + 1 FROM alf_authority") else: cursor.execute("SELECT ISNULL(MAX(id), 0) + 1 FROM LRSAdmin.alf_authority") user_id = cursor.fetchone()[0] crc = calculate_crc(username) # Insert user into alf_authority if db_type == "mysql": cursor.execute( "INSERT INTO alf_authority (id, version, authority, crc) VALUES (%s, 1, %s, %s)", (user_id, username, crc), ) else: cursor.execute( "INSERT INTO LRSAdmin.alf_authority (id, version, authority, crc) VALUES (?, 1, ?, ?)", (user_id, username, crc), ) # Store hashed password password_hash = auth_service.get_password_hash(body.password) if db_type == "mysql": cursor.execute( "INSERT INTO plg_user_credentials (username, password_hash) VALUES (%s, %s)", (username, password_hash), ) else: cursor.execute( "INSERT INTO LRSAdmin.plg_user_credentials (username, password_hash) VALUES (?, ?)", (username, password_hash), ) # Assign each role for role_name in body.roles or []: if not role_name or not isinstance(role_name, str): continue role_name = role_name.strip() if not role_name.startswith("ROLE_"): continue if db_type == "mysql": cursor.execute("SELECT id FROM alf_authority WHERE authority = %s", (role_name,)) else: cursor.execute("SELECT id FROM LRSAdmin.alf_authority WHERE authority = ?", (role_name,)) role_row = cursor.fetchone() if not role_row: continue role_id = role_row[0] if db_type == "mysql": cursor.execute("SELECT COALESCE(MAX(id), 0) + 1 FROM alf_authority_alias") else: cursor.execute("SELECT ISNULL(MAX(id), 0) + 1 FROM LRSAdmin.alf_authority_alias") alias_id = cursor.fetchone()[0] if db_type == "mysql": cursor.execute( "INSERT INTO alf_authority_alias (id, version, auth_id, alias_id) VALUES (%s, 1, %s, %s)", (alias_id, user_id, role_id), ) else: cursor.execute( "INSERT INTO LRSAdmin.alf_authority_alias (id, version, auth_id, alias_id) VALUES (?, 1, ?, ?)", (alias_id, user_id, role_id), ) conn.commit() cursor.close() conn.close() return {"message": f"User '{username}' created successfully", "username": username} except HTTPException: raise except Exception as e: logger.exception("POST /auth/users failed (username=%s)", (body.username or "").strip()) raise HTTPException(status_code=500, detail=f"Error creating user: {str(e)}") finally: if cursor is not None: try: cursor.close() except Exception: pass if conn is not None: try: conn.close() except Exception: pass @app.options("/auth/users/{username}") async def options_auth_users_username(username: str = FastAPIPath(...)): """CORS preflight for PATCH /auth/users/{username}. Must return 2xx.""" return {} @app.patch("/auth/users/{username}") async def update_user( username: str = FastAPIPath(...), body: UpdateUserRequest = None, current_user: UserInfo = Depends(require_admin), ): if body is None: body = UpdateUserRequest() """Update a user (admin only). Currently supports password reset.""" if not AUTH_AVAILABLE or not auth_service: raise HTTPException(status_code=status.HTTP_503_SERVICE_UNAVAILABLE, detail="Authentication service not available") username = username.strip() if not username: raise HTTPException(status_code=400, detail="Username is required") if not body or not body.password: raise HTTPException(status_code=400, detail="Password is required for update") if len(body.password) < 6: raise HTTPException(status_code=400, detail="Password must be at least 6 characters") # Password length guardrail: bcrypt_sha256 supports arbitrary length, but we still cap to prevent abuse. if len(body.password.encode("utf-8")) > 4096: raise HTTPException(status_code=400, detail="Password is too long (max 4096 bytes)") try: conn = auth_service._get_db_connection() cursor = conn.cursor() db_type = auth_service.db_config.get("type", "mssql").lower() if db_type == "mysql": cursor.execute("SELECT id FROM alf_authority WHERE authority = %s AND authority NOT LIKE 'ROLE_%%' AND authority NOT LIKE 'GROUP_%%'", (username,)) else: cursor.execute("SELECT id FROM LRSAdmin.alf_authority WHERE authority = ? AND authority NOT LIKE 'ROLE_%' AND authority NOT LIKE 'GROUP_%'", (username,)) if not cursor.fetchone(): cursor.close() conn.close() raise HTTPException(status_code=404, detail=f"User '{username}' not found") password_hash = auth_service.get_password_hash(body.password) if db_type == "mysql": cursor.execute( "INSERT INTO plg_user_credentials (username, password_hash) VALUES (%s, %s) ON DUPLICATE KEY UPDATE password_hash = VALUES(password_hash)", (username, password_hash), ) else: cursor.execute("UPDATE LRSAdmin.plg_user_credentials SET password_hash = ? WHERE username = ?", (password_hash, username)) if cursor.rowcount == 0: cursor.execute("INSERT INTO LRSAdmin.plg_user_credentials (username, password_hash) VALUES (?, ?)", (username, password_hash)) conn.commit() cursor.close() conn.close() return {"message": f"User '{username}' updated successfully"} except HTTPException: raise except Exception as e: raise HTTPException(status_code=500, detail=f"Error updating user: {str(e)}") @app.delete("/auth/users/{username}") async def delete_user( username: str = FastAPIPath(...), current_user: UserInfo = Depends(require_admin), ): """Delete a user (admin only). Removes from plg_user_credentials, alf_authority_alias, and alf_authority.""" if not AUTH_AVAILABLE or not auth_service: raise HTTPException(status_code=status.HTTP_503_SERVICE_UNAVAILABLE, detail="Authentication service not available") username = username.strip() if not username: raise HTTPException(status_code=400, detail="Username is required") try: conn = auth_service._get_db_connection() cursor = conn.cursor() db_type = auth_service.db_config.get("type", "mssql").lower() if db_type == "mysql": cursor.execute("SELECT id FROM alf_authority WHERE authority = %s AND authority NOT LIKE 'ROLE_%%' AND authority NOT LIKE 'GROUP_%%'", (username,)) else: cursor.execute("SELECT id FROM LRSAdmin.alf_authority WHERE authority = ? AND authority NOT LIKE 'ROLE_%' AND authority NOT LIKE 'GROUP_%'", (username,)) user_row = cursor.fetchone() if not user_row: cursor.close() conn.close() raise HTTPException(status_code=404, detail=f"User '{username}' not found") user_id = user_row[0] if db_type == "mysql": cursor.execute("DELETE FROM plg_user_credentials WHERE username = %s", (username,)) cursor.execute("DELETE FROM alf_authority_alias WHERE auth_id = %s", (user_id,)) cursor.execute("DELETE FROM alf_authority WHERE id = %s", (user_id,)) else: cursor.execute("DELETE FROM LRSAdmin.plg_user_credentials WHERE username = ?", (username,)) cursor.execute("DELETE FROM LRSAdmin.alf_authority_alias WHERE auth_id = ?", (user_id,)) cursor.execute("DELETE FROM LRSAdmin.alf_authority WHERE id = ?", (user_id,)) conn.commit() cursor.close() conn.close() return {"message": f"User '{username}' deleted successfully"} except HTTPException: raise except Exception as e: raise HTTPException(status_code=500, detail=f"Error deleting user: {str(e)}") @app.get("/auth/roles") async def get_all_roles(current_user: UserInfo = Depends(require_admin)): """ Get all roles (admin only) Args: current_user: Current authenticated admin user Returns: List of all roles """ return {"roles": auth_service.get_all_roles()} @app.post("/auth/roles", status_code=status.HTTP_201_CREATED) async def create_role( body: CreateRoleRequest, current_user: UserInfo = Depends(require_admin), ): """ Create a new role (admin only). Role name must start with ROLE_. """ if not AUTH_AVAILABLE or not auth_service: raise HTTPException( status_code=status.HTTP_503_SERVICE_UNAVAILABLE, detail="Authentication service not available", ) role_name = (body.role_name or "").strip() if not role_name: raise HTTPException(status_code=400, detail="Role name is required") if not role_name.startswith("ROLE_"): raise HTTPException( status_code=400, detail="Role name must start with ROLE_ (e.g. ROLE_CUSTOM_NAME)", ) try: import zlib def calculate_crc(authority: str) -> int: return zlib.crc32(authority.encode("utf-8")) & 0xFFFFFFFF # Use auth_service connection so we write to the same DB as get_all_roles (avoids # roles disappearing after logout/login when DEFAULT_DB_CONFIG differed from get_db_config()) conn = auth_service._get_db_connection() cursor = conn.cursor() db_type = auth_service.db_config.get("type", "mssql").lower() if db_type == "mysql": cursor.execute("SELECT id FROM alf_authority WHERE authority = %s", (role_name,)) else: cursor.execute("SELECT id FROM LRSAdmin.alf_authority WHERE authority = ?", (role_name,)) if cursor.fetchone(): cursor.close() conn.close() raise HTTPException(status_code=409, detail=f"Role '{role_name}' already exists") if db_type == "mysql": cursor.execute("SELECT COALESCE(MAX(id), 0) + 1 FROM alf_authority") else: cursor.execute("SELECT ISNULL(MAX(id), 0) + 1 FROM LRSAdmin.alf_authority") role_id = cursor.fetchone()[0] crc = calculate_crc(role_name) if db_type == "mysql": cursor.execute( "INSERT INTO alf_authority (id, version, authority, crc) VALUES (%s, 1, %s, %s)", (role_id, role_name, crc), ) else: cursor.execute( "INSERT INTO LRSAdmin.alf_authority (id, version, authority, crc) VALUES (?, 1, ?, ?)", (role_id, role_name, crc), ) conn.commit() cursor.close() conn.close() return {"message": f"Role '{role_name}' created successfully", "role_name": role_name} except HTTPException: raise except Exception as e: raise HTTPException(status_code=500, detail=f"Error creating role: {str(e)}") @app.get("/auth/groups") async def get_all_groups(current_user: UserInfo = Depends(require_admin)): """ Get all groups (admin only) Args: current_user: Current authenticated admin user Returns: List of all groups """ return {"groups": auth_service.get_all_groups()} @app.post("/auth/users/{username}/roles/{role_name}") async def assign_role_to_user( username: str = FastAPIPath(...), role_name: str = FastAPIPath(...), current_user: UserInfo = Depends(require_admin) ): """ Assign a role to a user (admin only) Args: username: Username role_name: Role name (e.g., 'ROLE_INDEXING_OFFICER') current_user: Current authenticated admin user Returns: Success message """ if not AUTH_AVAILABLE or not auth_service: raise HTTPException( status_code=status.HTTP_503_SERVICE_UNAVAILABLE, detail="Authentication service not available" ) conn = None cursor = None try: # Use auth_service DB config so role changes persist and don't depend on browser_service defaults. conn = auth_service._get_db_connection() cursor = conn.cursor() db_type = auth_service.db_config.get("type", "mssql").lower() # Get user ID if db_type == "mysql": cursor.execute("SELECT id FROM alf_authority WHERE authority = %s AND authority NOT LIKE 'ROLE_%%' AND authority NOT LIKE 'GROUP_%%'", (username,)) else: cursor.execute("SELECT id FROM LRSAdmin.alf_authority WHERE authority = ? AND authority NOT LIKE 'ROLE_%' AND authority NOT LIKE 'GROUP_%'", (username,)) user_row = cursor.fetchone() if not user_row: cursor.close() conn.close() raise HTTPException(status_code=404, detail=f"User '{username}' not found") user_id = user_row[0] # Get role ID if db_type == "mysql": cursor.execute("SELECT id FROM alf_authority WHERE authority = %s", (role_name,)) else: cursor.execute("SELECT id FROM LRSAdmin.alf_authority WHERE authority = ?", (role_name,)) role_row = cursor.fetchone() if not role_row: cursor.close() conn.close() raise HTTPException(status_code=404, detail=f"Role '{role_name}' not found") role_id = role_row[0] # Check if link already exists if db_type == "mysql": cursor.execute(""" SELECT id FROM alf_authority_alias WHERE auth_id = %s AND alias_id = %s """, (user_id, role_id)) else: cursor.execute(""" SELECT id FROM LRSAdmin.alf_authority_alias WHERE auth_id = ? AND alias_id = ? """, (user_id, role_id)) if cursor.fetchone(): cursor.close() conn.close() return {"message": f"User '{username}' already has role '{role_name}'"} # Get next alias ID if db_type == "mysql": cursor.execute("SELECT MAX(id) FROM alf_authority_alias") else: cursor.execute("SELECT MAX(id) FROM LRSAdmin.alf_authority_alias") max_alias_id = cursor.fetchone()[0] alias_id = (max_alias_id or 0) + 1 # Create link if db_type == "mysql": cursor.execute(""" INSERT INTO alf_authority_alias (id, version, auth_id, alias_id) VALUES (%s, 1, %s, %s) """, (alias_id, user_id, role_id)) else: cursor.execute(""" INSERT INTO LRSAdmin.alf_authority_alias (id, version, auth_id, alias_id) VALUES (?, 1, ?, ?) """, (alias_id, user_id, role_id)) conn.commit() cursor.close() conn.close() return {"message": f"Role '{role_name}' assigned to user '{username}' successfully"} except HTTPException: raise except Exception as e: raise HTTPException(status_code=500, detail=f"Error assigning role: {str(e)}") finally: if cursor is not None: try: cursor.close() except Exception: pass if conn is not None: try: conn.close() except Exception: pass @app.delete("/auth/users/{username}/roles/{role_name}") async def remove_role_from_user( username: str = FastAPIPath(...), role_name: str = FastAPIPath(...), current_user: UserInfo = Depends(require_admin) ): """ Remove a role from a user (admin only) Args: username: Username role_name: Role name to remove current_user: Current authenticated admin user Returns: Success message """ if not AUTH_AVAILABLE or not auth_service: raise HTTPException( status_code=status.HTTP_503_SERVICE_UNAVAILABLE, detail="Authentication service not available" ) conn = None cursor = None try: # Use auth_service DB config so role changes persist and don't depend on browser_service defaults. conn = auth_service._get_db_connection() cursor = conn.cursor() db_type = auth_service.db_config.get("type", "mssql").lower() # Get user ID if db_type == "mysql": cursor.execute("SELECT id FROM alf_authority WHERE authority = %s AND authority NOT LIKE 'ROLE_%%' AND authority NOT LIKE 'GROUP_%%'", (username,)) else: cursor.execute("SELECT id FROM LRSAdmin.alf_authority WHERE authority = ? AND authority NOT LIKE 'ROLE_%' AND authority NOT LIKE 'GROUP_%'", (username,)) user_row = cursor.fetchone() if not user_row: cursor.close() conn.close() raise HTTPException(status_code=404, detail=f"User '{username}' not found") user_id = user_row[0] # Get role ID if db_type == "mysql": cursor.execute("SELECT id FROM alf_authority WHERE authority = %s", (role_name,)) else: cursor.execute("SELECT id FROM LRSAdmin.alf_authority WHERE authority = ?", (role_name,)) role_row = cursor.fetchone() if not role_row: cursor.close() conn.close() raise HTTPException(status_code=404, detail=f"Role '{role_name}' not found") role_id = role_row[0] # Delete link if db_type == "mysql": cursor.execute(""" DELETE FROM alf_authority_alias WHERE auth_id = %s AND alias_id = %s """, (user_id, role_id)) else: cursor.execute(""" DELETE FROM LRSAdmin.alf_authority_alias WHERE auth_id = ? AND alias_id = ? """, (user_id, role_id)) if cursor.rowcount == 0: cursor.close() conn.close() return {"message": f"User '{username}' does not have role '{role_name}'"} conn.commit() cursor.close() conn.close() return {"message": f"Role '{role_name}' removed from user '{username}' successfully"} except HTTPException: raise except Exception as e: raise HTTPException(status_code=500, detail=f"Error removing role: {str(e)}") finally: if cursor is not None: try: cursor.close() except Exception: pass if conn is not None: try: conn.close() except Exception: pass @app.get("/dictionary/document-types") async def get_document_types(): """ Get document type labels from lr_dictionary. Returns a list of document types with IDs and labels. """ try: conn = service._get_db_connection() cursor = conn.cursor() db_type = service.db_config.get("type", "mssql").lower() if db_type == "mysql": cursor.execute(""" SELECT Id, code, label, descr FROM lr_dictionary WHERE category = 'document_type' ORDER BY label """) else: cursor.execute(""" SELECT Id, code, label, descr FROM LRSAdmin.lr_dictionary WHERE category = 'document_type' ORDER BY label """) types = [] for id_val, code, label, descr in cursor.fetchall(): types.append({ "id": int(id_val), "code": code or "", "label": label or "", "description": descr or "", "category": "document_type" }) cursor.close() conn.close() return types except Exception as e: raise HTTPException( status_code=500, detail=f"Error fetching document types: {str(e)}" ) @app.get("/dictionary/party-types") async def get_party_types(): """ Get all party types from lr_dictionary. Returns list of party types (Individual, Company, etc.) with their IDs and labels. """ try: conn = service._get_db_connection() cursor = conn.cursor() db_type = service.db_config.get("type", "mssql").lower() if db_type == "mysql": cursor.execute(""" SELECT Id, code, label, descr FROM lr_dictionary WHERE category = 'party_type' ORDER BY label """) else: cursor.execute(""" SELECT Id, code, label, descr FROM LRSAdmin.lr_dictionary WHERE category = 'party_type' ORDER BY label """) types = [] for row in cursor.fetchall(): types.append({ "id": int(row[0]), "code": row[1] or "", "label": row[2] or "", "description": row[3] or "", "category": "party_type" }) cursor.close() conn.close() return types except Exception as e: raise HTTPException(status_code=500, detail=f"Database error: {str(e)}") @app.get("/dictionary/legal-roles") async def get_legal_roles(): """ Get all legal roles (party role types) from lr_dictionary. Returns list of roles (Grantee, Grantor, etc.) with their IDs and labels. """ try: conn = service._get_db_connection() cursor = conn.cursor() db_type = service.db_config.get("type", "mssql").lower() if db_type == "mysql": cursor.execute(""" SELECT Id, code, label, descr FROM lr_dictionary WHERE category = 'party_role_type' ORDER BY label """) else: cursor.execute(""" SELECT Id, code, label, descr FROM LRSAdmin.lr_dictionary WHERE category = 'party_role_type' ORDER BY label """) roles = [] for row in cursor.fetchall(): roles.append({ "id": int(row[0]), "code": row[1] or "", "label": row[2] or "", "description": row[3] or "", "category": "party_role_type" }) cursor.close() conn.close() return roles except Exception as e: raise HTTPException(status_code=500, detail=f"Database error: {str(e)}") @app.get("/dictionary/genders") async def get_genders(): """ Get all gender types from lr_dictionary. Returns list of genders (Male, Female) with their IDs and labels. """ try: conn = service._get_db_connection() cursor = conn.cursor() db_type = service.db_config.get("type", "mssql").lower() if db_type == "mysql": cursor.execute(""" SELECT Id, code, label, descr FROM lr_dictionary WHERE category = 'gender' ORDER BY label """) else: cursor.execute(""" SELECT Id, code, label, descr FROM LRSAdmin.lr_dictionary WHERE category = 'gender' ORDER BY label """) genders = [] for row in cursor.fetchall(): genders.append({ "id": int(row[0]), "code": row[1] or "", "label": row[2] or "", "description": row[3] or "", "category": "gender" }) cursor.close() conn.close() return genders except Exception as e: raise HTTPException(status_code=500, detail=f"Database error: {str(e)}") @app.get("/dictionary/transaction-types") async def get_transaction_types(): """ Get all transaction types from lr_dictionary. Returns list of transaction types with their IDs and labels. """ try: conn = service._get_db_connection() cursor = conn.cursor() db_type = service.db_config.get("type", "mssql").lower() if db_type == "mysql": cursor.execute(""" SELECT Id, code, label, descr FROM lr_dictionary WHERE category = 'event_type' ORDER BY label """) else: cursor.execute(""" SELECT Id, code, label, descr FROM LRSAdmin.lr_dictionary WHERE category = 'event_type' ORDER BY label """) types = [] for row in cursor.fetchall(): types.append({ "id": int(row[0]), "code": row[1] or "", "label": row[2] or "", "description": row[3] or "", "category": "event_type" }) cursor.close() conn.close() return types except Exception as e: raise HTTPException(status_code=500, detail=f"Database error: {str(e)}") @app.get("/dictionary/transaction-statuses") async def get_transaction_statuses(): """ Get all transaction statuses from lr_dictionary. Returns list of transaction statuses with their IDs and labels. """ try: conn = service._get_db_connection() cursor = conn.cursor() db_type = service.db_config.get("type", "mssql").lower() if db_type == "mysql": cursor.execute(""" SELECT Id, code, label, descr FROM lr_dictionary WHERE category = 'transaction_status' ORDER BY label """) else: cursor.execute(""" SELECT Id, code, label, descr FROM LRSAdmin.lr_dictionary WHERE category = 'transaction_status' ORDER BY label """) statuses = [] for row in cursor.fetchall(): statuses.append({ "id": int(row[0]), "code": row[1] or "", "label": row[2] or "", "description": row[3] or "", "category": "transaction_status" }) cursor.close() conn.close() return statuses except Exception as e: raise HTTPException(status_code=500, detail=f"Database error: {str(e)}") @app.get("/dictionary/property-types") async def get_property_types(): """ Get all property types (spatial unit types) from lr_dictionary. Returns list of property types with their IDs and labels. """ try: conn = service._get_db_connection() cursor = conn.cursor() db_type = service.db_config.get("type", "mssql").lower() if db_type == "mysql": cursor.execute(""" SELECT Id, code, label, descr FROM lr_dictionary WHERE category = 'spatial_unit_type' ORDER BY label """) else: cursor.execute(""" SELECT Id, code, label, descr FROM LRSAdmin.lr_dictionary WHERE category = 'spatial_unit_type' ORDER BY label """) types = [] for row in cursor.fetchall(): types.append({ "id": int(row[0]), "code": row[1] or "", "label": row[2] or "", "description": row[3] or "", "category": "spatial_unit_type" }) cursor.close() conn.close() return types except Exception as e: raise HTTPException(status_code=500, detail=f"Database error: {str(e)}") @app.get("/dictionary/spatial-unit-types") async def get_spatial_unit_types(): """Get spatial unit types from dictionary (alias for property-types).""" return await get_property_types() @app.get("/dictionary/tenure-types") async def get_tenure_types(): """Get tenure types from dictionary.""" conn = service._get_db_connection() cursor = conn.cursor() db_type = service.db_config.get("type", "mssql").lower() try: if db_type == "mysql": query = "SELECT Id, code, label, descr, category FROM lr_dictionary WHERE category = 'tenure_type' ORDER BY sort_order, label" else: query = "SELECT Id, code, label, descr, category FROM LRSAdmin.lr_dictionary WHERE category = 'tenure_type' ORDER BY sort_order, label" cursor.execute(query) rows = cursor.fetchall() result = [{"id": int(r[0]), "code": r[1] or "", "label": r[2] or "", "description": r[3] or "", "category": r[4] or ""} for r in rows] cursor.close() conn.close() return result except Exception as e: cursor.close() conn.close() raise HTTPException(status_code=500, detail=f"Error fetching tenure types: {str(e)}") @app.get("/dictionary/purposes") async def get_purposes(): """Get property purposes from dictionary.""" conn = service._get_db_connection() cursor = conn.cursor() db_type = service.db_config.get("type", "mssql").lower() try: if db_type == "mysql": query = "SELECT Id, code, label, descr, category FROM lr_dictionary WHERE category = 'purpose' ORDER BY sort_order, label" else: query = "SELECT Id, code, label, descr, category FROM LRSAdmin.lr_dictionary WHERE category = 'purpose' ORDER BY sort_order, label" cursor.execute(query) rows = cursor.fetchall() result = [{"id": int(r[0]), "code": r[1] or "", "label": r[2] or "", "description": r[3] or "", "category": r[4] or ""} for r in rows] cursor.close() conn.close() return result except Exception as e: cursor.close() conn.close() raise HTTPException(status_code=500, detail=f"Error fetching purposes: {str(e)}") @app.get("/dictionary/zones") async def get_zones(): """Get zones from dictionary.""" conn = service._get_db_connection() cursor = conn.cursor() db_type = service.db_config.get("type", "mssql").lower() try: if db_type == "mysql": query = "SELECT Id, code, label, descr, category FROM lr_dictionary WHERE category = 'zone' ORDER BY sort_order, label" else: query = "SELECT Id, code, label, descr, category FROM LRSAdmin.lr_dictionary WHERE category = 'zone' ORDER BY sort_order, label" cursor.execute(query) rows = cursor.fetchall() result = [{"id": int(r[0]), "code": r[1] or "", "label": r[2] or "", "description": r[3] or "", "category": r[4] or ""} for r in rows] cursor.close() conn.close() return result except Exception as e: cursor.close() conn.close() raise HTTPException(status_code=500, detail=f"Error fetching zones: {str(e)}") @app.get("/dictionary/layouts") async def get_layouts(): """Get layouts from dictionary.""" conn = service._get_db_connection() cursor = conn.cursor() db_type = service.db_config.get("type", "mssql").lower() try: if db_type == "mysql": query = "SELECT Id, code, label, descr, category FROM lr_dictionary WHERE category = 'layout' ORDER BY sort_order, label" else: query = "SELECT Id, code, label, descr, category FROM LRSAdmin.lr_dictionary WHERE category = 'layout' ORDER BY sort_order, label" cursor.execute(query) rows = cursor.fetchall() result = [{"id": int(r[0]), "code": r[1] or "", "label": r[2] or "", "description": r[3] or "", "category": r[4] or ""} for r in rows] cursor.close() conn.close() return result except Exception as e: cursor.close() conn.close() raise HTTPException(status_code=500, detail=f"Error fetching layouts: {str(e)}") @app.get("/dictionary/area-unit-types") async def get_area_unit_types(): """Get area unit types from dictionary.""" conn = service._get_db_connection() cursor = conn.cursor() db_type = service.db_config.get("type", "mssql").lower() try: if db_type == "mysql": query = "SELECT Id, code, label, descr, category FROM lr_dictionary WHERE category = 'area_unit_type' ORDER BY sort_order, label" else: query = "SELECT Id, code, label, descr, category FROM LRSAdmin.lr_dictionary WHERE category = 'area_unit_type' ORDER BY sort_order, label" cursor.execute(query) rows = cursor.fetchall() result = [{"id": int(r[0]), "code": r[1] or "", "label": r[2] or "", "description": r[3] or "", "category": r[4] or ""} for r in rows] cursor.close() conn.close() return result except Exception as e: cursor.close() conn.close() raise HTTPException(status_code=500, detail=f"Error fetching area unit types: {str(e)}") @app.get("/dictionary/statuses") async def get_statuses(): """Get statuses from dictionary (for properties).""" conn = service._get_db_connection() cursor = conn.cursor() db_type = service.db_config.get("type", "mssql").lower() try: if db_type == "mysql": query = "SELECT Id, code, label, descr, category FROM lr_dictionary WHERE category = 'status' ORDER BY sort_order, label LIMIT 100" else: query = "SELECT TOP 100 Id, code, label, descr, category FROM LRSAdmin.lr_dictionary WHERE category = 'status' ORDER BY sort_order, label" cursor.execute(query) rows = cursor.fetchall() result = [{"id": int(r[0]), "code": r[1] or "", "label": r[2] or "", "description": r[3] or "", "category": r[4] or ""} for r in rows] cursor.close() conn.close() return result except Exception as e: cursor.close() conn.close() raise HTTPException(status_code=500, detail=f"Error fetching statuses: {str(e)}") @app.get("/schema", response_model=SchemaInfo) async def get_schema(): """ Discover and return Aumentum database schema. Useful for understanding property/document relationships. """ schema = service.discover_database_schema() return SchemaInfo(**schema) @app.get("/documents/lookup", response_model=DocumentInfo) async def lookup_document( content_path: str = Query(..., description="Content path from database"), convert_to_pdf: bool = Query(True, description="Convert to PDF for browser viewing") ): """ Lookup a document by its content path. Args: content_path: Relative path from database (e.g., "2014/11/03/file.bin") convert_to_pdf: If True, converts .bin to PDF Returns: Document information including PDF path """ result = service.lookup_document(content_path, convert_to_pdf) return DocumentInfo(**result) @app.get("/lrs/source-documents") async def get_lrs_source_documents(limit: int = Query(20, ge=1, le=200)): """Return recent rows from LRSAdmin.lr_source_document (MSSQL only).""" items = service.list_recent_lr_source_documents(limit=limit) return {"count": len(items), "items": items} @app.get("/lrs/content/recent") async def get_lrs_recent_content(limit: int = Query(20, ge=1, le=200)): """Return recent rows from LRSAdmin Alfresco content tables (MSSQL only).""" items = service.list_recent_alf_content(limit=limit) return {"count": len(items), "items": items} @app.get("/documents/pdf-by-store-url") async def pdf_by_store_url(store_url: str = Query(..., description="store://YYYY/M/D/H/M/UUID.bin")): """Convert a store:// URL to PDF and stream it.""" res = service.fetch_pdf_by_store_url(store_url) if res.get("error"): raise HTTPException(status_code=400, detail=res["error"]) if not res.get("pdf_path"): raise HTTPException(status_code=500, detail="PDF not generated") return FileResponse( res["pdf_path"], media_type="application/pdf", filename=os.path.basename(res["pdf_path"]) ) @app.get("/documents/pdf") async def get_pdf_stream( content_path: str = Query(..., description="Content path from database") ): """ Stream PDF file directly to browser. Args: content_path: Relative path from database Returns: PDF file stream """ result = service.lookup_document(content_path, convert_to_pdf=True) if not result.get("pdf_path") or not os.path.exists(result["pdf_path"]): raise HTTPException( status_code=404, detail=f"PDF not found for: {content_path}" ) return FileResponse( result["pdf_path"], media_type="application/pdf", filename=os.path.basename(content_path).replace(".bin", ".pdf") ) # @app.get("/documents/by-document-number") # async def get_by_document_number( # document_number: str = Query(..., description="LRSAdmin.lr_source_document.document_number") # ): # """Resolve one or more content URLs by document_number via alf_node_properties (MSSQL).""" # items = service.resolve_store_urls_by_document_number(document_number) # return { # "document_number": document_number, # "count": len(items), # "items": items, # } @app.get("/documents/by-document-number") async def get_by_document_number( document_number: str = Query(..., description="LRSAdmin.lr_source_document.document_number") ): """ Retrieve all document records from LRSAdmin.lr_source_document matching the given document_number. UPDATED: Now includes document type labels from lr_dictionary and available_images count. Returns: List of document IDs with metadata, type labels, and image counts """ conn = service._get_db_connection() cursor = conn.cursor() db_type = service.db_config.get("type", "mssql").lower() if db_type == "mysql": query = """ SELECT sd.id, sd.Hjtype, sd.document_number, sd.ref_number, sd.minute_number, sd.document_type, d.label AS document_type_label, d.code AS document_type_code, sd.duplicate, sd.access_type, sd.acceptance, sd.recordation, sd.submission, sd.issued_by, sd.page_count, sd.comments, sd.create_date, sd.modified_at, sd.modified_by, sd.property_file_page FROM lr_source_document sd LEFT JOIN lr_dictionary d ON d.Id = sd.document_type WHERE TRIM(sd.document_number) = %s ORDER BY sd.id """ cursor.execute(query, (document_number,)) else: query = """ SELECT sd.id, sd.Hjtype, sd.document_number, sd.ref_number, sd.minute_number, sd.document_type, d.label AS document_type_label, d.code AS document_type_code, sd.duplicate, sd.access_type, sd.acceptance, sd.recordation, sd.submission, sd.issued_by, sd.page_count, sd.comments, sd.create_date, sd.modified_at, sd.modified_by, sd.property_file_page FROM LRSAdmin.lr_source_document sd LEFT JOIN LRSAdmin.lr_dictionary d ON d.Id = sd.document_type WHERE RTRIM(LTRIM(sd.document_number)) = ? ORDER BY sd.id """ cursor.execute(query, (document_number,)) columns = [col[0] for col in cursor.description] rows = cursor.fetchall() cursor.close() conn.close() if not rows: raise HTTPException(status_code=404, detail=f"No records found for document_number '{document_number}'") items = [dict(zip(columns, row)) for row in rows] # Enhance with content information from Alfresco doc_groups = service.resolve_store_urls_by_document_number(document_number) content_map = {doc["document_id"]: len(doc["images"]) for doc in doc_groups} for item in items: item["available_images"] = content_map.get(item["id"], 0) return { "document_number": document_number, "count": len(items), "items": items, "note": "Each document ID may have different document types and page counts. Use 'available_images' to see which have content." } @app.get("/documents/pdf-by-document-number") async def pdf_by_document_number( document_number: str = Query(..., description="LRSAdmin.lr_source_document.document_number"), document_id: Optional[int] = Query(None, description="Optional specific document ID to retrieve") ): """ Resolve store URL by document_number and stream PDF to browser. FIXED: Now supports document_id parameter to handle multiple document IDs per document number. If document_id is not provided, uses the first document ID found. Args: document_number: Document number (e.g., 'BP703') document_id: Optional specific document ID to retrieve Returns: PDF file stream """ res = service.fetch_pdf_by_document_number(document_number, document_id=document_id) if res.get("error"): raise HTTPException(status_code=400, detail=res["error"]) if not res.get("pdf_path"): raise HTTPException(status_code=500, detail="PDF not generated") # Generate appropriate filename if res.get("document_id"): filename = f"{document_number}_doc{res['document_id']}.pdf" else: filename = f"{document_number}.pdf" return FileResponse( res["pdf_path"], media_type="application/pdf", filename=filename, headers={ "X-Document-Number": document_number, "X-Document-ID": str(res.get("document_id", "")), "X-Page-Count": str(res.get("page_count", 0)) } ) @app.get("/documents/by-property") async def get_documents_by_property( property_id: Optional[str] = Query(None, description="Property ID"), apn: Optional[str] = Query(None, description="Assessor's Parcel Number") ): """ Get all documents associated with a property. Args: property_id: Property ID apn: APN number Returns: List of document paths """ doc_paths = service.get_document_paths_by_property(property_id=property_id, apn=apn) return { "property_id": property_id, "apn": apn, "document_count": len(doc_paths), "documents": doc_paths } @app.post("/documents/convert") async def convert_document( content_path: str = Query(..., description="Content path to convert") ): """ Convert a document to PDF and return conversion result. Args: content_path: Relative path from database Returns: Conversion result with PDF path """ result = service.lookup_document(content_path, convert_to_pdf=True) if result.get("error"): raise HTTPException(status_code=400, detail=result["error"]) return { "success": result["pdf_path"] is not None, "content_path": content_path, "pdf_path": result.get("pdf_path") } # Add this to your existing FastAPI app in multipage_pdf_api.py @app.get("/documents/id/{document_id}") async def get_document_by_id( document_id: int = FastAPIPath(..., description="Document ID from lr_source_document") ): """ Get document details by ID (primary key from lr_source_document). This is needed because the LRS UI uses document ID in URLs, not document_number. URL pattern: #detail:SourceDocument:10000000000188 Args: document_id: Document ID (e.g., 10000000000188) Returns: Document metadata including document_number """ try: conn = service._get_db_connection() cursor = conn.cursor() db_type = service.db_config.get("type", "mssql").lower() if db_type == "mysql": query = """ SELECT id, document_number, document_type, page_count, acceptance, recordation, submission, issued_by, comments FROM lr_source_document WHERE id = %s """ cursor.execute(query, (document_id,)) else: query = """ SELECT id, document_number, document_type, page_count, acceptance, recordation, submission, issued_by, comments FROM LRSAdmin.lr_source_document WHERE id = ? """ cursor.execute(query, (document_id,)) row = cursor.fetchone() if not row: cursor.close() conn.close() raise HTTPException( status_code=404, detail=f"Document not found with ID: {document_id}" ) cols = [d[0] for d in cursor.description] result = dict(zip(cols, row)) cursor.close() conn.close() return result except HTTPException: raise except Exception as e: raise HTTPException( status_code=500, detail=f"Database error: {str(e)}" ) @app.get("/documents/id/{document_id}/pdf") async def stream_pdf_by_id( document_id: int = FastAPIPath(..., description="Document ID from lr_source_document"), regenerate: bool = Query(False, description="Force regeneration") ): """ Generate and stream PDF directly using document ID. This is the main endpoint for LRS UI integration. Usage from LRS UI: http://localhost:8001/documents/id/10000000000188/pdf Args: document_id: Document ID (e.g., 10000000000188) regenerate: Force regeneration Returns: Multi-page PDF file stream """ # First get the document_number from the ID try: conn = service._get_db_connection() cursor = conn.cursor() db_type = service.db_config.get("type", "mssql").lower() if db_type == "mysql": cursor.execute( "SELECT document_number, page_count FROM lr_source_document WHERE id = %s", (document_id,) ) else: cursor.execute( "SELECT document_number, page_count FROM LRSAdmin.lr_source_document WHERE id = ?", (document_id,) ) row = cursor.fetchone() if not row: cursor.close() conn.close() raise HTTPException( status_code=404, detail=f"Document not found with ID: {document_id}" ) document_number = row[0] page_count = row[1] cursor.close() conn.close() print(f"📋 Document ID {document_id} → Document Number: {document_number} (Pages: {page_count})") except HTTPException: raise except Exception as e: raise HTTPException( status_code=500, detail=f"Database error: {str(e)}" ) # Now generate/stream the PDF using document_number cached_pdf = os.path.join(TEMP_PDF_DIR, f"{document_number}.pdf") if os.path.exists(cached_pdf) and not regenerate: print(f"📦 Serving cached PDF for document ID {document_id} ({document_number})") return FileResponse( cached_pdf, media_type="application/pdf", filename=f"{document_number}.pdf", headers={ "Content-Disposition": f"inline; filename={document_number}.pdf", "Cache-Control": "public, max-age=3600", "X-Document-ID": str(document_id), "X-Document-Number": document_number } ) # Generate new PDF print(f"🔨 Generating PDF for document ID {document_id} ({document_number})") result = service.generate_pdf_for_document( document_number=document_number, cache_dir=TEMP_PDF_DIR ) if not result["success"]: raise HTTPException( status_code=500, detail={ "message": f"Failed to generate PDF for document {document_number}", "document_id": document_id, "errors": result["errors"] } ) if not result["pdf_path"] or not os.path.exists(result["pdf_path"]): raise HTTPException( status_code=500, detail="PDF was generated but file not found" ) # Stream the PDF return FileResponse( result["pdf_path"], media_type="application/pdf", filename=f"{document_number}.pdf", headers={ "Content-Disposition": f"inline; filename={document_number}.pdf", "X-Document-ID": str(document_id), "X-Document-Number": document_number, "X-Page-Count": str(result["actual_pages"]), "X-Expected-Pages": str(result["expected_pages"]) } ) # @app.get("/documents/{document_number}/generate", response_model=PDFGenerationResult) # async def generate_pdf( # document_number: str = FastAPIPath(..., description="Document number") # ): # """ # Generate a multi-page PDF and return generation metadata. # Does not stream the file, just returns status and path. # Args: # document_number: Document number # Returns: # PDF generation result with metadata # """ # result = service.generate_pdf_for_document( # document_number=document_number, # output_dir=TEMP_PDF_DIR # ) # return PDFGenerationResult(**result) @app.get("/documents/{document_number}/generate", response_model=PDFGenerationResult) async def generate_pdf( document_number: str = FastAPIPath(..., description="Document number") ): """ Generate a multi-page PDF and return generation metadata. Does not stream the file — only returns status and file path. Args: document_number: Document number (e.g., '10000000002503') Returns: JSON object containing PDF generation result and metadata """ try: result = service.generate_pdf_for_document( document_number=document_number, cache_dir=TEMP_PDF_DIR ) # Validate the result structure if not result.get("success"): raise HTTPException( status_code=500, detail={ "message": f"Failed to generate PDF for document {document_number}", "errors": result.get("errors", []) } ) # Ensure file path exists pdf_path = result.get("pdf_path") if not pdf_path or not os.path.exists(pdf_path): raise HTTPException( status_code=500, detail=f"PDF was generated but file not found: {pdf_path}" ) return PDFGenerationResult(**result) except HTTPException: raise # preserve existing HTTPException except Exception as e: raise HTTPException( status_code=500, detail=f"Unexpected error during PDF generation: {str(e)}" ) # @app.post("/documents/bulk-generate") # async def bulk_generate_pdfs( # document_numbers: List[str] = Query(..., description="List of document numbers") # ): # """ # Generate PDFs for multiple documents at once. # Useful for batch processing. # Args: # document_numbers: List of document numbers # Returns: # List of generation results # """ # if not document_numbers: # raise HTTPException( # status_code=400, # detail="No document numbers provided" # ) # if len(document_numbers) > 100: # raise HTTPException( # status_code=400, # detail="Maximum 100 documents per request" # ) # results = service.bulk_generate_pdfs( # document_numbers=document_numbers, # output_dir=TEMP_PDF_DIR # ) # success_count = sum(1 for r in results if r["success"]) # return { # "total": len(document_numbers), # "successful": success_count, # "failed": len(document_numbers) - success_count, # "results": [PDFGenerationResult(**r) for r in results] # } @app.delete("/documents/{document_number}/cache") async def clear_cached_pdf( document_number: str = FastAPIPath(..., description="Document number") ): """ Clear cached PDF for a document. Forces regeneration on next request. Args: document_number: Document number Returns: Deletion status """ cached_pdf = os.path.join(TEMP_PDF_DIR, f"{document_number}.pdf") if os.path.exists(cached_pdf): try: os.remove(cached_pdf) return { "success": True, "message": f"Cached PDF cleared for {document_number}" } except Exception as e: raise HTTPException( status_code=500, detail=f"Failed to delete cache: {str(e)}" ) else: return { "success": False, "message": f"No cached PDF found for {document_number}" } @app.delete("/cache/clear-all") async def clear_all_cache(): """ Clear all cached PDFs. Use with caution in production. Returns: Number of files deleted """ try: deleted_count = 0 for filename in os.listdir(TEMP_PDF_DIR): if filename.endswith(".pdf"): file_path = os.path.join(TEMP_PDF_DIR, filename) os.remove(file_path) deleted_count += 1 return { "success": True, "deleted_count": deleted_count, "message": f"Cleared {deleted_count} cached PDFs" } except Exception as e: raise HTTPException( status_code=500, detail=f"Failed to clear cache: {str(e)}" ) @app.get("/documents/id/{document_id}/metadata") async def get_metadata_by_id( document_id: int = FastAPIPath(..., description="Document ID") ): """ Get document metadata by ID. Args: document_id: Document ID Returns: Document metadata """ return await get_document_by_id(document_id) # =========================== # CORS Configuration for LRS Integration # =========================== """ Update your CORS middleware to allow requests from LRS server: """ # Replace the existing CORS middleware with this: app.add_middleware( CORSMiddleware, allow_origins=[ "http://10.10.10.3:8080", # Your LRS server "http://localhost:8080", "http://localhost:8001", "*" # Remove in production ], allow_credentials=True, allow_methods=["GET", "POST", "PUT", "PATCH", "DELETE", "OPTIONS", "HEAD"], allow_headers=["*"], expose_headers=["X-Document-ID", "X-Document-Number", "X-Page-Count"] ) @app.get("/documents/by-uuid", summary="Resolve content by Alfresco UUID") async def get_document_by_uuid(uuid: str = Query(..., description="Alfresco node UUID")): """ Resolve document metadata by Alfresco UUID (native Aumentum method). Mimics Aumentum Web Access: /documentPage.do?uuid= Example: dcb3a0d9-2a29-4441-afaa-919d32f1826d """ result = service.resolve_store_url_by_uuid(uuid) if not result: raise HTTPException(status_code=404, detail=f"No content found for UUID: {uuid}") return {"uuid": uuid, "content": result} @app.get("/documents/pdf-by-uuid", summary="Stream PDF by Alfresco UUID") async def stream_pdf_by_uuid(uuid: str = Query(..., description="Alfresco node UUID")): """ Convert and stream PDF by Alfresco UUID (native Aumentum method). Mimics Aumentum Web Access document viewer. Example: dcb3a0d9-2a29-4441-afaa-919d32f1826d """ result = service.fetch_pdf_by_uuid(uuid) if result.get("error"): raise HTTPException(status_code=404, detail=result["error"]) pdf_path = result["pdf_path"] if not pdf_path or not os.path.exists(pdf_path): raise HTTPException(status_code=500, detail="PDF conversion failed or file not found") return FileResponse(pdf_path, media_type="application/pdf", filename=f"{uuid}.pdf") @app.get("/documents/metadata-by-number") async def get_full_metadata_by_number( document_number: str = Query(..., description="Document number (e.g., BP703, PL035)") ): """ Get full document metadata from lr_source_document by document_number. Returns page_count, document_type, and all other metadata. This endpoint complements /documents/by-document-number which returns content URLs. This one returns the actual document metadata from the lr_source_document table. Args: document_number: Document number from lr_source_document Returns: Full document metadata including id, page_count, document_type, etc. Example: GET /documents/metadata-by-number?document_number=BP703 Response: { "id": 10000000000188, "document_number": "BP703", "document_type": 111, "page_count": 5, "acceptance": "2014-11-03T00:00:00", "recordation": null, "submission": "2014-11-03T00:00:00", "issued_by": "Ministry of Lands", "comments": null } """ try: conn = service._get_db_connection() cursor = conn.cursor() db_type = service.db_config.get("type", "mssql").lower() if db_type == "mysql": query = """ SELECT id, document_number, document_type, page_count, acceptance, recordation, submission, issued_by, comments, create_date, modified_at, modified_by FROM lr_source_document WHERE TRIM(document_number) = %s """ cursor.execute(query, (document_number,)) else: query = """ SELECT id, document_number, document_type, page_count, acceptance, recordation, submission, issued_by, comments, create_date, modified_at, modified_by FROM LRSAdmin.lr_source_document WHERE RTRIM(LTRIM(document_number)) = ? """ cursor.execute(query, (document_number,)) row = cursor.fetchone() if not row: cursor.close() conn.close() raise HTTPException( status_code=404, detail=f"Document not found: {document_number}" ) cols = [d[0] for d in cursor.description] result = dict(zip(cols, row)) # Convert datetime objects to strings for JSON serialization for key, value in result.items(): if hasattr(value, 'isoformat'): result[key] = value.isoformat() cursor.close() conn.close() return result except HTTPException: raise except Exception as e: raise HTTPException( status_code=500, detail=f"Database error: {str(e)}" ) # Optional: Add a combined endpoint that returns both metadata and content URLs @app.get("/documents/full-info-by-number") async def get_full_info_by_number( document_number: str = Query(..., description="Document number") ): """ Get complete document information including metadata and content URLs. Combines data from lr_source_document and Alfresco content. Args: document_number: Document number Returns: Complete document information with metadata and content URLs """ # Get metadata metadata_result = await get_full_metadata_by_number(document_number) # Get content URLs content_result = await get_by_document_number(document_number) return { "document_number": document_number, "metadata": metadata_result, "content": content_result } # debuging endpoint @app.get("/debug/document-numbers") async def debug_document_numbers( search_pattern: Optional[str] = Query(None, description="Search pattern (e.g., 'PL%' or 'PL11089')") ): """ Debug endpoint to see all document numbers in the database and identify potential matching issues. This helps diagnose why PL11089 might be matching PL689. Args: search_pattern: Optional SQL LIKE pattern (e.g., 'PL%' for all PL documents) Returns: List of document numbers with their IDs, showing exact values including whitespace """ try: conn = service._get_db_connection() cursor = conn.cursor() if search_pattern: query = """ SELECT id, document_number, LEN(document_number) as length, DATALENGTH(document_number) as byte_length, ASCII(LEFT(document_number, 1)) as first_char_ascii, ASCII(RIGHT(document_number, 1)) as last_char_ascii, document_type, page_count FROM LRSAdmin.lr_source_document WHERE document_number LIKE ? ORDER BY document_number """ cursor.execute(query, (search_pattern,)) else: query = """ SELECT TOP 100 id, document_number, LEN(document_number) as length, DATALENGTH(document_number) as byte_length, ASCII(LEFT(document_number, 1)) as first_char_ascii, ASCII(RIGHT(document_number, 1)) as last_char_ascii, document_type, page_count FROM LRSAdmin.lr_source_document ORDER BY document_number DESC """ cursor.execute(query) cols = [d[0] for d in cursor.description] rows = cursor.fetchall() results = [] for row in rows: doc = dict(zip(cols, row)) # Add visual representation of whitespace doc_num = doc['document_number'] doc['visual_representation'] = repr(doc_num) # Shows \n, \t, spaces, etc. doc['has_leading_space'] = doc_num != doc_num.lstrip() doc['has_trailing_space'] = doc_num != doc_num.rstrip() results.append(doc) cursor.close() conn.close() return { "count": len(results), "search_pattern": search_pattern, "documents": results, "note": "Check 'visual_representation' for hidden whitespace characters" } except Exception as e: raise HTTPException( status_code=500, detail=f"Database error: {str(e)}" ) @app.get("/debug/compare-document-numbers") async def compare_document_numbers( doc1: str = Query(..., description="First document number"), doc2: str = Query(..., description="Second document number") ): """ Compare two document numbers to see if they're matching incorrectly. Example: Compare PL11089 with PL689 to see why they might be conflicting. Args: doc1: First document number (e.g., 'PL11089') doc2: Second document number (e.g., 'PL689') Returns: Detailed comparison showing why they might be matching """ try: conn = service._get_db_connection() cursor = conn.cursor() # Get both documents cursor.execute(""" SELECT id, document_number, LEN(document_number) as length, DATALENGTH(document_number) as byte_length, document_type, page_count FROM LRSAdmin.lr_source_document WHERE document_number IN (?, ?) OR RTRIM(LTRIM(document_number)) IN (?, ?) """, (doc1, doc2, doc1.strip(), doc2.strip())) cols = [d[0] for d in cursor.description] rows = cursor.fetchall() results = [] for row in rows: doc = dict(zip(cols, row)) doc_num = doc['document_number'] doc['visual_representation'] = repr(doc_num) doc['trimmed'] = doc_num.strip() doc['has_whitespace'] = doc_num != doc_num.strip() results.append(doc) # Check for Alfresco properties cursor.execute(""" SELECT DISTINCT np.string_value, LEN(np.string_value) as length, q.local_name, n.id as node_id FROM LRSAdmin.alf_node_properties np JOIN LRSAdmin.alf_qname q ON q.id = np.qname_id JOIN LRSAdmin.alf_node n ON n.id = np.node_id AND n.node_deleted = 0 WHERE (np.string_value = ? OR np.string_value = ?) AND q.local_name IN ('targetRids','sourceRids') """, (doc1, doc2)) alf_props = [] for row in cursor.fetchall(): alf_props.append({ 'string_value': row[0], 'length': row[1], 'property_name': row[2], 'node_id': row[3], 'visual_representation': repr(row[0]) }) cursor.close() conn.close() return { "comparison": { "doc1_input": doc1, "doc2_input": doc2, "doc1_stripped": doc1.strip(), "doc2_stripped": doc2.strip(), "are_equal": doc1 == doc2, "are_equal_trimmed": doc1.strip() == doc2.strip(), "substring_match": doc1 in doc2 or doc2 in doc1 }, "database_records": results, "alfresco_properties": alf_props, "warning": "If database_records shows multiple matches, there's a data quality issue" } except Exception as e: raise HTTPException( status_code=500, detail=f"Database error: {str(e)}" ) @app.get("/debug/match-by-transaction-time") async def match_by_transaction_time( document_number: str = Query(..., description="Document number to match") ): """ Match documents to correct files using transaction timestamps and node creation times. This is the CORRECT way to identify which file belongs to which document. """ try: conn = service._get_db_connection() cursor = conn.cursor() # Get document creation time cursor.execute(""" SELECT id, document_number, create_date, submission, modified_at FROM LRSAdmin.lr_source_document WHERE document_number = ? ORDER BY id """, (document_number,)) doc_info = [] for row in cursor.fetchall(): doc_info.append({ 'id': row[0], 'document_number': row[1], 'create_date': str(row[2]) if row[2] else None, 'submission': str(row[3]) if row[3] else None, 'modified_at': str(row[4]) if row[4] else None }) if not doc_info: cursor.close() conn.close() raise HTTPException(status_code=404, detail=f"Document not found: {document_number}") doc_created = doc_info[0]['create_date'] # Find nodes created near this time cursor.execute(""" SELECT np.string_value AS node_labeled_as, n.id AS node_id, n.audit_created AS node_created_time, cu.content_url, cu.content_size, DATEDIFF(HOUR, ?, CAST(n.audit_created AS datetime)) AS hours_diff FROM LRSAdmin.alf_node_properties np JOIN LRSAdmin.alf_qname q ON q.id = np.qname_id JOIN LRSAdmin.alf_node n ON n.id = np.node_id AND n.node_deleted = 0 LEFT JOIN LRSAdmin.alf_content_data cd ON cd.id = n.id LEFT JOIN LRSAdmin.alf_content_url cu ON cu.id = cd.content_url_id WHERE q.local_name = 'sourceRids' AND n.audit_created BETWEEN DATEADD(DAY, -7, ?) AND DATEADD(DAY, 7, ?) AND cu.content_url IS NOT NULL ORDER BY ABS(DATEDIFF(HOUR, ?, CAST(n.audit_created AS datetime))) """, (doc_created, doc_created, doc_created, doc_created)) matches = [] for row in cursor.fetchall(): matches.append({ 'node_labeled_as': row[0], 'node_id': row[1], 'node_created_time': str(row[2]) if row[2] else None, 'content_url': row[3], 'content_size': row[4], 'hours_diff': row[5], 'confidence': 'STRONG_MATCH' if abs(row[5]) <= 24 else 'POSSIBLE_MATCH' if abs(row[5]) <= 72 else 'WEAK_MATCH' }) cursor.close() conn.close() return { 'document_number': document_number, 'document_info': doc_info, 'potential_matches': matches[:10], # Top 10 closest matches 'note': 'Nodes listed by timestamp proximity. STRONG_MATCH (within 24h) is most likely correct.' } except HTTPException: raise except Exception as e: raise HTTPException(status_code=500, detail=str(e)) @app.get("/debug/show-store-urls") async def show_store_urls( document_number: str = Query(..., description="Document number to analyze") ): """ Show actual store:// URLs from database for a document. Helps identify which files are associated with which documents. """ try: conn = service._get_db_connection() cursor = conn.cursor() cursor.execute(""" SELECT DISTINCT np.string_value AS document_number, n.id AS node_id, cu.content_url, cu.content_size, n.audit_created, q.local_name AS property_name FROM LRSAdmin.alf_node_properties np JOIN LRSAdmin.alf_qname q ON q.id = np.qname_id JOIN LRSAdmin.alf_node n ON n.id = np.node_id AND n.node_deleted = 0 LEFT JOIN LRSAdmin.alf_content_data cd ON cd.id = n.id LEFT JOIN LRSAdmin.alf_content_url cu ON cu.id = cd.content_url_id WHERE RTRIM(LTRIM(np.string_value)) = ? AND q.local_name IN ('targetRids','sourceRids') ORDER BY n.id, cu.content_url """, (document_number,)) cols = [d[0] for d in cursor.description] rows = cursor.fetchall() urls = [] for row in rows: url_dict = dict(zip(cols, row)) if url_dict.get('audit_created'): url_dict['audit_created'] = str(url_dict['audit_created']) urls.append(url_dict) cursor.close() conn.close() return { "document_number": document_number, "total_urls": len(urls), "urls": urls } except Exception as e: raise HTTPException(status_code=500, detail=str(e)) @app.get("/debug/document-number-stats") async def document_number_statistics(): """ Get statistics about document numbers to identify data quality issues. Returns: Statistics about document numbers including duplicates, whitespace issues, etc. """ try: conn = service._get_db_connection() cursor = conn.cursor() # Get statistics stats = {} # Total documents cursor.execute("SELECT COUNT(*) FROM LRSAdmin.lr_source_document") stats['total_documents'] = cursor.fetchone()[0] # Documents with leading/trailing whitespace cursor.execute(""" SELECT COUNT(*) FROM LRSAdmin.lr_source_document WHERE document_number != RTRIM(LTRIM(document_number)) """) stats['with_whitespace'] = cursor.fetchone()[0] # Duplicate document numbers (after trimming) cursor.execute(""" SELECT RTRIM(LTRIM(document_number)) as clean_num, COUNT(*) as count FROM LRSAdmin.lr_source_document GROUP BY RTRIM(LTRIM(document_number)) HAVING COUNT(*) > 1 ORDER BY count DESC """) duplicates = [] for row in cursor.fetchall(): duplicates.append({ 'document_number': row[0], 'count': row[1] }) stats['duplicates'] = duplicates # Substring matches (documents where one number contains another) cursor.execute(""" SELECT DISTINCT a.document_number as doc1, b.document_number as doc2 FROM LRSAdmin.lr_source_document a CROSS JOIN LRSAdmin.lr_source_document b WHERE a.id != b.id AND ( a.document_number LIKE '%' + b.document_number + '%' OR b.document_number LIKE '%' + a.document_number + '%' ) AND LEN(a.document_number) > 3 AND LEN(b.document_number) > 3 """) substring_matches = [] for row in cursor.fetchall()[:20]: # Limit to 20 examples substring_matches.append({ 'doc1': row[0], 'doc2': row[1] }) stats['substring_matches_sample'] = substring_matches cursor.close() conn.close() return { "statistics": stats, "recommendations": [ "If with_whitespace > 0: Run UPDATE to trim all document_number fields", "If duplicates exist: These are multiple documents with same number", "If substring_matches exist: This could explain PL11089 matching PL689" ] } except Exception as e: raise HTTPException( status_code=500, detail=f"Database error: {str(e)}" ) # updating method to ensure it ONLY processes the specified document_id: @app.get("/documents/by-document-number-enhanced") async def get_by_document_number_enhanced( document_number: str = Query(..., description="Document number (e.g., PL11089)") ): """ ENHANCED: Get all document IDs for a document number, showing which have available content. Returns a summary of all document types under this document number, with clear indication of which ones have PDF content available. Example Response: { "document_number": "PL11089", "total_document_ids": 3, "documents": [ { "document_id": 10000000013787, "document_type": 111, "document_type_label": "Building Plan", "page_count": 1, "available_images": 1, "has_content": true, "pdf_url": "/documents/pdf-by-document-number?document_number=PL11089&document_id=10000000013787" }, { "document_id": 10000000013791, "document_type": 103, "document_type_label": "Site Plan", "page_count": 46, "available_images": 46, "has_content": true, "pdf_url": "/documents/pdf-by-document-number?document_number=PL11089&document_id=10000000013791" }, ... ] } """ try: # Get all document records from lr_source_document conn = service._get_db_connection() cursor = conn.cursor() cursor.execute(""" SELECT sd.id, sd.document_type, d.label AS document_type_label, sd.page_count, sd.acceptance, sd.recordation, sd.issued_by, sd.comments FROM LRSAdmin.lr_source_document sd LEFT JOIN LRSAdmin.lr_dictionary d ON d.Id = sd.document_type WHERE RTRIM(LTRIM(sd.document_number)) = ? ORDER BY sd.id """, (document_number.strip(),)) cols = [d[0] for d in cursor.description] rows = cursor.fetchall() cursor.close() conn.close() if not rows: raise HTTPException( status_code=404, detail=f"No documents found for document_number: {document_number}" ) # Get content availability from Alfresco doc_groups = service.resolve_store_urls_by_document_number(document_number) content_map = {doc["document_id"]: len(doc["images"]) for doc in doc_groups} # Build response documents = [] for row in rows: doc_dict = dict(zip(cols, row)) doc_id = doc_dict["id"] available_images = content_map.get(doc_id, 0) documents.append({ "document_id": doc_id, "document_type": doc_dict["document_type"], "document_type_label": doc_dict.get("document_type_label"), "page_count": doc_dict["page_count"], "available_images": available_images, "has_content": available_images > 0, "acceptance": doc_dict["acceptance"].isoformat() if doc_dict.get("acceptance") else None, "recordation": doc_dict["recordation"].isoformat() if doc_dict.get("recordation") else None, "issued_by": doc_dict.get("issued_by"), "comments": doc_dict.get("comments"), "pdf_url": f"/documents/pdf-by-document-number?document_number={document_number}&document_id={doc_id}" if available_images > 0 else None }) return { "document_number": document_number, "total_document_ids": len(documents), "documents": documents, "note": "Use document_id parameter to retrieve specific document type" } except HTTPException: raise except Exception as e: raise HTTPException( status_code=500, detail=f"Database error: {str(e)}" ) @app.get("/documents/pdf-by-document-number-fixed") async def pdf_by_document_number_fixed( document_number: str = Query(..., description="Document number (e.g., PL11089)"), document_id: Optional[int] = Query(None, description="Specific document ID to retrieve") ): """ FIXED: Stream PDF with mandatory document_id selection for multi-document numbers. If document_number has multiple document IDs and document_id is not specified, returns a list of available document IDs with their types. Args: document_number: Document number (e.g., 'PL11089') document_id: Specific document ID (required if multiple exist) Returns: - PDF file stream if document_id is specified - List of available documents if document_id is missing and multiple exist """ document_number = document_number.strip() # Step 1: Get all document IDs for this document number try: conn = service._get_db_connection() cursor = conn.cursor() cursor.execute(""" SELECT sd.id, sd.document_type, d.label AS document_type_label, sd.page_count FROM LRSAdmin.lr_source_document sd LEFT JOIN LRSAdmin.lr_dictionary d ON d.Id = sd.document_type WHERE RTRIM(LTRIM(sd.document_number)) = ? ORDER BY sd.id """, (document_number,)) docs = cursor.fetchall() cursor.close() conn.close() if not docs: raise HTTPException( status_code=404, detail=f"No documents found for document_number: {document_number}" ) # Step 2: If multiple document IDs exist and none specified, return choices if len(docs) > 1 and document_id is None: choices = [] for doc in docs: choices.append({ "document_id": doc[0], "document_type": doc[1], "document_type_label": doc[2], "page_count": doc[3], "url": f"/documents/pdf-by-document-number-fixed?document_number={document_number}&document_id={doc[0]}" }) raise HTTPException( status_code=400, detail={ "error": "Multiple documents found. Please specify document_id.", "document_number": document_number, "available_documents": choices, "hint": "Add &document_id=XXXXX to your request" } ) # Step 3: Use specified document_id or default to first if only one exists if document_id is None: document_id = docs[0][0] print(f"📋 Only one document ID found, using: {document_id}") else: # Validate that document_id belongs to this document_number valid_ids = [doc[0] for doc in docs] if document_id not in valid_ids: raise HTTPException( status_code=400, detail=f"Document ID {document_id} does not belong to document number {document_number}. Valid IDs: {valid_ids}" ) except HTTPException: raise except Exception as e: raise HTTPException( status_code=500, detail=f"Database error: {str(e)}" ) # Step 4: Generate/retrieve PDF for specific document_id print(f"🔨 Generating PDF for {document_number} (Document ID: {document_id})") result = service.generate_pdf_for_document( document_number=document_number, document_id=document_id, cache_dir=TEMP_PDF_DIR ) if not result["success"]: raise HTTPException( status_code=500, detail={ "message": f"Failed to generate PDF for document {document_number}", "document_id": document_id, "errors": result["errors"] } ) if not result["pdf_path"] or not os.path.exists(result["pdf_path"]): raise HTTPException( status_code=500, detail="PDF was generated but file not found" ) # Step 5: Stream the PDF return FileResponse( result["pdf_path"], media_type="application/pdf", filename=f"{document_number}_doc{document_id}.pdf", headers={ "Content-Disposition": f"inline; filename={document_number}_doc{document_id}.pdf", "X-Document-ID": str(document_id), "X-Document-Number": document_number, "X-Page-Count": str(result["actual_pages"]), "X-Expected-Pages": str(result["expected_pages"]) } ) @app.get("/documents/list-by-document-number") async def list_documents_by_number( document_number: str = Query(..., description="Document number (e.g., PL11089)") ): """ List all document IDs and their types for a given document number. This endpoint helps users understand what documents are available before requesting the PDF. Args: document_number: Document number Returns: List of document IDs with metadata and PDF availability """ document_number = document_number.strip() try: conn = service._get_db_connection() cursor = conn.cursor() # Get all documents with type labels cursor.execute(""" SELECT sd.id, sd.document_type, d.label AS document_type_label, d.code AS document_type_code, sd.page_count, sd.acceptance, sd.recordation, sd.issued_by FROM LRSAdmin.lr_source_document sd LEFT JOIN LRSAdmin.lr_dictionary d ON d.Id = sd.document_type WHERE RTRIM(LTRIM(sd.document_number)) = ? ORDER BY CASE WHEN sd.document_type = 111 THEN 1 -- Building Plan first WHEN sd.document_type = 103 THEN 2 -- Site Plan second ELSE 3 END, sd.id """, (document_number,)) cols = [d[0] for d in cursor.description] rows = cursor.fetchall() cursor.close() conn.close() if not rows: raise HTTPException( status_code=404, detail=f"No documents found for document_number: {document_number}" ) # Get content availability doc_groups = service.resolve_store_urls_by_document_number(document_number) content_map = {doc["document_id"]: len(doc["images"]) for doc in doc_groups} # Build response documents = [] for row in rows: doc_dict = dict(zip(cols, row)) doc_id = doc_dict["id"] available_images = content_map.get(doc_id, 0) documents.append({ "document_id": doc_id, "document_type": doc_dict["document_type"], "document_type_label": doc_dict.get("document_type_label"), "document_type_code": doc_dict.get("document_type_code"), "page_count": doc_dict["page_count"], "available_images": available_images, "has_content": available_images > 0, "acceptance": doc_dict["acceptance"].isoformat() if doc_dict.get("acceptance") else None, "recordation": doc_dict["recordation"].isoformat() if doc_dict.get("recordation") else None, "issued_by": doc_dict.get("issued_by"), "view_url": f"/documents/pdf-by-document-number-fixed?document_number={document_number}&document_id={doc_id}" if available_images > 0 else None }) return { "document_number": document_number, "total_documents": len(documents), "documents": documents, "summary": { "with_content": sum(1 for d in documents if d["has_content"]), "without_content": sum(1 for d in documents if not d["has_content"]), "total_pages": sum(d["page_count"] or 0 for d in documents), "available_pages": sum(d["available_images"] for d in documents) } } except HTTPException: raise except Exception as e: raise HTTPException( status_code=500, detail=f"Database error: {str(e)}" ) # =========================== # DOCUMENT SEARCH ENDPOINT (Unified - searches both lr_source_document and boundary_document) # =========================== @app.get("/documents/search") async def search_documents( document_number: Optional[str] = Query(None, description="Document number to search"), document_type: Optional[int] = Query(None, description="Document type ID (for lr_source_document)"), date_issued_from: Optional[str] = Query(None, description="Date issued from (YYYY-MM-DD)"), date_issued_to: Optional[str] = Query(None, description="Date issued to (YYYY-MM-DD)"), transaction_number: Optional[str] = Query(None, description="Transaction number"), transaction_type: Optional[int] = Query(None, description="Transaction type ID"), transaction_status: Optional[int] = Query(None, description="Transaction status ID"), limit: int = Query(100, ge=1, le=500, description="Maximum number of results") ): """ Unified document search - searches both lr_source_document (PLAGIS) and boundary_document (Boundary Commission). Returns documents from both tables matching the search criteria. """ try: conn = service._get_db_connection() cursor = conn.cursor() all_documents = [] db_type = service.db_config.get("type", "mysql").lower() # Search boundary_document table (new Boundary Commission documents) if document_number: try: boundary_query = """ SELECT id, document_number, document_type, title, description, file_name, file_size, page_count, uploaded_date, uploaded_by, related_boundary_id, related_dispute_id, status, created_at FROM boundary_document WHERE document_number LIKE %s ORDER BY uploaded_date DESC LIMIT %s """ cursor.execute(boundary_query, (f"%{document_number}%", limit)) columns = [col[0] for col in cursor.description] for row in cursor.fetchall(): doc_dict = dict(zip(columns, row)) # Convert datetime to ISO strings for date_field in ['uploaded_date', 'created_at']: if doc_dict.get(date_field) and hasattr(doc_dict[date_field], 'isoformat'): doc_dict[date_field] = doc_dict[date_field].isoformat() # Add source indicator doc_dict['source'] = 'boundary_commission' doc_dict['document_type_label'] = doc_dict.get('document_type', 'N/A') all_documents.append(doc_dict) except Exception as e: print(f"⚠️ Error searching boundary_document: {e}") # Search lr_source_document table (old PLAGIS documents) try: where_clauses = [] params = [] if document_number: if db_type == "mysql": where_clauses.append("sd.document_number LIKE %s") else: where_clauses.append("sd.document_number LIKE ?") params.append(f"%{document_number}%") if document_type: if db_type == "mysql": where_clauses.append("sd.document_type = %s") else: where_clauses.append("sd.document_type = ?") params.append(document_type) if date_issued_from: if db_type == "mysql": where_clauses.append("sd.recordation >= %s") else: where_clauses.append("sd.recordation >= ?") params.append(date_issued_from) if date_issued_to: if db_type == "mysql": where_clauses.append("sd.recordation <= %s") else: where_clauses.append("sd.recordation <= ?") params.append(date_issued_to) # Transaction filtering if transaction_number or transaction_type or transaction_status: if db_type == "mysql": where_clauses.append(""" EXISTS ( SELECT 1 FROM lr_transaction_document td JOIN lr_transaction t ON t.id = td.transaction_id WHERE td.document_id = sd.id """) else: where_clauses.append(""" EXISTS ( SELECT 1 FROM LRSAdmin.lr_transaction_document td JOIN LRSAdmin.lr_transaction t ON t.id = td.transaction_id WHERE td.document_id = sd.id """) if transaction_number: if db_type == "mysql": where_clauses[-1] += " AND t.transaction_number LIKE %s" else: where_clauses[-1] += " AND t.transaction_number LIKE ?" params.append(f"%{transaction_number}%") if transaction_type: if db_type == "mysql": where_clauses[-1] += " AND t.transaction_type = %s" else: where_clauses[-1] += " AND t.transaction_type = ?" params.append(transaction_type) if transaction_status: if db_type == "mysql": where_clauses[-1] += " AND t.transaction_status = %s" else: where_clauses[-1] += " AND t.transaction_status = ?" params.append(transaction_status) where_clauses[-1] += ")" if where_clauses: where_clause = " AND ".join(where_clauses) if db_type == "mysql": table_prefix = "" query = f""" SELECT sd.id, sd.document_number, sd.document_type, sd.ref_number, sd.minute_number, sd.page_count, sd.acceptance, sd.recordation, sd.submission, d.label AS document_type_label FROM lr_source_document sd LEFT JOIN lr_dictionary d ON d.Id = sd.document_type WHERE {where_clause} ORDER BY sd.id DESC LIMIT %s """ params.append(limit) else: table_prefix = "LRSAdmin." query = f""" SELECT sd.id, sd.document_number, sd.document_type, sd.ref_number, sd.minute_number, sd.page_count, sd.acceptance, sd.recordation, sd.submission, d.label AS document_type_label FROM {table_prefix}lr_source_document sd LEFT JOIN {table_prefix}lr_dictionary d ON d.Id = sd.document_type WHERE {where_clause} ORDER BY sd.id DESC """ # MSSQL uses TOP instead of LIMIT query = query.replace("LIMIT ?", f"TOP ({limit})") cursor.execute(query, params) columns = [col[0] for col in cursor.description] for row in cursor.fetchall(): doc_dict = dict(zip(columns, row)) # Convert datetime to ISO strings for date_field in ['acceptance', 'recordation', 'submission']: if doc_dict.get(date_field) and hasattr(doc_dict[date_field], 'isoformat'): doc_dict[date_field] = doc_dict[date_field].isoformat() # Add source indicator doc_dict['source'] = 'plagis' all_documents.append(doc_dict) except Exception as e: print(f"⚠️ Error searching lr_source_document: {e}") import traceback traceback.print_exc() # Build search_params for display search_params = {} if document_number: search_params["Document Number"] = document_number if document_type: search_params["Document Type"] = str(document_type) if date_issued_from: search_params["Date Issued From"] = date_issued_from if date_issued_to: search_params["Date Issued To"] = date_issued_to if transaction_number: search_params["Transaction Number"] = transaction_number # Log results for debugging print(f"📊 Document search results: {len(all_documents)} documents found") if all_documents: for doc in all_documents[:3]: # Log first 3 print(f" - {doc.get('document_number')} ({doc.get('source', 'unknown')})") cursor.close() conn.close() return { "total_results": len(all_documents), "limit": limit, "documents": all_documents[:limit], # Limit final results "search_params": search_params } except HTTPException: raise except Exception as e: import traceback traceback.print_exc() raise HTTPException(status_code=500, detail=f"Database error: {str(e)}") # =========================== # PROPERTY SEARCH ENDPOINT # =========================== @app.get("/properties/search") async def search_properties( property_number: Optional[str] = Query(None, description="Property number to search"), property_type: Optional[int] = Query(None, description="Property type ID"), upin: Optional[str] = Query(None, description="UPIN"), state: Optional[str] = Query(None, description="State"), village: Optional[str] = Query(None, description="Village"), block_number: Optional[str] = Query(None, description="Block number"), plot_number: Optional[str] = Query(None, description="Plot number"), party_type: Optional[int] = Query(None, description="Party type ID to filter properties"), legal_role: Optional[int] = Query(None, description="Legal role ID to filter properties"), transaction_type: Optional[int] = Query(None, description="Transaction type ID to filter properties"), transaction_status: Optional[int] = Query(None, description="Transaction status ID to filter properties"), transaction_number: Optional[str] = Query(None, description="Transaction number to filter properties"), limit: int = Query(100, ge=1, le=500, description="Maximum number of results") ): """ Search for properties by property criteria, party criteria, and transaction criteria. Returns matching properties with their basic information and associated transactions. """ conn = service._get_db_connection() cursor = conn.cursor() db_type = service.db_config.get("type", "mssql").lower() try: # Build dynamic WHERE clause where_clauses = [] params = [] # Property filters if property_number: if db_type == "mysql": where_clauses.append("su.refnum LIKE %s") else: where_clauses.append("su.refnum LIKE ?") params.append(f"%{property_number}%") if property_type: if db_type == "mysql": where_clauses.append("su.spatial_unit_type = %s") else: where_clauses.append("su.spatial_unit_type = ?") params.append(property_type) if upin: if db_type == "mysql": where_clauses.append("su.upin LIKE %s") else: where_clauses.append("su.upin LIKE ?") params.append(f"%{upin}%") if state: if state.isdigit(): if db_type == "mysql": where_clauses.append("su.status = %s") else: where_clauses.append("su.status = ?") params.append(int(state)) else: if db_type == "mysql": where_clauses.append("state_dict.label LIKE %s") else: where_clauses.append("state_dict.label LIKE ?") params.append(f"%{state}%") if village: if db_type == "mysql": where_clauses.append("addr.cell LIKE %s") else: where_clauses.append("addr.cell LIKE ?") params.append(f"%{village}%") if block_number: if db_type == "mysql": where_clauses.append("su.block_number LIKE %s") else: where_clauses.append("su.block_number LIKE ?") params.append(f"%{block_number}%") if plot_number: if db_type == "mysql": where_clauses.append("su.lot_number LIKE %s") else: where_clauses.append("su.lot_number LIKE ?") params.append(f"%{plot_number}%") # Transaction filtering if transaction_type or transaction_status or transaction_number: if db_type == "mysql": where_clauses.append(""" EXISTS ( SELECT 1 FROM lr_transaction t_tx WHERE (t_tx.source_property_ids LIKE CONCAT('%%', su.refnum, '%%') OR t_tx.target_property_ids LIKE CONCAT('%%', su.refnum, '%%') OR t_tx.old_property_id = su.refnum) """) else: where_clauses.append(""" EXISTS ( SELECT 1 FROM LRSAdmin.lr_transaction t_tx WHERE (t_tx.source_property_ids LIKE '%' + su.refnum + '%' OR t_tx.target_property_ids LIKE '%' + su.refnum + '%' OR t_tx.old_property_id = su.refnum) """) if transaction_type: if db_type == "mysql": where_clauses[-1] += " AND t_tx.transaction_type = %s" else: where_clauses[-1] += " AND t_tx.transaction_type = ?" params.append(transaction_type) if transaction_status: if db_type == "mysql": where_clauses[-1] += " AND t_tx.transaction_status = %s" else: where_clauses[-1] += " AND t_tx.transaction_status = ?" params.append(transaction_status) if transaction_number: if db_type == "mysql": where_clauses[-1] += " AND t_tx.transaction_number LIKE %s" else: where_clauses[-1] += " AND t_tx.transaction_number LIKE ?" params.append(f"%{transaction_number}%") where_clauses[-1] += ")" # Party filtering if party_type or legal_role: if db_type == "mysql": where_clauses.append(""" EXISTS ( SELECT 1 FROM lr_transaction t_party JOIN lr_transaction_party tp ON tp.transaction_id = t_party.id JOIN lr_party party ON party.id = tp.party_id WHERE (t_party.source_property_ids LIKE CONCAT('%%', su.refnum, '%%') OR t_party.target_property_ids LIKE CONCAT('%%', su.refnum, '%%') OR t_party.old_property_id = su.refnum) """) else: where_clauses.append(""" EXISTS ( SELECT 1 FROM LRSAdmin.lr_transaction t_party JOIN LRSAdmin.lr_transaction_party tp ON tp.transaction_id = t_party.id JOIN LRSAdmin.lr_party party ON party.id = tp.party_id WHERE (t_party.source_property_ids LIKE '%' + su.refnum + '%' OR t_party.target_property_ids LIKE '%' + su.refnum + '%' OR t_party.old_property_id = su.refnum) """) if party_type: if db_type == "mysql": where_clauses[-1] += " AND party.party_type = %s" else: where_clauses[-1] += " AND party.party_type = ?" params.append(party_type) if legal_role: if db_type == "mysql": where_clauses[-1] += " AND party.party_role_type = %s" else: where_clauses[-1] += " AND party.party_role_type = ?" params.append(legal_role) where_clauses[-1] += ")" if not where_clauses: raise HTTPException( status_code=400, detail="At least one search parameter is required" ) where_clause = " AND ".join(where_clauses) # Build query with MySQL/MSSQL compatibility if db_type == "mysql": table_prefix = "" query = f""" SELECT DISTINCT su.id, su.refnum, su.upin, su.spatial_unit_type, sut_dict.label as property_type_label, su.area, su.area_unit_type, area_dict.label as measurement_unit_label, su.block_number, su.lot_number, su.plan_number, su.tracing_number, su.status, state_dict.label as state_label, su.property_area, prop_area_dict.label as property_area_label, su.local_number, lga_dict.label as local_government_area_label FROM lr_spatial_unit su LEFT JOIN lr_address addr ON addr.id = su.address LEFT JOIN lr_dictionary sut_dict ON sut_dict.Id = su.spatial_unit_type LEFT JOIN lr_dictionary area_dict ON area_dict.Id = su.area_unit_type LEFT JOIN lr_dictionary state_dict ON state_dict.Id = su.status LEFT JOIN lr_dictionary prop_area_dict ON prop_area_dict.Id = su.property_area LEFT JOIN lr_dictionary lga_dict ON lga_dict.Id = addr.local_government_area WHERE {where_clause} ORDER BY su.refnum LIMIT %s """ params.append(limit) else: table_prefix = "LRSAdmin." query = f""" SELECT DISTINCT TOP {limit} su.id, su.refnum, su.upin, su.spatial_unit_type, sut_dict.label as property_type_label, su.area, su.area_unit_type, area_dict.label as measurement_unit_label, su.block_number, su.lot_number, su.plan_number, su.tracing_number, su.status, state_dict.label as state_label, su.property_area, prop_area_dict.label as property_area_label, su.local_number, lga_dict.label as local_government_area_label FROM {table_prefix}lr_spatial_unit su LEFT JOIN {table_prefix}lr_address addr ON addr.id = su.address LEFT JOIN {table_prefix}lr_dictionary sut_dict ON sut_dict.Id = su.spatial_unit_type LEFT JOIN {table_prefix}lr_dictionary area_dict ON area_dict.Id = su.area_unit_type LEFT JOIN {table_prefix}lr_dictionary state_dict ON state_dict.Id = su.status LEFT JOIN {table_prefix}lr_dictionary prop_area_dict ON prop_area_dict.Id = su.property_area LEFT JOIN {table_prefix}lr_dictionary lga_dict ON lga_dict.Id = addr.local_government_area WHERE {where_clause} ORDER BY su.refnum """ cursor.execute(query, params) rows = cursor.fetchall() properties = [] for row in rows: property_info = { "id": int(row[0]), "property_number": row[1], "upin": row[2], "property_type": int(row[3]) if row[3] else None, "property_type_label": row[4] or "Land Parcel", "area": float(row[5]) if row[5] else None, "measurement_unit": int(row[6]) if row[6] else None, "measurement_unit_label": row[7] or "Square Meters", "block_number": row[8] or "", "plot_number": row[9] or "", "plan_number": row[10] or "", "tracing_number": row[11] or "", "state": int(row[12]) if row[12] else None, "state_label": row[13] or "", "property_area": int(row[14]) if row[14] else None, "property_area_label": row[15] or "", "local_government_area_label": row[17] or "" } # Get transaction count if db_type == "mysql": cursor.execute(""" SELECT COUNT(DISTINCT t.id) FROM lr_transaction t WHERE t.source_property_ids LIKE CONCAT('%%', %s, '%%') OR t.target_property_ids LIKE CONCAT('%%', %s, '%%') OR t.old_property_id = %s """, (row[1], row[1], row[1])) else: cursor.execute(""" SELECT COUNT(DISTINCT t.id) FROM LRSAdmin.lr_transaction t WHERE t.source_property_ids LIKE ? OR t.target_property_ids LIKE ? OR t.old_property_id = ? """, (f'%{row[1]}%', f'%{row[1]}%', row[1])) tx_count_row = cursor.fetchone() property_info["transaction_count"] = int(tx_count_row[0]) if tx_count_row else 0 properties.append(property_info) cursor.close() conn.close() return { "total_results": len(properties), "limit": limit, "properties": properties } except HTTPException: raise except Exception as e: import traceback traceback.print_exc() cursor.close() conn.close() raise HTTPException(status_code=500, detail=f"Error searching properties: {str(e)}") # =========================== # ANALYTICS DATA-CENTRIC ENDPOINT # =========================== @app.get("/analytics/data-centric") async def get_data_centric_dashboard( year: Optional[int] = Query(None, ge=1900, le=2100, description="Filter by issuance year"), start_year: Optional[int] = Query(None, ge=1900, le=2100, description="Start of issuance year range"), end_year: Optional[int] = Query(None, ge=1900, le=2100, description="End of issuance year range"), quarter: Optional[str] = Query(None, description="Quarter filter (Q1-Q4)"), month: Optional[str] = Query(None, description="Month filter (Jan, January, or 1-12)"), limit: int = Query(500, ge=10, le=5000, description="Maximum owner rows to return"), offset: int = Query(0, ge=0, le=50000, description="Offset for owner rows pagination"), document_scope: str = Query("cofo_only", pattern="^(cofo_only|all_documents)$", description="Choose between CofO-only data or all document types"), document_types: Optional[List[str]] = Query(None, description="Optional list of specific document_type codes to include"), owner_type: Optional[List[str]] = Query(None, description="Filter owner rows by party type label (e.g., Organization)"), owner_name: Optional[str] = Query(None, description="Case-insensitive substring to match owner name fields"), owner_gender: Optional[List[str]] = Query(None, description="Filter owners by gender label"), ownership_type: Optional[List[str]] = Query(None, description="Filter by derived ownership type (Single/Joint)"), ownership_category: Optional[List[str]] = Query(None, description="Filter by derived ownership category (e.g., Joint - Male-Female)"), property_unique_id: Optional[str] = Query(None, description="Filter by property unique identifier / LA unit UID"), issuance_date_from: Optional[str] = Query(None, description="Filter owner rows with issuance date on/after this date (YYYY-MM-DD)"), issuance_date_to: Optional[str] = Query(None, description="Filter owner rows with issuance date on/before this date (YYYY-MM-DD)"), registration_date_from: Optional[str] = Query(None, description="Filter owner rows with registration date on/after this date (YYYY-MM-DD)"), registration_date_to: Optional[str] = Query(None, description="Filter owner rows with registration date on/before this date (YYYY-MM-DD)"), reference_number: Optional[str] = Query(None, description="Filter by reference/instrument/page/volume numbers") ): """ Provide the analytics dataset that powers the Data Centric Center dashboard. Returns summary metrics, chart datasets, ownership category distributions, and a paginated owner table focused on Certificate of Occupancy transactions. """ conn = service._get_db_connection() cursor = conn.cursor() db_type = service.db_config.get("type", "mssql").lower() try: # Normalize document types normalized_doc_types = [dt.strip() for dt in (document_types or []) if dt.strip()] if document_scope == "cofo_only": doc_type_codes = normalized_doc_types or ["sd_Co_title"] else: doc_type_codes = normalized_doc_types # For now, return a simplified response since the full analytics query is complex # This will prevent 404 errors and allow the frontend to work return { "filters": { "year": year, "startYear": start_year, "endYear": end_year, "quarter": quarter, "month": month, "limit": limit, "offset": offset, "document_scope": document_scope, "document_types": document_types, "owner_type": owner_type, "property_unique_id": property_unique_id }, "availableYears": [], "summary": { "totalTransactions": 0, "ownerTypeBreakdown": [], "ownershipCategoryCards": [], "ownershipTrend": [] }, "owners": [], "message": "Analytics endpoint is available. Full implementation in progress." } except Exception as e: import traceback traceback.print_exc() cursor.close() conn.close() raise HTTPException(status_code=500, detail=f"Error getting analytics data: {str(e)}") # =========================== # PROPERTY INDEXING ENDPOINT # =========================== @app.post("/properties/index") async def index_property( property_number: Optional[str] = Form(None), upin: Optional[str] = Form(None), spatial_unit_type: int = Form(..., description="Property type ID from dictionary"), tenure_type: int = Form(..., description="Tenure type ID from dictionary"), purpose: int = Form(..., description="Purpose ID from dictionary"), zone: int = Form(..., description="Zone ID from dictionary"), layout: int = Form(..., description="Layout ID from dictionary"), area_unit_type: int = Form(..., description="Area unit type ID from dictionary"), status: int = Form(..., description="Status ID from dictionary"), refnum: Optional[str] = Form(None), block_number: Optional[str] = Form(None), lot_number: Optional[str] = Form(None), plan_number: Optional[str] = Form(None), tracing_number: Optional[str] = Form(None), area: Optional[float] = Form(None), area_planned: Optional[float] = Form(None), label: Optional[str] = Form(None), note: Optional[str] = Form(None), comments: Optional[str] = Form(None), property_kind: Optional[int] = Form(None), local_number: Optional[str] = Form(None), block_name: Optional[str] = Form(None), volume: Optional[str] = Form(None), folio: Optional[str] = Form(None), indexed_by: Optional[str] = Form(None), current_user: Any = Depends(get_current_user_optional) ): """ Index/Create a new landed property record in the system. This endpoint creates a new entry in lr_spatial_unit table. """ conn = service._get_db_connection() cursor = conn.cursor() db_type = service.db_config.get("type", "mssql").lower() try: # Generate property number if not provided if not property_number and not refnum: # Auto-generate property number (format: PL12345) import random random_number = random.randint(10000, 99999) property_number = f"PL{random_number}" refnum = property_number if not refnum: refnum = property_number # Get next ID (for MySQL, use AUTO_INCREMENT; for MSSQL, use IDENTITY) if db_type == "mysql": # Get max ID and add 1 cursor.execute("SELECT COALESCE(MAX(CAST(id AS UNSIGNED)), 0) + 1 FROM lr_spatial_unit") result = cursor.fetchone() new_id = int(result[0]) if result else 10000000000001 else: # For MSSQL, use IDENTITY_INSERT or get max ID cursor.execute("SELECT COALESCE(MAX(CAST(id AS BIGINT)), 0) + 1 FROM LRSAdmin.lr_spatial_unit") result = cursor.fetchone() new_id = int(result[0]) if result else 10000000000001 # Default values for required fields hjtype = "com.landsystems.lrs.model.SpatialUnit" # Insert property record if db_type == "mysql": insert_query = """ INSERT INTO lr_spatial_unit ( id, Hjtype, refnum, upin, spatial_unit_type, tenure_type, purpose, zone, layout, area_unit_type, areapl_unit_type, areabld_unit_type, status, block_number, lot_number, plan_number, tracing_number, area, area_planned, label, note, comments, property_kind, local_number, block_name, volume, folio, rec_dict1, rec_dict2, version, begin_lifespan_version ) VALUES ( %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, NOW() ) """ cursor.execute(insert_query, ( new_id, hjtype, refnum, upin, spatial_unit_type, tenure_type, purpose, zone, layout, area_unit_type, area_unit_type, area_unit_type, status, block_number, lot_number, plan_number, tracing_number, area, area_planned, label, note, comments, property_kind, local_number, block_name, volume, folio, 1, 1, 1 )) else: insert_query = """ INSERT INTO LRSAdmin.lr_spatial_unit ( id, Hjtype, refnum, upin, spatial_unit_type, tenure_type, purpose, zone, layout, area_unit_type, areapl_unit_type, areabld_unit_type, status, block_number, lot_number, plan_number, tracing_number, area, area_planned, label, note, comments, property_kind, local_number, block_name, volume, folio, rec_dict1, rec_dict2, version, begin_lifespan_version ) VALUES ( ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, GETDATE() ) """ cursor.execute(insert_query, ( new_id, hjtype, refnum, upin, spatial_unit_type, tenure_type, purpose, zone, layout, area_unit_type, area_unit_type, area_unit_type, status, block_number, lot_number, plan_number, tracing_number, area, area_planned, label, note, comments, property_kind, local_number, block_name, volume, folio, 1, 1, 1 )) conn.commit() # Fetch the created property if db_type == "mysql": cursor.execute(""" SELECT su.id, su.refnum, su.upin, su.spatial_unit_type, su.tenure_type, su.purpose, su.zone, su.layout, su.area, su.area_planned, su.block_number, su.lot_number, su.plan_number, su.tracing_number, su.label, su.note, su.comments, su.status FROM lr_spatial_unit su WHERE su.id = %s """, (new_id,)) else: cursor.execute(""" SELECT su.id, su.refnum, su.upin, su.spatial_unit_type, su.tenure_type, su.purpose, su.zone, su.layout, su.area, su.area_planned, su.block_number, su.lot_number, su.plan_number, su.tracing_number, su.label, su.note, su.comments, su.status FROM LRSAdmin.lr_spatial_unit su WHERE su.id = ? """, (new_id,)) row = cursor.fetchone() cols = [d[0] for d in cursor.description] property_data = dict(zip(cols, row)) cursor.close() conn.close() return { "success": True, "message": "Property indexed successfully", "property": property_data, "property_number": refnum } except Exception as e: import traceback traceback.print_exc() conn.rollback() cursor.close() conn.close() raise HTTPException(status_code=500, detail=f"Error indexing property: {str(e)}") # =========================== # GIS WORKSTATION ENDPOINTS # =========================== @app.get("/gis/properties/geojson") async def get_properties_geojson( bbox: Optional[str] = Query(None, description="Bounding box: minx,miny,maxx,maxy"), property_number: Optional[str] = Query(None), current_user: Any = Depends(get_current_user_optional) ): """ Get properties as GeoJSON for GIS viewers (QGIS, web maps). Returns spatial data in GeoJSON format. """ conn = service._get_db_connection() cursor = conn.cursor() db_type = service.db_config.get("type", "mssql").lower() try: # Build query based on filters where_clauses = [] params = [] if property_number: if db_type == "mysql": where_clauses.append("refnum = %s") else: where_clauses.append("refnum = ?") params.append(property_number) where_sql = " AND ".join(where_clauses) if where_clauses else "1=1" # Query properties (simplified - actual spatial queries would need geometry columns) if db_type == "mysql": query = f""" SELECT id, refnum, upin, block_number, lot_number, plan_number, area, area_planned, spatial_unit_type, tenure_type, purpose, zone, layout, status FROM lr_spatial_unit WHERE {where_sql} LIMIT 1000 """ else: query = f""" SELECT TOP 1000 id, refnum, upin, block_number, lot_number, plan_number, area, area_planned, spatial_unit_type, tenure_type, purpose, zone, layout, status FROM LRSAdmin.lr_spatial_unit WHERE {where_sql} """ cursor.execute(query, params) rows = cursor.fetchall() cols = [d[0] for d in cursor.description] # Convert to GeoJSON format features = [] for row in rows: prop = dict(zip(cols, row)) # Note: Actual geometry would come from spatial columns # This is a placeholder structure feature = { "type": "Feature", "properties": prop, "geometry": { "type": "Point", # Placeholder - would be actual geometry "coordinates": [0, 0] # Placeholder - would be actual coordinates } } features.append(feature) geojson = { "type": "FeatureCollection", "features": features } cursor.close() conn.close() return geojson except Exception as e: cursor.close() conn.close() raise HTTPException(status_code=500, detail=f"Error fetching GIS data: {str(e)}") @app.get("/gis/boundaries/geojson") async def get_boundaries_geojson( boundary_number: Optional[str] = Query(None), current_user: Any = Depends(get_current_user_optional) ): """ Get boundary records as GeoJSON for GIS viewers. """ # Similar implementation to properties geojson return { "type": "FeatureCollection", "features": [] } @app.get("/gis/qgis/connect") async def qgis_connect( current_user: Any = Depends(get_current_user_optional) ): """ QGIS extension connection endpoint. Returns connection info and available layers. """ return { "status": "connected", "api_version": "2.0.0", "layers": [ { "id": "properties", "name": "Properties", "type": "vector", "endpoint": "/gis/properties/geojson" }, { "id": "boundaries", "name": "Boundaries", "type": "vector", "endpoint": "/gis/boundaries/geojson" } ] } # =========================== # CASHIER WORKSTATION ENDPOINTS # =========================== class PaymentRequest(BaseModel): transaction_number: str property_number: Optional[str] = None amount: float payment_method: str # cash, bank_transfer, cheque, card, online description: Optional[str] = None service_type: Optional[str] = None class PaymentResponse(BaseModel): success: bool payment_id: Optional[str] = None receipt_number: Optional[str] = None message: str @app.post("/cashier/payment", response_model=PaymentResponse) async def process_payment( payment: PaymentRequest, current_user: Any = Depends(require_auth) ): """ Process a payment transaction. Creates payment record and generates receipt. """ # TODO: Implement actual payment processing with database storage # For now, return a mock response import random from datetime import datetime payment_id = f"PAY-{datetime.now().strftime('%Y%m%d-%H%M%S')}-{random.randint(1000, 9999)}" receipt_number = f"RCP-{datetime.now().strftime('%Y%m%d')}-{random.randint(10000, 99999)}" # In production, this would: # 1. Create payment record in database # 2. Update transaction status # 3. Generate receipt # 4. Log payment activity return PaymentResponse( success=True, payment_id=payment_id, receipt_number=receipt_number, message=f"Payment of ₦{payment.amount:,.2f} processed successfully" ) @app.get("/cashier/fees") async def get_fee_schedule( service_type: Optional[str] = Query(None), current_user: Any = Depends(get_current_user_optional) ): """ Get fee schedule for various services. """ fees = [ { "service": "Property Registration", "service_type": "registration", "amount": 50000.00, "currency": "NGN", "description": "New property registration" }, { "service": "Document Indexing", "service_type": "indexing", "amount": 5000.00, "currency": "NGN", "description": "Per document indexing" }, { "service": "Boundary Survey", "service_type": "survey", "amount": 100000.00, "currency": "NGN", "description": "Boundary survey processing" }, { "service": "Certificate of Occupancy", "service_type": "certificate", "amount": 200000.00, "currency": "NGN", "description": "C of O issuance" } ] if service_type: fees = [f for f in fees if f["service_type"] == service_type] return {"fees": fees} @app.get("/cashier/payments") async def get_payment_history( start_date: Optional[str] = Query(None), end_date: Optional[str] = Query(None), transaction_number: Optional[str] = Query(None), current_user: Any = Depends(require_auth) ): """ Get payment history with optional filters. """ # TODO: Implement actual database query return { "payments": [], "total": 0, "count": 0 } # =========================== # MAIN ENTRY POINT # =========================== if __name__ == "__main__": import uvicorn print("🚀 Starting Aumentum Browser API Server") print("📡 API will be available at: http://localhost:8001") print("📚 Interactive docs at: http://localhost:8001/docs") print("\n") uvicorn.run(app, host="0.0.0.0", port=8001)