""" core.py — Qualora Core Infrastructure ====================================== Consolidated: config.py + db.py + logger.py + utils.py Strictly hardened for Vercel (Serverless) & Enterprise MVC standards. """ import os import sys import time import json import logging import re import hashlib import secrets import threading import certifi from datetime import datetime, timezone from typing import Optional, Dict, Any, Union from urllib.parse import urlparse from dotenv import load_dotenv from werkzeug.security import generate_password_hash from werkzeug.utils import secure_filename from pymongo import MongoClient, ASCENDING, DESCENDING from pymongo.errors import ConnectionFailure, OperationFailure import gridfs from bson import ObjectId # ══════════════════════════════════════════════════════════════════════════════ # CONFIGURATION # ══════════════════════════════════════════════════════════════════════════════ load_dotenv(override=True) # Ensure ChromaDB telemetry is disabled early to avoid PostHog failures # (some chromadb versions call PostHog with incompatible signatures). os.environ.setdefault("CHROMADB_DISABLE_TELEMETRY", "1") os.environ.setdefault("CHROMA_DISABLE_TELEMETRY", "1") os.environ.setdefault("ANONYMIZED_TELEMETRY", "True") # Project paths # `BASE_DIR` points to the templates directory used by Flask's `template_folder`. BASE_DIR = os.path.abspath(os.path.join(os.path.dirname(__file__), "templates")) # General DEBUG = os.environ.get("FLASK_ENV") == "development" PORT = int(os.environ.get("PORT", "8000")) VERCEL_ENV = os.environ.get("VERCEL_ENV", "") IS_PRODUCTION = bool(VERCEL_ENV or os.environ.get("PRODUCTION", "")) # Auth & Security JWT_SECRET = os.environ.get("JWT_SECRET") if not JWT_SECRET: import warnings if IS_PRODUCTION: warnings.warn("JWT_SECRET missing in production — app started in degraded mode", RuntimeWarning) JWT_SECRET = "dev_insafe_key_change_me" # Fallback; HF Spaces will inject at runtime else: warnings.warn("Using dev JWT_SECRET — NEVER use in production", DeprecationWarning) JWT_SECRET = "dev_insafe_key_change_me" JWT_EXPIRATION_SECONDS = int(os.environ.get("JWT_EXPIRATION_SECONDS", "3600")) ALLOWED_ORIGIN = os.environ.get("ALLOWED_ORIGIN") if ALLOWED_ORIGIN == "*": import warnings if IS_PRODUCTION: warnings.warn("Wildcard CORS detected in production — resetting to safe default", RuntimeWarning) ALLOWED_ORIGIN = "http://localhost:5173" if DEBUG else "https://qualora.io" if not ALLOWED_ORIGIN: ALLOWED_ORIGIN = "http://localhost:5173" if DEBUG else "https://qualora.io" # Database MONGODB_URI = os.environ.get("MONGODB_URI", "") if not MONGODB_URI: import warnings if IS_PRODUCTION: warnings.warn("MONGODB_URI is missing in production — database operations will fail until configured", RuntimeWarning) else: warnings.warn("MONGODB_URI not configured — database operations will fail", UserWarning) # Service API Keys GROQ_API_KEY = os.environ.get("GROQ_API_KEY", "") OPENROUTER_API_KEY = os.environ.get("OPENROUTER_API_KEY", "") ELEVENLABS_API_KEY = os.environ.get("ELEVENLABS_API_KEY", "") ELEVENLABS_WEBHOOK_SECRET = os.environ.get("ELEVENLABS_WEBHOOK_SECRET") if not ELEVENLABS_WEBHOOK_SECRET: import warnings if IS_PRODUCTION: warnings.warn("ELEVENLABS_WEBHOOK_SECRET missing in production — webhook validation will use fallback", RuntimeWarning) ELEVENLABS_WEBHOOK_SECRET = "dev_webhook_fallback" DEEPGRAM_API_KEY = os.environ.get("DEEPGRAM_API_KEY", "") HF_SPACE_URL = os.environ.get("HF_SPACE_URL", "") HF_SPACE_TOKEN = os.environ.get("HF_SPACE_TOKEN", "") VOYAGE_API_KEY = os.environ.get("VOYAGE_API_KEY", "") MURF_API_KEY = os.environ.get("MURF_API_KEY", "") # Sentinel used by RAG pipeline to indicate no policy context was found. RAG_NULL_SENTINEL = "[NO_CONTEXT]" # External Integrations WEBHOOK_URL = os.environ.get("WEBHOOK_URL", "") SLACK_URL = os.environ.get("SLACK_URL", "") DISCORD_URL = os.environ.get("DISCORD_URL", "") GITHUB_URL = os.environ.get("GITHUB_URL", "https://github.com/prathamamritkar/genAI-qualityBot") # App Constraints & Vercel Storage Compatibility MAX_CONTENT_LENGTH = 50 * 1024 * 1024 # 50MB def get_abs_path(rel_path: str) -> str: """Helper to resolve paths relative to project root, robust on Windows/Vercel.""" if os.path.isabs(rel_path): return rel_path project_root = os.path.dirname(os.path.abspath(__file__)) # Handle /tmp special case for Vercel if rel_path.startswith("/tmp"): return rel_path return os.path.normpath(os.path.join(project_root, rel_path)) CHROMA_PATH = get_abs_path(os.environ.get("CHROMA_PATH", "./data/chroma")) UPLOAD_FOLDER = get_abs_path(os.environ.get("UPLOAD_FOLDER", "./data/uploads")) # Ensure directories exist try: os.makedirs(CHROMA_PATH, exist_ok=True) os.makedirs(UPLOAD_FOLDER, exist_ok=True) except Exception as e: # Fallback to local 'data' if system paths fail backup_data = os.path.join(os.path.dirname(os.path.abspath(__file__)), "data") os.makedirs(backup_data, exist_ok=True) print(f"Warning: Could not create primary storage paths ({e}). Using {backup_data}") # Rate Limiting AUTH_RATE_LIMIT_PER_MIN = int(os.environ.get("AUTH_RATE_LIMIT_PER_MIN", "10")) AUDIT_RATE_LIMIT_PER_MIN = int(os.environ.get("AUDIT_RATE_LIMIT_PER_MIN", "30")) ABUSE_BLOCK_THRESHOLD = int(os.environ.get("ABUSE_BLOCK_THRESHOLD", "5")) ABUSE_BLOCK_DURATION_SEC = int(os.environ.get("ABUSE_BLOCK_DURATION_SEC", "300")) # Logging # Support a special 'OFF' level to silence all logs across the process. # Default to INFO so both terminal and browser logging are enabled by default. LOG_LEVEL = os.environ.get("LOG_LEVEL", "INFO").upper() LOG_FORMAT = os.environ.get("LOG_FORMAT", "json" if VERCEL_ENV else "pretty") # ══════════════════════════════════════════════════════════════════════════════ # LOGGING # ══════════════════════════════════════════════════════════════════════════════ # Pre-compile regex for high-performance log scrubbing _SENSITIVE_PATTERNS = [ (re.compile(r'(password|passwd|pwd)\s*=\s*["\']?[^"\',\s]+', re.IGNORECASE), r'\1='), (re.compile(r'(api[_-]?key|token|secret|auth)\s*=\s*["\']?[^"\',\s]+', re.IGNORECASE), r'\1='), (re.compile(r'Bearer\s+[A-Za-z0-9_\-\.]+'), 'Bearer '), (re.compile(r'mongodb(\+srv)?://[^\s"\',]+'), r'mongodb\1://') ] class _JsonFormatter(logging.Formatter): def format(self, record: logging.LogRecord) -> str: doc: dict = { "ts": datetime.now(timezone.utc).isoformat(timespec="milliseconds"), "level": record.levelname, "logger": record.name, "msg": record.getMessage(), } _reserved = { "asctime", "created", "exc_info", "exc_text", "filename", "funcName", "levelname", "levelno", "lineno", "message", "module", "msecs", "msg", "name", "pathname", "process", "processName", "relativeCreated", "stack_info", "taskName", "thread", "threadName", "args", } for key, val in vars(record).items(): if key not in _reserved: doc[key] = val if record.exc_info and record.exc_info[0] is not None: traceback_str = self.formatException(record.exc_info) doc["traceback"] = self._scrub_sensitive_data(traceback_str) return json.dumps(doc, default=str, ensure_ascii=False) def _scrub_sensitive_data(self, text: str) -> str: for pattern, replacement in _SENSITIVE_PATTERNS: text = pattern.sub(replacement, text) return text _COLOURS = { "DEBUG": "\033[36m", "INFO": "\033[32m", "WARNING": "\033[33m", "ERROR": "\033[31m", "CRITICAL": "\033[35m", } _RESET = "\033[0m" class _PrettyFormatter(logging.Formatter): def format(self, record: logging.LogRecord) -> str: ts = datetime.now(timezone.utc).strftime("%H:%M:%S") col = _COLOURS.get(record.levelname, "") level = f"{col}{record.levelname:<8}{_RESET}" name = f"\033[90m{record.name}\033[0m" msg = record.getMessage() _reserved = { "asctime", "created", "exc_info", "exc_text", "filename", "funcName", "levelname", "levelno", "lineno", "message", "module", "msecs", "msg", "name", "pathname", "process", "processName", "relativeCreated", "stack_info", "taskName", "thread", "threadName", "args", } extras = {k: v for k, v in vars(record).items() if k not in _reserved} extra_str = (" " + " ".join(f"\033[90m{k}=\033[0m{v!r}" for k, v in extras.items())) if extras else "" line = f"{ts} {level} {name} {msg}{extra_str}" if record.exc_info and record.exc_info[0] is not None: line += "\n" + self.formatException(record.exc_info) return line def _setup_root(level: str, fmt: str) -> None: root = logging.getLogger() if root.handlers: return # Special-case: disable all logging output when level is OFF/NONE/DISABLED. if isinstance(level, str) and level.upper() in ("OFF", "NONE", "DISABLED"): # Prevent further log output from all standard loggers. logging.disable(logging.CRITICAL) return handler = logging.StreamHandler(sys.stdout) handler.setFormatter(_JsonFormatter() if fmt == "json" else _PrettyFormatter()) root.addHandler(handler) try: root.setLevel(getattr(logging, level)) except Exception: root.setLevel(logging.INFO) for noisy in ("urllib3", "httpcore", "httpx", "groq", "watchdog", "pymongo"): logging.getLogger(noisy).setLevel(logging.DEBUG if level == "DEBUG" else logging.WARNING) _setup_root(LOG_LEVEL, LOG_FORMAT) # Allow targeted control over PyMongo verbosity without changing global LOG_LEVEL. # This lets operators keep DEBUG for app internals while silencing repetitive pymongo DEBUG logs. MONGODB_LOG_LEVEL = os.environ.get("MONGODB_LOG_LEVEL", "").upper() if not MONGODB_LOG_LEVEL: MONGODB_LOG_LEVEL = "WARNING" try: pymongo_level = getattr(logging, MONGODB_LOG_LEVEL, logging.WARNING) logging.getLogger("pymongo").setLevel(pymongo_level) except Exception: logging.getLogger("pymongo").setLevel(logging.WARNING) def get_logger(name: str) -> logging.Logger: return logging.getLogger(name) def log_request_middleware(app): _req_log = get_logger("qualora.request") @app.before_request def _before(): from flask import g g._req_start = time.monotonic() @app.after_request def _after(response): from flask import g, request latency_ms = float(f"{(time.monotonic() - getattr(g, '_req_start', time.monotonic())) * 1000:.1f}") _req_log.info( f"{request.method} {request.path}", extra={"method": request.method, "path": request.path, "status": response.status_code, "latency_ms": latency_ms} ) return response return app # ══════════════════════════════════════════════════════════════════════════════ # DATABASE # ══════════════════════════════════════════════════════════════════════════════ _db_lock = threading.Lock() _db_client = None _db_instance = None _gridfs_instance = None _indexes_created = False def _extract_db_name(mongodb_uri: str) -> str: if not mongodb_uri: return "qualora_enterprise" try: parsed = urlparse(mongodb_uri) db_name = parsed.path.lstrip('/') if db_name: return db_name.split('?')[0] or "qualora_enterprise" except Exception: pass return "qualora_enterprise" def _ensure_indexes(db) -> None: try: db.users.create_index([("email", ASCENDING)], unique=True) db.users.create_index([("org_id", ASCENDING)]) db.orgs.create_index([("name", ASCENDING)]) db.audits.create_index([("org_id", ASCENDING)]) db.audits.create_index([("submitted_by", ASCENDING)]) db.audits.create_index([("created_at", DESCENDING)]) db.audits.create_index([("org_id", ASCENDING), ("created_at", DESCENDING)]) db.kb_documents.create_index([("org_id", ASCENDING)]) db.kb_documents.create_index([("gridfs_id", ASCENDING)]) db.kb_chunks.create_index([("org_id", ASCENDING)]) db.kb_chunks.create_index([("doc_id", ASCENDING)]) db.kb_chunks.create_index([("org_id", ASCENDING), ("doc_id", ASCENDING)]) db.alerts.create_index([("created_at", ASCENDING)], expireAfterSeconds=7776000) db.alerts.create_index([("org_id", ASCENDING)]) db.alerts.create_index([("read", ASCENDING)]) db.audit_logs.create_index([("created_at", ASCENDING)], expireAfterSeconds=15552000) db.audit_logs.create_index([("org_id", ASCENDING)]) db.audit_criteria.create_index([("org_id", ASCENDING)]) db.org_settings.create_index([("org_id", ASCENDING)], unique=True) db.agent_profiles.create_index([("org_id", ASCENDING)]) db.agent_profiles.create_index([("agent_id", ASCENDING)], unique=True) db.agent_profiles.create_index([("org_id", ASCENDING), ("agent_id", ASCENDING)]) db.agent_profiles.create_index([("updated_at", DESCENDING)]) logging.info("✅ All MongoDB indexes verified/created successfully.") except OperationFailure as e: logging.warning("Index creation skipped (likely pre-existing): %s", str(e)) def _seed_initial_accounts(db) -> None: admin_email = os.environ.get("ADMIN_EMAIL") admin_pass = os.environ.get("ADMIN_PASSWORD") demo_login = os.environ.get("DEMO_USER_LOGIN") demo_pass = os.environ.get("DEMO_USER_PASS") sys_org_id = ObjectId("000000000000000000000000") if admin_email and admin_pass: if db.users.count_documents({"email": admin_email}, limit=1) == 0: db.users.insert_one({ "email": admin_email, "password": generate_password_hash(admin_pass), "name": "System Administrator", "org_id": sys_org_id, "role": "admin" }) logging.info("🔒 Default Admin account securely provisioned.") if demo_login and demo_pass: demo_email = f"{demo_login}@demo.local" if db.users.count_documents({"email": demo_email}, limit=1) == 0: db.users.insert_one({ "email": demo_email, "password": generate_password_hash(demo_pass), "name": "Demo Agent", "org_id": sys_org_id, "role": "agent" }) logging.info("🔒 Default Demo account securely provisioned.") def get_db(): global _db_client, _db_instance, _indexes_created if _db_instance is not None: return _db_instance with _db_lock: if _db_instance is not None: return _db_instance mongo_uri = os.environ.get("MONGODB_URI") if not mongo_uri: logging.error("CRITICAL: MONGODB_URI is missing.") return None max_retries = 3 retry_delay = 1 for attempt in range(1, max_retries + 1): try: # Optimized for Vercel Serverless (Low connection pool, quick idle drops) _db_client = MongoClient( mongo_uri, maxPoolSize=10, minPoolSize=0, maxIdleTimeMS=5000, serverSelectionTimeoutMS=5000, connectTimeoutMS=10000, socketTimeoutMS=20000, retryWrites=True, retryReads=True, tlsCAFile=certifi.where() ) _db_client.admin.command('ping') db_name = _extract_db_name(mongo_uri) _db_instance = _db_client[db_name] if not _indexes_created: _ensure_indexes(_db_instance) _seed_initial_accounts(_db_instance) _indexes_created = True logging.info(f"✅ MongoDB connected to: {db_name} (attempt {attempt}/{max_retries})") return _db_instance except ConnectionFailure as e: if attempt < max_retries: wait_time = retry_delay * (2 ** (attempt - 1)) logging.warning(f"MongoDB connection attempt {attempt}/{max_retries} failed: {str(e)}. Retrying in {wait_time}s...") time.sleep(wait_time) else: logging.error(f"MongoDB connection FATAL error after {max_retries} attempts: {str(e)}") return None except Exception as e: logging.error(f"MongoDB connection unexpected error on attempt {attempt}: {str(e)}") if attempt >= max_retries: return None time.sleep(retry_delay * (2 ** (attempt - 1))) def attempt_atlas_vector_index(db) -> bool: """Safely handle Atlas Vector Search index creation.""" index_name = "kb_chunks_vector_index" collection_name = "kb_chunks" try: if hasattr(db[collection_name], "list_search_indexes"): existing_indexes = list(db[collection_name].list_search_indexes()) if any(idx.get("name") == index_name for idx in existing_indexes): return True index_definition = { "name": index_name, "definition": { "mappings": { "dynamic": False, "fields": { "embedding_voyage": {"type": "knnVector", "dimensions": 1024, "similarity": "cosine"}, "org_id": {"type": "filter"} } } } } db[collection_name].create_search_index(index_definition) logging.info(f"✅ Atlas Vector index '{index_name}' creation initiated.") return True except OperationFailure as e: logging.debug(f"Atlas Vector Search skip (OperationFailure): {str(e)}") return False except AttributeError: logging.debug("PyMongo version does not support list_search_indexes. Skipping programmatic creation.") return False def get_gridfs(): global _gridfs_instance if _gridfs_instance is not None: return _gridfs_instance db = get_db() if db is None: logging.warning("GridFS unavailable — database connection failed.") return None _gridfs_instance = gridfs.GridFS(db) logging.info("✅ GridFS initialized.") return _gridfs_instance def get_db_status() -> dict: db = get_db() if db is None: return {"connected": False, "collections": {}, "atlas_vector_index": False} try: return { "connected": True, "collections": { "audits": db.audits.estimated_document_count(), "users": db.users.estimated_document_count(), "kb_documents": db.kb_documents.estimated_document_count(), "kb_chunks": db.kb_chunks.estimated_document_count(), "alerts": db.alerts.estimated_document_count(), "agent_profiles": db.agent_profiles.estimated_document_count(), }, "atlas_vector_index": attempt_atlas_vector_index(db) } except Exception as e: logging.error("get_db_status failed: %s", str(e)) return {"connected": True, "collections": {}, "atlas_vector_index": False, "error": str(e)} # ══════════════════════════════════════════════════════════════════════════════ # UTILITIES # ══════════════════════════════════════════════════════════════════════════════ def sanitize_filename(filename: str) -> str: """Uses Werkzeug's secure_filename for robust security against traversal attacks.""" if not filename: return "unnamed_file" secured = secure_filename(filename) return secured if secured else "unnamed_file" def extract_text_from_file(filepath: str, filename: str) -> str: if not os.path.exists(filepath): logging.error("File not found for extraction", extra={"filepath": filepath, "requested_file": filename}) raise ValueError(f"File not found: {filename}") ext = os.path.splitext(filename)[1].lower() try: if ext == '.pdf': # Lazy import to save memory in serverless environments try: try: from pypdf import PdfReader except ImportError: from PyPDF2 import PdfReader # Fallback except ImportError: # Critical check: Inform the user exactly which libraries are missing logging.error("PDF Extraction failed: Neither 'pypdf' nor 'PyPDF2' is installed.") raise ImportError("PDF extraction requires 'pypdf' or 'PyPDF2'. Please run 'pip install pypdf'.") try: reader = PdfReader(filepath) text = '\n'.join(page.extract_text() or '' for page in reader.pages).strip() if not text: raise ValueError(f"PDF file contains no extractable text (might be scanned/image-only): {filename}") return text except Exception as pdf_err: logging.error(f"Low-level PDF read error for {filename}: {pdf_err}") raise ValueError(f"Could not parse PDF {filename}: {str(pdf_err)}") elif ext == '.json': with open(filepath, 'r', encoding='utf-8', errors='replace') as f: data = json.load(f) result = json.dumps(data, indent=2) if isinstance(data, (dict, list)) else str(data) if not result: raise ValueError(f"JSON file is empty: {filename}") return result else: with open(filepath, 'r', encoding='utf-8', errors='replace') as f: text = f.read().strip() if not text: raise ValueError(f"File contains no readable text: {filename}") return text except json.JSONDecodeError as e: raise ValueError(f"Invalid JSON format in file {filename}: {str(e)}") except Exception as e: raise ValueError(f"Failed to extract text from {filename}: {str(e)}") def generate_csrf_token() -> str: return secrets.token_urlsafe(32) def repair_json(raw: str) -> dict: """Iterative strategy-based JSON recovery for LLM outputs.""" raw = raw.strip() # Strip markdown fences for fence in ("```json", "```JSON", "```"): if raw.startswith(fence): raw = raw[len(fence):] break if raw.endswith("```"): raw = raw[:-3] raw = raw.strip() # Find boundaries start_obj, start_arr = raw.find('{'), raw.find('[') start_idx = start_arr if (start_arr != -1 and (start_obj == -1 or start_arr < start_obj)) else start_obj end_idx = raw.rfind(']') if start_idx == start_arr else raw.rfind('}') if start_idx != -1 and end_idx != -1: raw = raw[start_idx:end_idx + 1] # Repair Strategies Pipeline strategies = [ lambda x: x, # Strategy 1: As-is lambda x: re.sub(r',\s*([}\]])', r'\1', x), # Strategy 2: Remove trailing commas lambda x: re.sub(r'(? float: f1 = audit.get("agent_f1_score") if f1 is not None and isinstance(f1, (int, float)) and f1 > 0: return float(f1) qm = audit.get("quality_matrix") or {} # Safely convert to float, defaulting to 5 if missing or malformed def get_score(key: str) -> float: try: return float(qm.get(key, 5)) except (TypeError, ValueError): return 5.0 precision = (get_score("language_proficiency") + get_score("efficiency") + get_score("bias_reduction")) / 30.0 recall = (get_score("cognitive_empathy") + get_score("active_listening")) / 20.0 if precision + recall > 0: return round((2 * precision * recall) / (precision + recall), 2) return 0.50