Scrape เว็บ 7,000 หน้าใน 3 นาทีด้วย Python และ Olostep

%pip -q install olostep==1.2.2 requests tqdm

import asyncio, csv, gzip, hashlib, json, os, random, time
from datetime import datetime, timezone
from getpass import getpass
from pathlib import Path
from xml.etree import ElementTree as ET
 
import requests
from olostep import AsyncOlostep
from olostep.errors import OlostepClientError_ResponseValidationFailed
from olostep.models.response import BatchInfoResponse
from tqdm import tqdm
 
N = 10000
MAX_WAIT_SECONDS = 2400
DATA = Path("data"); DATA.mkdir(exist_ok=True)
client = AsyncOlostep(api_key=os.getenv("OLOSTEP_API_KEY") or getpass("Olostep API key: "))
 
def write_json(path, value):
    path.write_text(json.dumps(value, indent=2), encoding="utf-8")
 
def valid_file(path):
    try:
        return path.is_file() and bool(path.read_text(encoding="utf-8").strip())
    except (OSError, UnicodeError):
        return False
 
async def batch_info(batch_id):
    try:
        return await client.batches.info(batch_id)
    except OlostepClientError_ResponseValidationFailed as error:
        return BatchInfoResponse.model_validate({"parser": "none", **json.loads(error.response.body)})

source = "https://developer.mozilla.org/sitemaps/en-us/sitemap.xml.gz"
 
if not (DATA / "urls.csv").exists():
    response = requests.get(source, timeout=60); response.raise_for_status()
    payload = response.content
    xml = gzip.decompress(payload) if payload.startswith(b"\x1f\x8b") else payload
    (DATA / "mdn_sitemap.xml").write_bytes(xml)
    ns = {"s": "http://www.sitemaps.org/schemas/sitemap/0.9"}
    nodes = ET.fromstring(xml).findall("s:url", ns)
    urls = sorted({node.findtext("s:loc", namespaces=ns) for node in nodes})
    urls = [url for url in urls if url and url.startswith("https://developer.mozilla.org/en-US/docs/")]
    random.Random(42).shuffle(urls)
    assert len(urls) >= N, f"Only {len(urls)} unique documentation URLs available."
    with (DATA / "urls.csv").open("w", newline="", encoding="utf-8") as f:
        writer = csv.writer(f); writer.writerow(["url"]); writer.writerows([url] for url in urls[:N])
    write_json(DATA / "dataset.json", {"source": source, "downloaded_utc": datetime.now(timezone.utc).isoformat(),
                                     "available_urls": len(urls), "selected_urls": N, "random_seed": 42})
 
with (DATA / "urls.csv").open(encoding="utf-8", newline="") as f:
    urls = [row["url"] for row in csv.DictReader(f)]
 
assert len(urls) == len(set(urls)) == N
fingerprint = hashlib.sha256("\n".join(urls).encode()).hexdigest()
expected = {f"{i:05d}": url for i, url in enumerate(urls)}
print(f"Ready: {len(urls):,} unique URLs; source: {source}")

Ready: 10,000 unique URLs; source: https://developer.mozilla.org/sitemaps/en-us/sitemap.xml.gz

checkpoint = Path("batch.json")
resumed = checkpoint.exists()
 
if resumed:
    run = json.loads(checkpoint.read_text(encoding="utf-8"))
    assert run["urls_sha256"] == fingerprint, "URL list changed; preserve the checkpoint before starting another batch."
else:
    started = time.time()
    batch = await client.batches.create(urls=[{"url": url, "custom_id": key} for key, url in expected.items()])
    run = {"batch_id": batch.id, "requested": N, "urls_sha256": fingerprint, "started_wall": started}
    write_json(checkpoint, run)
 
OUT = Path("output") / run["batch_id"]; OUT.mkdir(parents=True, exist_ok=True)
print("Resuming" if resumed else "Submitted", run["batch_id"])
wait_started = time.monotonic()
 
with tqdm(total=N, desc="Scraping", unit="page") as bar:
    while True:
        info = await batch_info(run["batch_id"])
        bar.update(max(0, info.completed_urls - bar.n))
        write_json(OUT / "progress.json", {"phase": "scraping", "status": info.status,
                   "processed": info.completed_urls, "total": N, "elapsed_seconds": round(time.time() - run["started_wall"], 2)})
        if info.status in ("completed", "failed"):
            break
        if time.monotonic() - wait_started > MAX_WAIT_SECONDS:
            raise TimeoutError("Batch still running. Rerun this cell to continue the same batch.")
        await asyncio.sleep(10)
 
run.setdefault("scrape_seconds", time.time() - run["started_wall"])
write_json(checkpoint, run)
print(f"Scraping stage finished: {info.status}; {info.completed_urls:,}/{N:,} completed.")

download_started = time.perf_counter()
limit = asyncio.Semaphore(96)
errors, scheduled, tasks = {}, set(), []
saved = set()
bar = tqdm(total=N, desc="Markdown saved", unit="file")
 
async def save_markdown(item):
    key = item.custom_id
    if key not in expected:
        return
    async with limit:
        path = OUT / f"{key}.md"
        if valid_file(path):
            saved.add(key); bar.update(1)
            return
        for attempt in range(2):
            try:
                async with asyncio.timeout(40):
                    content = (await item.retrieve(["markdown"])).markdown_content
                if not content or not content.strip():
                    errors[key] = "No Markdown content returned"
                    return
                path.write_text(content, encoding="utf-8")
                saved.add(key); bar.update(1)
                if len(saved) % 50 == 0:
                    write_json(OUT / "progress.json", {"phase": "downloading", "saved": len(saved), "total": N,
                               "elapsed_seconds": round(time.time() - run["started_wall"], 2)})
                return
            except Exception as error:
                errors[key] = str(error)
                if attempt == 0:
                    await asyncio.sleep(1)
 
async for item in client.batches.items(run["batch_id"], status="completed", batch_size=50, wait_for_completion=False):
    if item.retrieve_id and item.custom_id not in scheduled:
        scheduled.add(item.custom_id)
        tasks.append(asyncio.create_task(save_markdown(item)))
 
await asyncio.gather(*tasks)
bar.close()
download_seconds = time.perf_counter() - download_started

missing = {key for key in expected if not valid_file(OUT / f"{key}.md")}
 
with (OUT / "manifest.csv").open("w", newline="", encoding="utf-8") as f:
    writer = csv.writer(f); writer.writerow(["file", "url", "saved"])
    writer.writerows((f"{key}.md", url, key not in missing) for key, url in expected.items())
 
with (OUT / "failures.csv").open("w", newline="", encoding="utf-8") as f:
    writer = csv.writer(f); writer.writerow(["url", "error"])
    writer.writerows((expected[key], errors.get(key, "Missing result or failed scrape")) for key in sorted(missing))
 
total_seconds = time.time() - run["started_wall"]
summary = {"batch_id": run["batch_id"], "requested": N, "batch_status": info.status,
           "api_completed_urls": info.completed_urls, "saved_markdown": N - len(missing),
           "missing_markdown": len(missing), "all_saved": not missing, "resumed": resumed,
           "scrape_seconds": round(run["scrape_seconds"], 2), "download_seconds": round(download_seconds, 2),
           "total_seconds": round(total_seconds, 2),
           "all_saved_under_5_minutes": not resumed and not missing and total_seconds < 300}
 
write_json(OUT / "summary.json", summary)
write_json(OUT / "progress.json", {"phase": "finished", **summary})
print(json.dumps(summary, indent=2))
print("Files:", OUT.resolve())
print("All 10,000 Markdown files saved." if not missing else f"{len(missing):,} missing; see failures.csv.")
await client.close()