cormort/procurement-crawler-py

1"""舊做法 vs 新做法的實測對照（會打真實 API，請禮貌執行）。2 3跑法：4    python3 bench_api.py            # 預設 10 筆5    python3 bench_api.py 206 7量什麼：8  A. 舊做法：搜尋 1 次 + 逐筆 `/api/tender`（原程式的流程，沒有客戶端限速）→ 請求數、429 次數、耗時9  B. 新做法：搜尋 1 次（帶 columns[]）→ 請求數、耗時、拿到幾個詳細欄位10  C. 完整欄位：新做法下逐筆取 86 欄（有客戶端限速）→ 驗證欄位數與 Excel 產出11 12注意：pcc-api 的速率限制是「10 秒 >10 次或 60 秒 >60 次就 429」，所以 A 很容易被擋13——那正是原程式在真實使用時會遇到的狀況（429 之後使用者只看到失敗訊息）。14"""15 16import sys17import time18 19import requests20 21from pcc_core import LIST_COLUMNS, PccApi, RateLimiter, flatten_detail22 23COUNT = int(sys.argv[1]) if len(sys.argv) > 1 else 1024QUERY = "農田水利署"25 26 27def pick_targets(count: int) -> list[dict]:28    payload = requests.get(f"https://pcc-api.openfun.app/api/searchbytitle",29                           params={"query": QUERY, "page": 1}, timeout=25).json()30    return (payload.get("records") or [])[:count]31 32 33def old_way(targets: list[dict]) -> dict:34    """原程式流程：搜尋一次（不帶 columns），再逐筆打 /api/tender，無客戶端限速。"""35    started = time.time()36    requests_made, blocked, ok = 0, 0, 037    for item in targets:38        url = f"https://pcc-api.openfun.app/api/tender?unit_id={item['unit_id']}&job_number={item['job_number']}"39        response = requests.get(url, timeout=20)40        requests_made += 141        if response.status_code == 429:42            blocked += 143            continue44        detail = (response.json().get("records") or [{}])[0].get("detail", {})45        if detail:46            ok += 147        time.sleep(0.05)      # 原程式沒有延遲；這裡只留一點避免瞬間灌爆48    return {"requests": requests_made, "blocked": blocked, "ok": ok, "seconds": time.time() - started}49 50 51def new_way(targets: list[dict]) -> dict:52    """新做法：一次搜尋請求就把整頁加上指定的詳細欄位帶回來。"""53    api = PccApi()            # 內建自我限速54    started = time.time()55    payload = api.search_by_title(QUERY, 1, LIST_COLUMNS)56    seconds = time.time() - started57    records = payload.get("records") or []58    with_detail = sum(1 for r in records if r.get("detail"))59    sample = next((r for r in records if r.get("detail")), {}).get("detail", {})60    return {61        "requests": api.requests_made,62        "records": len(records),63        "with_detail": with_detail,64        "columns": len(sample),65        "seconds": seconds,66        "sample": sample,67    }68 69 70def full_detail(targets: list[dict]) -> dict:71    """新做法下的「完整欄位」：逐筆取 86 欄（有客戶端限速，因此不會被 429）。"""72    limiter = RateLimiter()73    api = PccApi(limiter=limiter)74    started = time.time()75    rows, blocked = [], 076    for item in targets[:3]:      # 只抽 3 筆示範77        try:78            payload = api.tender(item["unit_id"], item["job_number"])79        except Exception as exc:      # noqa: BLE00180            blocked += 181            print("   失敗:", exc)82            continue83        records = payload.get("records") or []84        if records:85            rows.append(flatten_detail(records[0].get("detail", {})))86    return {87        "requests": api.requests_made,88        "rows": len(rows),89        "fields": len(rows[0]) if rows else 0,90        "seconds": time.time() - started,91        "waited": limiter.waited,92    }93 94 95def main() -> None:96    print(f"=== 取樣：{QUERY} 前 {COUNT} 筆 ===")97    targets = pick_targets(COUNT)98    print(f"取得 {len(targets)} 筆標的（此請求本身不算在比較內）\n")99 100    old = old_way(targets)101    print("【舊做法】搜尋 + 逐筆 /api/tender（無客戶端限速）")102    print(f"  請求 {old['requests']} 次｜成功 {old['ok']} 筆｜被 429 擋 {old['blocked']} 次｜耗時 {old['seconds']:.1f}s")103    print(f"  每次請求平均 {old['seconds'] / max(1, old['requests']):.2f}s\n")104 105    # 讓 10 秒窗口清空，避免把前一段的 429 狀態帶進下一段106    print("（等待 15 秒讓 API 的限速窗口清空…）")107    time.sleep(15)108 109    new = new_way(targets)110    print("【新做法】一次搜尋請求（帶 columns[]）")111    print(f"  請求 {new['requests']} 次｜回傳 {new['records']} 筆（其中 {new['with_detail']} 筆含詳細欄位）"112          f"｜耗時 {new['seconds']:.1f}s")113    print(f"  單筆帶回欄位數：{new['columns']}")114    if new["sample"]:115        for key, value in list(new["sample"].items())[:6]:116            print(f"    {key} = {value}")117    print()118 119    print("（等待 15 秒…）")120    time.sleep(15)121    full = full_detail(targets)122    print("【新做法】完整欄位（86 欄，逐筆、有限速）")123    print(f"  請求 {full['requests']} 次｜{full['rows']} 列 × {full['fields']} 欄｜耗時 {full['seconds']:.1f}s"124          f"（限速等待 {full['waited']:.1f}s）")125    print()126    print("=== 結論 ===")127    saved = old["requests"] - new["requests"]128    print(f"  同樣取得 {len(targets)} 筆的金額／截止日：舊做法 {old['requests']} 次請求（其中 {old['blocked']} 次被擋），"129          f"新做法 {new['requests']} 次 → 省下 {saved} 次（{(1 - new['requests'] / max(1, old['requests'])) * 100:.0f}%）")130 131 132if __name__ == "__main__":133    main()134