← back to Wallco Ai
scripts/crawl_pd.py
350 lines
#!/usr/bin/env python3
"""
Multi-source public-domain pattern corpus crawler for wallco.ai.
Sources:
- Wikimedia Commons (no key, polite 2 req/s)
Categories: Textile_patterns, Floral_patterns, Decorative_patterns,
Geometric_patterns, Damask, Paisley_patterns, Toile, Brocade,
Files_from_the_Smithsonian, Files_from_the_Cooper_Hewitt,_Smithsonian_Design_Museum
- Met Museum Open Access (no key, no advertised rate limit)
Filter: isPublicDomain=true, classification in our pattern set
- Smithsonian / Cooper Hewitt (DEMO_KEY 30/hr — used sparingly)
Query: unit_code=CHNDM AND wallpaper
Writes to dw_unified.pd_source_designs. Idempotent on (source, source_id).
Downloads images to ~/Projects/wallco-ai/data/images/<source>/<id>.<ext>.
Run:
python3 scripts/crawl_pd.py --source wikimedia --limit 200
python3 scripts/crawl_pd.py --source met --limit 500
python3 scripts/crawl_pd.py --source smithsonian --limit 50
"""
import os, sys, json, time, argparse, urllib.request, ssl, subprocess
from pathlib import Path
from urllib.parse import quote
ROOT = Path.home() / 'Projects' / 'wallco-ai'
IMGROOT = ROOT / 'data' / 'images'
LOG = ROOT / 'logs' / 'crawl_pd.log'
IMGROOT.mkdir(parents=True, exist_ok=True)
LOG.parent.mkdir(parents=True, exist_ok=True)
CTX = ssl.create_default_context()
WIKIMEDIA_CATEGORIES = [
'Textile_patterns', 'Floral_patterns', 'Decorative_patterns',
'Geometric_patterns', 'Damask', 'Paisley_patterns', 'Toile_de_Jouy',
'Brocade', 'Wallpaper_(material)', 'Embroidery_patterns',
# Tick 10 additions — broader pattern provenance
'Tapa_cloth_designs', 'Ikat', 'Batik_patterns', 'Indian_block_prints',
'Chinoiserie', 'William_Morris', 'Art_Nouveau_patterns', 'Pattern_books',
'Trellis_patterns', 'Lattice_patterns'
]
MET_PATTERN_QUERIES = ['wallpaper', 'textile', 'damask', 'brocade', 'pattern', 'design', 'ornament']
PD_LICENSES = {'CC0', 'PD', 'Public domain', 'Public domain mark', 'PD-old', 'PD-self', 'PD-Art', 'PDM'}
def http_get_json(url, timeout=20):
req = urllib.request.Request(url, headers={'User-Agent': 'wallco.ai-crawler/0.1 (research; contact: steve@designerwallcoverings.com)'})
with urllib.request.urlopen(req, timeout=timeout, context=CTX) as r:
return json.loads(r.read().decode('utf-8'))
def http_download(url, dest, timeout=60):
req = urllib.request.Request(url, headers={'User-Agent': 'wallco.ai-crawler/0.1'})
with urllib.request.urlopen(req, timeout=timeout, context=CTX) as r, open(dest, 'wb') as f:
f.write(r.read())
def psql(sql):
r = subprocess.run(['psql', 'dw_unified', '-At', '-q'], input=sql, capture_output=True, text=True)
if r.returncode != 0:
raise RuntimeError(f"psql failed: {r.stderr}")
return r.stdout.strip()
def esc(s):
if s is None: return 'NULL'
return "'" + str(s).replace("'", "''") + "'"
def upsert(source, source_id, **fields):
cols = ['source', 'source_id'] + list(fields.keys())
vals_clean = [esc(source), esc(source_id)]
for k, v in fields.items():
if v is None:
vals_clean.append('NULL')
elif isinstance(v, list):
arr = "ARRAY[" + ",".join(esc(x) for x in v) + "]::text[]"
vals_clean.append(arr)
elif isinstance(v, dict):
vals_clean.append(f"{esc(json.dumps(v))}::jsonb")
elif isinstance(v, bool):
vals_clean.append('TRUE' if v else 'FALSE')
elif isinstance(v, (int, float)):
vals_clean.append(str(v))
else:
vals_clean.append(esc(v))
update_set = ", ".join(f"{k}=EXCLUDED.{k}" for k in fields.keys())
sql = f"""
INSERT INTO pd_source_designs ({", ".join(cols)})
VALUES ({", ".join(vals_clean)})
ON CONFLICT (source, source_id) DO UPDATE SET {update_set}, ingested_at=NOW()
RETURNING id;
"""
return psql(sql)
# -------- Wikimedia Commons --------
def crawl_wikimedia(limit=None, polite_s=0.5):
saved = 0
skipped_license = 0
for cat in WIKIMEDIA_CATEGORIES:
cont = ''
while True:
url = f'https://commons.wikimedia.org/w/api.php?action=query&list=categorymembers&cmtitle=Category:{cat}&cmtype=file&cmlimit=200&format=json' + cont
try:
d = http_get_json(url)
except Exception as e:
print(f" cat {cat} fetch err: {e}")
break
members = d.get('query', {}).get('categorymembers', []) or []
for m in members:
title = m['title'] # e.g. 'File:foo.jpg'
if not title.startswith('File:'):
continue
# Fetch imageinfo (request 1280px thumbnail to avoid wikimedia.org rate limits on full-res)
ii_url = f'https://commons.wikimedia.org/w/api.php?action=query&titles={quote(title)}&prop=imageinfo&iiprop=url|extmetadata|size|mime&iiurlwidth=1280&format=json'
try:
ii = http_get_json(ii_url)
except Exception as e:
print(f" imageinfo err for {title}: {e}")
continue
pages = ii.get('query', {}).get('pages', {}) or {}
page = next(iter(pages.values()), None)
if not page: continue
infos = page.get('imageinfo') or []
if not infos: continue
info = infos[0]
exm = info.get('extmetadata', {}) or {}
license_short = (exm.get('LicenseShortName', {}) or {}).get('value', '')
if not any(lic.lower() in license_short.lower() for lic in PD_LICENSES):
skipped_license += 1
time.sleep(polite_s)
continue
source_id = title
creator = (exm.get('Artist', {}) or {}).get('value', '')
date_created = (exm.get('DateTimeOriginal', {}) or {}).get('value', '') or (exm.get('DateTime', {}) or {}).get('value', '')
desc = (exm.get('ImageDescription', {}) or {}).get('value', '')
cats = (exm.get('Categories', {}) or {}).get('value', '')
# strip HTML
import re
creator_clean = re.sub(r'<[^>]+>', '', creator).strip()[:250]
desc_clean = re.sub(r'<[^>]+>', '', desc).strip()
tags = [t.strip() for t in cats.split('|') if t.strip()][:30]
# Prefer thumbnail URL over full-res — Wikimedia rate-limits full-res hard.
img_url = info.get('thumburl') or info.get('url')
width = info.get('thumbwidth') or info.get('width')
height = info.get('thumbheight') or info.get('height')
bytes_ = info.get('size')
# Download
ext = (img_url or '').rsplit('.', 1)[-1].lower()[:5] or 'jpg'
dest_dir = IMGROOT / 'wikimedia'
dest_dir.mkdir(parents=True, exist_ok=True)
safe = title.replace('File:', '').replace(' ', '_').replace('/', '_')[:120]
dest = dest_dir / f'{safe}'
if not dest.exists() and img_url:
try:
http_download(img_url, dest)
except Exception as e:
print(f" dl err {title}: {e}")
time.sleep(polite_s)
continue
row_id = upsert(
'wikimedia',
source_id,
title=desc_clean[:300] or title.replace('File:', ''),
creator=creator_clean,
date_created=date_created[:60] if date_created else None,
license=license_short or 'PD',
source_url=f'https://commons.wikimedia.org/wiki/{quote(title)}',
image_url=img_url,
local_path=str(dest),
width_px=width,
height_px=height,
bytes=bytes_,
raw_metadata=exm,
tags=tags,
category=cat.replace('_', '-').lower()
)
saved += 1
if saved % 10 == 0:
print(f" [{saved}] wikimedia saved · cat={cat} · last={title[:60]}")
if limit and saved >= limit:
print(f" hit limit {limit}, stopping")
return saved, skipped_license
time.sleep(polite_s)
cont_obj = d.get('continue')
if not cont_obj:
break
cont = '&cmcontinue=' + cont_obj['cmcontinue']
return saved, skipped_license
# -------- Met Museum Open Access --------
def crawl_met(limit=None, polite_s=0.4):
saved = 0
skipped_non_pd = 0
seen_ids = set()
for q in MET_PATTERN_QUERIES:
try:
ids_resp = http_get_json(f'https://collectionapi.metmuseum.org/public/collection/v1/search?q={quote(q)}&hasImages=true')
except Exception as e:
print(f" met search err for {q}: {e}")
continue
ids = ids_resp.get('objectIDs') or []
print(f" Met query '{q}' returned {len(ids)} candidates")
for oid in ids:
if oid in seen_ids: continue
seen_ids.add(oid)
try:
o = http_get_json(f'https://collectionapi.metmuseum.org/public/collection/v1/objects/{oid}')
except Exception as e:
print(f" object {oid} err: {e}")
time.sleep(polite_s); continue
if not o.get('isPublicDomain'):
skipped_non_pd += 1
time.sleep(polite_s); continue
primary = o.get('primaryImage') or o.get('primaryImageSmall')
if not primary:
time.sleep(polite_s); continue
title = o.get('title') or f"Met #{oid}"
ext = primary.rsplit('.', 1)[-1].split('?')[0][:5] or 'jpg'
dest_dir = IMGROOT / 'met'
dest_dir.mkdir(parents=True, exist_ok=True)
dest = dest_dir / f'met_{oid}.{ext}'
if not dest.exists():
try:
http_download(primary, dest)
except Exception as e:
print(f" met dl err {oid}: {e}")
time.sleep(polite_s); continue
tags = [t.get('term') for t in (o.get('tags') or []) if t.get('term')]
row = upsert(
'met',
str(oid),
title=title[:300],
creator=o.get('artistDisplayName') or '',
date_created=o.get('objectDate') or None,
license='CC0',
source_url=o.get('objectURL') or f'https://www.metmuseum.org/art/collection/search/{oid}',
image_url=primary,
local_path=str(dest),
raw_metadata={k: o.get(k) for k in ('classification','medium','culture','department','dimensions','period','objectName')},
tags=tags,
category=(o.get('classification') or '').lower().replace(' ', '-') or 'pattern'
)
saved += 1
if saved % 5 == 0:
print(f" [{saved}] met saved · oid={oid} · {title[:50]}")
if limit and saved >= limit:
return saved, skipped_non_pd
time.sleep(polite_s)
return saved, skipped_non_pd
# -------- Smithsonian / Cooper Hewitt --------
def crawl_smithsonian(limit=50, polite_s=2.5, api_key='DEMO_KEY'):
# DEMO_KEY = 30 req/hour. With polite_s=2.5 we do ~24 req/min — too fast.
# 30 req/hour = 1 every 120s. Stick to that or set a real api_key in .env.
# For initial seeding we just sample.
saved = 0
start = 0
while True:
url = f'https://api.si.edu/openaccess/api/v1.0/search?q=unit_code%3ACHNDM+AND+wallpaper&start={start}&rows=20&api_key={api_key}'
try:
d = http_get_json(url)
except Exception as e:
print(f" si search err: {e}")
break
rows = (d.get('response') or {}).get('rows') or []
if not rows: break
for r in rows:
dnr = (r.get('content') or {}).get('descriptiveNonRepeating') or {}
om = dnr.get('online_media') or {}
media = om.get('media') or []
ind = (r.get('content') or {}).get('indexedStructured') or {}
for m in media[:5]: # keep at most 5 images per record
usage = (m.get('usage') or {}).get('access')
if usage != 'CC0':
continue
img_url = m.get('content') or m.get('thumbnail')
if not img_url: continue
source_id = f"{r.get('id')}|{m.get('idsId') or img_url[-40:]}"
dest_dir = IMGROOT / 'smithsonian'
dest_dir.mkdir(parents=True, exist_ok=True)
ext = 'jpg'
safe = source_id.replace('|','_').replace('/','_')[:120]
dest = dest_dir / f'{safe}.{ext}'
if not dest.exists():
try:
http_download(img_url, dest)
except Exception as e:
print(f" si dl err {source_id}: {e}")
continue
upsert(
'smithsonian',
source_id,
title=(r.get('title') or '')[:300],
creator=(', '.join((ind.get('name') or [])))[:250],
date_created=(', '.join(ind.get('date') or []))[:60],
license='CC0',
source_url=f"https://www.si.edu/object/{r.get('id')}",
image_url=img_url,
local_path=str(dest),
raw_metadata={k: ind.get(k) for k in ('object_type','topic','culture','place','date')},
tags=(ind.get('topic') or [])[:20],
category=(((ind.get('object_type') or [None])[0]) or 'wallpaper').lower().replace(' ', '-')
)
saved += 1
if saved % 5 == 0:
print(f" [{saved}] smithsonian saved · {r.get('title','')[:40]}")
if limit and saved >= limit:
return saved
time.sleep(polite_s)
start += 20
if start >= 500: break
return saved
def main():
ap = argparse.ArgumentParser()
ap.add_argument('--source', choices=['wikimedia', 'met', 'smithsonian', 'all'], default='all')
ap.add_argument('--limit', type=int, default=200, help='per-source cap')
args = ap.parse_args()
if args.source in ('wikimedia', 'all'):
n, sk = crawl_wikimedia(limit=args.limit)
print(f"WIKIMEDIA: saved {n}, skipped non-PD {sk}")
if args.source in ('met', 'all'):
n, sk = crawl_met(limit=args.limit)
print(f"MET: saved {n}, skipped non-PD {sk}")
if args.source in ('smithsonian', 'all'):
n = crawl_smithsonian(limit=min(args.limit, 50))
print(f"SMITHSONIAN: saved {n}")
print("\n--- DB counts ---")
print(psql("SELECT source, COUNT(*) FROM pd_source_designs GROUP BY source ORDER BY 1;"))
if __name__ == '__main__':
main()