Update get_vpn.py
This commit is contained in:
+78
-25
@@ -1,57 +1,110 @@
|
|||||||
|
import base64
|
||||||
import json
|
import json
|
||||||
import os
|
import os
|
||||||
import sys
|
import sys
|
||||||
import urllib.parse
|
|
||||||
import requests
|
import requests
|
||||||
|
|
||||||
|
|
||||||
def get_vpn_servers():
|
def get_vpn_servers():
|
||||||
# Забираем ссылку из переменных окружения (туда её подставит Gitea)
|
# Получаем ссылку из секретов Gitea
|
||||||
sub_url = os.getenv("VPN_SUB_URL")
|
sub_url = os.getenv("VPN_SUB_URL")
|
||||||
|
|
||||||
if not sub_url:
|
if not sub_url:
|
||||||
print("Ошибка: Переменная окружения VPN_SUB_URL не задана!")
|
print("Ошибка: Переменная окружения VPN_SUB_URL не задана!")
|
||||||
sys.exit(1)
|
sys.exit(1)
|
||||||
|
|
||||||
parsed_url = urllib.parse.urlparse(sub_url)
|
# 1. Сначала пробуем сделать запрос по исходной ссылке
|
||||||
base_url = f"{parsed_url.scheme}://{parsed_url.netloc}"
|
# Имитируем заголовки приложения Happ
|
||||||
query_params = urllib.parse.parse_qs(parsed_url.query)
|
|
||||||
token = query_params.get("token", [""])
|
|
||||||
|
|
||||||
if not token or not token[0]:
|
|
||||||
print("Ошибка: В ссылке не найден token!")
|
|
||||||
sys.exit(1)
|
|
||||||
|
|
||||||
api_url = f"{base_url}/api/v1/client/appConfig?token={token[0]}"
|
|
||||||
|
|
||||||
headers = {
|
headers = {
|
||||||
"User-Agent": "Happ/2.1.0 (iPhone; iOS 15.5; Scale/3.00)",
|
"User-Agent": "Happ/2.1.0 (iPhone; iOS 15.5; Scale/3.00)",
|
||||||
"Accept": "*/*",
|
"Accept": "*/*",
|
||||||
"Accept-Language": "ru-RU,ru;q=0.9,en-US;q=0.8,en;q=0.7",
|
|
||||||
}
|
}
|
||||||
|
|
||||||
try:
|
try:
|
||||||
response = requests.get(api_url, headers=headers, timeout=10)
|
print("Отправка запроса к серверу VPN...")
|
||||||
|
response = requests.get(sub_url, headers=headers, timeout=15)
|
||||||
|
|
||||||
if response.status_code != 200:
|
if response.status_code != 200:
|
||||||
print(f"Ошибка сервера: Статус {response.status_code}")
|
print(f"Ошибка сервера: Статус {response.status_code}")
|
||||||
sys.exit(1)
|
sys.exit(1)
|
||||||
|
|
||||||
data = response.json()
|
raw_content = response.text.strip()
|
||||||
servers = data.get("data", {}).get("server_user", [])
|
|
||||||
|
|
||||||
if not servers:
|
# 2. Проверяем, не JSON ли это (на случай, если в секрет сразу зашит appConfig URL)
|
||||||
print("Серверы не найдены в ответе API. Сохраняем сырой ответ для отладки.")
|
try:
|
||||||
servers = data
|
data = json.loads(raw_content)
|
||||||
|
print("Обнаружен формат JSON (панель Happ API).")
|
||||||
|
servers = data.get("data", {}).get("server_user", data)
|
||||||
|
|
||||||
# Записываем результат в файл
|
with open("servers.json", "w", encoding="utf-8") as f:
|
||||||
with open("servers.json", "w", encoding="utf-8") as f:
|
json.dump(servers, f, indent=2, ensure_ascii=False)
|
||||||
json.dump(servers, f, indent=2, ensure_ascii=False)
|
print("Данные сохранены в файл servers.json")
|
||||||
|
return
|
||||||
|
except json.JSONDecodeError:
|
||||||
|
# Если не JSON, идем дальше
|
||||||
|
pass
|
||||||
|
|
||||||
print(f"Успешно обновлено! Найдено серверов: {len(servers)}")
|
# 3. Проверяем, не Base64 ли это (стандартная подписка)
|
||||||
|
try:
|
||||||
|
# Добавляем padding, если строка обрезана (стандартный трюк для base64 в Python)
|
||||||
|
padded_content = raw_content + "=" * (4 - len(raw_content) % 4)
|
||||||
|
decoded_bytes = base64.b64decode(padded_content)
|
||||||
|
decoded_text = decoded_bytes.decode("utf-8", errors="ignore").strip()
|
||||||
|
|
||||||
|
# Если внутри есть vless/ss/ssr/trojan/vmess ссылки
|
||||||
|
if any(
|
||||||
|
proto in decoded_text
|
||||||
|
for proto in ["vless://", "vmess://", "ss://", "trojan://"]
|
||||||
|
):
|
||||||
|
print(
|
||||||
|
"Обнаружена Base64 подписка. Успешно декодировано в список ссылок."
|
||||||
|
)
|
||||||
|
|
||||||
|
# Вариант А: Сохраняем как простой текстовый список серверов
|
||||||
|
with open("servers.txt", "w", encoding="utf-8") as f:
|
||||||
|
f.write(decoded_text)
|
||||||
|
print("Ссылки сохранены в файл servers.txt")
|
||||||
|
|
||||||
|
# Вариант Б: Для удобства сохраняем этот же список в JSON массив
|
||||||
|
links_list = [
|
||||||
|
line.strip()
|
||||||
|
for line in decoded_text.splitlines()
|
||||||
|
if line.strip()
|
||||||
|
]
|
||||||
|
with open("servers.json", "w", encoding="utf-8") as f:
|
||||||
|
json.dump(links_list, f, indent=2, ensure_ascii=False)
|
||||||
|
print("Ссылки также структурированы в servers.json")
|
||||||
|
return
|
||||||
|
except Exception as b64_err:
|
||||||
|
print(f"Попытка декодировать Base64 не удалась: {b64_err}")
|
||||||
|
|
||||||
|
# 4. Если это и не JSON, и не Base64, возможно это уже готовый сырой текст со ссылками
|
||||||
|
if any(
|
||||||
|
proto in raw_content
|
||||||
|
for proto in ["vless://", "vmess://", "ss://", "trojan://"]
|
||||||
|
):
|
||||||
|
print("Сервер вернул сырой текст со ссылками.")
|
||||||
|
with open("servers.txt", "w", encoding="utf-8") as f:
|
||||||
|
f.write(raw_content)
|
||||||
|
|
||||||
|
links_list = [
|
||||||
|
line.strip() for line in raw_content.splitlines() if line.strip()
|
||||||
|
]
|
||||||
|
with open("servers.json", "w", encoding="utf-8") as f:
|
||||||
|
json.dump(links_list, f, indent=2, ensure_ascii=False)
|
||||||
|
print("Данные сохранены в servers.txt и servers.json")
|
||||||
|
return
|
||||||
|
|
||||||
|
# 5. Если вообще ничего не подошло, сохраняем как есть для анализа
|
||||||
|
print(
|
||||||
|
"Неизвестный формат ответа. Сохраняем raw-текст в raw_response.txt для отладки."
|
||||||
|
)
|
||||||
|
with open("raw_response.txt", "w", encoding="utf-8") as f:
|
||||||
|
f.write(raw_content)
|
||||||
|
sys.exit(1)
|
||||||
|
|
||||||
except Exception as e:
|
except Exception as e:
|
||||||
print(f"Произошла ошибка: {e}")
|
print(f"Произошла критическая ошибка: {e}")
|
||||||
sys.exit(1)
|
sys.exit(1)
|
||||||
|
|
||||||
|
|
||||||
|
|||||||
Reference in New Issue
Block a user