| |
| """Weekly refresh of the Etimad tender-titles dataset. |
| |
| python scripts/weekly_update.py [--dry-run] [--max-pages 40] [--no-upload] |
| |
| 1. known refs <- data/tenders_all.csv |
| 2. crawl the unfiltered newest-first visitor listing exactly like scripts/crawl.py (same endpoint, |
| headers, >= 8.6 s between requests, sequential) until two consecutive pages add no new ref, or --max-pages |
| 3. normalise like build_dataset.py / make_splits.py (parent activity, spelling fold, frozen v1 label set) |
| 4. append to data/tenders_all.csv, write data/published/updates/<date>.jsonl + .csv (train/test never touched) |
| 5. refresh the "## Updates" table in DATASET-CARD.md |
| 6. upload updates/ + card (as README.md) to the dataset repo and tag v<YYYY.MM.DD> (skipped with --dry-run) |
| Idempotent: refs already in tenders_all.csv are never added twice.""" |
| import argparse, csv, json, os, re, sys, time, datetime, logging |
| from collections import Counter |
| HERE = os.path.dirname(os.path.abspath(__file__)); ROOT = os.path.abspath(os.path.join(HERE, "..")) |
| sys.path.insert(0, HERE) |
| from common import norm_text, sector_of, kind_of, okey |
|
|
| REPO = "chaliasif/etimad-tender-titles" |
| BASE = "https://tenders.etimad.sa/Tender/AllSupplierTendersForVisitorAsync" |
| H = {"X-Requested-With": "XMLHttpRequest", "Referer": "https://tenders.etimad.sa/Tender/AllTendersForVisitor"} |
| MIN_GAP = max(8.6, float(os.environ.get("MIN_GAP", "8.6"))) |
| ALL_FIELDS = ["ref", "title", "agency", "branch", "activity", "ttype", "publish_date", "source"] |
| PUB_FIELDS = ["ref", "title", "agency", "branch", "activity", "activity_detail", "activity_label", "ttype", "sector", "kind", "publish_date", "source"] |
| ALL_CSV = os.path.join(ROOT, "data", "tenders_all.csv") |
| UPD_DIR = os.path.join(ROOT, "data", "published", "updates") |
| CARD = os.path.join(ROOT, "DATASET-CARD.md") |
| log = logging.getLogger("weekly") |
|
|
| def setup_log(): |
| os.makedirs(os.path.join(ROOT, "logs"), exist_ok=True) |
| fmt = logging.Formatter("%(asctime)s %(message)s") |
| log.setLevel(logging.INFO) |
| for h in (logging.FileHandler(os.path.join(ROOT, "logs", "weekly_update.log"), encoding="utf-8"), logging.StreamHandler()): |
| h.setFormatter(fmt); log.addHandler(h) |
|
|
| def frozen_labels(): |
| """v1 class list + surface forms, read from the published splits (never recomputed from counts).""" |
| import pandas as pd |
| df = pd.concat([pd.read_csv(os.path.join(ROOT, "data", "published", f"{n}.csv"), dtype=str).fillna("") for n in ("train", "test")]) |
| classes = set(df.activity_label) - {"other"} |
| surf = {} |
| for a, _ in Counter(df.activity).most_common(): surf.setdefault(okey(a), a) |
| for a in classes: surf.setdefault(okey(a), a) |
| return classes, surf |
|
|
| def normalise(t, classes, surf, src): |
| g = lambda k: norm_text(t.get(k)) |
| full = g("tenderActivityName") |
| parent = re.split(r"\s*-\s*", full, maxsplit=1)[0].strip() |
| parent = surf.get(okey(parent), parent) |
| row = {"ref": g("referenceNumber"), "title": g("tenderName"), "agency": g("agencyName"), "branch": g("branchName"), |
| "activity": parent, "activity_detail": full, "activity_label": parent if parent in classes else "other", |
| "ttype": g("tenderTypeName"), "sector": sector_of(parent), "publish_date": (t.get("submitionDate") or "")[:10], "source": src} |
| row["kind"] = kind_of(parent, row["ttype"], row["title"]) |
| return row |
|
|
| def fetch(pg, state): |
| from curl_cffi import requests |
| qs = ("MultipleSearch=&TenderCategory=&ReferenceNumber=&TenderNumber=&AgencyCode=&PublishDateId=&TenderTypeId=" |
| "&TenderActivityId=&SortDirection=DESC&Sort=SubmitionDate&PageSize=24&PageNumber=%d&IsSearch=false" % pg) |
| for attempt in range(3): |
| wait = MIN_GAP - (time.time() - state["last"]) |
| if wait > 0: time.sleep(wait) |
| state["last"] = time.time() |
| try: |
| j = requests.get(BASE + "?" + qs, impersonate="safari", timeout=45, headers=H).json() |
| return j.get("data") or [] |
| except Exception as e: |
| log.info(f"page {pg} attempt {attempt+1} failed: {str(e)[:80]}"); time.sleep(15 * (attempt + 1)) |
| raise RuntimeError(f"BLOCKED at page {pg}: 3 failures, stopping without retrying around it") |
|
|
| def update_card(date, new_total, total, tag): |
| txt = open(CARD, encoding="utf-8").read() |
| hdr = "| Date | New rows | Total rows | Tag |\n|---|---|---|---|\n" |
| m = re.search(r"## Updates\n(.*?)(?=\n## )", txt, re.S) |
| rows = re.findall(r"^\| (\d{4}-\d\d-\d\d) \| ([\d,]+) \| ([\d,]+) \| (\S+) \|$", m.group(1), re.M) if m else [] |
| if not rows: rows = [("2026-10-13", "0", f"{total - new_total:,}", "v1.0")] |
| rows = [r for r in rows if r[0] != date] + [(date, f"{new_total:,}", f"{total:,}", tag)] |
| block = ("## Updates\nWeekly refresh adds newly listed tenders under `updates/` (one file per date); `train` and `test` never change. " |
| "Each tag freezes the repository at that date.\n\n" + hdr + "".join(f"| {a} | {b} | {c} | {d} |\n" for a, b, c, d in rows) + "\n") |
| if m: txt = txt[:m.start()] + block.rstrip("\n") + "\n" + txt[m.end():] |
| else: txt = txt.replace("## ملخص بالعربية", block + "## ملخص بالعربية", 1) |
| open(CARD, "w", encoding="utf-8").write(txt) |
|
|
| def main(): |
| ap = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter) |
| ap.add_argument("--dry-run", action="store_true"); ap.add_argument("--max-pages", type=int, default=40) |
| ap.add_argument("--no-upload", action="store_true") |
| a = ap.parse_args(); setup_log() |
| today = datetime.date.today().isoformat(); src = f"weekly-{today}" |
| log.info(f"start {today} dry_run={a.dry_run} max_pages={a.max_pages}") |
| with open(ALL_CSV, encoding="utf-8", newline="") as f: known = {r["ref"] for r in csv.DictReader(f)} |
| classes, surf = frozen_labels() |
| log.info(f"known refs {len(known)}, frozen classes {len(classes)} (+other)") |
| new, seen, state, empty_streak, pages = [], set(), {"last": 0.0}, 0, 0 |
| try: |
| for pg in range(1, a.max_pages + 1): |
| data = fetch(pg, state); pages += 1 |
| if not data: log.info(f"page {pg}: empty, end of listing"); break |
| added = 0 |
| for t in data: |
| r = normalise(t, classes, surf, src) |
| if not (r["ref"] and r["title"] and r["activity"] and r["ttype"]) or r["ref"] in known or r["ref"] in seen: continue |
| seen.add(r["ref"]); new.append(r); added += 1 |
| log.info(f"page {pg}: {len(data)} rows, {added} new (total new {len(new)})") |
| empty_streak = 0 if added else empty_streak + 1 |
| if empty_streak >= 2: log.info("two consecutive pages with no new refs, stopping"); break |
| except RuntimeError as e: |
| log.info(str(e)); return 2 |
| log.info(f"crawled {pages} pages, {len(new)} new rows") |
| if a.dry_run: |
| log.info("dry run: nothing written"); [log.info(" " + json.dumps(r, ensure_ascii=False)) for r in new[:3]]; return 0 |
| if new: |
| with open(ALL_CSV, "a", encoding="utf-8", newline="") as f: |
| w = csv.DictWriter(f, ALL_FIELDS, extrasaction="ignore"); w.writerows(new) |
| os.makedirs(UPD_DIR, exist_ok=True) |
| jp, cp = os.path.join(UPD_DIR, today + ".jsonl"), os.path.join(UPD_DIR, today + ".csv") |
| with open(jp, "a", encoding="utf-8") as f: |
| for r in new: f.write(json.dumps({k: r[k] for k in PUB_FIELDS}, ensure_ascii=False) + "\n") |
| rows = [json.loads(l) for l in open(jp, encoding="utf-8")] |
| with open(cp, "w", encoding="utf-8", newline="") as f: |
| w = csv.DictWriter(f, PUB_FIELDS); w.writeheader(); w.writerows(rows) |
| jp = os.path.join(UPD_DIR, today + ".jsonl") |
| day_new = sum(1 for _ in open(jp, encoding="utf-8")) if os.path.exists(jp) else 0 |
| with open(ALL_CSV, encoding="utf-8", newline="") as f: total = sum(1 for _ in csv.DictReader(f)) |
| tag = "v" + today.replace("-", ".") |
| update_card(today, day_new, total, tag) |
| log.info(f"day new rows {day_new}, total {total}, card updated, tag {tag}") |
| if a.no_upload: return 0 |
| from huggingface_hub import HfApi |
| api = HfApi() |
| api.upload_file(path_or_fileobj=CARD, path_in_repo="README.md", repo_id=REPO, repo_type="dataset", commit_message=f"Card: update table {today}") |
| if os.path.isdir(UPD_DIR) and os.listdir(UPD_DIR): |
| api.upload_folder(repo_id=REPO, repo_type="dataset", folder_path=UPD_DIR, path_in_repo="updates", commit_message=f"Weekly update {today}: {day_new} new rows") |
| api.create_tag(REPO, tag=tag, repo_type="dataset", tag_message=f"Weekly refresh {today}: +{day_new}, total {total}", exist_ok=True) |
| log.info(f"uploaded and tagged {tag}") |
| return 0 |
| if __name__ == "__main__": sys.exit(main()) |
|
|
