"""Runtime settings (env / .env) - single source of truth for configuration. Symbols are NOT configured here; see `us2kr.symbols`. """ from __future__ import annotations from datetime import date from pathlib import Path from typing import Annotated from pydantic import Field, field_validator from pydantic_settings import BaseSettings, NoDecode, SettingsConfigDict KST = "Asia/Seoul" ET = "America/New_York" class Settings(BaseSettings): model_config = SettingsConfigDict( env_file=".env", env_file_encoding="utf-8", extra="ignore", populate_by_name=True ) # --- MiniMax (OpenAI-compatible endpoint) --- minimax_api_key: str = Field(default="", alias="MINIMAX_API_KEY") minimax_base_url: str = Field(default="https://api.minimax.io/v1", alias="MINIMAX_BASE_URL") minimax_model: str = Field(default="MiniMax-M3", alias="MINIMAX_MODEL") minimax_fallback_models: Annotated[list[str], NoDecode] = Field( default_factory=lambda: ["MiniMax-M2.7", "MiniMax-M2.5"], alias="MINIMAX_FALLBACK_MODELS" ) minimax_max_completion_tokens: int = Field(default=16384, alias="MINIMAX_MAX_COMPLETION_TOKENS") minimax_timeout_seconds: float = Field(default=180.0, alias="MINIMAX_TIMEOUT_SECONDS") # --- Delivery --- telegram_bot_token: str | None = Field(default=None, alias="TELEGRAM_BOT_TOKEN") telegram_chat_id: str | None = Field(default=None, alias="TELEGRAM_CHAT_ID") # --- Blogger (optional; issue the refresh token once with `us2kr blogger-auth`) --- blogger_client_id: str | None = Field(default=None, alias="BLOGGER_CLIENT_ID") blogger_client_secret: str | None = Field(default=None, alias="BLOGGER_CLIENT_SECRET") blogger_refresh_token: str | None = Field(default=None, alias="BLOGGER_REFRESH_TOKEN") blogger_blog_id: str | None = Field(default=None, alias="BLOGGER_BLOG_ID") blogger_labels: Annotated[list[str], NoDecode] = Field( default_factory=lambda: ["미국증시", "마감브리핑"], alias="BLOGGER_LABELS" ) # --- Optional data keys --- fred_api_key: str | None = Field(default=None, alias="FRED_API_KEY") # --- SearXNG (optional news search; instance must enable the json output format) --- searxng_base_url: str | None = Field(default=None, alias="SEARXNG_BASE_URL") # Empty -> collector defaults (us2kr.collectors.searx.DEFAULT_QUERIES). searxng_queries: Annotated[list[str], NoDecode] = Field( default_factory=list, alias="US2KR_SEARXNG_QUERIES" ) # --- Paths --- data_dir: Path = Field(default=Path("data"), alias="US2KR_DATA_DIR") reports_dir: Path = Field(default=Path("reports"), alias="US2KR_REPORTS_DIR") log_dir: Path = Field(default=Path("logs"), alias="US2KR_LOG_DIR") # --- Scheduling policy --- run_offset_minutes: int = Field(default=75, alias="US2KR_RUN_OFFSET_MINUTES") catchup_window_hours: int = Field(default=3, alias="US2KR_CATCHUP_WINDOW_HOURS") # Skip only briefings that become pointless: a KRX holiday where the *next* US session closes before # KRX trades again (e.g. mid-week Chuseok). Friday sessions are always delivered (used for Monday open). skip_when_krx_closed: bool = Field(default=False, alias="US2KR_SKIP_WHEN_KRX_CLOSED") # Ad-hoc KRX closures missing from exchange_calendars (e.g. election days), CSV of YYYY-MM-DD. krx_extra_holidays: Annotated[list[date], NoDecode] = Field(default_factory=list, alias="US2KR_KRX_EXTRA_HOLIDAYS") premarket_update_enabled: bool = Field(default=False, alias="US2KR_PREMARKET_UPDATE_ENABLED") premarket_time_kst: str = Field(default="08:20", alias="US2KR_PREMARKET_TIME_KST") # --- Collection policy --- news_lookback_hours: int = Field(default=18, alias="US2KR_NEWS_LOOKBACK_HOURS") news_max_items: int = Field(default=40, alias="US2KR_NEWS_MAX_ITEMS") # --- LLM verification (fact-check pass over the generated report) --- verify_enabled: bool = Field(default=True, alias="US2KR_VERIFY_ENABLED") log_level: str = Field(default="INFO", alias="US2KR_LOG_LEVEL") @field_validator( "minimax_fallback_models", "krx_extra_holidays", "searxng_queries", "blogger_labels", mode="before" ) @classmethod def _split_csv(cls, v): if isinstance(v, str): return [s.strip() for s in v.split(",") if s.strip()] return v @property def db_path(self) -> Path: return self.data_dir / "us2kr.sqlite3" @property def telegram_enabled(self) -> bool: return bool(self.telegram_bot_token and self.telegram_chat_id) @property def blogger_enabled(self) -> bool: return bool( self.blogger_client_id and self.blogger_client_secret and self.blogger_refresh_token and self.blogger_blog_id ) def get_settings() -> Settings: """Load settings from environment / .env (called once per process by the CLI).""" return Settings()