#!/usr/bin/env python3 """ Natural histories and travel accounts in the Internet Archive. Usage: python ia_nature_travel.py # -> ia_nature_travel.csv python ia_nature_travel.py --cached # reuse cache/ia/scrape.jsonl instead of querying again Queries the Internet Archive's scrape API for texts dated up to 1929 in English, German, French, Spanish, Portuguese or Italian whose subject headings name natural history, nature study, naturalists, description and travel, voyages, exploration or scientific expeditions (and their equivalents in the other five languages), leaving out IA's mirrors of the Biodiversity Heritage Library and Project Gutenberg, which the catalogue takes from those libraries directly, and manuscript ships' logbooks. For Spanish, Portuguese and Italian, whose scans from European libraries often lack subject headings, a search on title words (viaje, viagem, viaggio, historia natural, storia naturale, ...) runs alongside. Libraries scanned many works several times, and multi-volume works are one item per volume, so items with the same first author's surname, title and language are merged into one work that links to the most downloaded copy and counts the others. Works with the same excluded subjects as in bhl_nature_travel.py (catalogues, periodicals, juvenile literature, fiction, ...) are dropped. As with BHL, the LLM on innkube (qwen38-27b) then sorts each work into natural history, travel, both, or neither; answers are cached in ia_labels.json. Needs INNKUBE_API_KEY. """ import argparse import csv import json import re import sys import time import unicodedata import urllib.parse import urllib.request from collections import Counter, defaultdict from pathlib import Path from bhl_nature_travel import EXCLUDE, label SCRAPE = "https://archive.org/services/search/v1/scrape" UA = {"User-Agent": "NaturalHistoriesCatalogue/1.0 (https://thomashaider.net/natural-histories/)"} FIELDS = "identifier,title,creator,year,date,language,subject,collection,volume,downloads" SUBJECTS = ["natural history", "nature study", "naturalists", "description and travel", "voyages and travels", "discovery and exploration", "voyages around the world", "scientific expeditions", "histoire naturelle", "voyages et voyages", "récits de voyages", "naturgeschichte", "reisebericht", "reisebeschreibung", "entdeckungsreisen", "historia natural", "viajes", "descripciones y viajes", "descripción y viajes", "exploraciones", "viagens", "história natural", "viaggi", "storia naturale", "descrizioni e viaggi"] LANGS = {"eng": "en", "english": "en", "en": "en", "ger": "de", "german": "de", "deu": "de", "de": "de", "fre": "fr", "french": "fr", "fra": "fr", "fr": "fr", "spa": "es", "spanish": "es", "es": "es", "por": "pt", "portuguese": "pt", "pt": "pt", "ita": "it", "italian": "it", "it": "it"} SKIP_COLLECTIONS = {"biodiversity", "gutenberg", "microfilm", "periodicals", "serials", "pub_", # manuscript ships' logs (whaling voyages): handwritten scans without text "providencepublicnicholson", "newbedfordwhalingmuseumlogbooksandjournals", "newbedfordfreemaritimelogbooksandjournals"} LOGBOOK = re.compile(r"^\[?(Logbook|Log|Journal) of .*\((Ship|Bark|Brig|Schooner)\)") # Spanish, Portuguese and Italian scans from European libraries often have no subject headings, so for # these languages a title search runs alongside the subject search. ("lettere" and "descrizione" are left # out: they bring in thousands of letters and topographies.) TITLE_QUERIES = { "spa OR spanish": 'viaje OR viajes OR viage OR viages OR "historia natural" OR expedición OR navegación OR ' 'excursión OR excursiones OR peregrinación OR aves OR plantas OR animales OR insectos', "por OR portuguese": 'viagem OR viagens OR viajem OR "historia natural" OR "história natural" OR expedição OR ' 'expedicao OR navegação OR navegacao OR roteiro OR aves OR plantas OR animaes OR animais', "ita OR italian": 'viaggio OR viaggi OR "storia naturale" OR "istoria naturale" OR spedizione OR navigazione OR ' 'escursione OR escursioni OR peregrinazione OR uccelli OR piante OR animali OR insetti', } def title_queries(max_year): return [f"mediatype:texts AND year:[1450 TO {max_year}] AND language:({langs}) AND title:({words})" " AND -collection:biodiversity AND -collection:gutenberg" for langs, words in TITLE_QUERIES.items()] def query(max_year): subj = " OR ".join(f'subject:"{s}"' for s in SUBJECTS) langs = " OR ".join(sorted(LANGS)) return (f"mediatype:texts AND year:[1450 TO {max_year}] AND ({subj}) AND language:({langs})" " AND -collection:biodiversity AND -collection:gutenberg") def scrape(q, out, mode="w"): params = {"q": q, "fields": FIELDS, "count": 10000} n = 0 with open(out, mode, encoding="utf-8") as f: while True: url = SCRAPE + "?" + urllib.parse.urlencode(params) for attempt in range(6): try: with urllib.request.urlopen(urllib.request.Request(url, headers=UA), timeout=600) as r: d = json.load(r) break except Exception as e: print(f" ! {e}; retrying", file=sys.stderr) time.sleep(10 * 2 ** attempt) else: raise RuntimeError(url) for item in d.get("items", []): f.write(json.dumps(item, ensure_ascii=False) + "\n") n += len(d.get("items", [])) print(f" {n}/{d.get('total')}", file=sys.stderr) if not d.get("cursor"): return params["cursor"] = d["cursor"] def as_list(v): return v if isinstance(v, list) else [v] if v else [] def title_of(item): return " ".join(as_list(item.get("title"))[:1]) if item.get("title") else "" def norm(s): s = unicodedata.normalize("NFKD", s.replace("ſ", "s")) return re.sub(r"[^a-z0-9 ]", "", "".join(c for c in s if not unicodedata.combining(c)).lower()) def clean_title(t): t = re.sub(r"\s+([:;,])", r"\1", " ".join(t.split())) return t.rstrip(" ./:;,") VOLUME = re.compile(r"[\s.,;:(\[-]+\b(v|vol|volume|t|tome|bd|band|th|theil|teil|part|pt)\.?\s*([0-9]+|[ivxlc]+)\b.*$", re.I) def work_key(item, lang): creator = as_list(item.get("creator")) surname = norm(re.split(r"[,(]", creator[0])[0]).strip() if creator else "" title = norm(VOLUME.sub("", clean_title(title_of(item)))) title = re.split(r"\s{2,}|\b(or|ou|oder)\b", title)[0] return lang, surname, " ".join(title.split()[:8]) def main(): ap = argparse.ArgumentParser() ap.add_argument("--dir", default="cache/ia") ap.add_argument("--cached", action="store_true", help="reuse the last scrape") ap.add_argument("--labels", default="ia_labels.json") ap.add_argument("--workers", type=int, default=4) ap.add_argument("--max-year", type=int, default=1929) ap.add_argument("-o", "--out", default="ia_nature_travel.csv") args = ap.parse_args() raw = Path(args.dir) / "scrape.jsonl" raw.parent.mkdir(parents=True, exist_ok=True) if not (args.cached and raw.exists()): scrape(query(args.max_year), raw) for q in title_queries(args.max_year): scrape(q, raw, mode="a") works = defaultdict(list) n_items = n_excluded = 0 seen = set() for line in open(raw, encoding="utf-8"): item = json.loads(line) if item["identifier"] in seen: # found by both the subject and the title search continue seen.add(item["identifier"]) n_items += 1 langs = [LANGS[l.lower()] for l in as_list(item.get("language")) if l.lower() in LANGS] subjects = {s.strip().rstrip(".") for x in as_list(item.get("subject")) for s in re.split(r"\s*(?:--|;)\s*", x)} if not langs or not title_of(item) or subjects & EXCLUDE or LOGBOOK.search(title_of(item)) \ or {c for c in as_list(item.get("collection")) if any(c.startswith(s) for s in SKIP_COLLECTIONS)}: n_excluded += 1 continue works[work_key(item, langs[0])].append(item) print(f"{n_items} items; {n_excluded} excluded; {len(works)} works after merging copies and volumes", file=sys.stderr) rows = [] for (lang, _, _), items in works.items(): best = max(items, key=lambda i: int(i.get("downloads") or 0)) years = [int(str(i["year"])[:4]) for i in items if str(i.get("year", ""))[:4].isdigit()] subj = sorted({s for i in items for x in as_list(i.get("subject")) for s in re.split(r"\s*;\s*", x) if s}, key=str.lower) vols = {str(i.get("volume")) for i in items if i.get("volume")} | \ {m.group(2).lower() for i in items if (m := VOLUME.search(title_of(i)))} rows.append({ "id": best["identifier"], "title": clean_title(title_of(best)), "author": "; ".join(as_list(best.get("creator")))[:200], "year": min(years) if years else "", "language": lang, "subjects": " | ".join(subj)[:600], "copies": len(items), "volumes": len(vols), "downloads": sum(int(i.get("downloads") or 0) for i in items), "url": f"https://archive.org/details/{best['identifier']}", }) rows.sort(key=lambda r: -r["downloads"]) lines = [(r["id"], f"{r['title'][:160]} | {r['author'][:60] or 'anonymous'} | {r['year']} | {r['subjects'][:200]}") for r in rows] labels = label(lines, args.labels, args.workers, what="book in the Internet Archive") out = [] for r in rows: v = labels.get(r["id"]) if v and v != "neither": r["category"] = "natural_history+travel" if v == "both" else v out.append(r) with open(args.out, "w", encoding="utf-8", newline="") as f: w = csv.DictWriter(f, fieldnames=list(out[0].keys())) w.writeheader() w.writerows(out) unlabelled = sum(1 for r in rows if r["id"] not in labels) print(f"Wrote {len(out)} of {len(rows)} works to {args.out}" + (f" ({unlabelled} still unlabelled; rerun with --cached to retry)" if unlabelled else ""), file=sys.stderr) for k, n in sorted(Counter((r["language"], r["category"]) for r in out).items()): print(" ", *k, n, file=sys.stderr) if __name__ == "__main__": main()