Spis treści

Wprowadzenie: dlaczego długie pobieranie prawie zawsze się zrywa i to normalne

Jeśli choć raz uruchamiałeś duże pobieranie danych, znasz to uczucie. Proces trwał kilka godzin, doszedł do dziewięćdziesięciu procent i się zerwał. Połączenie padło, serwer zwrócił błąd, laptop zasnął. I wszystko trzeba zaczynać od nowa. Ten przewodnik powstał po to, żeby to się więcej nie powtarzało.

Co zyskasz. Nauczysz się budować downloader, który przetrwa zerwania. Dokończy pliki od środka, zapamięta, na której stronie się zatrzymał, nie stworzy duplikatów przy powtórzeniu i potrafi wznowić nawet po długiej przerwie. Otrzymasz gotowy szkielet w Pythonie, który dostosujesz do swojego zadania.

Dla kogo jest ten przewodnik. Dla inżynierów, analityków i programistów, którzy pobierają dane z API, ściągają duże pliki lub zbierają wyniki strona po stronie przez proxy. Poziom średni. Powinieneś rozumieć podstawy HTTP i umieć czytać kod w Pythonie. Głęboka wiedza o programowaniu sieciowym nie jest wymagana.

Co trzeba wiedzieć wcześniej. Podstawy Pythona, pojęcie zapytania i odpowiedzi HTTP, czym są nagłówki i kody statusu. Jeśli pracowałeś z biblioteką requests, to wystarczy.

Ile czasu to zajmie. Przeczytanie i zrozumienie koncepcji zajmie około czterdziestu minut. Zbudowanie działającego downloadera na naszym szkielecie od jednej do trzech godzin, w zależności od twojego źródła danych.

Ważne doprecyzowanie tematu. Nie będziemy omawiać kodów statusu typu 429 ani strategii ponawiania z opóźnieniem (backoff). Na to jest osobny materiał. Tutaj skupiamy się tylko na jednym: stanie procesu i jego wznawianiu. Jak zapisywać postęp, jak nie zgubić i nie zdublować danych, jak kontynuować od miejsca, w którym się zatrzymałeś.

Wskazówka: Miej pod ręką notatnik lub osobny plik, do którego będziesz wypisywać parametry swojego źródła: czy obsługuje wznawianie, czy ma nawigację strona po stronie, jaki ma format kursora. Te notatki przydadzą się na każdym kroku.

Przygotowanie wstępne: narzędzia, dostępy i środowisko

Zanim zaczniemy pisać kod, zbierzmy środowisko pracy. Zajmie to dziesięć minut, ale oszczędzi godziny debugowania.

Co trzeba zainstalować

  1. Zainstaluj Python w wersji 3.10 lub nowszej. Sprawdź wersję komendą python --version w terminalu.
  2. Utwórz wirtualne środowisko komendą python -m venv venv, żeby zależności projektu nie mieszały się z systemowymi.
  3. Aktywuj środowisko. W Windows to venv\Scripts\activate, w macOS i Linux to source venv/bin/activate.
  4. Zainstaluj bibliotekę do zapytań HTTP komendą pip install requests.
  5. Do szybszej pracy z bazą stanu nie musisz nic dodatkowo instalować: moduł sqlite3 jest już częścią standardowej biblioteki Pythona.

Co trzeba mieć z dostępów

  • Dostęp do twojego źródła danych: URL, token lub klucz API, jeśli jest wymagany.
  • Proxy od Proxeon z adresem, portem i danymi do autoryzacji. Bez stabilnego proxy stabilne pobieranie traci sens, bo to właśnie proxy rozkłada obciążenie i czyni połączenia przewidywalnymi.
  • Miejsce na dysku na plik stanu i na same pobierane dane.

Sprawdzenie proxy Proxeon

  1. Weź ciąg połączenia w postaci http://login:hasło@adres:port.
  2. Sprawdź go prostym zapytaniem. W terminalu wykonaj curl -x http://login:hasło@adres:port https://api.ipify.org i upewnij się, że zwrócił adres IP proxy, a nie twój własny.

Wskazówka: Zapisz ciąg połączenia do proxy w zmiennej środowiskowej, a nie w kodzie. Tak przypadkiem nie wyślesz hasła do systemu kontroli wersji. W kodzie czytaj je przez os.environ.

⚠️ Uwaga: Zawsze pracuj tylko z tymi źródłami danych, do których masz dostęp na podstawie zgodnych z prawem podstaw. Przestrzegaj warunków korzystania z serwisu i limitów wskazanych przez jego właściciela. Proxy Proxeon służy do legalnej pracy inżynierskiej: rozkładania obciążenia, stabilności połączeń i poprawnego pobierania danych.

✅ Sprawdzenie: Środowisko jest gotowe, jeśli komenda python -c "import requests, sqlite3" zadziałała bez błędów, a zapytanie przez proxy zwróciło adres serwera proxy.

Podstawowe pojęcia: słownik stabilnego pobierania prostymi słowami

Zanim napiszemy kod, omówmy kluczowe terminy. Bez nich kolejne kroki będą brzmiały jak zaklęcia.

Wznawianie

Wznawianie to kontynuowanie pobierania pliku od tego bajtu, na którym zostało przerwane. Zamiast ściągać plik od nowa, prosisz serwer o wydanie tylko brakującego fragmentu. Działa przez nagłówek HTTP Range.

Checkpoint

Checkpoint to zapisany punkt postępu. Wyobraź sobie zapis w grze komputerowej. Jeśli coś poszło nie tak, wracasz do ostatniego zapisu, a nie do początku gry. W pobieraniu checkpoint przechowuje informację, na której stronie lub rekordzie się zatrzymałeś.

Kursor

Kursor to znacznik, który API ci daje, żebyś mógł poprosić o następną porcję danych. Często to ciąg znaków w rodzaju eyJvZmZzZXQiOjEwMH0. Wysyłasz go z powrotem, a serwer rozumie, od czego kontynuować.

Idempotentność

Idempotentność to właściwość operacji, przy której jej powtórzenie nie zmienia wyniku. Jeśli dwa razy zapisałeś ten sam rekord z tym samym kluczem, w efekcie rekord jest jeden, a nie dwa. To ochrona przed duplikatami przy powtórzeniach.

Deduplikacja

Deduplikacja to odsiewanie powtarzających się rekordów. Nawet przy starannej pracy ten sam obiekt może przyjść dwa razy. Deduplikacja gwarantuje, że w twoim końcowym zbiorze pozostanie tylko jeden.

Podstawowa zasada

Stabilny downloader opiera się na jednym pomyśle: postęp trzeba zapisywać stale, a nie tylko na końcu. Każdy krok może być ostatnim przed zerwaniem. Znaczy to, że po każdym udanym fragmencie pracy stan powinien zostać zapisany na dysku. Wtedy wznowienie to po prostu odczyt stanu i kontynuacja.

Wskazówka: Zapamiętaj zasadę trzech pytań dla każdego pobierania. Pierwsze: gdzie się zatrzymałem? Drugie: jak nie zdublować już otrzymanego? Trzecie: co się przeterminuje, zanim wrócę? Odpowiedzi na nie składają się właśnie na stabilność.

Krok 1: Wznawianie przez HTTP za pomocą nagłówka Range

Cel etapu. Nauczyć się pobierać duży plik tak, żeby po zerwaniu kontynuować od niedokończonego bajtu, a nie od zera.

Jak to działa

HTTP pozwala zażądać nie całego pliku, a jego części. W tym celu do zapytania dodaje się nagłówek Range. Na przykład Range: bytes=1048576- znaczy: wydaj mi wszystko, począwszy od bajtu numer 1048576. Ale najpierw trzeba się upewnić, że serwer to potrafi.

  1. Wyślij na plik zapytanie metodą HEAD lub zwykłe GET i spójrz na nagłówki odpowiedzi.
  2. Znajdź nagłówek Accept-Ranges. Jeśli ma wartość bytes, serwer obsługuje wznawianie.
  3. Jeśli nagłówka nie ma lub ma wartość none, wznawianie jest niemożliwe. W tym wypadku trzeba ściągnąć plik w całości za jednym razem albo poszukać alternatywnego źródła.

Sprawdzenie obsługi wznawiania

Oto kod, który sprawdza, czy serwer umie wydawać części pliku.

import requests
def supports_resume(url, proxies):
resp = requests.head(url, proxies=proxies, timeout=30, allow_redirects=True)
accept = resp.headers.get("Accept-Ranges", "none")
total = resp.headers.get("Content-Length")
return accept.lower() == "bytes", total

Wznawianie pliku od środka

Teraz główny kod. Patrzy, ile bajtów już ściągnięto lokalnie, i prosi serwer tylko o resztę.

import os
import requests
def download_resumable(url, dest, proxies):
already = 0
if os.path.exists(dest):
already = os.path.getsize(dest)
headers = {}
if already > 0:
headers["Range"] = f"bytes={already}-"
mode = "ab" if already > 0 else "wb"
with requests.get(url, headers=headers, proxies=proxies,
 stream=True, timeout=60) as r:
if already > 0 and r.status_code == 200:
mode = "wb"
already = 0
with open(dest, mode) as f:
for chunk in r.iter_content(chunk_size=65536):
if chunk:
f.write(chunk)
return os.path.getsize(dest)

Omówmy ważne momenty. Jeśli serwer zwrócił status 206, uczciwie wydał część pliku i dopisanie przejdzie poprawnie. Jeśli serwer zwrócił 200 mimo nagłówka Range, znaczy to, że zignorował wznawianie i wydaje plik w całości. W tym wypadku przełączamy się w tryb pełnego nadpisania, żeby nie skleić starego fragmentu z nowym i nie uszkodzić pliku.

⚠️ Uwaga: Nigdy nie dopisuj danych w trybie ab, jeśli nie masz pewności, że serwer odpowiedział kodem 206. Inaczej dostaniesz uszkodzony plik, gdzie początek to resztka z poprzedniej próby, a dalsza część to nowy pełny plik. Taki plik otworzy się z błędem, a ty stracisz czas na szukanie przyczyny.

Wskazówka: Ściągaj nie od razu do docelowego pliku, a do tymczasowego z rozszerzeniem .part. Kiedy pobieranie całkowicie się zakończy, zmień jego nazwę na finalną. Tak nigdy nie pomylisz gotowego pliku z niedokończonym.

Kontrola integralności

Po pełnym pobraniu dobrze byłoby sprawdzić, czy plik się nie uszkodził. Jeśli serwer wydawał nagłówek Content-Length, porównaj go z rzeczywistym rozmiarem pliku na dysku. Jeśli rozmiary się zgadzają, plik doszedł w całości.

def verify_size(dest, expected):
if expected is None:
return True
return os.path.getsize(dest) == int(expected)

✅ Sprawdzenie: Przerwij pobieranie w połowie, zamykając program. Uruchom go ponownie. W logach powinieneś zobaczyć, że zapytanie poszło z nagłówkiem Range, a plik dopisał się, a nie zaczął od nowa. Końcowy rozmiar zgadza się z oczekiwanym.

Krok 2: Checkpointy dla pobierania strona po stronie

Cel etapu. Skonfigurować zapisywanie postępu dla API, które wydają dane stronami, żeby po zerwaniu kontynuować od właściwej strony.

Co dokładnie zapisywać

Plik wznawia się bajt po bajcie, a pobieranie strona po stronie rządzi się zupełnie inną logiką. Tu nie ma bajtów, są strony i rekordy. Znaczy to, że w checkpoincie trzeba przechowywać coś innego.

  • Kursor, jeśli API działa na kursorach. To najpewniejszy wariant, bo kursor sam wie, od czego kontynuować.
  • Numer strony lub przesunięcie, jeśli API działa na offset i limit. Przechowuj numer ostatniej pomyślnie przetworzonej strony.
  • Identyfikator ostatniego rekordu, jeśli można sortować po rosnącym ID lub dacie. Wtedy następne zapytanie prosi o rekordy z ID większym niż zapisane.
  • Licznik przetworzonych rekordów do kontroli i raportowania.

Gdzie przechowywać stan

Masz trzy główne warianty, od prostego do niezawodnego.

  1. Plik JSON. Najprościej. Zapisujesz słownik z kursorem i licznikiem do pliku po każdej stronie. Nadaje się do pojedynczych, nie równoległych pobierań.
  2. Baza SQLite. Niezawodniej. Daje transakcje, więc stan nie zepsuje się przy zerwaniu w momencie zapisu. Dobra, gdy danych jest dużo i potrzebna jest deduplikacja.
  3. Zewnętrzna baza danych. Dla dużych rozproszonych pobierań, gdy kilka procesów dzieli jedną pracę.

Zapisywanie checkpointu w JSON

import json
import os
def save_checkpoint(path, cursor, page, last_id, count):
tmp = path + ".tmp"
data = {
"cursor": cursor,
"page": page,
"last_id": last_id,
"count": count,
}
with open(tmp, "w") as f:
json.dump(data, f)
os.replace(tmp, path)
def load_checkpoint(path):
if not os.path.exists(path):
return {"cursor": None, "page": 0, "last_id": None, "count": 0}
with open(path) as f:
return json.load(f)

Zwróć uwagę na trik z plikiem tymczasowym. Piszemy do pliku z przyrostkiem .tmp, a potem atomowo zmieniamy jego nazwę przez os.replace. To chroni przed sytuacją, gdy program padł dokładnie w trakcie zapisu checkpointu. Stary checkpoint przy tym pozostaje cały, a nie zamienia się w połowę JSON-a, którego nie da się odczytać.

⚠️ Uwaga: Nigdy nie zapisuj checkpointu bezpośrednio do tego samego pliku na stary, bez pliku tymczasowego. Zerwanie w środku zapisu zostawi ci uszkodzony checkpoint i wznowienie stanie się niemożliwe. Atomowa zamiana rozwiązuje ten problem całkowicie.

Wskazówka: Zapisuj checkpoint dopiero po tym, jak dane strony naprawdę zostaną zapisane w magazynie. Kolejność jest taka: otrzymałeś stronę, zapisałeś dane, potem zaktualizowałeś checkpoint. Jeśli zamienisz kolejność, przy zerwaniu pominiesz stronę i stracisz dane.

Główna pętla z checkpointami

def paginate(fetch_page, save_data, cp_path, proxies):
cp = load_checkpoint(cp_path)
cursor = cp["cursor"]
count = cp["count"]
while True:
items, next_cursor = fetch_page(cursor, proxies)
if not items:
break
save_data(items)
count += len(items)
last_id = items[-1].get("id")
save_checkpoint(cp_path, next_cursor, cp["page"] + 1,
last_id, count)
cursor = next_cursor
if next_cursor is None:
break
return count

✅ Sprawdzenie: Uruchom pobieranie, pozwól mu przetworzyć kilka stron, przerwij. Otwórz plik checkpointu i upewnij się, że zapisany jest w nim aktualny kursor i licznik. Uruchom ponownie: pobieranie powinno kontynuować od zapisanego kursora, a nie od pierwszej strony.

Krok 3: Idempotentność, żeby powtórzenie nie tworzyło duplikatów

Cel etapu. Sprawić, żeby ponowne uruchomienie lub powtórzenie konkretnego zapytania nie prowadziło do pojawienia się takich samych rekordów w twoim magazynie.

Dlaczego powstają duplikaty

Wyobraź sobie: otrzymałeś stronę danych, zapisałeś ją do pliku, ale program padł przed zaktualizowaniem checkpointu. Przy następnym uruchomieniu poprosisz o tę samą stronę ponownie. Dane przyjdą powtórnie i zapiszą się drugi raz. Tak rodzą się duplikaty. To nieunikniona konsekwencja zerwań i trzeba z nią walczyć na poziomie architektury.

Klucz deduplikacji

Główne narzędzie idempotentności to klucz deduplikacji. To pole lub kombinacja pól, które jednoznacznie określają rekord. Właściwy wybór klucza rozwiązuje połowę problemu.

  • Naturalne ID. Jeśli rekord ma unikalny identyfikator od źródła, użyj go. To idealny klucz.
  • Kombinacja pól. Jeśli nie ma jednego ID, złóż klucz z kilku stabilnych pól. Na przykład email plus data rejestracji.
  • Hash zawartości. Jeśli stabilnych pól nie ma w ogóle, policz hash z całego rekordu. To ostateczność, bo każda zmiana pola stworzy nowy klucz.

Zapis bez duplikatów przez UPSERT

Jeśli przechowujesz wynik w SQLite lub innej bazie, użyj wstawiania z ignorowaniem konfliktu. Wtedy ponowny zapis z tym samym kluczem po prostu nic nie zrobi.

import sqlite3
def init_db(path):
conn = sqlite3.connect(path)
conn.execute(
"CREATE TABLE IF NOT EXISTS records ("
"dedup_key TEXT PRIMARY KEY, payload TEXT)"
)
conn.commit()
return conn
def save_records(conn, items):
rows = [(item["id"], json.dumps(item)) for item in items]
conn.executemany(
"INSERT OR IGNORE INTO records (dedup_key, payload) "
"VALUES (?, ?)", rows
)
conn.commit()

Kluczowy szczegół to PRIMARY KEY na polu dedup_key. Baza sama odrzuci ponowne wstawienie z tym samym kluczem, bo INSERT OR IGNORE połknie konflikt po cichu. Nie musisz ręcznie sprawdzać, czy taki rekord już jest. Baza robi to za ciebie i robi to szybko.

Wskazówka: Wybierz klucz deduplikacji raz na początku projektu i zapisz go w dokumentacji. Zmiana klucza w środku pobierania oznacza, że stare i nowe rekordy przestaną się dopasowywać i duplikaty jednak się pojawią. Stabilność klucza jest ważniejsza niż jego uroda.

✅ Sprawdzenie: Uruchom pobieranie dwa razy pod rząd na tym samym zakresie danych. Policz liczbę wierszy w bazie komendą SELECT COUNT(*) FROM records. Liczba powinna być taka sama po pierwszym i po drugim uruchomieniu.

Krok 4: Deduplikacja wyników bez rozdymania pamięci

Cel etapu. Odsiewać powtarzające się rekordy na milionach wierszy, nie ładując całej pamięci komputera mnóstwem już widzianych kluczy.

Naiwne podejście i jego problem

Najprostsza deduplikacja: trzymać w pamięci zbiór set ze wszystkich widzianych kluczy. Dla każdego nowego rekordu sprawdzać, czy klucz jest w zbiorze. Działa świetnie na setkach tysięcy wierszy. Ale na milionach i dziesiątkach milionów zbiór rozrasta się i zjada gigabajty pamięci operacyjnej. Program zwalnia lub pada.

Rozwiązanie pierwsze: zdać się na bazę

Najprostszy i najpewniejszy sposób na dużych wolumenach to nie trzymać widzianego w pamięci wcale, a powierzyć sprawdzanie bazie przez PRIMARY KEY, jak zrobiliśmy w poprzednim kroku. Baza trzyma indeks na dysku, a nie w pamięci twojego procesu. Poradzi sobie z dziesiątkami milionów kluczy bez obciążania twojej pamięci RAM.

Rozwiązanie drugie: hash rekordu

Kiedy nie ma naturalnego klucza, licz kompaktowy hash z rekordu. Hash zajmuje stały i niewielki rozmiar niezależnie od rozmiaru samego rekordu.

import hashlib
import json
def record_hash(item):
raw = json.dumps(item, sort_keys=True, ensure_ascii=False)
return hashlib.sha256(raw.encode("utf-8")).hexdigest()

Parametr sort_keys=True jest tu krytyczny. Gwarantuje, że rekordy o identycznej zawartości dadzą identyczny hash, nawet jeśli pola w nich szły w innej kolejności. Bez tego sortowania dwa identyczne obiekty mogą dostać różne hashe i przejść jako różne rekordy.

Rozwiązanie trzecie: filtr Blooma dla oszczędności pamięci

Jeśli jednak potrzebujesz szybkiego sprawdzania w pamięci na ogromnych wolumenach, stosuje się filtr Blooma. To struktura, która zajmuje mało miejsca i szybko odpowiada, czy widzieliśmy klucz, czy na pewno nie. Ma pewną cechę: może czasem błędnie powiedzieć, że klucz już był, choć go nie było. Dlatego filtr Blooma stosuje się jako szybki wstępny odsiew, a ostateczne sprawdzenie zostawia się bazie.

  1. Sprawdzamy klucz filtrem Blooma.
  2. Jeśli filtr mówi, że na pewno nie widzieliśmy, od razu piszemy do bazy.
  3. Jeśli filtr mówi, że być może widzieliśmy, robimy dokładne sprawdzenie w bazie.

⚠️ Uwaga: Nie próbuj deduplikować dziesiątek milionów wierszy zwykłym zbiorem w pamięci. Na typowym laptopie doprowadzi to do wyczerpania pamięci i awarii procesu w połowie pobierania. Przenieś obciążenie na dysk przez bazę lub użyj filtra Blooma.

Wskazówka: Jeśli pobierasz dane porcjami i wewnątrz jednej porcji możliwe są duplikaty, deduplikuj porcję w pamięci zwykłym zbiorem przed zapisem do bazy. Porcja jest niewielka, pamięć nie ucierpi, a do bazy pójdzie mniej zbędnych wstawień.

✅ Sprawdzenie: Uruchom deduplikację na dużym zestawie testowym z celowymi powtórzeniami. Sprawdź, że końcowa liczba unikalnych rekordów jest poprawna, a zużycie pamięci przez proces pozostaje stabilne i nie rośnie liniowo z liczbą wierszy.

Krok 5: Równoległość bez strat

Cel etapu. Przyspieszyć pobieranie dzięki równoległym zapytaniom, nie tracąc przy tym ani jednego zadania i poprawnie powtarzając te, które padły.

Kolejka zadań

Podstawą bezpiecznej równoległości jest kolejka zadań. Z góry dzielisz pracę na niezależne kawałki. Na przykład listę stron lub zakresów. Wrzucasz je do kolejki. Kilku workerów bierze zadania z kolejki, wykonuje je i składa wynik. Jeśli worker padł, jego zadanie można zwrócić do kolejki i oddać innemu.

Ograniczenie jednoczesności

Nie można uruchamiać nieskończonej liczby równoległych zapytań. To przeciąży źródło i twoje proxy. Właściwe podejście to ograniczyć liczbę jednoczesnych workerów do rozsądnej wartości. Zacznij od niewielkiej liczby i zwiększaj, obserwując stabilność.

from concurrent.futures import ThreadPoolExecutor, as_completed
def run_parallel(tasks, worker, proxies, max_workers=5):
results = []
failed = []
with ThreadPoolExecutor(max_workers=max_workers) as pool:
future_map = {
pool.submit(worker, t, proxies): t for t in tasks
}
for future in as_completed(future_map):
task = future_map[future]
try:
results.append(future.result())
except Exception:
failed.append(task)
return results, failed

Powtórzenie zadań, które padły

Zebrana lista failed to nie utracone dane, a lista tego, co trzeba powtórzyć. Po pierwszym przejściu przepuszczasz zadania, które padły, jeszcze raz. Zwykle to wystarczy, żeby dobić resztę.

def run_with_retry(tasks, worker, proxies, rounds=3):
remaining = tasks
for _ in range(rounds):
done, remaining = run_parallel(remaining, worker, proxies)
if not remaining:
break
return remaining

Rola proxy Proxeon w równoległości. Przy pracy równoległej proxy rozkłada połączenia, co czyni pobieranie stabilniejszym i bardziej przewidywalnym. Każdy worker pracuje przez swoje połączenie, a obciążenie nie koncentruje się w jednym punkcie.

⚠️ Uwaga: Przy równoległym zapisie do jednego pliku lub do jednego checkpointu powstają wyścigi danych. Dwóch workerów może nadpisać stan drugiego. Zapisuj wyniki tylko do bazy z transakcjami albo używaj osobnego pliku dla każdego workera, a zbiorczy checkpoint zbieraj osobnym wątkiem.

Wskazówka: Rób zadania małe i niezależne. Jeśli jedno zadanie obejmuje zbyt duży zakres, jego zerwanie odrzuci dużo pracy. Małe zadania powtarza się tanio i niemal niezauważalnie.

✅ Sprawdzenie: Uruchom równoległe pobieranie, sztucznie uwal część workerów. Po kolejnych rundach lista remaining powinna stać się pusta, a końcowy zestaw danych kompletny. Porównaj liczbę otrzymanych rekordów z oczekiwaną.

Krok 6: Wznawianie po długiej przerwie

Cel etapu. Poprawnie kontynuować pobieranie, jeśli między próbami minęło dużo czasu, i zrozumieć, co mogło się przeterminować w tym okresie.

Co przeterminowuje się z czasem

Zerwanie na minutę i pauza na dobę to różne sytuacje. Przez długą przerwę część twojego stanu może stać się nieważna.

  • Sesja. Wiele serwisów trzyma sesję przez ograniczony czas. Po długiej pauzie serwer ją zapomni i zapytania zaczną zwracać błąd autoryzacji.
  • Token dostępu. Tokeny API często mają czas życia w minutach lub godzinach. Przeterminowany token trzeba odświeżyć przed kontynuacją.
  • Kursor. Niektóre kursory żyją krótko. Jeśli kursor się przeterminował, trzeba zacząć od najbliższego stabilnego punktu, na przykład po identyfikatorze ostatniego rekordu.
  • Same dane. W czasie pauzy w źródle mogły pojawić się nowe rekordy lub zmienić się stare. To wpływa na przesunięcia przy nawigacji strona po stronie po offset.

Strategia bezpiecznego wznawiania

  1. Przy starcie sprawdź wiek checkpointu. Jeśli jest stary, bądź gotów na to, że część stanu się zestarzała.
  2. Odśwież token dostępu i utwórz nową sesję przed pierwszym zapytaniem. Nie polegaj na starych.
  3. Przedkładaj wznawianie po identyfikatorze ostatniego rekordu, a nie po numerze strony. ID jest stabilne, a numer strony przesuwa się, jeśli dane się zmieniły.
  4. Zrób próbne zapytanie z zapisanym kursorem. Jeśli zwróciło błąd nieprawidłowego kursora, przełącz się na wznawianie po last_id.
def resume(cp, fetch_by_id, fetch_by_cursor, proxies):
if cp["cursor"]:
try:
return fetch_by_cursor(cp["cursor"], proxies)
except CursorExpired:
pass
return fetch_by_id(cp["last_id"], proxies)

Dlaczego wznawianie po ID jest pewniejsze. Wyobraź sobie, że zatrzymałeś się na stronie 50 przy sortowaniu po dacie. Podczas twojej nieobecności dodano nowe rekordy na początek. Teraz strona 50 zawiera zupełnie inne dane, a część rekordów pominiesz. Wznawianie po identyfikatorze ostatniego rekordu na tym nie cierpi: po prostu prosisz o wszystko, co większe od zapisanego ID.

Wskazówka: Zawsze zapisuj w checkpoincie i kursor, i identyfikator ostatniego rekordu jednocześnie. Kursor jest szybszy, ale ID to twoja lina ratunkowa na wypadek, gdyby kursor przeterminował się w czasie długiej pauzy.

✅ Sprawdzenie: Zatrzymaj pobieranie, poczekaj wystarczająco długo, żeby token lub kursor się zestarzały, i uruchom ponownie. Downloader powinien odświeżyć token, wykryć przeterminowany kursor i kontynuować po identyfikatorze bez utraty i bez dublowania rekordów.

Krok 7: Gotowy szkielet stabilnego downloadera w Pythonie

Cel etapu. Zebrać wszystko, czego się nauczyliśmy, w jedną działającą ramę, którą dostosujesz do swojego źródła danych.

Poniżej zebrano ramę łączącą checkpointy, deduplikację przez bazę, odświeżanie tokenu i wznawianie. Funkcje pobierania strony podstawiasz swoje, pod konkretne API.

import os
import json
import sqlite3
import requests
class ResilientLoader:
def __init__(self, cp_path, db_path, proxies):
self.cp_path = cp_path
self.proxies = proxies
self.conn = sqlite3.connect(db_path)
self.conn.execute(
"CREATE TABLE IF NOT EXISTS records ("
"dedup_key TEXT PRIMARY KEY, payload TEXT)"
)
self.conn.commit()
def load_cp(self):
if not os.path.exists(self.cp_path):
return {"cursor": None, "last_id": None, "count": 0}
with open(self.cp_path) as f:
return json.load(f)
def save_cp(self, cp):
tmp = self.cp_path + ".tmp"
with open(tmp, "w") as f:
json.dump(cp, f)
os.replace(tmp, self.cp_path)
def save_records(self, items):
rows = [(str(i["id"]), json.dumps(i)) for i in items]
self.conn.executemany(
"INSERT OR IGNORE INTO records "
"(dedup_key, payload) VALUES (?, ?)", rows
)
self.conn.commit()
def run(self, fetch_page):
cp = self.load_cp()
while True:
items, next_cursor = fetch_page(
cp["cursor"], cp["last_id"], self.proxies
)
if not items:
break
self.save_records(items)
cp["count"] += len(items)
cp["last_id"] = items[-1]["id"]
cp["cursor"] = next_cursor
self.save_cp(cp)
if next_cursor is None:
break
return cp["count"]

Przykład funkcji pobierania strony pod twoje źródło. Tutaj realizujesz logikę zapytania i rozbiór odpowiedzi.

def fetch_page(cursor, last_id, proxies):
params = {"limit": 100}
if cursor:
params["cursor"] = cursor
elif last_id:
params["after_id"] = last_id
r = requests.get(
"https://example-source/api/records",
params=params, proxies=proxies, timeout=60
)
r.raise_for_status()
data = r.json()
return data["items"], data.get("next_cursor")

Uruchomienie całego mechanizmu wygląda prosto.

proxies = {
"http": os.environ["PROXEON_URL"],
"https": os.environ["PROXEON_URL"],
}
loader = ResilientLoader("state.json", "out.db", proxies)
total = loader.run(fetch_page)
print("Wszystkich rekordów:", total)

Wskazówka: Dodaj do pętli logowanie co setnego rekordu: czas, licznik, aktualny kursor. Tak będziesz widział postęp i łatwo zrozumiesz, jeśli pobieranie utknie w jednym miejscu.

✅ Sprawdzenie: Uruchom ramę na realnym źródle, przerwij w połowie, uruchom ponownie. Końcowa liczba rekordów po dokończeniu zgadza się z pełną liczbą rekordów źródła, a ponowne uruchomienie nie zwiększy licznika unikalnych wierszy.

Sprawdzenie wyniku: lista kontrolna stabilnego pobierania

Przejdź przez tę listę. Jeśli wszystkie punkty się spełniają, twój downloader naprawdę jest stabilny.

  • Wznawianie pliku kontynuuje od niedokończonego bajtu, a nie od zera.
  • Downloader poprawnie obsługuje przypadek, gdy serwer ignoruje nagłówek Range.
  • Checkpoint zapisuje się po każdej przetworzonej stronie, a nie tylko na końcu.
  • Checkpoint zapisuje się atomowo przez plik tymczasowy i zmianę nazwy.
  • Ponowne uruchomienie nie tworzy duplikatów w końcowym zestawie.
  • Deduplikacja nie rośnie w pamięci liniowo z liczbą wierszy.
  • Równolegli workerzy nie tracą zadań, które padły, i powtarzają je.
  • Po długiej pauzie token się odświeża, a przeterminowany kursor zastępuje wznawianie po ID.

Jak przetestować

  1. Uruchom pełne pobieranie niewielkiego zestawu i zapamiętaj liczbę rekordów.
  2. Uruchom jeszcze raz na tym samym zestawie i upewnij się, że liczba się nie zmieniła.
  3. Przerwij pobieranie w różnych miejscach: na początku, w środku, bliżej końca.
  4. Po każdym przerwaniu uruchom ponownie i sprawdź, że wynik jest taki sam.

Wskaźniki sukcesu. Liczba unikalnych rekordów jest stabilna między uruchomieniami. Zużycie pamięci nie rośnie niekontrolowanie. Wznawianie zawsze kontynuuje od zapisanego punktu. Uszkodzonych plików po dokończeniu nie ma.

Typowe błędy i ich rozwiązania

Problem: plik po dokończeniu się nie otwiera. Przyczyna: dane dopisywano w trybie ab, choć serwer odpowiedział kodem 200 i wydał plik w całości. Rozwiązanie: sprawdzaj status odpowiedzi, przy 200 przełącz się na pełne nadpisanie pliku od zera.

Problem: po zerwaniu pobieranie zaczyna się od pierwszej strony. Przyczyna: checkpoint zapisywał się tylko na końcu pracy albo nie zapisywał się wcale. Rozwiązanie: zapisuj checkpoint po każdej przetworzonej stronie, zaraz po zapisie danych.

Problem: checkpointu nie da się odczytać, JSON jest uszkodzony. Przyczyna: program padł w trakcie zapisu bezpośrednio do docelowego pliku. Rozwiązanie: pisz do pliku tymczasowego i zamieniaj atomowo przez os.replace.

Problem: w końcowym zestawie są duplikaty. Przyczyna: brak klucza deduplikacji albo jest niestabilny. Rozwiązanie: ustaw PRIMARY KEY na pewnym kluczu i używaj INSERT OR IGNORE.

Problem: proces pada z powodu braku pamięci na dużych wolumenach. Przyczyna: wszystkie widziane klucze trzymane są w zbiorze w pamięci. Rozwiązanie: przenieś sprawdzanie unikalności do bazy albo zastosuj filtr Blooma.

Problem: po długiej pauzie zapytania zwracają błąd autoryzacji. Przyczyna: token lub sesja przeterminowały się przez przerwę. Rozwiązanie: odświeżaj token i twórz nową sesję przy każdym starcie.

Problem: po pauzie część rekordów została pominięta lub zdublowana. Przyczyna: wznawianie szło po numerze strony, a dane w źródle się zmieniły. Rozwiązanie: wznawiaj po identyfikatorze ostatniego rekordu, a nie po offset.

Problem: równolegli workerzy tracą część danych. Przyczyna: kilku workerów pisało do jednego checkpointu i nadpisywało się nawzajem. Rozwiązanie: zapisuj wyniki do bazy z transakcjami, a nie do wspólnego pliku stanu.

Dodatkowe możliwości i optymalizacja

Zapis wsadowy

Nie zapisuj do bazy po jednym wierszu. Zbieraj paczkę z kilkuset rekordów i wstawiaj naraz przez executemany. To wielokrotnie przyspiesza zapis na dużych wolumenach.

Okresowe zatwierdzanie

Wywołuj commit nie przy każdym zapisie, a raz na kilkaset wierszy. Zbyt częsty commit spowalnia bazę. Zbyt rzadki ryzykuje utratę większej ilości danych przy zerwaniu. Znajdź równowagę pod swoje obciążenie.

Raport o postępie

Dodaj szacowanie pozostałego czasu. Znając szybkość przetwarzania stron i łączną liczbę rekordów, przykmiesz, ile jeszcze czekać. To wygodne przy długich pobieraniach.

Osobne magazyny danych surowych i przetworzonych

Trzymaj surowe odpowiedzi osobno od rozebranych rekordów. Jeśli później zmienisz logikę rozbioru, nie będziesz musiał ponownie ściągać danych. Wystarczy przepuścić surowe odpowiedzi przez nowy parser.

Wskazówka: Skonfiguruj proxy Proxeon tak, żeby połączenia były stabilne przez całe pobieranie. Stabilne połączenie zmniejsza liczbę zerwań, a więc twój downloader rzadziej wchodzi we wznawianie i pracuje szybciej.

FAQ: częste pytania o stabilne pobieranie

Jak poznać, czy serwer obsługuje wznawianie pliku? Wyślij zapytanie HEAD i spójrz na nagłówek Accept-Ranges. Wartość bytes oznacza obsługę. Brak nagłówka lub none oznacza, że wznawianie jest niemożliwe.

Co robić, jeśli API nie wydaje kursora, tylko strony? Zapisuj numer strony i, jeśli to możliwe, identyfikator ostatniego rekordu. Wznawiaj najlepiej po ID, bo numery stron przesuwają się przy zmianie danych.

Jak często zapisywać checkpoint? Po każdej pomyślnie przetworzonej i zapisanej stronie. Tak przy zerwaniu stracisz maksymalnie jedną stronę pracy, a nie całe pobieranie.

Czy można deduplikować bez bazy danych? Na małych wolumenach tak, zwykłym zbiorem w pamięci. Na milionach wierszy to niebezpieczne z powodu pamięci. Lepiej użyć bazy z PRIMARY KEY albo filtra Blooma.

Co wybrać jako klucz deduplikacji? Naturalne unikalne ID źródła, jeśli jest. Jeśli nie, kombinację stabilnych pól. W ostateczności hash całego rekordu z sortowaniem kluczy.

Dlaczego wznawianie po ID jest pewniejsze niż po numerze strony? Bo dane w źródle mogą się zmieniać. Nowe rekordy przesuwają strony i po numerze pominiesz lub zdublujesz dane. ID od tego nie zależy.

Ile równoległych workerów ustawić? Zacznij od niewielkiej liczby i zwiększaj, obserwując stabilność i limity źródła. Nadmierna równoległość szkodzi bardziej niż pomaga.

Jak bezpiecznie przechowywać ciąg połączenia do proxy? W zmiennej środowiskowej, a nie w kodzie. Czytaj ją przez os.environ. Tak hasło nie trafi do systemu kontroli wersji.

Co robić, jeśli kursor przeterminował się przez długą pauzę? Wyłap błąd nieprawidłowego kursora i przełącz się na wznawianie po zapisanym identyfikatorze ostatniego rekordu.

Czy trzeba sprawdzać integralność ściągniętego pliku? Tak. Porównaj rzeczywisty rozmiar pliku z nagłówkiem Content-Length. Jeśli serwer wydaje sumę kontrolną, sprawdź też ją.

Zakończenie: co teraz umiesz i dokąd się rozwijać

Przeszedłeś drogę od kruchego pobierania, które wali się przy pierwszym zerwaniu, do stabilnego downloadera. Teraz masz wszystkie narzędzia, żeby zerwanie przestało być katastrofą, a stało się zwykłą sytuacją roboczą.

Co opanowałeś. Wznawianie plików przez nagłówek Range ze sprawdzaniem Accept-Ranges. Zapisywanie postępu przez atomowe checkpointy. Idempotentność na poziomie klucza deduplikacji. Deduplikację na milionach wierszy bez rozdymania pamięci. Równoległość z powtarzaniem zadań, które padły. Wznawianie po długiej pauzie z odświeżaniem tokenów i zastępowaniem przeterminowanych kursorów. I najważniejsze, gotową ramę w Pythonie, która łączy to wszystko.

Co robić dalej. Weź swoje realne źródło danych i dostosuj pod nie funkcję pobierania strony. Zacznij od małego wolumenu, zdebuguj wznawianie na przerwaniach, a potem skaluj. Skonfiguruj stabilne proxy Proxeon, żeby połączenia były przewidywalne przez całe pobieranie.

Dokąd się rozwijać. Zgłęb zapis wsadowy i filtry Blooma. Dodaj monitoring postępu i szacowanie czasu. Rozdziel przechowywanie danych surowych i przetworzonych. Stopniowo

O autorze

Roman Melnikov

Roman Melnikov

Technical Writer and System Administrator

Doświadczenie zawodowe: Technical writer and DevOps engineer with 9 years of experience. Created over 50 detailed guides on system configuration and administration. His instructions helped thousands of professionals successfully solve technical tasks. Popular author on Habr and YouTube.
Wykształcenie: Bauman Moscow State Technical University. Information Systems and Technologies
Ekspertyza:
Technical Documentation DevOps System Administration Linux Docker and Kubernetes CI/CD Infrastructure Automation Cloud Technologies System Monitoring Bash and Python Scripting

Podziel się artykułem: