Jak stworzyć bota Telegram do monitorowania proxy: krok po kroku dla początkujących

Wprowadzenie

W tym przewodniku krok po kroku stworzysz od zera w pełni działającego bota Telegram, który automatycznie monitoruje listę proxy, mierzy czas odpowiedzi, wykrywa zmiany IP po rotacji, sprawdza geolokalizację przez serwis ip-api.com i wysyła widoczne powiadomienia w Telegramie w przypadku problemów lub degradacji. Przeprowadzimy Cię krok po kroku od pomysłu do wdrożenia na VPS z automatycznym uruchamianiem, historią w SQLite i prostym panelem z statystykami dostępności i eksportem logów.

Przewodnik jest przeznaczony dla początkujących, z elementami dla bardziej zaawansowanych. Jeśli nigdy nie pisałeś bota, nie martw się: omówimy każdy krok, wyjaśnimy, dlaczego jest potrzebny i jak sprawdzić wynik. Na koniec otrzymasz stabilne narzędzie do monitorowania proxy, które działa według harmonogramu co 5–15 minut, potrafi grupować proxy według regionów i typów oraz wysyła informacyjne alerty do Twojego Telegramu.

Przed rozpoczęciem warto wiedzieć, jak otworzyć terminal, jak instalować programy oraz jak tworzyć foldery i pliki. Wszystko inne wyjaśnimy. Wykonanie wszystkich kroków zajmie orientacyjnie od 4 do 8 godzin, w tym konfigurację środowiska, pisanie kodu, testy i wdrożenie. Jeśli masz już VPS i podstawowe umiejętności w Pythonie, dasz radę szybciej.

Na koniec przewodnika otrzymasz: działającego bota Telegrama w aiogram 3, asynchroniczny moduł do sprawdzeń na aiohttp, planner APScheduler do zadań cyklicznych, bazę danych SQLite z logami i statystykami, prosty web-dashboard z głównymi wskaźnikami i eksportem do CSV, instrukcje dotyczące utrzymania i rozszerzania funkcjonalności.

Porada: Lepiej przechodzić przez przewodnik krok po kroku, nie pomijając etapów. Po każdym etapie dodaliśmy kontrole, abyś od razu upewnił się, że wszystko działa.

Wstępne przygotowanie

Przed rozpoczęciem pisania kodu przygotujemy wszystkie niezbędne narzędzia i środowisko. To pozwoli uniknąć przypadkowych błędów i zaoszczędzi czas.

Niezbędne narzędzia, programy i dostęp

• Konto Telegram do stworzenia bota i otrzymywania powiadomień.
• VPS lub lokalny komputer z systemem Linux (preferowane Ubuntu 22.04 lub 24.04). Można także na Windows i macOS, ale wdrożenie pokażemy na Ubuntu.
• Python 3.11 lub nowszy. Sprawdzimy wersję poleceniem python3 --version.
• Terminal i podstawowe uprawnienia sudo do instalacji pakietów.
• Lista proxy do monitorowania w formacie host:port, plus login i hasło, jeśli wymagane. Obsługiwane są HTTP, HTTPS i SOCKS5.

Wymagania systemowe

• 1 CPU, 512–1024 MB RAM wystarcza dla 50–200 proxy przy interwale 5–15 minut. Dla tysięcy proxy lepiej 2 CPU i 2 GB RAM.
• Wolne miejsce na dysku od 1 GB. SQLite jest lekka, ale logi rosną.
• Otwarty dostęp do internetu. Dostęp do API Telegrama i Twoich proxy jest obowiązkowy. Dla ip-api.com wymagany jest wychodzący dostęp HTTP.

Co należy zainstalować

• Python 3.11+ i venv dla wirtualnego środowiska.
• pip do instalacji zależności.
• Biblioteki: aiogram w wersji 3, aiohttp, APScheduler, aiosqlite, python-dotenv (dla wygodnej konfiguracji), uvloop (opcjonalnie dla Linux), yarl (często jako zależność).

Tworzenie kopii zapasowych

Na początku tworzenie kopii zapasowych nie jest krytyczne. Po uruchomieniu bota koniecznie skonfigurujemy prostą kopię zapasową bazy SQLite z logami i ustawieniami. To pozwoli na cofnięcie się w przypadku błędnej aktualizacji lub uszkodzenia pliku.

⚠️ Uwaga: SQLite to jeden plik. Jeśli ulegnie on uszkodzeniu przy niespodziewanym braku zasilania, można stracić świeże zapisy. Dlatego na VPS włącz automatyczne codzienne tworzenie kopii zapasowej pliku bazy za pomocą cron lub timerów systemd. Pokażemy prosty skrypt do tego na końcu.

Podstawowe pojęcia

Aby się nie pogubić, ustalmy kluczowe terminy i zasady działania.

• Proxy (Proxy) — serwer pośredniczący, przez który przechodzą Twoje zapytania. Mogą być HTTP, HTTPS i SOCKS5. Proxy może wymagać autoryzacji login:hasło lub pracy przez przypisanie IP.
• Latency (opóźnienie) — czas, w którym wykonujemy zapytanie przez proxy i otrzymujemy odpowiedź. Mierzymy w milisekundach. Im mniej, tym lepiej.
• Rotacja IP — niektóre proxy okresowo zmieniają zewnętrzny adres IP. Nasz bot powinien monitorować te zmiany i informować, jeśli IP zmieniło się niespodziewanie lub nie zgadza się z oczekiwanym regionem.
• Geolokalizacja (Geo) — kraj, miasto i operator adresu IP. Otrzymujemy przez serwis ip-api.com na podstawie IP. To pomoże upewnić się, że Twoje proxy znajduje się w odpowiednim kraju.
• Degradacja — proxy odpowiada, ale znacznie wolniej niż normalnie lub z częstymi błędami. Będziemy rozróżniać DOWN (całkowicie niedostępny) i DEGRADED (działa, ale źle).
• aiogram 3 — nowoczesna asynchroniczna biblioteka do botów Telegram. Prosta, szybka, obsługuje wszystkie aktualne funkcje Telegramu w 2026 roku.
• aiohttp — asynchroniczne zapytania HTTP z czasem oczekiwania i wsparciem dla proxy.
• APScheduler — planista zadań, który uruchomi kontrole zgodnie z harmonogramem co N minut.
• SQLite — wbudowana baza danych. Jedno plik, szybko i bez instalacji serwera. Doskonała do lokalnych logów i statystyk.

Princip działania bota jest taki: zgodnie z harmonogramem bot pobiera listę proxy, wykonuje szybkie testowe zapytania przez każde proxy, mierzy czas, uzyskuje zewnętrzny IP i dane geograficzne, zapisuje wyniki w SQLite. Jeśli proxy padło lub znacznie się pogorszyło, bot formuje starannie skonstruowaną wiadomość i wysyła ją do Telegramu. Dodatkowo bot reaguje na polecenia z czatu: pokazuje status, dostępność, ostatnie błędy, eksportuje CSV, dodaje lub wyłącza proxy, zmienia interwał.

Porada: Aby nie przekroczyć limitów ip-api.com (zazwyczaj 45 zapytań na minutę dla darmowego poziomu), wdrożymy pamięć geodanych i ograniczenie równoległych zapytań geolokalizacji. To nie spowolni działania, ale ochroni przed zablokowaniem.

Krok 1: Tworzymy bota w Telegramie i konfigurujemy dostęp

Cel etapu

Uzyskać token bota Telegrama i stworzyć prywatny czat do powiadomień, aby później bot mógł wysyłać wiadomości o incydentach.

Instrukcja krok po kroku

1. Otwórz aplikację Telegram na telefonie lub komputerze.
2. W pasku wyszukiwania wpisz BotFather i otwórz oficjalne konto z niebieską zakładką.
3. Naciśnij Start lub wpisz komendę /start, następnie wpisz /newbot.
4. Wymyśl nazwę bota. Na przykład: ProxyMonitorBot.
5. Wymyśl unikalną nazwę użytkownika bota, kończącą się na bot. Na przykład: proxy_monitor_helper_bot.
6. Skopiuj wydany token. Wygląda jak zestaw znaków i cyfr oddzielonych dwukropkiem.
7. Stwórz prywatny czat lub grupę, do której bot będzie wysyłał powiadomienia. Dodaj bota do czatu.
8. Napisz w czacie dowolną wiadomość, aby czat pojawił się na liście dla Twojego konta.
9. Otwórz u bota okno dialogowe i naciśnij Start, aby go aktywować.

Ważne kwestie: Token przechowuj w tajemnicy. Nie przekazuj nikomu. Przy wycieku tokenu, osoba niepowołana może zarządzać Twoim botem. Jeśli token został skompromitowany, wybierz /revoke w BotFatherze i uzyskaj nowy.

Porada: Utwórz osobną grupę dla alertów, w której będziesz tylko Ty i bot. Tak powiadomienia nie zginą wśród innych wiadomości.

✅ Weryfikacja: Otrzymałeś token i dodałeś bota do wymaganego czatu. Bot odpowiada na komendę /start w osobnym dialogu.

Możliwe problemy i rozwiązania

• Problem: Bot nie odpowiada na /start. Przyczyna: Nie nacisnąłeś Start lub token został podany niepoprawnie w kodzie. Rozwiązanie: Sprawdź, czy bot jest aktywowany, a później prawidłowo wskażemy token w pliku .env.
• Problem: Nie widzisz chat ID. Przyczyna: Jeszcze go nie otrzymaliśmy. Rozwiązanie: Później komenda w bocie automatycznie pokaże chat_id.

Krok 2: Przygotowujemy środowisko Pythona i strukturę projektu

Cel etapu

Stworzyć roboczy folder projektu, wirtualne środowisko Python 3.11+, zainstalować zależności, skonfigurować ustawienia i strukturę katalogów.

Instrukcja krok po kroku

1. Otwórz terminal na swoim komputerze lub połącz się z VPS przez SSH.
2. Sprawdź wersję Pythona poleceniem: python3 --version. Jeśli wersja jest niższa niż 3.11, zainstaluj aktualną.
3. Na Ubuntu wykonaj: sudo apt update; sudo apt install -y python3.11 python3.11-venv python3-pip.
4. Utwórz folder projektu: mkdir proxy-monitor-bot; cd proxy-monitor-bot.
5. Utwórz wirtualne środowisko: python3.11 -m venv .venv.
6. Aktywuj środowisko: source .venv/bin/activate.
7. Zaktualizuj pip: pip install --upgrade pip.
8. Zainstaluj zależności: pip install aiogram aiohttp aiosqlite APScheduler python-dotenv uvloop.
9. Utwórz strukturę folderów: mkdir app app/modules app/db app/bot app/web.
10. Utwórz pliki: touch app/__init__.py app/config.py app/main.py app/modules/proxy_checker.py app/db/models.py app/db/migrations.py app/bot/handlers.py app/bot/notifications.py app/scheduler.py app/web/server.py .env .env.example README.md.

Podstawowa konfiguracja

Otwórz plik .env i dodaj klucze (zmień wartości na swoje):

BOT_TOKEN=twój_token_z_BotFather TELEGRAM_CHAT_ID=numer_lub_zostaw_puste DB_PATH=./data/monitor.db PROBE_TIMEOUT=8 PROBE_RETRIES=2 CHECK_INTERVAL_MINUTES=10 GEO_CACHE_TTL_HOURS=24 IP_API_URL=http://ip-api.com/json PROXY_CONCURRENCY=20 GEO_CONCURRENCY=10 DASHBOARD_HOST=0.0.0.0 DASHBOARD_PORT=8080

Skopiuj .env do .env.example i usuń sekrety, aby wygodnie dzielić się projektem bez tokenów.

Porada: Przechowuj plik .env poza systemem kontroli wersji. Dodaj .env do .gitignore, aby przypadkowo nie załadować tokenu do repozytorium.

Minimalna struktura kodu

Plik app/config.py będzie odczytywał zmienne środowiskowe i dostarczał konfigurację. Plik app/main.py uruchomi bota, planistę i web-dashboard. Moduł proxy_checker.py przeprowadzi wszystkie kontrole. W folderze db umieścimy modele i migracje SQLite. W bot napiszemy obsługiwacze komend i wysyłanie powiadomień. W web uruchomimy prosty serwer webowy z statystykami i eksportem.

✅ Weryfikacja: Masz wirtualne środowisko, zainstalowane zależności, utworzony plik .env, a struktura projektu odpowiada wymienionej. Komenda python -c "import aiogram, aiohttp, aiosqlite; print('ok')" выводит ok.

Możliwe problemy i rozwiązania

• Problem: pip zgłasza błąd SSL lub niedostępność. Przyczyna: Brak dostępu do internetu. Rozwiązanie: Sprawdź sieć lub użyj lokalnego lustra pip.
• Problem: Python 3.11 niedostępny. Przyczyna: Stary system. Rozwiązanie: Zaktualizuj system lub zainstaluj Pythona przez deadsnakes PPA lub pyenv.

Krok 3: Pisanie modułu sprawdzania proxy: połączenie, latencja, IP, geolokalizacja

Cel etapu

Stworzyć asynchroniczny moduł, który szybko i niezawodnie sprawdza proxy: potrafi wykonać testowe zapytanie HTTP przez proxy, zmierzyć opóźnienie, uzyskać zewnętrzny IP, wykonać zapytanie do ip-api.com, skasować dane geograficzne i poprawnie obsługiwać błędy.

Kluczowe pomysły

• Aby sprawdzić dostępność, wykonaj zapytanie do lekkiego zasobu, na przykład https://api.ipify.org lub dowolnego szybkiego echo-endpointu. Wybierzemy uzyskanie IP przez serwis, a następnie geo przez ip-api.com. Można także w drugą stronę.
• Czas oczekiwania ustawiamy na 8 sekund domyślnie, prób 2. Te parametry można dostosować przez .env.
• aiohttp wygodnie działa z proxy: przekazujemy parametr proxy w ClientSession.request.
• Dla geolokalizacji kasujemy dane na 24 godziny na podstawie IP, aby nie tracić limitu.
• Równolegle sprawdzamy, ale ograniczamy jednoczesność semaforem, aby nie przeciążać połączeń i nie wpadać w limity ip-api.

Szczegółowa instrukcja

1. Otwórz plik app/modules/proxy_checker.py.
2. Opisz model proxy jako słownik: id, label, type (http, https, socks5), endpoint (host:port), username, password, region, group, expected_country, last_ip.
3. Stwórz asynchroniczną funkcję build_proxy_url, która z type, host, port i opcjonalnie username:password formuje ciąg w formacie scheme://user:pass@host:port.
4. Stwórz asynchroniczną funkcję fetch_ip(session, proxy_url) do uzyskiwania zewnętrznego IP przez proxy. Użyj url https://api.ipify.org?format=json lub podobnego serwisu, który zwraca Twoje IP w JSON. Zmierz czas do wezwania i po odpowiedzi, aby uzyskać latency.
5. Stwórz pamięć geodanych: słownik w pamięci i tabelę w SQLite, aby nie tracić pamięci między restarami. Klucz — IP, wartość — JSON kraju, miasta, organizacji i czasu kasowania.
6. Stwórz funkcję fetch_geo(session, ip), która robi GET zapytanie do ip-api.com/json/{ip}?fields=status,country,countryCode,regionName,city,org,query. Przy statusie fail zwróć oznakę błędu. Przy sukcesie zwróć dane geograficzne.
7. Realizuj funkcję check_proxy(proxy) z logiką: zbuduj proxy_url, spróbuj uzyskać IP z czasem oczekiwania, zmierz latency, następnie uzyskaj dane geograficzne (z pamięci podręcznej lub sieci), sformułuj wynik: dostępność (ok), opóźnienie w milisekundach, zewnętrzne IP, kraj i miasto, oraz typ wyniku: UP, DEGRADED lub DOWN.
8. Określ kryteria DEGRADED: jeśli latency przekracza Twój próg, na przykład 1500–2000 ms, lub jeśli część błędów w ostatnich kontrolach jest zauważalna. Dla uproszczenia: jeśli odpowiedź jest, ale latency jest większe niż 2 sekundy — DEGRADED, w przeciwnym razie UP. Jeśli odpowiedzi nie ma — DOWN.
9. Zwróć szczegółowy wynik do logowania: timestamp, proxy_id, status, latency_ms lub None, ip lub None, geo lub None, error_reason, changed_ip (True lub False jeśli IP różni się od poprzedniego), changed_geo (jeśli kraj się zmienił), region, group, type.
10. Dodaj obróbkę wyjątków: asyncio.TimeoutError, aiohttp.ClientConnectorError, aiohttp.ClientHttpProxyError, aiohttp.ClientProxyConnectionError, błędy sock. W polu error_reason zapisz krótkie opisy.

Przykład kodu (upraszczając, w jednej linii)

Przykład logiki w kompaktowej formie: import asyncio, time, json, aiohttp; async def check_proxy(proxy, cfg, geo_cache, sem): async with sem: start=time.perf_counter(); purl=f"{proxy['type']}://{proxy.get('username', '')+(':'+proxy.get('password','') if proxy.get('username') else '')+'@' if proxy.get('username') else ''}{proxy['endpoint']}"; timeout=aiohttp.ClientTimeout(total=cfg['PROBE_TIMEOUT']); try: async with aiohttp.ClientSession(timeout=timeout) as s: t0=time.perf_counter(); async with s.get('https://api.ipify.org?format=json', proxy=purl, ssl=False) as r: data=await r.json(); ip=data.get('ip'); latency=int((time.perf_counter()-t0)*1000); if not ip: raise Exception('no-ip'); geo=geo_cache.get(ip); if not geo or geo['ts']

W rzeczywistym kodzie dodamy logowanie w SQLite i semafory dla GEO-zapytań, a także eleganckie czasy oczekiwania i retry.

⚠️ Uwaga: Nie rób zbyt wielu równoległych zapytań do ip-api.com. Jeśli masz listę ze setkami proxy, kasuj wyniki i ogranicz równoległość zapytań. To ochroni przed limitami i błędami.

✅ Weryfikacja: Ręcznie wywołaj funkcję check_proxy dla jednego testowego proxy i wydrukuj wynik. Powinieneś zobaczyć status UP lub DEGRADED z zmierzonym opóźnieniem i zewnętrznym IP.

Możliwe problemy i rozwiązania

• Problem: Błąd ClientHttpProxyError. Przyczyna: Proxy nie obsługuje żądanego typu lub błędna autoryzacja. Rozwiązanie: Sprawdź typ i dane logowania, zaktualizuj endpoint.
• Problem: Geo zawsze unknown. Przyczyna: Błąd zapytania do ip-api lub blokada. Rozwiązanie: Sprawdź IP_API_URL i sieć, włącz logowanie wyjątków.

Krok 4: Konfiguracja bazy SQLite i logowanie historii

Cel etapu

Stworzyć tabele w SQLite do przechowywania listy proxy, historii sprawdzeń, incydentów, pamięci geograficznej i ustawień. Da to szczegółową statystykę oraz możliwość budowania wykresów.

Struktura bazy

• proxies: id, label, type, endpoint, username, password, region, group_name, expected_country, last_ip, enabled, created_at, updated_at.
• checks: id, proxy_id, ts, status, latency_ms, ip, country, city, org, error_reason.
• incidents: id, proxy_id, opened_at, closed_at, kind (DOWN lub DEGRADED), last_status, last_message, active.
• geo_cache: ip, country, city, org, ts.
• settings: key, value (na przykład domyślny interwał sprawdzania).

Szczegółowa instrukcja

1. Otwórz app/db/models.py. Opisz stałe SQL do tworzenia tabel. Dodaj indeksy według proxy_id i ts, aby przyspieszyć wybory.
2. Stwórz funkcję init_db(db_path), która tworzy folder data w razie potrzeby, otwiera połączenie aiosqlite.connect i wykonuje create table if not exists dla wszystkich tabel.
3. Dodaj funkcje do zapisywania wyników: add_proxy, update_proxy_last_ip, add_check, open_incident, close_incident_if_needed, upsert_geo_cache, get_recent_checks, get_uptime_stats.
4. Realizuj migracje w app/db/migrations.py. Dla pierwszego uruchomienia wystarczy wykonać wszystkie create table. Przy aktualizacjach wersji skrypt będzie dodawać brakujące kolumny.

Przykład SQL (w jednej linii)

CREATE TABLE IF NOT EXISTS proxies (id INTEGER PRIMARY KEY AUTOINCREMENT, label TEXT, type TEXT, endpoint TEXT, username TEXT, password TEXT, region TEXT, group_name TEXT, expected_country TEXT, last_ip TEXT, enabled INTEGER DEFAULT 1, created_at INTEGER, updated_at INTEGER); CREATE TABLE IF NOT EXISTS checks (id INTEGER PRIMARY KEY AUTOINCREMENT, proxy_id INTEGER, ts INTEGER, status TEXT, latency_ms INTEGER, ip TEXT, country TEXT, city TEXT, org TEXT, error_reason TEXT, FOREIGN KEY(proxy_id) REFERENCES proxies(id)); CREATE INDEX IF NOT EXISTS idx_checks_proxy_ts ON checks(proxy_id, ts); CREATE TABLE IF NOT EXISTS incidents (id INTEGER PRIMARY KEY AUTOINCREMENT, proxy_id INTEGER, opened_at INTEGER, closed_at INTEGER, kind TEXT, last_status TEXT, last_message TEXT, active INTEGER, FOREIGN KEY(proxy_id) REFERENCES proxies(id)); CREATE TABLE IF NOT EXISTS geo_cache (ip TEXT PRIMARY KEY, country TEXT, city TEXT, org TEXT, ts INTEGER); CREATE TABLE IF NOT EXISTS settings (key TEXT PRIMARY KEY, value TEXT);

Porada: Jeśli już przechowujesz listę proxy w CSV lub JSON, dodaj wygodny import do bazy: prosty skrypt czyta plik i dodaje wpisy do proxies.

✅ Weryfikacja: Uruchom funkcję init_db i upewnij się, że plik bazy danych pojawił się w ścieżce z DB_PATH. Sprawdź, czy tabele zostały utworzone, wykonując proste zapytanie select count(*) from proxies i uzyskując odpowiedź 0.

Możliwe problemy i rozwiązania

• Problem: Błąd no such table. Przyczyna: Inicjalizacja nie została wykonana. Rozwiązanie: Wywołaj init_db przy starcie aplikacji przed wszystkimi operacjami.
• Problem: Zablokowania na zapis. Przyczyna: Równoczesne transakcje. Rozwiązanie: Jeśli to możliwe, grupuj zapisy wyników w pakietach lub używaj krótkich transakcji.

Krok 5: Realizujemy bota Telegrama w aiogram 3

Cel etapu

Uruchomić bota, który reaguje na komendy, potrafi wysyłać powiadomienia do Twojego czatu oraz odczytywać ustawienia z .env. Skonfiguruj podstawowe komendy: /start, /status, /chatid, /addproxy, /list, /enable, /disable, /interval, /export.

Szczegółowa instrukcja

1. Otwórz app/bot/handlers.py. Stwórz trasy komend. Dla aiogram 3 użyj Router, Dispatcher i asynchroniczne handlery.
2. Realizuj /start: odpowiada z krótkim opisem możliwości i aktualnym interwałem sprawdzeń.
3. Realizuj /chatid: wysyła identyfikator czatu, abyś mógł zapisać TELEGRAM_CHAT_ID w .env. To wygodne, jeśli uruchamiasz bota z prywatnego czatu lub grupy.
4. Realizuj /status: znajduje ostatnie N sprawdzeń dla każdego proxy i wyświetla krótkie podsumowanie: status, latencja, kraj, czas aktualizacji.
5. Realizuj /list: lista proxy z id, label, type, region, enabled.
6. Realizuj /addproxy: parametrami przekazuj type, endpoint, username, password, region, group, expected_country. Na pierwszym etapie można zrobić prostą wersję, która przyjmuje ciąg w formacie type=http endpoint=host:port region=EU i parsuje parametry.
7. Realizuj /enable i /disable z parametrem id, aby szybko włączać i wyłączać konkretne proxy.
8. Realizuj /interval N dla zmiany interwału sprawdzania w minutach. Zapisz wartość w tabeli settings.
9. Realizuj /export do generowania CSV według ostatnich kontroli za okres. Później dodamy podobny przycisk w web-dashboardzie.

Wysyłanie powiadomień

W app/bot/notifications.py stwórz funkcję notify_change, która formuje wiadomość o zdarzeniach. Przykład wiadomości: Proxy: EU-1 (http) Region: EU Grupa: EUROPA Status: DOWN Powód: timeout Ostatni IP: 203.0.113.45 Oczekiwany: kraj DE Otrzymany: kraj NL Czas: 2026-02-01 12:34:56 Latencja: n/a. Dla UP użyj etykiety UP, dla degradacji DEGRADED. Dodaj sortowanie według ważności: najpierw DOWN, następnie DEGRADED, następnie wiadomości o zmianie IP. Dodaj emoji-indykatory dla przejrzystości, np. ✅ dla UP, ⚠️ dla DEGRADED i ❌ dla DOWN. Jeśli TELEGRAM_CHAT_ID nie jest ustawiony, wysyłaj do bieżącego czatu, z którego przyszła komenda. W produkcji lepiej zablokować chat_id w .env.

Porada: Dla grup czat-bot może potrzebować uprawnień do wysyłania wiadomości. Sprawdź ustawienia grupy, aby bot był uczestnikiem z prawem pisania.

✅ Weryfikacja: Uruchom bota lokalnie i wyślij mu /chatid. Powinieneś otrzymać numeryczny identyfikator. Ustaw TELEGRAM_CHAT_ID w .env. Wyślij /start i /status. Bot powinien odpowiedzieć. Komenda /list na razie może być pusta, to normalne.

Możliwe problemy i rozwiązania

• Problem: Bot się nie uruchamia. Przyczyna: Błędny token. Rozwiązanie: Sprawdź BOT_TOKEN w .env, zbędne spacje i znaki nowej linii.
• Problem: Komendy nie są wykonywane. Przyczyna: Router nie jest podłączony. Rozwiązanie: Upewnij się, że dodałeś router do Dispatcher i uruchamiasz polling.

Krok 6: Planner zadań na APScheduler

Cel etapu

Skonfigurować cykliczne kontrole wszystkich włączonych proxy co 5–15 minut z uwzględnieniem bieżącego interwału z ustawień, równolegle sprawdzać bez przeciążania, logować historię i wysyłać alerty przy zmianach.

Szczegółowa instrukcja

1. Otwórz app/scheduler.py. Zainicjalizuj AsyncIOScheduler.
2. Stwórz zadanie run_checks, które wybiera z bazy wszystkie proxy z enabled=1, wysyła je do sprawdzenia w puli asynchronicznych zadań z semaforem PROXY_CONCURRENCY, zbiera wyniki.
3. Dla każdego wyniku zapisuj ciąg w checks. Aktualizuj last_ip w proxies w przypadku zmiany.
4. Logika incydentów: jeśli status DOWN i incydentu nie ma, utwórz zapis incydentów (active=1). Jeśli status znowu UP, zamknij incydent (active=0, closed_at=ts). Dla DEGRADED analogicznie, ale można łączyć DEGRADED w jeden aktywny incydent lub zamykać przy przywróceniu.
5. Wysyłaj powiadomienia: przy pierwszej zmianie na DOWN, przy DEGRADED, przy przywróceniu na UP, przy zmianie IP lub kraju. Grupuj wiadomości, jeśli jednocześnie wiele proxy zmieniło status: wyślij jeden nagłówek i listę elementów.
6. Stwórz zadanie scheduler.add_job(run_checks, 'interval', minutes=CHECK_INTERVAL_MINUTES, id='checks', replace_existing=True). Dodaj reakcję na /interval, aby ponownie uruchomić zadanie z nowym interwałem bez ponownego uruchamiania bota.
7. Uruchom scheduler.start() razem z uruchomieniem bota i serwera webowego w main.

Porada:

Porada: Dodaj losowy jitter 0–60 sekund do harmonogramu, aby zredukować szczytowe obciążenie i uniknąć synchronizacji zapytań.

✅ Weryfikacja: Ustaw CHECK_INTERVAL_MINUTES=1 tymczasowo i uruchom aplikację. W logach powinieneś zobaczyć, że kontrole wykonywane są co minutę, a nowe wiersze dodawane są do tabeli checks.

Możliwe problemy i rozwiązania

• Problem: Kontrole się nie uruchamiają. Przyczyna: Scheduler się nie uruchamia. Rozwiązanie: Upewnij się, że wywołano scheduler.start() i nie ma wyjątków przy tworzeniu zadania.
• Problem: Zablokowania bazy. Przyczyna: Zbyt częste zapisy. Rozwiązanie: Skróć interwał lub dodaj buforowanie zapisów, zapisując pakietowo.

Krok 7: Integracja wszystkich części w punkcie wejścia main.py

Cel etapu

Zbierz aplikację całościowo: konfiguracja, Baza Danych, bot, sprawdzacz, planner i web-dashboard. Zapewnij prawidłowy start i zatrzymanie, przetwórz sygnały kończenia.

Szczegółowa instrukcja

1. Otwórz app/main.py. Importuj asyncio, logging, uvloop (jeśli Linux), aiogram, funkcje init_db, router, scheduler, serwer webowy.
2. Załaduj konfigurację z .env przez python-dotenv. Sprawdź obowiązkowe klucze: BOT_TOKEN i DB_PATH.
3. Inicjalizuj Baza Danych: await init_db(DB_PATH).
4. Stwórz Dispatcher i Router z aiogram 3, dodaj handlery komend.
5. Stwórz planner APScheduler i dodaj zadanie run_checks zgodnie z interwałem.
6. Uruchom jednocześnie trzy komponenty: polling bota, scheduler.start(), web-dashboard. Użyj asyncio.gather dla równoległego uruchomienia.
7. Przetwórz sygnały SIGINT i SIGTERM: prawidłowo zatrzymaj scheduler, zamknij połączenia z BD i zakończ polling.

Uproszczona koncepcja uruchomienia w jednej linii: import asyncio, logging; async def main(): await init_db(...); dp=Dispatcher(); dp.include_router(router); scheduler=AsyncIOScheduler(); scheduler.add_job(run_checks, 'interval', minutes=interval); scheduler.start(); await asyncio.gather(dp.start_polling(bot), start_web_server(), wait_for_signals(scheduler)); asyncio.run(main()).

Porada: Dodaj szczegółowe logowanie na poziomie INFO i WARN. To pomoże szybko zrozumieć, co się dzieje podczas uruchamiania i podczas kontrolek.

✅ Weryfikacja: Uruchom python app/main.py. Bot powinien odpowiedzieć na /start, web-dashboard powinien otwierać się pod adresem http://localhost:8080, a planner powinien rozpocząć kontrole zgodnie z interwałem.

Możliwe problemy i rozwiązania

• Problem: Adres 0.0.0.0:8080 zajęty. Przyczyna: Port jest używany przez inny proces. Rozwiązanie: Zmień DASHBOARD_PORT w .env lub zatrzymaj zajęty proces.
• Problem: Błąd podczas uruchamiania polling. Przyczyna: Błędny token lub sieć. Rozwiązanie: Sprawdź BOT_TOKEN i dostęp do internetu.

Krok 8: Format powiadomień i grupowanie proxy

Cel etapu

Uczynić powiadomienia zrozumiałymi, zwięzłymi i informacyjnymi. Skonfiguruj grupowanie proxy według regionów i typów, aby wiadomości były bardziej usystematyzowane.

Format powiadomień

• Nagłówek: Zmiany statusu proxy.
• Podtytuł: data i godzina.
• Sekcje według priorytetu: DOWN, następnie DEGRADED, następnie Przywrócenie, następnie Zmiana IP, następnie Zmiana geo.
• Każdy punkt: [Region] [Grupa] [Label] ([type]) status, latency, IP, kraj, powód.
• Ikony: DOWN oznaczamy ❌, DEGRADED — ⚠️, UP — ✅.

Grupowanie proxy

• Według pola region: EU, US, ASIA i inne.
• Według pola group_name: logiczne grupowanie, na przykład EUROPA, PAID, TESTING.
• Według typu: http, https, socks5.

Realizacja

1. W app/bot/notifications.py stwórz funkcje group_by_region i group_by_type, aby pogrupować wyniki przed wysłaniem. To pozwoli wysyłać mniej wiadomości i zwiększy czytelność.
2. Dla każdego bloku formułuj krótkie ciągi z kluczowymi danymi: "EU, EUROPE, EU-1 (http): ❌ DOWN, timeout" lub "US, AMERICA, US-2 (socks5): ⚠️ DEGRADED, 2300ms, country US".
3. Jeśli zmian jest wiele, wyślij jeden podsumowujący nagłówek i kilka wiadomości według regionów. To lepsze niż 100 osobnych wiadomości.

✅ Weryfikacja: Na chwilę zamień jedno proxy na na pewno niedziałający endpoint. Czekaj na kontrolę. Otrzymasz powiadomienie z blokiem DOWN. Przywróć proxy do pracy, a zobaczysz powiadomienie o przywróceniu.

Możliwe problemy i rozwiązania

• Problem: Zbyt długie wiadomości. Przyczyna: Dużo szczegółów. Rozwiązanie: Skróć pola, w szczegóły wchodzić po komendzie /status lub /details id.
• Problem: Duplikaty alertów. Przyczyna: Powtórne wysłanie przy niezmienionym statusie. Rozwiązanie: Wysyłaj powiadomienie tylko przy przejściu statusów i przy wyraźnej zmianie IP lub kraju.

Krok 9: Wdrożenie na VPS (Ubuntu) i automatyczne uruchamianie

Cel etapu

Rozpocząć bota na zdalnym serwerze, skonfigurować automatyczne uruchamianie przez systemd, zapewnić logowanie i prostą aktualizację.

Przygotowanie VPS

1. Połącz się z VPS: ssh user@server_ip.
2. Zaktualizuj system: sudo apt update && sudo apt upgrade -y.
3. Zainstaluj Pythona, jeśli nie jest zainstalowany: sudo apt install -y python3.11 python3.11-venv python3-pip.
4. Utwórz użytkownika dla aplikacji, jeśli to konieczne: sudo adduser botuser; następnie zaloguj się pod nim.
5. Skopiuj projekt na serwer przez scp lub git clone Twojego repozytorium.
6. Utwórz wirtualne środowisko oraz zainstaluj zależności jak na kroku 2.
7. Sprawdź .env. Wypełnij rzeczywiste wartości BOT_TOKEN i TELEGRAM_CHAT_ID.

systemd serwis

1. Utwórz plik serwisu: sudo nano /etc/systemd/system/proxy-monitor.service.
2. Przykład jednostki (jedna linia): [Unit] Description=Proxy Monitor Bot After=network.target [Service] Type=simple User=botuser WorkingDirectory=/home/botuser/proxy-monitor-bot ExecStart=/home/botuser/proxy-monitor-bot/.venv/bin/python -m app.main Environment=PYTHONUNBUFFERED=1 Restart=on-failure RestartSec=5 [Install] WantedBy=multi-user.target.
3. Przeładuj demona: sudo systemctl daemon-reload.
4. Włącz automatyczne uruchamianie: sudo systemctl enable proxy-monitor.service.
5. Uruchom: sudo systemctl start proxy-monitor.service.
6. Sprawdź log: journalctl -u proxy-monitor.service -f.

Porada: Jeśli używasz uvloop, upewnij się, że jest zainstalowany w wirtualnym środowisku. Na Windows uvloop nie jest używany, na Linux poprawia wydajność.

✅ Weryfikacja: Bot odpowiada na /start z Twojego Telegrama przy uruchomionym serwisie. W logach widać uruchomienie planisty, a web-dashboard nasłuchuje portu 8080.

Możliwe problemy i rozwiązania

• Problem: Serwis pada od razu. Przyczyna: Błędna ścieżka do Pythona lub roboczej katalogu. Rozwiązanie: Sprawdź ExecStart i WorkingDirectory, uprawnienia użytkownika.
• Problem: Brak dostępu do portu 8080 z zewnątrz. Przyczyna: UFW lub zapora w chmurze. Rozwiązanie: Otwórz port: sudo ufw allow 8080 lub skonfiguruj reguły w panelu dostawcy.

Sprawdzanie wyników

Checklist

• Bot przyjmuje komendy /start, /status, /list, /chatid.
• Planner uruchamia kontrole zgodnie z interwałem, zapisuje pojawiają się w tabeli checks.
• Gdy proxy pada, bot wysyła powiadomienie do Telegramu.
• Zmiana IP jest rejestrowana i zaznaczana w wiadomościach.
• Geolokalizacja jest określana i porównywana z oczekiwanym krajem.
• Dashboard jest wyświetlany na wskazanym porcie, pokazuje dostępność i ostatnie wydarzenia.
• Eksport CSV działa i generuje poprawny plik.

Jak przetestować

1. Wyłącz jedno z proxy lub podaj błędny port. Czekaj na alert DOWN.
2. Tymczasowo zmień endpoint na inny, aby wywołać zmianę IP. Upewnij się, że bot informuje o zmianie IP i kraju, jeśli został zmieniony.
3. Sztucznie zwiększ latencję przez ograniczenie sieci (jeśli to możliwe) lub przetestuj ciężkim zapytaniem, aby uzyskać DEGRADED.
4. Sprawdź komendy: /interval 5, /list, /enable, /disable i upewnij się w odzwierciedleniu statusu.

Wskaźniki sukcesu

• Stabilne kontrole bez błędów w logach.
• Prawidłowo przychodzące alerty przy realnych problemach.
• Poprawne działanie logiki incydentów: otwieranie przy DOWN i zamykanie przy UP.

✅ Weryfikacja: Jeśli wszystkie punkty checklisty przechodzą i scenariusze testowe działają, Twój bot jest prawidłowo wdrożony i wykonuje działania monitorujące.

Typowe błędy i rozwiązania

• Problem: Brak powiadomień w Telegramie → Przyczyna: Błędny TELEGRAM_CHAT_ID lub bot nie jest członkiem grupy → Rozwiązanie: Wykonaj /chatid w odpowiednim czacie, zapisz ID w .env, dodaj bota do czatu i daj mu prawo pisania.
• Problem: Wszystkie proxy są DOWN, mimo że są działające → Przyczyna: Błędny format URL proxy lub typ → Rozwiązanie: Sprawdź type=http, https lub socks5, sprawdź autoryzację user:pass i endpoint host:port, przetestuj jedno proxy osobno.
• Problem: Geolokalizacja nie jest określana → Przyczyna: Limit ip-api lub błąd sieci → Rozwiązanie: Włącz kasowanie, zmniejsz częstotliwość zapytań do ip-api, sprawdź IP_API_URL.
• Problem: Baza rośnie zbyt szybko → Przyczyna: Zbyt częste sprawdzenia i duża liczba proxy → Rozwiązanie: Zwiększ interwał, włącz czyszczenie starych rekordów, eksportuj archiwa i usuwaj starsze niż 30–90 dni.
• Problem: Wysokie opóźnienie wszystkich proxy → Przyczyna: Serwer przeciążony lub problemy sieciowe → Rozwiązanie: Zwiększ zasoby VPS, zmniejsz PROXY_CONCURRENCY, użyj bliższego regionu.
• Problem: Duplikujące alerty o tym samym incydencie → Przyczyna: Nie przechowuje się stan incydentu → Rozwiązanie: Użyj tabeli incydentów i wysyłaj alert tylko przy zmianie statusu.
• Problem: Bot często pada przy wyjątkach → Przyczyna: Nieuchwycone błędy → Rozwiązanie: Owiń operacje sieciowe w try-except, dodaj logowanie i retry.

Dodatkowe możliwości

Zaawansowane ustawienia

• Elastyczne progi degradacji: przechowuj progi latencji według grup w tabeli settings.
• Różne interwały według grup: ważne proxy sprawdzaj co 5 minut, drugorzędne co 15–30.
• Rozszerzone parsowanie komend: użyj formatu key=value i rygorystycznej walidacji parametrów.

Optymalizacja

• Włącz uvloop na Linuxie dla przyspieszenia cyklu zdarzeń.
• Użyj zapisu batch w SQLite lub trybu WAL dla przyspieszenia równoległych zapisów.
• Paremetrzyzuj semafory dla zrównoważenia między prędkością a obciążeniem.

Prosty dashboard

W app/web/server.py uruchom serwer aiohttp.web. Na stronie głównej wyświetl podsumowanie: liczba proxy, ile UP, DOWN, DEGRADED, średnie opóźnienie, dostępność przez 24 godziny. Zrób endpoint /export?from=ts&to=ts do eksportowania CSV. Minimalny HTML może być prostą listą i tabelą ze statusem. Wizualizację wykresów można dodać później przez proste SVG.

Porada: Dodaj przycisk Zmień interwał bezpośrednio w dashboardzie, który wywołuje endpoint i zmienia wartość w ustawieniach. To przyspieszy zarządzanie bez komend Telegram.

Kopie zapasowe

Skrypt do backupu bazy: tar -czf backup-$(date +%F).tar.gz data/monitor.db. Zaplanuj codzienny backup przez cron lub timery systemd. Przechowuj przynajmniej ostatnie 7 archiwów.

⚠️ Uwaga: Nie edytuj pliku bazy na żywo. W przypadku ręcznych poprawek zatrzymaj serwis, zrób kopię, edytuj i uruchom ponownie.

FAQ

• Jak dodać proxy z autoryzacją? W komendzie /addproxy podaj username i password, добавить user:pass туда IP. Przykład: type=http endpoint=user:pass@host:port.
• Jak ograniczyć sprawdzenia w nocy? Zrealizuj harmonogram cron-like w APScheduler, dodając wyzwalacz z oknem czasowym, lub po prostu zwiększ interwał w nocy przez automatyczną regułę.
• Co zrobić, jeśli ip-api.com jest niedostępny? Użyj pamięci podręcznej i tymczasowo pomijaj zapytania geo. Później można dodać zapasowe źródło geohistory.
• Jak filtrować powiadomienia? Dodaj poziomy istotności i ustawienia: wysyłać tylko DOWN, a DEGRADED wysyłać podsumowująco raz na godzinę.
• Czy można używać PostgreSQL? Tak. Zastąp aiosqlite asyncpg, dostosuj SQL i połączenie. Dla dużych wolumenów to bardziej wygodne.
• Jak w bezpieczny sposób przechowywać tokeny? W .env na serwerze z prawami 600. Okresowo aktualizuj token w razie potrzeby.
• Dlaczego bot czasami uzyskuje timeout? Sieci są niestabilne. Zwiększ PROBE_TIMEOUT, zmniejsz równoległość, sprawdź trasy.
• Czy można sprawdzić proxy IPv6? Tak, jeśli docelowe serwisy wspierają IPv6. Upewnij się, że endpoint zwraca IPv6, a ip-api prawidłowo przetwarza adres.
• Jak śledzić konkretnego operatora telekomunikacyjnego? W wiadomości pamiętaj o org z ip-api i porównuj z oczekiwanym wartością na podstawie zasady.
• Jak tymczasowo wyłączyć proxy? Użyj /disable id, aby tymczasowo wyłączyć proxy z kontroli bez usuwania z bazy.

Podsumowanie

Przeszedłeś całą drogę od pomysłu do działającego rozwiązania: stworzyłeś bota Telegrama w aiogram 3, napisałeś moduł do sprawdzania proxy na aiohttp, skonfigurowałeś planista APScheduler, zapisałeś historię w SQLite, zrealizowałeś powiadomienia o upadkach, degradacjach, zmianach IP i geolokalizacji, wdrożyłeś aplikację na VPS z automatycznym uruchamianiem przez systemd i dodałeś prosty web-dashboard z eksportem logów. Teraz masz elastyczne narzędzie, które można skalować: dzielić grupy, ustawiać progi, łączyć wykresy i raporty, przenosić na PostgreSQL przy wzroście obciążenia. Następny krok — automatyzacja konserwacji: ustawienia rotacji logów, codzienne kopie zapasowe bazy, monitorowanie samego bota przez metryki systemowe VPS. Możesz dalej się rozwijać: dodać rolę dostępu do komend bota, wprowadzić SLA-raporty dla grup, połączyć alternatywne źródła danych geo, stworzyć interfejs webowy na dowolnej wygodnej dla Ciebie bibliotece. Powodzenia i stabilnego uptime dla Twoich proxy!