mirror of
https://gitverse.ru/kpa39l/chronicle.nixg.ru.git
synced 2026-09-29 09:55:08 +00:00
88 lines
2.6 KiB
Python
88 lines
2.6 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
|
|
from pathlib import Path
|
|
from typing import Optional
|
|
|
|
|
|
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
|
|
api_id: int = Field(..., description="Telegram API ID from my.telegram.org")
|
|
api_hash: str = Field(..., description="Telegram API Hash from my.telegram.org")
|
|
phone: str = Field(..., 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"
|
|
|
|
def validate_credentials(self) -> bool:
|
|
"""Validate that required credentials are present."""
|
|
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()
|
|
settings.validate_credentials()
|
|
return settings
|