#!/usr/bin/env python3 """ Natural histories and travel accounts in the Biodiversity Heritage Library (BHL). Usage: python bhl_nature_travel.py # -> bhl_nature_travel.csv Reads BHL's public data export (https://www.biodiversitylibrary.org/data/TSV/: title, subject, creator and item tables; downloaded once into cache/bhl/) and keeps English, German, French, Spanish, Portuguese and Italian monographs published up to 1929 whose subject headings include natural history, nature study, or description and travel, voyages, exploration and the like, leaving out nursery and seed catalogues, field notebooks, periodicals, textbooks and juvenile works. BHL holds mostly scientific literature, so an LLM on innkube (qwen38-27b, Uni Passau) then sorts each candidate into natural history, travel, both, or neither (technical reports, taxonomic monographs, expedition results, proceedings, manuals, local histories). Answers are cached in bhl_labels.json. Needs INNKUBE_API_KEY. """ import argparse import csv import json import os import re import sys import urllib.request from collections import defaultdict from concurrent.futures import ThreadPoolExecutor from pathlib import Path DATA = "https://www.biodiversitylibrary.org/data/TSV/" UA = {"User-Agent": "NaturalHistoriesCatalogue/1.0 (https://thomashaider.net/natural-histories/)"} INNKUBE = "https://llms.innkube.fim.uni-passau.de/v1/chat/completions" LLM_MODEL = "qwen38-27b" BATCH = 40 TRAVEL = {"Description and travel", "Voyages and travels", "Discovery and exploration", "Voyages around the world", "Scientific expeditions", "Travel", "Explorers", "Arctic regions", "Antarctica"} NATURE = {"Natural history", "Nature study", "Naturalists", "Outdoor life", "Outdoor books", "Popular works", "Pre-Linnean works"} EXCLUDE = {"Catalogs", "Nursery stock", "Seeds", "Prices", "Sales catalogs", "Commercial catalogs", "Field notes", "Seed industry and trade", "Wholesale trade", "Periodicals", "Juvenile literature", "Juvenile fiction", "Fiction", "Textbooks", "Bibliography", "Congresses", "Laboratory manuals", "Identification", "Classification", "Nomenclature"} LANGS = {"ENG": "en", "GER": "de", "FRE": "fr", "SPA": "es", "POR": "pt", "ITA": "it"} PROMPT = ( "Each numbered line is a {what}: {fields}. " "Label each one for a catalogue of natural histories and travel writing:\n" "- natural_history: a descriptive or narrative work about animals, plants, landscapes or the nature of a " "place, written for general or learned readers: classic systematic natural histories (Buffon, Blumenbach), " "regional natural histories, nature essays, naturalists' rambles and observations.\n" "- travel: an account of a journey, voyage or expedition, or a description of a country or region as seen " "by a traveller.\n" "- both: a naturalist's travel account, or a voyage with substantial natural history.\n" "- neither: technical scientific reports, taxonomic monographs, species lists and catalogues, 'scientific " "results' or 'zoological results' volumes, proceedings and journals, manuals and handbooks for practice, " "civil or local histories, biographies, collected papers, fiction (novels, stories) and poems.\n" "Answer only with JSON mapping each number to one of: natural_history, travel, both, neither.\n\n") def table(path, name, refresh): f = path / name if refresh or not f.exists(): print(f"downloading {name} …", file=sys.stderr) path.mkdir(parents=True, exist_ok=True) with urllib.request.urlopen(urllib.request.Request(DATA + name, headers=UA), timeout=900) as r: f.write_bytes(r.read()) csv.field_size_limit(10 ** 8) return csv.DictReader(open(f, encoding="utf-8-sig", newline=""), delimiter="\t", quoting=csv.QUOTE_NONE) def clean_title(t): t = re.sub(r"\s+([:;,])", r"\1", t.strip()) return t.rstrip(" ./:;,") def label(lines, cache_path, workers, what="book in the Biodiversity Heritage Library", fields="title | author | year | subject headings"): """Sort works into natural_history/travel/both/neither with the LLM; answers are cached by key.""" cache = json.loads(Path(cache_path).read_text(encoding="utf-8")) if Path(cache_path).exists() else {} todo = [(k, line) for k, line in lines if k not in cache] batches = [todo[i:i + BATCH] for i in range(0, len(todo), BATCH)] print(f"{len(todo)} titles to label ({len(batches)} requests)", file=sys.stderr) def ask(batch): prompt = PROMPT.format(what=what, fields=fields) + "\n".join(f"{n}. {line}" for n, (_, line) in enumerate(batch, 1)) body = {"model": LLM_MODEL, "messages": [{"role": "user", "content": prompt}], "temperature": 0, "max_tokens": 3000, "chat_template_kwargs": {"enable_thinking": False}} req = urllib.request.Request(INNKUBE, data=json.dumps(body).encode(), headers={ "Authorization": "Bearer " + os.environ["INNKUBE_API_KEY"], "Content-Type": "application/json"}) try: with urllib.request.urlopen(req, timeout=300) as r: content = json.load(r)["choices"][0]["message"]["content"] content = re.sub(r".*?|```(json)?", "", content, flags=re.S) return batch, json.loads(content[content.index("{"):content.rindex("}") + 1]) except Exception as e: # leave uncached; a rerun retries print(f" ! batch failed: {e}", file=sys.stderr) return batch, None with ThreadPoolExecutor(workers) as pool: for n, (batch, answer) in enumerate(pool.map(ask, batches), 1): if answer: for i, (k, _) in enumerate(batch, 1): v = answer.get(str(i)) if v in ("natural_history", "travel", "both", "neither"): cache[k] = v if n % 10 == 0 or n == len(batches): Path(cache_path).write_text(json.dumps(cache, ensure_ascii=False), encoding="utf-8") print(f" {n}/{len(batches)}", file=sys.stderr) return cache def main(): ap = argparse.ArgumentParser() ap.add_argument("--dir", default="cache/bhl") ap.add_argument("--refresh", action="store_true", help="download the BHL tables again") ap.add_argument("--labels", default="bhl_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="bhl_nature_travel.csv") args = ap.parse_args() d = Path(args.dir) titles = {} for r in table(d, "title.txt", args.refresh): y = r["StartYear"] if r["LanguageCode"] in LANGS and r["MARCLeader"][7:8] == "m" and y.isdigit() and int(y) <= args.max_year: titles[r["TitleID"]] = r subjects = defaultdict(set) for r in table(d, "subject.txt", args.refresh): if r["TitleID"] in titles: subjects[r["TitleID"]].add(r["Subject"].strip().rstrip(".")) keep = {k for k, s in subjects.items() if s & (TRAVEL | NATURE) and not s & EXCLUDE} print(f"{len(titles)} English/German/French/Spanish/Portuguese/Italian monographs to {args.max_year}; {len(keep)} with nature or travel subjects", file=sys.stderr) authors = defaultdict(list) for r in table(d, "creator.txt", args.refresh): if r["TitleID"] in keep and r["CreatorType"].startswith("Main"): authors[r["TitleID"]].append(r["CreatorName"].strip().rstrip(".,")) items = defaultdict(list) for r in table(d, "item.txt", args.refresh): if r["TitleID"] in keep: items[r["TitleID"]].append(int(r["ItemID"])) rows = [] for k in sorted(keep, key=int): t = titles[k] author = "; ".join(authors[k]) or t["TL2Author"].strip().rstrip(".,") subj = sorted(subjects[k], key=str.lower) rows.append({"id": k, "title": clean_title(t["FullTitle"]), "author": author, "year": t["StartYear"], "language": LANGS[t["LanguageCode"]], "subjects": " | ".join(subj), "volumes": len(items[k]), "url": t["TitleURL"].strip(), "read_url": f"https://www.biodiversitylibrary.org/item/{min(items[k])}" if items[k] else t["TitleURL"].strip()}) 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) 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)} candidates to {args.out}" + (f" ({unlabelled} still unlabelled; rerun to retry)" if unlabelled else ""), file=sys.stderr) if __name__ == "__main__": main()