# ABM Engine configuration — all optional at boot, degrade gracefully. import os from typing import List from pydantic import ConfigDict from pydantic_settings import BaseSettings class Settings(BaseSettings): app_name: str = "FortyDN ABM Engine" app_version: str = "1.0.0" debug: bool = False # ── Supabase identity (same project as the validator) ──────────────── supabase_url: str = "" supabase_service_role_key: str = "" license_enforcement: str = "" # "true" gates paid routes on licenses # ── Aiven OpenSearch (primary store + search + k-NN vectors) ───────── os_url: str = "" os_user: str = "" os_password: str = "" os_index_prefix: str = "abm" # ── Apify enrichment ───────────────────────────────────────────────── apify_token: str = "" apify_actor_google_maps: str = "sitcod3.lab/fdn-google-map-master-scraper" apify_actor_linkedin: str = "sitcod3.lab/linkedin-company-email-scraper" apify_run_timeout_s: int = 900 # ── Optional graph mirror: BetterDB (managed Neo4j) or direct Aura ─── betterdb_url: str = "" betterdb_token: str = "" neo4j_uri: str = "" neo4j_user: str = "" neo4j_password: str = "" # ── Optional Upstash Redis cache (REST) ───────────────────────────── upstash_rest_url: str = "" upstash_rest_token: str = "" # ── Embeddings ─────────────────────────────────────────────────────── embedding_provider: str = "auto" # auto | fastembed | hash embedding_dim: int = 384 # ── ICP scoring defaults (override via env) ───────────────────────── icp_industries: str = "software,saas,technology,manufacturing,logistics,healthcare,finance,professional services" icp_countries: str = "united states,usa,canada,united kingdom,germany,netherlands,australia,denmark,sweden" icp_min_headcount: int = 10 icp_max_headcount: int = 2000 # ── Access ─────────────────────────────────────────────────────────── api_key: str = "" # optional X-API-Key for service-to-service calls @property def icp_industries_list(self) -> List[str]: return [x.strip().lower() for x in self.icp_industries.split(",") if x.strip()] @property def icp_countries_list(self) -> List[str]: return [x.strip().lower() for x in self.icp_countries.split(",") if x.strip()] @property def os_configured(self) -> bool: return bool(self.os_url) @property def apify_configured(self) -> bool: return bool(self.apify_token) @property def graph_configured(self) -> bool: return bool((self.betterdb_url and self.betterdb_token) or self.neo4j_uri) @property def upstash_configured(self) -> bool: return bool(self.upstash_rest_url and self.upstash_rest_token) model_config = ConfigDict(env_file=".env", extra="ignore") settings = Settings()