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("Unexpected API response: expected products[] and integer total")
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)
# Advance by the number returned, not the requested limit.
skip += len(batch)
if skip >= total:
break
time.sleep(0.1)
else:
raise RuntimeError(f"在{MAX_PAGES}页的安全限制处停止")
if total is not None and len(products) < total:
print(f"警告：收到{len(products)}个唯一产品；API报告{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"已将{len(products)}个唯一行保存到products.csv")
if __name__ == "__main__":
main()
from urllib.parse import urljoin
from bs4 import BeautifulSoup
import requests
def get_next_url(page_url: str) -> str | None:
response = requests.get(page_url, timeout=(5, 20))
response.raise_for_status()
soup = BeautifulSoup(response.text, "html.parser")
link = soup.select_one("a.next")
if link is None:
return None
href = link.get("href")
return urljoin(response.url, href) if href else None
