109 lines
4.3 KiB
Python
109 lines
4.3 KiB
Python
import asyncio
|
||
import random
|
||
import re
|
||
|
||
import aiohttp
|
||
|
||
from config import (
|
||
HEADERS,
|
||
TG_JITTER,
|
||
TG_MAX_HEAD_BYTES,
|
||
TG_MIN_INTERVAL,
|
||
TG_REQUEST_TIMEOUT,
|
||
TG_RETRIES,
|
||
)
|
||
from rate_limiter import RateLimiter
|
||
|
||
OG_TITLE_RE = re.compile(rb'property=["\']og:title["\']\s+content=["\']([^"\']*)["\']')
|
||
OG_IMAGE_RE = re.compile(rb'property=["\']og:image["\']\s+content=["\']([^"\']*)["\']')
|
||
OG_DESC_RE = re.compile(
|
||
rb'property=["\']og:description["\']\s+content=["\']([^"\']*)["\']'
|
||
)
|
||
|
||
# один общий лимитер на всё приложение: t.me не должен видеть больше одного
|
||
# запроса раз в TG_MIN_INTERVAL секунд, сколько бы задач ни ждало своей очереди
|
||
_rate_limiter = RateLimiter(min_interval=TG_MIN_INTERVAL, jitter=TG_JITTER)
|
||
|
||
|
||
async def _fetch_head(session, url: str, max_bytes: int = TG_MAX_HEAD_BYTES) -> bytes:
|
||
"""Читает страницу кусками и останавливается сразу после </head> -
|
||
сам текст страницы не нужен, только og-теги в шапке."""
|
||
timeout = aiohttp.ClientTimeout(total=TG_REQUEST_TIMEOUT)
|
||
async with session.get(url, timeout=timeout, headers=HEADERS) as resp:
|
||
if resp.status != 200:
|
||
raise aiohttp.ClientResponseError(
|
||
resp.request_info, resp.history, status=resp.status
|
||
)
|
||
|
||
buf = b""
|
||
async for chunk in resp.content.iter_chunked(1024):
|
||
buf += chunk
|
||
if b"</head>" in buf or len(buf) >= max_bytes:
|
||
break
|
||
return buf
|
||
|
||
|
||
def _extract(pattern, html: bytes) -> str:
|
||
match = pattern.search(html)
|
||
return match.group(1).decode() if match else ""
|
||
|
||
|
||
def _looks_unregistered(
|
||
username: str, title: str, image: str, description: str
|
||
) -> bool:
|
||
"""Для несуществующих/удалённых юзернеймов телега всегда рисует одну и
|
||
ту же болванку: заголовок "Telegram: Contact @username", дефолтный
|
||
логотип и пустое описание. Если все три совпали - юзернейм свободен."""
|
||
expected_title = f"telegram: contact @{username}".lower()
|
||
return (
|
||
title.lower() == expected_title
|
||
and image == "https://telegram.org/img/t_logo_2x.png"
|
||
and description == ""
|
||
)
|
||
|
||
|
||
async def check_telegram(session, tg_url: str) -> str | None:
|
||
"""None - юзернейм занят или проверить не вышло. Строка - похоже свободен."""
|
||
username = tg_url.rstrip("/").split("/")[-1]
|
||
if not username:
|
||
print(f"Кривая ссылка на тг: {tg_url}")
|
||
return None
|
||
|
||
html = None
|
||
for attempt in range(1, TG_RETRIES + 1):
|
||
await _rate_limiter.wait()
|
||
try:
|
||
html = await _fetch_head(session, f"https://t.me/{username}")
|
||
break
|
||
except (aiohttp.ClientError, asyncio.TimeoutError) as e:
|
||
backoff = 2**attempt + random.uniform(0, 1)
|
||
print(
|
||
f" @{username}: попытка {attempt}/{TG_RETRIES} не удалась ({e}), жду {backoff:.1f}с"
|
||
)
|
||
await asyncio.sleep(backoff)
|
||
|
||
if html is None:
|
||
print(
|
||
f" @{username}: не смог получить страницу за {TG_RETRIES} попыток, пропускаю"
|
||
)
|
||
return None
|
||
|
||
title = _extract(OG_TITLE_RE, html)
|
||
image = _extract(OG_IMAGE_RE, html)
|
||
description = _extract(OG_DESC_RE, html)
|
||
|
||
print(f" @{username}: title={title!r} image={image!r} desc={description!r}")
|
||
|
||
if not title and not image and not description:
|
||
# пустые og-теги почти всегда значат, что нас притормозили,
|
||
# а не что страница правда пустая - за "свободно" это не считаем
|
||
print(f" @{username}: пустой ответ, похоже попали под лимит - пропускаю")
|
||
return None
|
||
|
||
if _looks_unregistered(username, title, image, description):
|
||
print(f" @{username}: похоже свободен")
|
||
return username
|
||
|
||
print(f" @{username}: занят")
|
||
return None
|