izzzzzi
379 lines
16 KiB
Python
379 lines
16 KiB
Python
import asyncio
|
||
import base64
|
||
import random
|
||
import re
|
||
import time
|
||
import urllib.parse
|
||
from datetime import datetime
|
||
|
||
import aiohttp
|
||
import asyncpg
|
||
import pytz
|
||
from aiohttp import web
|
||
|
||
from config import (
|
||
DATABASE_URL,
|
||
PROJECT_NAME,
|
||
SUB_MESSAGE,
|
||
SUPERNODE,
|
||
SUPPORT_CHAT_URL,
|
||
TOTAL_GB,
|
||
TRANSITION_DATE_STR,
|
||
USE_COUNTRY_SELECTION,
|
||
USERNAME_BOT,
|
||
)
|
||
from database import get_key_details, get_servers
|
||
from handlers.utils import convert_to_bytes
|
||
from logger import logger
|
||
|
||
|
||
async def fetch_url_content(url: str, identifier: str) -> list[str]:
|
||
"""
|
||
Получает содержимое подписки по URL и декодирует его.
|
||
|
||
Args:
|
||
url: URL для получения содержимого
|
||
identifier: Идентификатор пользователя (tg_id или email)
|
||
|
||
Returns:
|
||
Список строк из подписки
|
||
"""
|
||
try:
|
||
logger.info(f"Получение URL: {url} для идентификатора: {identifier}")
|
||
timeout = aiohttp.ClientTimeout(total=5)
|
||
async with aiohttp.ClientSession(timeout=timeout) as session:
|
||
async with session.get(url, ssl=False) as response:
|
||
if response.status == 200:
|
||
content = await response.text()
|
||
logger.info(f"Успешно получен контент с {url} для идентификатора: {identifier}")
|
||
return base64.b64decode(content).decode("utf-8").split("\n")
|
||
else:
|
||
logger.error(
|
||
f"Не удалось получить {url} для идентификатора: {identifier}, статус: {response.status}"
|
||
)
|
||
return []
|
||
except TimeoutError:
|
||
logger.error(f"Таймаут при получении {url} для идентификатора: {identifier}")
|
||
return []
|
||
except Exception as e:
|
||
logger.error(f"Ошибка при получении {url} для идентификатора: {identifier}: {e}")
|
||
return []
|
||
|
||
|
||
async def combine_unique_lines(urls: list[str], identifier: str, query_string: str) -> list[str]:
|
||
"""
|
||
Объединяет строки подписки из нескольких URL, удаляя дубликаты.
|
||
|
||
Args:
|
||
urls: Список URL для получения подписок
|
||
identifier: Идентификатор пользователя (tg_id или email)
|
||
query_string: Строка запроса для добавления к URL
|
||
|
||
Returns:
|
||
Список уникальных строк из всех подписок
|
||
"""
|
||
if SUPERNODE:
|
||
logger.info(f"Режим SUPERNODE активен. Возвращаем первую ссылку для идентификатора: {identifier}")
|
||
if not urls:
|
||
return []
|
||
url_with_query = f"{urls[0]}?{query_string}" if query_string else urls[0]
|
||
return await fetch_url_content(url_with_query, identifier)
|
||
|
||
logger.info(f"Начинаем объединение подписок для идентификатора: {identifier}, запрос: {query_string}")
|
||
urls_with_query = [f"{url}?{query_string}" if query_string else url for url in urls]
|
||
logger.info(f"Составлены URL-адреса: {urls_with_query}")
|
||
|
||
tasks = [fetch_url_content(url, identifier) for url in urls_with_query]
|
||
results = await asyncio.gather(*tasks, return_exceptions=True)
|
||
all_lines = set()
|
||
for lines in results:
|
||
all_lines.update(filter(None, lines))
|
||
logger.info(
|
||
f"Объединено {len(all_lines)} строк после фильтрации и удаления дубликатов для идентификатора: {identifier}"
|
||
)
|
||
return list(all_lines)
|
||
|
||
|
||
async def get_subscription_urls(server_id: str, email: str, conn) -> list[str]:
|
||
"""
|
||
Получает список URL-адресов для подписки в зависимости от режима выбора страны.
|
||
|
||
Args:
|
||
server_id: Идентификатор сервера или кластера
|
||
email: Email пользователя
|
||
conn: Соединение с базой данных
|
||
|
||
Returns:
|
||
Список URL-адресов для подписки
|
||
"""
|
||
if USE_COUNTRY_SELECTION:
|
||
logger.info(f"Режим выбора страны активен. Ищем сервер {server_id} в БД.")
|
||
server_data = await conn.fetchrow("SELECT subscription_url FROM servers WHERE server_name = $1", server_id)
|
||
if not server_data:
|
||
logger.warning(f"Не найден сервер {server_id} в БД!")
|
||
return []
|
||
subscription_url = server_data["subscription_url"]
|
||
urls = [f"{subscription_url}/{email}"]
|
||
logger.info(f"Используем подписку {urls[0]}")
|
||
return urls
|
||
|
||
servers = await get_servers(conn)
|
||
logger.info(f"Режим выбора страны отключен. Используем кластер {server_id}.")
|
||
cluster_servers = servers.get(server_id, [])
|
||
if not cluster_servers:
|
||
logger.warning(f"Не найдены сервера для {server_id}")
|
||
return []
|
||
urls = [f"{server['subscription_url']}/{email}" for server in cluster_servers]
|
||
logger.info(f"Найдено {len(urls)} URL-адресов в кластере {server_id}")
|
||
return urls
|
||
|
||
|
||
def get_transition_timestamp() -> int:
|
||
"""
|
||
Получает временную метку перехода с учетом часового пояса Москвы.
|
||
|
||
Returns:
|
||
Временная метка перехода в миллисекундах
|
||
"""
|
||
moscow_tz = pytz.timezone("Europe/Moscow")
|
||
transition_date_naive = datetime.strptime(TRANSITION_DATE_STR, "%Y-%m-%d %H:%M:%S")
|
||
transition_date = moscow_tz.localize(transition_date_naive)
|
||
transition_timestamp_ms = int(transition_date.timestamp() * 1000)
|
||
return transition_timestamp_ms
|
||
|
||
|
||
def calculate_traffic(cleaned_subscriptions: list[str], expiry_time_ms: int | None) -> str:
|
||
"""
|
||
Рассчитывает информацию о трафике на основе подписок.
|
||
|
||
Args:
|
||
cleaned_subscriptions: Список строк подписки
|
||
expiry_time_ms: Время истечения подписки в миллисекундах
|
||
|
||
Returns:
|
||
Строка с информацией о трафике
|
||
"""
|
||
expire_timestamp = int(expiry_time_ms / 1000) if expiry_time_ms else 0
|
||
|
||
if TOTAL_GB != 0:
|
||
country_remaining = {}
|
||
for line in cleaned_subscriptions:
|
||
if "#" not in line:
|
||
continue
|
||
|
||
try:
|
||
_, meta = line.split("#", 1)
|
||
except ValueError:
|
||
continue
|
||
|
||
parts = meta.split("-")
|
||
country = parts[0].strip()
|
||
remaining_str = parts[1].strip() if len(parts) == 2 else ""
|
||
|
||
if remaining_str:
|
||
remaining_str = remaining_str.replace(",", ".")
|
||
m_total = re.search(r"([\d\.]+)\s*([GMKTB]B)", remaining_str, re.IGNORECASE)
|
||
if m_total:
|
||
value = float(m_total.group(1))
|
||
unit = m_total.group(2).upper()
|
||
remaining_bytes = convert_to_bytes(value, unit)
|
||
country_remaining[country] = remaining_bytes
|
||
|
||
num_countries = len(country_remaining)
|
||
issued_per_country = TOTAL_GB
|
||
total_traffic_bytes = issued_per_country * num_countries
|
||
consumed_traffic_bytes = total_traffic_bytes - sum(country_remaining.values())
|
||
|
||
if consumed_traffic_bytes < 0:
|
||
consumed_traffic_bytes = 0
|
||
else:
|
||
consumed_traffic_bytes = 1
|
||
total_traffic_bytes = 0
|
||
|
||
return f"upload=0; download={consumed_traffic_bytes}; total={total_traffic_bytes}; expire={expire_timestamp}"
|
||
|
||
|
||
def clean_subscription_line(line: str) -> str:
|
||
"""
|
||
Очищает строку подписки, оставляя только нужную информацию.
|
||
|
||
Args:
|
||
line: Исходная строка подписки
|
||
|
||
Returns:
|
||
Очищенная строка подписки
|
||
"""
|
||
if "#" not in line:
|
||
return line
|
||
|
||
try:
|
||
base, meta = line.split("#", 1)
|
||
except ValueError:
|
||
return line
|
||
|
||
parts = meta.split("-")
|
||
country = parts[0].strip() if parts else ""
|
||
traffic = ""
|
||
|
||
for part in parts[1:]:
|
||
part_decoded = urllib.parse.unquote(part).strip()
|
||
if re.search(r"\d+(?:[.,]\d+)?\s*(?:GB|MB|KB|TB)", part_decoded, re.IGNORECASE):
|
||
traffic = part_decoded
|
||
break
|
||
|
||
meta_clean = f"{country} - {traffic}" if traffic else country
|
||
return base + "#" + meta_clean
|
||
|
||
|
||
def format_time_left(expiry_time_ms: int | None) -> str:
|
||
"""
|
||
Форматирует оставшееся время подписки.
|
||
|
||
Args:
|
||
expiry_time_ms: Время истечения подписки в миллисекундах
|
||
|
||
Returns:
|
||
Отформатированная строка с оставшимся временем
|
||
"""
|
||
if not expiry_time_ms:
|
||
return "N/A"
|
||
|
||
now_ms = int(time.time() * 1000)
|
||
remaining_sec = max((expiry_time_ms - now_ms) / 1000, 0)
|
||
days = int(remaining_sec // 86400)
|
||
hours = int((remaining_sec % 86400) // 3600)
|
||
|
||
return f"{days}D,{hours}H ⏳" if days else f"{hours}H ⏳"
|
||
|
||
|
||
def prepare_headers(
|
||
user_agent: str, project_name: str, subscription_info: str, subscription_userinfo: str
|
||
) -> dict[str, str]:
|
||
"""
|
||
Подготавливает заголовки ответа в зависимости от User-Agent клиента.
|
||
|
||
Args:
|
||
user_agent: User-Agent клиента
|
||
project_name: Название проекта
|
||
subscription_info: Информация о подписке
|
||
subscription_userinfo: Информация о трафике
|
||
|
||
Returns:
|
||
Словарь с заголовками ответа
|
||
"""
|
||
if "Happ" in user_agent:
|
||
encoded_project_name = f"{project_name}"
|
||
announce_str = f"↖️Бот | {subscription_info} | Поддержка↗️"
|
||
return {
|
||
"Content-Type": "text/plain; charset=utf-8",
|
||
"Content-Disposition": "inline",
|
||
"profile-update-interval": "3",
|
||
"profile-title": "base64:" + base64.b64encode(encoded_project_name.encode("utf-8")).decode("utf-8"),
|
||
"support-url": SUPPORT_CHAT_URL,
|
||
"announce": "base64:" + base64.b64encode(announce_str.encode("utf-8")).decode("utf-8"),
|
||
"profile-web-page-url": f"https://t.me/{USERNAME_BOT}",
|
||
"subscription-userinfo": subscription_userinfo,
|
||
}
|
||
elif "Hiddify" in user_agent:
|
||
parts = subscription_info.split(' - ')[0].split(': ')
|
||
key_info = parts[1] if len(parts) > 1 else parts[0]
|
||
|
||
encoded_project_name = f"{project_name}\n📄 Подписка: {key_info}"
|
||
return {
|
||
"profile-update-interval": "3",
|
||
"profile-title": "base64:" + base64.b64encode(encoded_project_name.encode("utf-8")).decode("utf-8"),
|
||
"subscription-userinfo": subscription_userinfo,
|
||
}
|
||
else:
|
||
encoded_project_name = f"{project_name}\n{subscription_info}"
|
||
return {
|
||
"Content-Type": "text/plain; charset=utf-8",
|
||
"Content-Disposition": "inline",
|
||
"profile-update-interval": "3",
|
||
"profile-title": "base64:" + base64.b64encode(encoded_project_name.encode("utf-8")).decode("utf-8"),
|
||
}
|
||
|
||
|
||
async def handle_subscription(request: web.Request, old_subscription: bool = False) -> web.Response:
|
||
"""
|
||
Обрабатывает запрос на подписку (старую или новую).
|
||
|
||
Args:
|
||
request: Объект запроса
|
||
old_subscription: Флаг, указывающий на тип подписки (старая или новая)
|
||
|
||
Returns:
|
||
Ответ с подпиской или сообщением об ошибке
|
||
"""
|
||
email = request.match_info.get("email")
|
||
tg_id = request.match_info.get("tg_id") if not old_subscription else None
|
||
|
||
if not email or (not old_subscription and not tg_id):
|
||
logger.warning("Получен запрос с отсутствующими параметрами")
|
||
return web.Response(text="❌ Неверные параметры запроса.", status=400)
|
||
|
||
logger.info(
|
||
f"Обработка запроса для {'старого' if old_subscription else 'нового'} клиента: email={email}, tg_id={tg_id}"
|
||
)
|
||
|
||
conn = await asyncpg.connect(DATABASE_URL)
|
||
try:
|
||
client_data = await get_key_details(email, conn)
|
||
if not client_data:
|
||
logger.warning(f"Клиент с email {email} не найден в базе.")
|
||
return web.Response(text="❌ Клиент с таким email не найден.", status=404)
|
||
|
||
stored_tg_id = client_data.get("tg_id")
|
||
server_id = client_data["server_id"]
|
||
|
||
if not old_subscription and str(tg_id) != str(stored_tg_id):
|
||
logger.warning(f"Неверный tg_id для клиента с email {email}.")
|
||
return web.Response(text="❌ Неверные данные. Получите свой ключ в боте.", status=403)
|
||
|
||
if old_subscription:
|
||
created_at_ms = client_data["created_at"]
|
||
created_at_datetime = datetime.utcfromtimestamp(created_at_ms / 1000)
|
||
logger.info(f"created_at для {email}: {created_at_datetime}, server_id: {server_id}")
|
||
|
||
transition_timestamp_ms = get_transition_timestamp()
|
||
logger.info(f"Время перехода (с учетом часового пояса Москвы): {transition_timestamp_ms}")
|
||
|
||
if created_at_ms >= transition_timestamp_ms:
|
||
logger.info(f"Клиент с email {email} является новым.")
|
||
return web.Response(text="❌ Эта ссылка устарела. Пожалуйста, обновите ссылку.", status=400)
|
||
|
||
expiry_time_ms = client_data.get("expiry_time")
|
||
time_left = format_time_left(expiry_time_ms)
|
||
|
||
urls = await get_subscription_urls(server_id, email, conn)
|
||
if not urls:
|
||
return web.Response(text="❌ Сервер не найден.", status=404)
|
||
|
||
query_string = request.query_string if not old_subscription else ""
|
||
combined_subscriptions = await combine_unique_lines(urls, tg_id or email, query_string)
|
||
random.shuffle(combined_subscriptions)
|
||
|
||
cleaned_subscriptions = [clean_subscription_line(line) for line in combined_subscriptions]
|
||
|
||
base64_encoded = base64.b64encode("\n".join(cleaned_subscriptions).encode("utf-8")).decode("utf-8")
|
||
subscription_info = f"📄 Подписка: {email} - {time_left}"
|
||
|
||
user_agent = request.headers.get("User-Agent", "")
|
||
subscription_userinfo = calculate_traffic(cleaned_subscriptions, expiry_time_ms)
|
||
headers = prepare_headers(user_agent, PROJECT_NAME, subscription_info, subscription_userinfo)
|
||
|
||
logger.info(f"Возвращаем объединенные подписки для email: {email}")
|
||
return web.Response(text=base64_encoded, headers=headers)
|
||
finally:
|
||
await conn.close()
|
||
|
||
|
||
async def handle_old_subscription(request: web.Request) -> web.Response:
|
||
"""Обработка запроса для старых клиентов."""
|
||
return await handle_subscription(request, old_subscription=True)
|
||
|
||
|
||
async def handle_new_subscription(request: web.Request) -> web.Response:
|
||
"""Обработка запроса для новых клиентов."""
|
||
return await handle_subscription(request, old_subscription=False)
|