Files
catalogue_backend/app/infrastructure/settings.py
2026-08-11 19:16:01 +05:30

150 lines
7.2 KiB
Python

"""
Centralized configuration for the AI Product Catalog + RAG backend.
SECURITY NOTE
--------------
The previous version of this project had a serious problem: `settings.py`
hard-coded a *live* database host, port and password as Python literal
fallbacks (`os.getenv("DB_HOST", "<real ip>")`, etc.). That means the
real production credentials shipped inside the source code itself - in
every copy, every zip export, and every git commit - regardless of
whether a `.env` file was present.
This rewrite removes every hard-coded secret. Every credential
(DB_PASSWORD, S3 keys, Google API key, ...) is read ONLY from the
environment (via a local `.env` file, loaded through python-dotenv, or
real OS/container environment variables). Non-secret values (ports,
feature flags, model names) keep sane, publicly-safe defaults so the
project still boots out of the box for local development.
If a secret-shaped variable is required for a feature that is enabled
(e.g. `USE_PGVECTOR=true` but no `DB_PASSWORD` set), we raise a clear
`RuntimeError` at settings-load time instead of silently connecting
with an empty/placeholder password. Fail loudly, not insecurely.
"""
from __future__ import annotations
import os
from pathlib import Path
try:
from dotenv import load_dotenv
# backend/.env (one level up from this file: app/infrastructure/settings.py)
_env_path = Path(__file__).resolve().parents[2] / ".env"
load_dotenv(_env_path)
except ImportError:
# python-dotenv not installed - fall back to whatever is already in the
# process environment (e.g. set by the shell, Docker, systemd, CI, etc.)
pass
def _bool(name: str, default: str) -> bool:
return os.getenv(name, default).strip().lower() in {"1", "true", "yes"}
def _require(name: str, *, feature_flag: str) -> str:
"""Read a required secret. Raises if missing and the owning feature is enabled."""
value = os.getenv(name)
if not value:
raise RuntimeError(
f"Missing required environment variable '{name}'. It is required because "
f"'{feature_flag}' is enabled. Set it in backend/.env (copy from "
f".env.example) or disable the feature by setting {feature_flag}=false."
)
return value
# ---------------------------------------------------------------------------
# Ollama (local LLM)
# ---------------------------------------------------------------------------
USE_OLLAMA = _bool("USE_OLLAMA", "true")
OLLAMA_BASE_URL = os.getenv("OLLAMA_BASE_URL", "http://localhost:11434")
OLLAMA_MODEL_NAME = os.getenv("OLLAMA_MODEL_NAME", "qwen2.5:1.5b")
# Per-request generation timeout (seconds). Small CPU-only models on modest
# hardware (e.g. 8GB RAM, no GPU) can take a while for longer RAG contexts.
OLLAMA_TIMEOUT_SECONDS = int(os.getenv("OLLAMA_TIMEOUT_SECONDS", "120"))
# ---------------------------------------------------------------------------
# Embeddings (sentence-transformers, CPU-friendly)
# ---------------------------------------------------------------------------
USE_EMBEDDINGS = _bool("USE_EMBEDDINGS", "true")
EMBEDDINGS_MODEL = os.getenv("EMBEDDINGS_MODEL", "sentence-transformers/all-MiniLM-L6-v2")
EMBEDDINGS_DIM = int(os.getenv("EMBEDDINGS_DIM", "384"))
# ---------------------------------------------------------------------------
# Postgres / pgvector
# ---------------------------------------------------------------------------
USE_PGVECTOR = _bool("USE_PGVECTOR", "true")
DB_HOST = os.getenv("DB_HOST", "localhost")
DB_PORT = os.getenv("DB_PORT", "5432")
DB_NAME = os.getenv("DB_NAME", "pgvector")
DB_USER = os.getenv("DB_USER", "postgres")
DB_PASSWORD = _require("DB_PASSWORD", feature_flag="USE_PGVECTOR") if USE_PGVECTOR else os.getenv("DB_PASSWORD", "")
DATABASE_URL = os.getenv(
"DATABASE_URL",
f"postgresql://{DB_USER}:{DB_PASSWORD}@{DB_HOST}:{DB_PORT}/{DB_NAME}",
)
# ---------------------------------------------------------------------------
# S3 / DigitalOcean Spaces (product image storage) - optional
# ---------------------------------------------------------------------------
USE_S3 = _bool("USE_S3", "false")
S3_ACCESS_KEY = _require("S3_ACCESS_KEY", feature_flag="USE_S3") if USE_S3 else os.getenv("S3_ACCESS_KEY")
S3_SECRET_KEY = _require("S3_SECRET_KEY", feature_flag="USE_S3") if USE_S3 else os.getenv("S3_SECRET_KEY")
S3_ENDPOINT = _require("S3_ENDPOINT", feature_flag="USE_S3") if USE_S3 else os.getenv("S3_ENDPOINT")
S3_BUCKET = _require("S3_BUCKET", feature_flag="USE_S3") if USE_S3 else os.getenv("S3_BUCKET")
S3_REGION = os.getenv("S3_REGION", "sgp1")
# ---------------------------------------------------------------------------
# Google Custom Search (OPTIONAL image source - leave disabled if unset)
# ---------------------------------------------------------------------------
USE_GOOGLE_CSE = _bool("USE_GOOGLE_CSE", "false")
GOOGLE_API_KEY = _require("GOOGLE_API_KEY", feature_flag="USE_GOOGLE_CSE") if USE_GOOGLE_CSE else os.getenv("GOOGLE_API_KEY")
GOOGLE_CSE_ID = _require("GOOGLE_CSE_ID", feature_flag="USE_GOOGLE_CSE") if USE_GOOGLE_CSE else os.getenv("GOOGLE_CSE_ID")
# ---------------------------------------------------------------------------
# Open-source image sources (no API key needed for any of these three)
# ---------------------------------------------------------------------------
USE_DDG_IMAGES = _bool("USE_DDG_IMAGES", "true")
USE_OPEN_FACTS = _bool("USE_OPEN_FACTS", "true")
USE_WIKIMEDIA = _bool("USE_WIKIMEDIA", "true")
# Last-resort headless-browser (Python Playwright) image fallback. Requires
# `pip install playwright && playwright install chromium`; automatically
# skipped (logged once) if that hasn't been done.
USE_PLAYWRIGHT_FALLBACK = _bool("USE_PLAYWRIGHT_FALLBACK", "true")
# Minimum byte size for a downloaded image to be accepted as "real" (filters
# out 1x1 tracking pixels / broken placeholder images)
MIN_IMAGE_BYTES = int(os.getenv("MIN_IMAGE_BYTES", "3000"))
# ---------------------------------------------------------------------------
# HTTP client defaults
# ---------------------------------------------------------------------------
USER_AGENT = os.getenv("USER_AGENT", "CatalogBot/1.0 (+https://example.com)")
REQUEST_TIMEOUT_SECONDS = int(os.getenv("REQUEST_TIMEOUT_SECONDS", "20"))
# ---------------------------------------------------------------------------
# FastAPI / web server
# ---------------------------------------------------------------------------
API_CORS_ORIGINS = [
origin.strip()
for origin in os.getenv("API_CORS_ORIGINS", "http://localhost:5173,http://127.0.0.1:5173").split(",")
if origin.strip()
]
# Default RAG behaviour
RAG_DEFAULT_TOP_K = int(os.getenv("RAG_DEFAULT_TOP_K", "5"))
RAG_MAX_TOP_K = int(os.getenv("RAG_MAX_TOP_K", "15"))
RAG_MAX_CONTEXT_CHARS = int(os.getenv("RAG_MAX_CONTEXT_CHARS", "4000"))
# Optional pgvector cosine-distance ceiling (0 = identical, 2 = opposite)
# used to drop weak semantic matches before they reach the LLM. Left
# unset (None) by default since the primary relevance guardrail is the
# category-aware retrieval in rag_service.py; set RAG_MAX_DISTANCE (e.g.
# "0.9") once you've inspected real distance scores for your embeddings
# if you want an extra cutoff on top of that.
_raw_max_distance = os.getenv("RAG_MAX_DISTANCE", "").strip()
RAG_MAX_DISTANCE = float(_raw_max_distance) if _raw_max_distance else None