import os from dataclasses import dataclass from dotenv import load_dotenv load_dotenv() GOOGLE_SCOPES = ( "https://www.googleapis.com/auth/gmail.modify", "https://www.googleapis.com/auth/gmail.send", "https://www.googleapis.com/auth/calendar.readonly", ) @dataclass(frozen=True) class Settings: google_client_secrets_file: str google_token_file: str gmail_scan_interval_minutes: int gmail_query: str agent_api_key: str auth_mode: str auth_jwt_secret: str auth_jwt_issuer: str | None auth_jwt_audience: str | None strands_api_key: str strands_model_id: str strands_base_url: str | None strands_timeout_seconds: float strands_temperature: float llm_fallback_to_rules: bool unsubscribe_digest_interval_minutes: int unsubscribe_query: str unsubscribe_max_results: int unsubscribe_state_file: str unsubscribe_digest_recipient: str | None unsubscribe_send_empty_digest: bool unsubscribe_hil_query: str unsubscribe_hil_max_results: int unsubscribe_hil_state_file: str unsubscribe_auto_enabled: bool unsubscribe_auto_interval_minutes: int unsubscribe_http_timeout_seconds: float unsubscribe_user_agent: str a2a_public_base_url: str | None a2a_agent_name: str a2a_agent_description: str log_level: str def get_settings() -> Settings: strands_base_url = _first_set_env("STRANDS_OPENAI_BASE_URL", "LLM_BASE_URL").strip() unsubscribe_digest_recipient = os.getenv("UNSUBSCRIBE_DIGEST_RECIPIENT", "").strip() return Settings( google_client_secrets_file=os.getenv("GOOGLE_CLIENT_SECRETS_FILE", "credentials.json"), google_token_file=os.getenv("GOOGLE_TOKEN_FILE", "token.json"), gmail_scan_interval_minutes=int(os.getenv("GMAIL_SCAN_INTERVAL_MINUTES", "5")), gmail_query=os.getenv( "GMAIL_QUERY", "in:inbox is:unread -label:AgentProcessed" ), agent_api_key=os.getenv("AGENT_API_KEY", ""), auth_mode=_normalize_auth_mode(os.getenv("AUTH_MODE", "api_key")), auth_jwt_secret=os.getenv("AUTH_JWT_SECRET", "").strip(), auth_jwt_issuer=os.getenv("AUTH_JWT_ISSUER", "").strip() or None, auth_jwt_audience=os.getenv("AUTH_JWT_AUDIENCE", "").strip() or None, strands_api_key=_first_set_env("STRANDS_OPENAI_API_KEY", "LLM_API_KEY"), strands_model_id=_first_set_env("STRANDS_MODEL_ID", "LLM_MODEL") or "gpt-4.1-mini", strands_base_url=strands_base_url or None, strands_timeout_seconds=float( _first_set_env("STRANDS_TIMEOUT_SECONDS", "LLM_TIMEOUT_SECONDS") or "20" ), strands_temperature=float(os.getenv("STRANDS_TEMPERATURE", "0")), llm_fallback_to_rules=_as_bool(os.getenv("LLM_FALLBACK_TO_RULES", "false")), unsubscribe_digest_interval_minutes=int( os.getenv("UNSUBSCRIBE_DIGEST_INTERVAL_MINUTES", "1440") ), unsubscribe_query=os.getenv("UNSUBSCRIBE_QUERY", "label:Advertising"), unsubscribe_max_results=int(os.getenv("UNSUBSCRIBE_MAX_RESULTS", "500")), unsubscribe_state_file=os.getenv( "UNSUBSCRIBE_STATE_FILE", "data/sent_unsubscribe_links.json" ), unsubscribe_digest_recipient=unsubscribe_digest_recipient or None, unsubscribe_send_empty_digest=_as_bool( os.getenv("UNSUBSCRIBE_SEND_EMPTY_DIGEST", "false") ), unsubscribe_hil_query=os.getenv("UNSUBSCRIBE_HIL_QUERY", "label:Advertising"), unsubscribe_hil_max_results=int(os.getenv("UNSUBSCRIBE_HIL_MAX_RESULTS", "500")), unsubscribe_hil_state_file=os.getenv( "UNSUBSCRIBE_HIL_STATE_FILE", "data/unsubscribed_methods.json" ), unsubscribe_auto_enabled=_as_bool(os.getenv("UNSUBSCRIBE_AUTO_ENABLED", "true")), unsubscribe_auto_interval_minutes=int( os.getenv("UNSUBSCRIBE_AUTO_INTERVAL_MINUTES", "720") ), unsubscribe_http_timeout_seconds=float( os.getenv("UNSUBSCRIBE_HTTP_TIMEOUT_SECONDS", "12") ), unsubscribe_user_agent=os.getenv( "UNSUBSCRIBE_USER_AGENT", "Mozilla/5.0 (compatible; PersonalAgentUnsubscribe/1.0; +https://example.local)", ), a2a_public_base_url=os.getenv("A2A_PUBLIC_BASE_URL", "").strip() or None, a2a_agent_name=os.getenv("A2A_AGENT_NAME", "Personal Agent"), a2a_agent_description=os.getenv( "A2A_AGENT_DESCRIPTION", "Personal productivity agent for calendar availability and email operations.", ), log_level=os.getenv("LOG_LEVEL", "INFO"), ) def _as_bool(value: str) -> bool: return value.strip().lower() in {"1", "true", "yes", "on"} def _first_set_env(*names: str) -> str: for name in names: value = os.getenv(name) if value: return value.strip() return "" def _normalize_auth_mode(value: str) -> str: normalized = value.strip().lower() if normalized in {"api_key", "jwt", "hybrid"}: return normalized return "api_key"