first commit
This commit is contained in:
70
app/config.py
Normal file
70
app/config.py
Normal file
@@ -0,0 +1,70 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
from dataclasses import dataclass
|
||||
from pathlib import Path
|
||||
|
||||
from dotenv import load_dotenv
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class Settings:
|
||||
telegram_bot_token: str
|
||||
telegram_channel_id: str
|
||||
telegram_proxy_url: str | None
|
||||
scheduler_interval_minutes: int
|
||||
kwork_category_ids: list[int]
|
||||
keywords_include: list[str]
|
||||
keywords_exclude: list[str]
|
||||
min_price_amount: int | None
|
||||
requests_timeout_seconds: int
|
||||
state_file: Path
|
||||
|
||||
|
||||
def load_settings() -> Settings:
|
||||
load_dotenv()
|
||||
|
||||
telegram_bot_token = _require_env("TELEGRAM_BOT_TOKEN")
|
||||
telegram_channel_id = _require_env("TELEGRAM_CHANNEL_ID")
|
||||
|
||||
return Settings(
|
||||
telegram_bot_token=telegram_bot_token,
|
||||
telegram_channel_id=telegram_channel_id,
|
||||
telegram_proxy_url=_parse_optional_text(os.getenv("TELEGRAM_PROXY_URL", "")),
|
||||
scheduler_interval_minutes=int(os.getenv("SCHEDULER_INTERVAL_MINUTES", "5")),
|
||||
kwork_category_ids=_parse_int_list(os.getenv("KWORK_CATEGORY_IDS", "41")),
|
||||
keywords_include=_parse_text_list(os.getenv("KEYWORDS_INCLUDE", "")),
|
||||
keywords_exclude=_parse_text_list(os.getenv("KEYWORDS_EXCLUDE", "")),
|
||||
min_price_amount=_parse_optional_int(os.getenv("MIN_PRICE_AMOUNT", "")),
|
||||
requests_timeout_seconds=int(os.getenv("REQUESTS_TIMEOUT_SECONDS", "20")),
|
||||
state_file=Path(os.getenv("STATE_FILE", "data/sent_offers.json")),
|
||||
)
|
||||
|
||||
|
||||
def _require_env(name: str) -> str:
|
||||
value = os.getenv(name)
|
||||
if not value:
|
||||
raise ValueError(f"Environment variable {name} is required")
|
||||
return value
|
||||
|
||||
|
||||
def _parse_text_list(raw_value: str) -> list[str]:
|
||||
return [item.strip().lower() for item in raw_value.split(",") if item.strip()]
|
||||
|
||||
|
||||
def _parse_int_list(raw_value: str) -> list[int]:
|
||||
return [int(item.strip()) for item in raw_value.split(",") if item.strip()]
|
||||
|
||||
|
||||
def _parse_optional_int(raw_value: str) -> int | None:
|
||||
value = raw_value.strip()
|
||||
if not value:
|
||||
return None
|
||||
return int(value)
|
||||
|
||||
|
||||
def _parse_optional_text(raw_value: str) -> str | None:
|
||||
value = raw_value.strip()
|
||||
if not value:
|
||||
return None
|
||||
return value
|
||||
Reference in New Issue
Block a user