diff --git a/tools/dc.py b/tools/dc.py index f142baf..d431966 100644 --- a/tools/dc.py +++ b/tools/dc.py @@ -1,121 +1,204 @@ +"""Telegram datacenter server checker.""" + +from __future__ import annotations + +import asyncio +from dataclasses import dataclass, field +from itertools import groupby +from operator import attrgetter +from pathlib import Path +from typing import TYPE_CHECKING + from telethon import TelegramClient from telethon.tl.functions.help import GetConfigRequest -import asyncio -api_id = '' -api_hash = '' +if TYPE_CHECKING: + from telethon.tl.types import DcOption -async def get_all_servers(): - print("🔄 Подключаемся к Telegram...") - client = TelegramClient('session', api_id, api_hash) - - await client.start() - print("✅ Подключение установлено!\n") - - print("📡 Запрашиваем конфигурацию серверов...") - config = await client(GetConfigRequest()) - - print(f"📊 Получено серверов: {len(config.dc_options)}\n") - print("="*80) - - # Группируем серверы по DC ID - dc_groups = {} - for dc in config.dc_options: - if dc.id not in dc_groups: - dc_groups[dc.id] = [] - dc_groups[dc.id].append(dc) - - # Выводим все серверы, сгруппированные по DC - for dc_id in sorted(dc_groups.keys()): - servers = dc_groups[dc_id] - print(f"\n🌐 DATACENTER {dc_id} ({len(servers)} серверов)") - print("-" * 80) - - for dc in servers: - # Собираем флаги - flags = [] - if dc.ipv6: - flags.append("IPv6") - if dc.media_only: - flags.append("🎬 MEDIA-ONLY") - if dc.cdn: - flags.append("📦 CDN") - if dc.tcpo_only: - flags.append("🔒 TCPO") - if dc.static: - flags.append("📌 STATIC") - - flags_str = f" [{', '.join(flags)}]" if flags else " [STANDARD]" - - # Форматируем IP (выравниваем для читаемости) - ip_display = f"{dc.ip_address:45}" - - print(f" {ip_display}:{dc.port:5}{flags_str}") - - # Статистика - print("\n" + "="*80) - print("📈 СТАТИСТИКА:") - print("="*80) - - total = len(config.dc_options) - ipv4_count = sum(1 for dc in config.dc_options if not dc.ipv6) - ipv6_count = sum(1 for dc in config.dc_options if dc.ipv6) - media_count = sum(1 for dc in config.dc_options if dc.media_only) - cdn_count = sum(1 for dc in config.dc_options if dc.cdn) - tcpo_count = sum(1 for dc in config.dc_options if dc.tcpo_only) - static_count = sum(1 for dc in config.dc_options if dc.static) - - print(f" Всего серверов: {total}") - print(f" IPv4 серверы: {ipv4_count}") - print(f" IPv6 серверы: {ipv6_count}") - print(f" Media-only: {media_count}") - print(f" CDN серверы: {cdn_count}") - print(f" TCPO-only: {tcpo_count}") - print(f" Static: {static_count}") - - # Дополнительная информация из config - print("\n" + "="*80) - print("ℹ️ ДОПОЛНИТЕЛЬНАЯ ИНФОРМАЦИЯ:") - print("="*80) - print(f" Дата конфигурации: {config.date}") - print(f" Expires: {config.expires}") - print(f" Test mode: {config.test_mode}") - print(f" This DC: {config.this_dc}") - - # Сохраняем в файл - print("\n💾 Сохраняем результаты в файл telegram_servers.txt...") - with open('telegram_servers.txt', 'w', encoding='utf-8') as f: - f.write("TELEGRAM DATACENTER SERVERS\n") - f.write("="*80 + "\n\n") - - for dc_id in sorted(dc_groups.keys()): - servers = dc_groups[dc_id] - f.write(f"\nDATACENTER {dc_id} ({len(servers)} servers)\n") - f.write("-" * 80 + "\n") - - for dc in servers: - flags = [] - if dc.ipv6: - flags.append("IPv6") - if dc.media_only: - flags.append("MEDIA-ONLY") - if dc.cdn: - flags.append("CDN") - if dc.tcpo_only: - flags.append("TCPO") - if dc.static: - flags.append("STATIC") - - flags_str = f" [{', '.join(flags)}]" if flags else " [STANDARD]" - f.write(f" {dc.ip_address}:{dc.port}{flags_str}\n") - - f.write(f"\n\nTotal servers: {total}\n") - f.write(f"Generated: {config.date}\n") - - print("✅ Результаты сохранены в telegram_servers.txt") - - await client.disconnect() - print("\n👋 Отключились от Telegram") +API_ID: int = 123456 +API_HASH: str = "" +SESSION_NAME: str = "session" +OUTPUT_FILE: Path = Path("telegram_servers.txt") -if __name__ == '__main__': - asyncio.run(get_all_servers()) \ No newline at end of file +_CONSOLE_FLAG_MAP: dict[str, str] = { + "IPv6": "IPv6", + "MEDIA-ONLY": "🎬 MEDIA-ONLY", + "CDN": "📦 CDN", + "TCPO": "🔒 TCPO", + "STATIC": "📌 STATIC", +} + + +@dataclass(frozen=True, slots=True) +class DCServer: + """Typed representation of a Telegram DC server. + + Attributes: + dc_id: Datacenter identifier. + ip: Server IP address. + port: Server port. + flags: Active flag labels (plain, without emoji). + """ + + dc_id: int + ip: str + port: int + flags: frozenset[str] = field(default_factory=frozenset) + + @classmethod + def from_option(cls, dc: DcOption) -> DCServer: + """Create from a Telethon DcOption. + + Args: + dc: Raw DcOption object. + + Returns: + Parsed DCServer instance. + """ + checks: dict[str, bool] = { + "IPv6": dc.ipv6, + "MEDIA-ONLY": dc.media_only, + "CDN": dc.cdn, + "TCPO": dc.tcpo_only, + "STATIC": dc.static, + } + return cls( + dc_id=dc.id, + ip=dc.ip_address, + port=dc.port, + flags=frozenset(k for k, v in checks.items() if v), + ) + + def flags_display(self, *, emoji: bool = False) -> str: + """Formatted flags string. + + Args: + emoji: Whether to include emoji prefixes. + + Returns: + Bracketed flags or '[STANDARD]'. + """ + if not self.flags: + return "[STANDARD]" + labels = sorted( + _CONSOLE_FLAG_MAP[f] if emoji else f for f in self.flags + ) + return f"[{', '.join(labels)}]" + + +class TelegramDCChecker: + """Fetches and displays Telegram DC configuration. + + Attributes: + _client: Telethon client instance. + _servers: Parsed server list. + """ + + def __init__(self) -> None: + """Initialize the checker.""" + self._client = TelegramClient(SESSION_NAME, API_ID, API_HASH) + self._servers: list[DCServer] = [] + + async def run(self) -> None: + """Connect, fetch config, display and save results.""" + print("🔄 Подключаемся к Telegram...") # noqa: T201 + try: + await self._client.start() + print("✅ Подключение установлено!\n") # noqa: T201 + + print("📡 Запрашиваем конфигурацию серверов...") # noqa: T201 + config = await self._client(GetConfigRequest()) + self._servers = [DCServer.from_option(dc) for dc in config.dc_options] + + self._print(config) + self._save(config) + finally: + await self._client.disconnect() + print("\n👋 Отключились от Telegram") # noqa: T201 + + def _grouped(self) -> dict[int, list[DCServer]]: + """Group servers by DC ID. + + Returns: + Ordered mapping of DC ID to servers. + """ + ordered = sorted(self._servers, key=attrgetter("dc_id")) + return {k: list(g) for k, g in groupby(ordered, key=attrgetter("dc_id"))} + + def _print(self, config: object) -> None: + """Print results to stdout in original format. + + Args: + config: Raw Telegram config. + """ + sep = "=" * 80 + dash = "-" * 80 + total = len(self._servers) + + print(f"📊 Получено серверов: {total}\n") # noqa: T201 + print(sep) # noqa: T201 + + for dc_id, servers in self._grouped().items(): + print(f"\n🌐 DATACENTER {dc_id} ({len(servers)} серверов)") # noqa: T201 + print(dash) # noqa: T201 + for s in servers: + print(f" {s.ip:45}:{s.port:5} {s.flags_display(emoji=True)}") # noqa: T201 + + ipv4 = total - self._flag_count("IPv6") + print(f"\n{sep}") # noqa: T201 + print("📈 СТАТИСТИКА:") # noqa: T201 + print(sep) # noqa: T201 + print(f" Всего серверов: {total}") # noqa: T201 + print(f" IPv4 серверы: {ipv4}") # noqa: T201 + print(f" IPv6 серверы: {self._flag_count('IPv6')}") # noqa: T201 + print(f" Media-only: {self._flag_count('MEDIA-ONLY')}") # noqa: T201 + print(f" CDN серверы: {self._flag_count('CDN')}") # noqa: T201 + print(f" TCPO-only: {self._flag_count('TCPO')}") # noqa: T201 + print(f" Static: {self._flag_count('STATIC')}") # noqa: T201 + + print(f"\n{sep}") # noqa: T201 + print("ℹ️ ДОПОЛНИТЕЛЬНАЯ ИНФОРМАЦИЯ:") # noqa: T201 + print(sep) # noqa: T201 + print(f" Дата конфигурации: {config.date}") # noqa: T201 # type: ignore[attr-defined] + print(f" Expires: {config.expires}") # noqa: T201 # type: ignore[attr-defined] + print(f" Test mode: {config.test_mode}") # noqa: T201 # type: ignore[attr-defined] + print(f" This DC: {config.this_dc}") # noqa: T201 # type: ignore[attr-defined] + + def _flag_count(self, flag: str) -> int: + """Count servers with a given flag. + + Args: + flag: Flag name. + + Returns: + Count of matching servers. + """ + return sum(1 for s in self._servers if flag in s.flags) + + def _save(self, config: object) -> None: + """Save results to file in original format. + + Args: + config: Raw Telegram config. + """ + parts: list[str] = [] + parts.append("TELEGRAM DATACENTER SERVERS\n") + parts.append("=" * 80 + "\n\n") + + for dc_id, servers in self._grouped().items(): + parts.append(f"\nDATACENTER {dc_id} ({len(servers)} servers)\n") + parts.append("-" * 80 + "\n") + for s in servers: + parts.append(f" {s.ip}:{s.port} {s.flags_display(emoji=False)}\n") + + parts.append(f"\n\nTotal servers: {len(self._servers)}\n") + parts.append(f"Generated: {config.date}\n") # type: ignore[attr-defined] + + OUTPUT_FILE.write_text("".join(parts), encoding="utf-8") + + print(f"\n💾 Сохраняем результаты в файл {OUTPUT_FILE}...") # noqa: T201 + print(f"✅ Результаты сохранены в {OUTPUT_FILE}") # noqa: T201 + + +if __name__ == "__main__": + asyncio.run(TelegramDCChecker().run())