mirror of
https://gitverse.ru/kpa39l/chronicle.nixg.ru.git
synced 2026-09-29 18:05:07 +00:00
112 lines
3.3 KiB
Python
112 lines
3.3 KiB
Python
"""
|
|
Configuration management for Telegram Archiver.
|
|
Uses pydantic-settings for .env file parsing and validation.
|
|
"""
|
|
|
|
from pydantic_settings import BaseSettings, SettingsConfigDict
|
|
from pydantic import Field, ValidationError
|
|
from pathlib import Path
|
|
from typing import Optional
|
|
import logging
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class SettingsError(Exception):
|
|
"""Raised when settings cannot be loaded."""
|
|
pass
|
|
|
|
|
|
class Settings(BaseSettings):
|
|
"""Application settings loaded from environment variables and .env file."""
|
|
|
|
model_config = SettingsConfigDict(
|
|
env_file=".env",
|
|
env_file_encoding="utf-8",
|
|
case_sensitive=False,
|
|
extra="ignore",
|
|
)
|
|
|
|
# Telegram API credentials - required for archiving to work
|
|
api_id: Optional[int] = Field(None, description="Telegram API ID from my.telegram.org")
|
|
api_hash: Optional[str] = Field(None, description="Telegram API Hash from my.telegram.org")
|
|
phone: Optional[str] = Field(None, description="Phone number for Telegram account")
|
|
|
|
# Session configuration
|
|
session_name: str = Field(
|
|
default="telegram-archiver",
|
|
description="Session name for Telethon (creates .session file)"
|
|
)
|
|
|
|
# File download limits
|
|
max_file_size: int = Field(
|
|
default=209715200, # 200 MB
|
|
description="Maximum file size to download in bytes"
|
|
)
|
|
|
|
# Output configuration
|
|
output_dir: Path = Field(
|
|
default=Path("./archives"),
|
|
description="Directory to store archived channels"
|
|
)
|
|
|
|
# Logging configuration
|
|
log_level: str = Field(
|
|
default="INFO",
|
|
description="Logging level: DEBUG, INFO, WARNING, ERROR"
|
|
)
|
|
|
|
# OpenTelemetry (optional, for future)
|
|
otel_service_name: str = Field(
|
|
default="telegram-archiver",
|
|
description="Service name for OpenTelemetry traces"
|
|
)
|
|
otel_enabled: bool = Field(
|
|
default=False,
|
|
description="Enable OpenTelemetry tracing"
|
|
)
|
|
|
|
@property
|
|
def session_path(self) -> Path:
|
|
"""Return full path to session file."""
|
|
return Path(__file__).parent / f"{self.session_name}.session"
|
|
|
|
@property
|
|
def is_configured(self) -> bool:
|
|
"""Check if credentials are configured."""
|
|
return bool(self.api_id and self.api_hash and self.phone)
|
|
|
|
def validate_credentials(self) -> bool:
|
|
"""Validate that required credentials are present and valid."""
|
|
if not self.is_configured:
|
|
raise SettingsError(
|
|
"Missing Telegram credentials. "
|
|
"Set API_ID, API_HASH, and PHONE in .env file or environment variables."
|
|
)
|
|
if self.api_id <= 0:
|
|
raise ValueError("API_ID must be a positive integer")
|
|
if len(self.api_hash) < 10:
|
|
raise ValueError("API_HASH seems too short")
|
|
if not self.phone.startswith("+"):
|
|
raise ValueError("PHONE must start with +")
|
|
return True
|
|
|
|
|
|
# Global settings instance
|
|
_settings: Optional[Settings] = None
|
|
|
|
|
|
def get_settings() -> Settings:
|
|
"""Get or create settings instance."""
|
|
global _settings
|
|
if _settings is None:
|
|
_settings = Settings()
|
|
return _settings
|
|
|
|
|
|
def get_required_settings() -> Settings:
|
|
"""Get settings and ensure credentials are configured."""
|
|
settings = get_settings()
|
|
settings.validate_credentials()
|
|
return settings
|