timeweb-loodka/tw.py

import ipaddress
import requests
from dotenv import load_dotenv
import os
import sys
import io

if isinstance(sys.stdout, io.TextIOWrapper):
    sys.stdout.reconfigure(encoding="utf-8")


load_dotenv()

# pip install python-dotenv requests
# Прокси назначаются аккаунтам по кругу — если прокси меньше чем аккаунтов, они переиспользуются.
# Если файл proxies.txt отсутствует, все запросы идут без прокси. Каждый аккаунт использует отдельную requests.Session, что изолирует соединения.

BASE_URL = "https://api.timeweb.cloud/api/v1"

AVAILABILITY_ZONE = os.getenv("AVAILABILITY_ZONE", "spb-3")
IS_DDOS_GUARD = os.getenv("IS_DDOS_GUARD", "false").lower() == "true"
MAX_ATTEMPTS = int(os.getenv("MAX_ATTEMPTS", "10"))
ACCOUNTS_FILE = os.getenv("ACCOUNTS_FILE", "accounts.txt")
PROXIES_FILE = os.getenv("PROXIES_FILE", "proxies.txt")

TG_BOT_TOKEN = os.getenv("TG_BOT_TOKEN", "")
TG_CHAT_ID = os.getenv("TG_CHAT_ID", "")

DESIRED_SUBNETS = [
    "81.200.148.0/24",
    "81.200.149.0/24",
    "81.200.150.0/24",
    "81.200.151.0/24",
    "94.228.117.0/24",
    "185.200.242.0/24",
]


def send_telegram(text: str):
    if not TG_BOT_TOKEN or not TG_CHAT_ID:
        return
    try:
        requests.post(
            f"https://api.telegram.org/bot{TG_BOT_TOKEN}/sendMessage",
            json={"chat_id": TG_CHAT_ID, "text": text, "parse_mode": "HTML"},
            timeout=10,
        )
    except Exception as e:
        print(f"  [TG] Не удалось отправить сообщение: {e}")


def load_lines(filepath: str) -> list[str]:
    with open(filepath, "r", encoding="utf-8") as f:
        return [line.strip() for line in f if line.strip()]


def ip_in_desired_subnets(ip: str) -> bool:
    addr = ipaddress.ip_address(ip)
    return any(addr in ipaddress.ip_network(subnet) for subnet in DESIRED_SUBNETS)


def make_session(proxy: str | None) -> requests.Session:
    session = requests.Session()
    if proxy:
        session.proxies = {"http": proxy, "https": proxy}
    return session


def create_floating_ip(session: requests.Session, headers: dict) -> dict:
    payload = {
        "availability_zone": AVAILABILITY_ZONE,
        "is_ddos_guard": IS_DDOS_GUARD,
    }
    response = session.post(f"{BASE_URL}/floating-ips", headers=headers, json=payload)
    if not response.ok:
        print(f"  Ошибка {response.status_code}: {response.text}")
        response.raise_for_status()
    return response.json()["ip"]


def delete_floating_ip(session: requests.Session, headers: dict, floating_ip_id: str):
    response = session.delete(
        f"{BASE_URL}/floating-ips/{floating_ip_id}", headers=headers
    )
    if not response.ok:
        print(
            f"  Не удалось удалить IP {floating_ip_id}: {response.status_code} {response.text}"
        )


def get_ip_from_desired_range(
    session: requests.Session, headers: dict, account_label: str
) -> dict | None:
    for attempt in range(1, MAX_ATTEMPTS + 1):
        ip_data = create_floating_ip(session, headers)
        ip_addr = ip_data["ip"]
        print(f"  [{account_label}] Попытка {attempt}: получен IP {ip_addr}")

        if ip_in_desired_subnets(ip_addr):
            print(f"  [{account_label}] ✅ IP {ip_addr} входит в нужный диапазон!")
            return ip_data
        else:
            print(f"  [{account_label}] ❌ IP {ip_addr} не подходит, удаляем...")
            delete_floating_ip(session, headers, ip_data["id"])

    print(f"  [{account_label}] Не удалось получить IP за {MAX_ATTEMPTS} попыток.")
    return None


if __name__ == "__main__":
    tokens = load_lines(ACCOUNTS_FILE)
    proxies = load_lines(PROXIES_FILE) if os.path.exists(PROXIES_FILE) else []

    print(f"Загружено аккаунтов: {len(tokens)}, прокси: {len(proxies)}\n")
    send_telegram(
        f"🚀 <b>tw.py запущен</b>\n"
        f"Аккаунтов: {len(tokens)}, прокси: {len(proxies)}\n"
        f"Зона: {AVAILABILITY_ZONE}, DDoS-Guard: {IS_DDOS_GUARD}, попыток: {MAX_ATTEMPTS}"
    )

    found = []
    failed = []

    for i, token in enumerate(tokens):
        proxy = proxies[i % len(proxies)] if proxies else None
        label = f"аккаунт {i + 1}"

        print(f"▶ Обработка {label} | прокси: {proxy or 'нет'}")

        headers = {
            "Authorization": f"Bearer {token}",
            "Content-Type": "application/json",
        }
        session = make_session(proxy)

        try:
            result = get_ip_from_desired_range(session, headers, label)
            if result:
                print(f"  Итоговый IP: {result['ip']}, ID: {result['id']}\n")
                found.append((label, result["ip"], result["id"]))
                send_telegram(
                    f"✅ <b>{label}</b>: получен IP <code>{result['ip']}</code>\n"
                    f"ID: <code>{result['id']}</code>"
                )
            else:
                failed.append(label)
                send_telegram(
                    f"❌ <b>{label}</b>: не удалось получить подходящий IP "
                    f"за {MAX_ATTEMPTS} попыток"
                )
        except Exception as e:
            print(f"  Критическая ошибка для {label}: {e}\n")
            failed.append(label)
            send_telegram(f"💥 <b>{label}</b>: критическая ошибка — {e}")

    summary_lines = [f"📊 <b>Итог:</b> найдено {len(found)}, не найдено {len(failed)}"]
    if found:
        summary_lines.append("\n<b>Успешные:</b>")
        for lbl, ip, fid in found:
            summary_lines.append(f"  • {lbl}: <code>{ip}</code>")
    if failed:
        summary_lines.append("\n<b>Неудачные:</b>")
        for lbl in failed:
            summary_lines.append(f"  • {lbl}")
    send_telegram("\n".join(summary_lines))