TL;DR

Phân trang là một hợp đồng giữa khách hàng của bạn và nguồn dữ liệu. Kiểm tra phản hồi để tìm số trang, offset, con trỏ hoặc liên kết tiếp theo trước khi viết vòng lặp.
Đối với API giới hạn/offset, hãy tiến bộ theo số bản ghi thực sự được trả về và dừng lại ở tổng số đã báo cáo hoặc một trang trống. Thêm một bảo vệ trang tối đa và loại bỏ bản sao theo ID ổn định.
API con trỏ là mờ đục theo thiết kế. Gửi con trỏ tiếp theo chính xác được trả về bởi máy chủ; không chế tạo số trang từ nó.
Hướng dẫn này sử dụng DummyJSON làm API demo. Nó hữu ích để học các cơ chế phân trang, không phải như một tập dữ liệu kinh doanh sản xuất hoặc hợp đồng đảm bảo tính khả dụng.

Ý Nghĩa Phân Trang Python Đối Với Thu Thập Dữ Liệu

Phân trang Python là logic yêu cầu một tập kết quả lớn trong các trang nhỏ hơn. Các API và trang web sử dụng phân trang vì việc trả về mỗi bản ghi trong một phản hồi có thể chậm, tốn bộ nhớ hoặc vượt quá giới hạn của máy chủ.

Một scraper chỉ lấy phản hồi đầu tiên có thể trông khỏe mạnh trong khi im lặng bỏ lỡ hầu hết tập dữ liệu. Vòng lặp chính xác phụ thuộc vào kiểu phân trang của nguồn dữ liệu: số trang, offset và giới hạn, token con trỏ, URL trang tiếp theo, hoặc nội dung được tải khi người dùng cuộn. Luôn đọc tài liệu API của nhà cung cấp và kiểm tra một phản hồi mẫu trước khi chọn một vòng lặp. Đối với phát hiện URL ở cấp độ trang web và trích xuất trang, Nstdata Crawl tách biệt với phân trang con trỏ hoặc offset cụ thể cho ứng dụng.

Các Mẫu Phân Trang Phổ Biến

Mẫu	Ví dụ yêu cầu	Làm thế nào để biết khi nào dừng	Thất bại thường gặp
Số trang	`?page=3`	Trang trống, không có trường trang tiếp theo, hoặc tổng số đã biết	Giả định mọi điểm cuối bắt đầu từ trang 1

import csv
import time
from typing import Any

import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry


API_URL = "https://dummyjson.com/products"
PAGE_SIZE = 30
MAX_PAGES = 100


def make_session() -> requests.Session:
    retry = Retry(
        total=3,
        connect=3,
        read=2,
        status=3,
        backoff_factor=0.5,
        status_forcelist=(429, 500, 502, 503, 504),
        allowed_methods=frozenset({"GET"}),
        respect_retry_after_header=True,
    )
    session = requests.Session()
    session.mount("https://", HTTPAdapter(max_retries=retry))
    session.headers.update({"User-Agent": "pagination-demo/1.0"})
    return session


def fetch_all_products() -> list[dict[str, Any]]:
    session = make_session()
    products: list[dict[str, Any]] = []
    seen_ids: set[int] = set()
    skip = 0
    total: int | None = None

    for page_number in range(1, MAX_PAGES + 1):
        response = session.get(
            API_URL,
            params={"limit": PAGE_SIZE, "skip": skip},
            timeout=(5, 20),
        )
        response.raise_for_status()

        payload = response.json()
        batch = payload.get("products")
        page_total = payload.get("total")

        if not isinstance(batch, list) or not isinstance(page_total, int):
            raise ValueError("Phản hồi API không như mong đợi: mong đợi products[] và tổng số nguyên")

        total = page_total
        if not batch:
            break

        for item in batch:
            item_id = item.get("id")
            if not isinstance(item_id, int):
                continue
            if item_id not in seen_ids:
                seen_ids.add(item_id)
                products.append(item)

        # Tiến bộ theo số bản ghi đã trả về, không phải theo giới hạn yêu cầu.
        skip += len(batch)
        if skip >= total:
            break

        time.sleep(0.1)

    else:
raise RuntimeError(f"Ngừng lại ở giới hạn an toàn của {MAX_PAGES} trang")

    if total is not None and len(products) < total:
        print(f"Cảnh báo: nhận được {len(products)} sản phẩm độc nhất; API đã báo cáo {total}")

    return products


def main() -> None:
    products = fetch_all_products()
    fields = ["id", "title", "category", "price"]

    with open("products.csv", "w", newline="", encoding="utf-8") as output:
        writer = csv.DictWriter(output, fieldnames=fields, extrasaction="ignore")
        writer.writeheader()
        writer.writerows(products)

    print(f"Đã lưu {len(products)} hàng độc nhất vào products.csv")


if __name__ == "__main__":
    main()

Python Phân Trang: Cách Thu Thập Dữ Liệu Qua Các Trang Phân Trang

TL;DR

Ý Nghĩa Phân Trang Python Đối Với Thu Thập Dữ Liệu

Các Mẫu Phân Trang Phổ Biến

Bai viet pho bien

Dun & Bradstreet Scraper và D&B API: Hướng dẫn Dữ liệu Kinh doanh

Python Phân Trang: Cách Thu Thập Dữ Liệu Qua Các Trang Phân Trang

Xây Dựng Kịch Bản Phân Trang Giới Hạn/Offset Trong Python

Tại sao Vòng lặp Tiến lên theo Số lượng Được Trả về

Cách Xử Lý API Số Trang

Cách Xử Lý Phân Trang Dựa Trên Con Trỏ

Cách Theo Dõi Liên Kết Phân Trang HTML

Cuộn Vô Hạn và Các Trang Được Phân Trang Bằng JavaScript

Ngăn Chặn Thiếu Hồ Sơ và Vòng Lặp Phân Trang

Khi Nào Nên Sử Dụng Một Khung Thu Thập Thay Vì Phân Trang Thủ Công

Kết Luận

Giữ Cho Thu Thập Trang Tách Biệt Với Phân Trang API

Câu Hỏi Thường Gặp