Mockup

.gitignore

@@ -7,13 +7,38 @@ data/cdx-index/*.tmp
 
 # Raw scraped HTML pages
 data/pilot/html/
+data/pilot/html-recent/
+data/pilot/html-recent-test/
+
+# Transient pipeline stdout logs
+data/pilot/*.out
+runs/**/*.out
+
+# Derived title-flag panel (33 MB) — regenerated by code/57-ai-diffusion-titles.py
+data/pilot/ai-title-flags.csv
 
 # Large pilot manifests (small derived CSVs/summaries are kept)
 data/pilot/*-manifest*.tsv
 
+# Scratch / working artifacts (not deliverables)
+scratchpad/
+
 # === Python / OS cruft ===
 __pycache__/
 *.pyc
 *.pyo
 .DS_Store
 .ipynb_checkpoints/
+data/pilot/html-balanced/
+data/pilot/html-expanded/
+
+# === Enlarged-collection outputs (2026-08) ===
+# Regenerable from the manifests + code/38, code/41 and the pipeline drivers.
+# 85 MB / 25 MB price files and their logs would exceed GitHub's limits on a
+# public repo; the small pilot-scale equivalents above stay tracked.
+data/pilot/balanced-prices.csv
+data/pilot/expanded-prices.csv
+data/pilot/gig-status-ledger.tsv
+data/pilot/expanded-gig-status.tsv
+data/pilot/*-download-log.tsv
+data/pilot/*-download-checkpoint.txt

code/08-download-html.py

@@ -16,6 +16,7 @@
 import argparse
 import asyncio
 import aiohttp
+import gzip
 import time
 import csv
 import os
@@ -41,6 +42,15 @@ def parse_args():
     p.add_argument("--max-rate", type=float, default=12.0,
                    help="Max requests per second")
     p.add_argument("--max-retries", type=int, default=3)
+    p.add_argument("--gzip", action="store_true",
+                   help="Store pages gzipped as <name>.html.gz. Measured 5.0x on "
+                        "this corpus (22 GB -> ~4.4 GB per 15k pages); "
+                        "09-extract-prices.py reads both forms transparently.")
+    p.add_argument("--chunk", type=int, default=5000,
+                   help="Schedule this many downloads at a time instead of "
+                        "creating one coroutine per manifest row up front.")
+    p.add_argument("--limit", type=int, default=0,
+                   help="Stop after N new downloads (0 = no limit). For pilots.")
     return p.parse_args()
 
 
@@ -56,12 +66,26 @@ def extract_seller_slug(url):
     return "unknown", "unknown"
 
 
-def build_output_path(html_dir, seller, slug, timestamp):
-    """Build output path: html/<seller>/<YYYYMMDD>_<slug>.html"""
+def build_output_path(html_dir, seller, slug, timestamp, use_gzip=False):
+    """Build output path: html/<seller>/<YYYYMMDD>_<slug>.html[.gz]
+
+    Does NOT create the directory — with ~90k gigs this is called once per
+    manifest row during the existence check, and mkdir on every row is a
+    measurable cost. The writer creates the parent when it actually writes.
+    """
     date = timestamp[:8]
-    out_dir = html_dir / seller
-    out_dir.mkdir(parents=True, exist_ok=True)
-    return out_dir / f"{date}_{slug}.html"
+    return html_dir / seller / f"{date}_{slug}.html{'.gz' if use_gzip else ''}"
+
+
+def existing_path(html_dir, seller, slug, timestamp):
+    """Return an already-downloaded page for this snapshot in either storage
+    form, or None. Lets a gzipped run reuse the plain files from earlier runs
+    instead of re-fetching 15,150 pages that are already on disk."""
+    for gz in (False, True):
+        p = build_output_path(html_dir, seller, slug, timestamp, gz)
+        if p.exists() and p.stat().st_size > 0:
+            return p
+    return None
 
 
 def load_manifest(manifest_path):
@@ -109,16 +133,18 @@ async def acquire(self):
 
 
 async def download_one(session, timestamp, url, html_dir, rate_limiter,
-                       max_retries, log_file, checkpoint_file, stats):
+                       max_retries, log_file, checkpoint_file, stats,
+                       use_gzip=False):
     """Download a single Wayback snapshot."""
     seller, slug = extract_seller_slug(url)
-    out_path = build_output_path(html_dir, seller, slug, timestamp)
 
-    # Skip if already exists on disk
-    if out_path.exists() and out_path.stat().st_size > 0:
+    # Skip if already on disk in either storage form
+    if existing_path(html_dir, seller, slug, timestamp) is not None:
         stats["skipped"] += 1
         return
 
+    out_path = build_output_path(html_dir, seller, slug, timestamp, use_gzip)
+
     wayback_url = WAYBACK_TPL.format(timestamp=timestamp, url=url)
 
     for attempt in range(max_retries):
@@ -128,8 +154,12 @@ async def download_one(session, timestamp, url, html_dir, rate_limiter,
                 if resp.status == 200:
                     content = await resp.read()
                     out_path.parent.mkdir(parents=True, exist_ok=True)
-                    with open(out_path, "wb") as f:
-                        f.write(content)
+                    if use_gzip:
+                        with gzip.open(out_path, "wb", compresslevel=6) as f:
+                            f.write(content)
+                    else:
+                        with open(out_path, "wb") as f:
+                            f.write(content)
 
                     stats["ok"] += 1
                     size = len(content)
@@ -171,10 +201,15 @@ async def main_async(args):
 
     # Filter out already done
     todo = [(ts, url) for ts, url in entries if f"{ts}\t{url}" not in done]
+    n_pending = len(todo)
+    if args.limit:
+        todo = todo[:args.limit]
 
     print(f"Manifest:    {len(entries):,} snapshots")
-    print(f"Already done:{len(entries) - len(todo):,}")
-    print(f"To download: {len(todo):,}")
+    print(f"Already done:{len(entries) - n_pending:,}")
+    print(f"To download: {len(todo):,}" +
+          (f" (--limit of {n_pending:,} pending)" if args.limit else ""))
+    print(f"Storage:     {'gzip' if args.gzip else 'plain'}")
     print(f"Concurrency: {args.concurrency}")
     print(f"Rate limit:  {args.max_rate} req/s")
     print()
@@ -205,7 +240,8 @@ async def bounded_download(ts, url):
                 async with sem:
                     await download_one(session, ts, url, args.html_dir,
                                        rate_limiter, args.max_retries,
-                                       log_file, checkpoint_file, stats)
+                                       log_file, checkpoint_file, stats,
+                                       use_gzip=args.gzip)
 
                 # Progress update
                 total_done = stats["ok"] + stats["failed"] + stats["skipped"]
@@ -218,8 +254,12 @@ async def bounded_download(ts, url):
                           f"skip={stats['skipped']:,} "
                           f"| {rate:.1f}/s | {elapsed:.0f}s")
 
-            tasks = [bounded_download(ts, url) for ts, url in todo]
-            await asyncio.gather(*tasks)
+            # Schedule in chunks. One coroutine per manifest row up front is
+            # fine at 15k rows and not at 100k on a 5 GB box.
+            for i in range(0, len(todo), args.chunk):
+                batch = todo[i:i + args.chunk]
+                await asyncio.gather(*(bounded_download(ts, url)
+                                       for ts, url in batch))
 
     elapsed = time.time() - start_time
     print()
@@ -230,9 +270,11 @@ async def bounded_download(ts, url):
     print(f"  Skipped: {stats['skipped']:,}")
     print(f"  Retries: {stats['retries']:,}")
 
-    # Check disk usage
-    total_size = sum(f.stat().st_size for f in args.html_dir.rglob("*.html"))
+    # Check disk usage (both storage forms)
+    total_size = sum(f.stat().st_size for f in args.html_dir.rglob("*.html*"))
     print(f"  Disk:    {total_size / 1024 / 1024 / 1024:.2f} GB")
+    if stats["ok"]:
+        print(f"  Rate:    {stats['ok'] / elapsed:.2f} pages/s sustained")
 
 
 def main():

code/09-extract-prices.py

@@ -16,6 +16,7 @@
 """
 
 import csv
+import gzip
 import json
 import os
 import re
@@ -244,8 +245,13 @@ def extract_rating(html):
 
 def process_file(filepath):
     """Extract all data from a single HTML file."""
-    # Parse filename: <YYYYMMDD>_<slug>.html
-    fname = filepath.stem  # e.g., "20230801_write-content-for-your-website"
+    # Parse filename: <YYYYMMDD>_<slug>.html or <YYYYMMDD>_<slug>.html.gz
+    # (.stem on a .html.gz leaves the ".html", which would land in the slug)
+    fname = filepath.name
+    for suffix in (".html.gz", ".html"):
+        if fname.endswith(suffix):
+            fname = fname[:-len(suffix)]
+            break
     seller = filepath.parent.name
 
     parts = fname.split("_", 1)
@@ -255,8 +261,9 @@ def process_file(filepath):
     year = date[:4] if len(date) >= 4 else ""
     month = date[4:6] if len(date) >= 6 else ""
 
+    opener = gzip.open if filepath.name.endswith(".gz") else open
     try:
-        with open(filepath, "r", encoding="utf-8", errors="replace") as f:
+        with opener(filepath, "rt", encoding="utf-8", errors="replace") as f:
             html = f.read()
     except Exception:
         return None, "read_error"
@@ -293,7 +300,10 @@ def process_file(filepath):
     result["year"] = year
     result["month"] = month
     result["title"] = extract_title(html)
-    result["file_path"] = str(filepath.relative_to(BASE_DIR))
+    try:
+        result["file_path"] = str(Path(filepath).resolve().relative_to(BASE_DIR))
+    except ValueError:
+        result["file_path"] = str(filepath)
 
     rating, review_count = extract_rating(html)
     result["rating"] = rating if rating is not None else ""
@@ -303,9 +313,20 @@ def process_file(filepath):
 
 
 def main():
+    import argparse
+    global HTML_DIR, OUTPUT, ERRORS_LOG
+    ap = argparse.ArgumentParser()
+    ap.add_argument("--html-dir", type=Path, default=HTML_DIR)
+    ap.add_argument("--output", type=Path, default=OUTPUT)
+    ap.add_argument("--errors-log", type=Path, default=ERRORS_LOG)
+    a = ap.parse_args()
+    HTML_DIR, OUTPUT, ERRORS_LOG = a.html_dir, a.output, a.errors_log
+
     print("Extracting prices from downloaded HTML...")
+    print(f"  Input: {HTML_DIR}")
 
-    html_files = sorted(HTML_DIR.rglob("*.html"))
+    html_files = sorted(list(HTML_DIR.rglob("*.html")) +
+                        list(HTML_DIR.rglob("*.html.gz")))
     print(f"  Found {len(html_files):,} HTML files")
 
     success = 0

code/12-panel-ipi.py

@@ -29,6 +29,9 @@
 import numpy as np
 from scipy import stats as sp_stats
 
+sys.path.insert(0, str(Path(__file__).resolve().parent))
+from gigfilter import is_gig
+
 BASE_DIR = Path(__file__).resolve().parent.parent
 PRICES_FILE = BASE_DIR / "data" / "pilot" / "pilot-prices.csv"
 ITEMS_FILE = BASE_DIR / "data" / "pilot" / "gig-items.csv"
@@ -111,6 +114,8 @@ def main():
 
     with open(PRICES_FILE) as f:
         for row in csv.DictReader(f):
+            if not is_gig(row["seller"]):      # /hire/, /agencies/ landing pages
+                continue
             key = (row["seller"], row["slug"])
             item = item_map.get(key)
             if not item:
@@ -196,17 +201,25 @@ def main():
         # Initialize at base quarter = 100
         index = {base_q: 100.0}
 
-        # Chain forward from base
+        # Chain forward from base. Gap-tolerant: the matched-model relative at a
+        # quarter is p_t / p_{gig's previous observed quarter}, which already spans
+        # any gap, so we chain from the most recent quarter already in the index
+        # rather than requiring the immediately-preceding grid quarter. This keeps a
+        # category alive across coverage gaps (e.g. translation is empty in 2019Q2–Q4
+        # right after the base but has 18 solid transitions over 2020Q1–2024Q3; the
+        # old "prev grid quarter must be in index" rule silently discarded all of it).
         forward_qs = [q for q in all_quarters if q >= base_q]
+        last_q = base_q
         for i in range(1, len(forward_qs)):
             q = forward_qs[i]
             if q in quarter_relatives and len(quarter_relatives[q]) >= 3:
                 geo_mean = np.exp(np.mean(np.log(quarter_relatives[q])))
-                prev_q = forward_qs[i-1]
-                if prev_q in index:
-                    index[q] = index[prev_q] * geo_mean
+                index[q] = index[last_q] * geo_mean
+                last_q = q
 
-        # Chain backward from base
+        # Chain backward from base. Kept strict (immediately-preceding grid quarter
+        # must already be in the index) so pre-base historical levels — which feed
+        # the paper's peak/start figures — are unchanged by this fix.
         backward_qs = [q for q in reversed(all_quarters) if q <= base_q]
         for i in range(1, len(backward_qs)):
             q = backward_qs[i]