Zotero Collections Ingest Implementation Plan

For agentic workers: REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (- [ ]) syntax for tracking.

Goal: Build zotero/ingest_to_wiki.py to ingest the Buckyverse Zotero group's curated collections (live local API) into .wiki/raw/** as idempotent raw-source records.

Architecture: A single Python module exposing small, pure functions (slugify, classify, frontmatter, dedup) plus a thin ZoteroClient for the local HTTP API and a main() orchestrator with --dry-run/--limit/--no-fetch. Pure logic is unit-tested with pytest; network + filesystem effects are isolated behind injectable seams.

Tech Stack: Python 3.12 stdlib (urllib, html.parser, json, argparse, pathlib), pytest for tests, pdftotext (poppler CLI) for best-effort PDF text. No third-party Python deps.

Reference constants (used across tasks — keep identical)

GROUP_ID = "5189518"
API_BASE = "http://localhost:23119/api"
RAW_DIR = ".wiki/raw"
REPO_ROOT_BOOK_GLOB = "*/index.md"          # depth-1 book source dirs
SNAPSHOT_PATH = "zotero/buckyverse_collections.json"
REPORT_PATH = "zotero/last-ingest-report.md"

# (name, key) — order defines fetch order
COLLECTIONS = [
    ("Bucky Shelf", "L84AV9YL"),
    ("Books",       "BK23LBDC"),
    ("By Bucky",    "YAAHI5VV"),
    ("Wish List",   "HLYVN3EP"),
    ("People",      "R5ID4MAH"),
]
KEY_TO_NAME = {k: n for n, k in COLLECTIONS}
WISH_LIST_KEY = "HLYVN3EP"
BY_BUCKY_KEY = "YAAHI5VV"

TYPE_MAP = {
    "book": "repos", "bookSection": "repos",
    "journalArticle": "papers", "report": "papers",
    "webpage": "articles", "blogPost": "articles",
    "newspaperArticle": "articles", "magazineArticle": "articles",
    "document": "articles", "radioBroadcast": "articles",
    "tvBroadcast": "articles", "encyclopediaArticle": "articles",
    "videoRecording": "notes", "film": "notes",
}
DEFAULT_RAW_TYPE = "articles"

Tests run with: python3 -m pytest zotero/test_ingest_to_wiki.py -v

Task 1: Module scaffold + `slugify`

Files:

Create: zotero/ingest_to_wiki.py
Test: zotero/test_ingest_to_wiki.py
Step 1: Write the failing test

# zotero/test_ingest_to_wiki.py
import os, sys
sys.path.insert(0, os.path.dirname(__file__))
import ingest_to_wiki as ing


def test_slugify_basic():
    assert ing.slugify("Operating Manual for Spaceship Earth") == \
        "operating-manual-for-spaceship-earth"

def test_slugify_punctuation_and_ampersand():
    assert ing.slugify("Earth, Inc. & Friends") == "earth-inc-and-friends"

def test_slugify_collapses_and_trims_dashes():
    assert ing.slugify("  --Nine   Chains-- to the Moon--  ") == \
        "nine-chains-to-the-moon"

def test_slugify_truncates():
    s = ing.slugify("word " * 50, maxlen=20)
    assert len(s) <= 20
    assert not s.endswith("-")

Step 2: Run test to verify it fails

Run: python3 -m pytest zotero/test_ingest_to_wiki.py -v
Expected: FAIL — ModuleNotFoundError: No module named 'ingest_to_wiki'

Step 3: Write minimal implementation

#!/usr/bin/env python3
"""Ingest the Buckyverse Zotero group's curated collections into .wiki/raw/**.

Reads the live local Zotero API (http://localhost:23119/api), walks the
configured collections, and writes idempotent raw-source records. See
docs/superpowers/specs/2026-06-30-zotero-collections-ingest-design.md.
"""
import re

GROUP_ID = "5189518"
API_BASE = "http://localhost:23119/api"
RAW_DIR = ".wiki/raw"
SNAPSHOT_PATH = "zotero/buckyverse_collections.json"
REPORT_PATH = "zotero/last-ingest-report.md"

COLLECTIONS = [
    ("Bucky Shelf", "L84AV9YL"),
    ("Books",       "BK23LBDC"),
    ("By Bucky",    "YAAHI5VV"),
    ("Wish List",   "HLYVN3EP"),
    ("People",      "R5ID4MAH"),
]
KEY_TO_NAME = {k: n for n, k in COLLECTIONS}
WISH_LIST_KEY = "HLYVN3EP"
BY_BUCKY_KEY = "YAAHI5VV"

TYPE_MAP = {
    "book": "repos", "bookSection": "repos",
    "journalArticle": "papers", "report": "papers",
    "webpage": "articles", "blogPost": "articles",
    "newspaperArticle": "articles", "magazineArticle": "articles",
    "document": "articles", "radioBroadcast": "articles",
    "tvBroadcast": "articles", "encyclopediaArticle": "articles",
    "videoRecording": "notes", "film": "notes",
}
DEFAULT_RAW_TYPE = "articles"


def slugify(text, maxlen=60):
    text = (text or "").lower().replace("&", " and ")
    text = re.sub(r"[^a-z0-9\s-]", "", text)
    text = re.sub(r"[\s_]+", "-", text)
    text = re.sub(r"-+", "-", text).strip("-")
    if len(text) > maxlen:
        text = text[:maxlen].rstrip("-")
    return text

Step 4: Run test to verify it passes

Run: python3 -m pytest zotero/test_ingest_to_wiki.py -v
Expected: PASS (4 passed)

Step 5: Commit

git add zotero/ingest_to_wiki.py zotero/test_ingest_to_wiki.py
git commit -m "feat(zotero): module scaffold + slugify"

Task 2: `classify` + `is_unknown_type` + `target_path`

Files:

Modify: zotero/ingest_to_wiki.py
Test: zotero/test_ingest_to_wiki.py
Step 1: Write the failing test

def test_classify_known_types():
    assert ing.classify({"itemType": "book"}) == "repos"
    assert ing.classify({"itemType": "journalArticle"}) == "papers"
    assert ing.classify({"itemType": "webpage"}) == "articles"
    assert ing.classify({"itemType": "encyclopediaArticle"}) == "articles"
    assert ing.classify({"itemType": "videoRecording"}) == "notes"

def test_classify_unknown_falls_back_to_articles():
    assert ing.classify({"itemType": "podcast"}) == "articles"
    assert ing.is_unknown_type({"itemType": "podcast"}) is True
    assert ing.is_unknown_type({"itemType": "book"}) is False

def test_target_path_per_type():
    assert ing.target_path("repos", "critical-path", "2026-06-30") == \
        ".wiki/raw/repos/critical-path.md"
    assert ing.target_path("papers", "some-paper", "2026-06-30") == \
        ".wiki/raw/papers/some-paper.md"
    assert ing.target_path("notes", "a-film", "2026-06-30") == \
        ".wiki/raw/notes/a-film.md"
    assert ing.target_path("articles", "a-page", "2026-06-30") == \
        ".wiki/raw/articles/2026-06-30-a-page.md"

Step 2: Run test to verify it fails

Run: python3 -m pytest zotero/test_ingest_to_wiki.py -k "classify or target_path" -v
Expected: FAIL — AttributeError: module 'ingest_to_wiki' has no attribute 'classify'

Step 3: Write minimal implementation

def classify(data):
    return TYPE_MAP.get(data.get("itemType", ""), DEFAULT_RAW_TYPE)


def is_unknown_type(data):
    return data.get("itemType", "") not in TYPE_MAP


def target_path(raw_type, slug, ingested):
    if raw_type == "articles":
        return f"{RAW_DIR}/articles/{ingested}-{slug}.md"
    return f"{RAW_DIR}/{raw_type}/{slug}.md"

Step 4: Run test to verify it passes

Run: python3 -m pytest zotero/test_ingest_to_wiki.py -k "classify or target_path" -v
Expected: PASS

Step 5: Commit

git add zotero/ingest_to_wiki.py zotero/test_ingest_to_wiki.py
git commit -m "feat(zotero): item type classification + target paths"

Task 3: `authorship_tag` + `pick_source_url`

Files:

Modify: zotero/ingest_to_wiki.py
Test: zotero/test_ingest_to_wiki.py
Step 1: Write the failing test

def test_authorship_tag_by_bucky_collection():
    data = {"collections": ["YAAHI5VV"], "creators": []}
    assert ing.authorship_tag(data) == "buckminster-fuller"

def test_authorship_tag_creator_is_fuller():
    data = {"collections": ["L84AV9YL"],
            "creators": [{"creatorType": "author", "firstName": "R. Buckminster",
                          "lastName": "Fuller"}]}
    assert ing.authorship_tag(data) == "buckminster-fuller"

def test_authorship_tag_default_about():
    data = {"collections": ["L84AV9YL"],
            "creators": [{"creatorType": "author", "firstName": "Hugh",
                          "lastName": "Kenner"}]}
    assert ing.authorship_tag(data) == "about-buckminster-fuller"

def test_pick_source_url_prefers_url():
    assert ing.pick_source_url({"key": "K1", "url": "http://x"}) == "http://x"

def test_pick_source_url_doi_then_zotero():
    assert ing.pick_source_url({"key": "K1", "DOI": "10.1/abc"}) == \
        "https://doi.org/10.1/abc"
    assert ing.pick_source_url({"key": "K1"}) == \
        "zotero://select/groups/5189518/items/K1"

Step 2: Run test to verify it fails

Run: python3 -m pytest zotero/test_ingest_to_wiki.py -k "authorship or source_url" -v
Expected: FAIL — attribute errors

Step 3: Write minimal implementation

def authorship_tag(data):
    cols = set(data.get("collections", []))
    if BY_BUCKY_KEY in cols:
        return "buckminster-fuller"
    for c in data.get("creators", []):
        last = (c.get("lastName") or c.get("name") or "").strip().lower()
        first = (c.get("firstName") or "").strip().lower()
        if last == "fuller" and ("buckminster" in first or first.startswith("r")):
            return "buckminster-fuller"
    return "about-buckminster-fuller"


def pick_source_url(data):
    if data.get("url"):
        return data["url"]
    if data.get("DOI"):
        return f"https://doi.org/{data['DOI']}"
    return f"zotero://select/groups/{GROUP_ID}/items/{data.get('key', '')}"

Step 4: Run test to verify it passes

Run: python3 -m pytest zotero/test_ingest_to_wiki.py -k "authorship or source_url" -v
Expected: PASS

Step 5: Commit

git add zotero/ingest_to_wiki.py zotero/test_ingest_to_wiki.py
git commit -m "feat(zotero): authorship tagging + source URL selection"

Task 4: `build_meta` + `dump_frontmatter` + `render_record`

Files:

Modify: zotero/ingest_to_wiki.py
Test: zotero/test_ingest_to_wiki.py
Step 1: Write the failing test

def _sample_book():
    return {
        "key": "ABC123", "itemType": "book", "title": "Critical Path",
        "collections": ["YAAHI5VV"], "ISBN": "978-0",
        "abstractNote": "A book.", "url": "",
        "tags": [{"tag": "design science"}],
        "creators": [{"creatorType": "author", "firstName": "R. Buckminster",
                      "lastName": "Fuller"}],
    }

def test_build_meta_book():
    m = ing.build_meta(_sample_book(), ingested="2026-06-30")
    assert m["title"] == "Critical Path"
    assert m["type"] == "repos"
    assert m["zotero_key"] == "ABC123"
    assert m["held"] is True
    assert m["isbn"] == "978-0"
    assert "buckminster-fuller" in m["tags"]
    assert "design science" in m["tags"]
    assert "By Bucky" in m["zotero_collections"]
    assert m["source"] == "zotero://select/groups/5189518/items/ABC123"

def test_build_meta_wishlist_is_not_held():
    d = _sample_book(); d["collections"] = ["HLYVN3EP"]
    m = ing.build_meta(d, ingested="2026-06-30")
    assert m["held"] is False
    assert "wish-list" in m["tags"]

def test_dump_frontmatter_roundtrip_fields():
    m = ing.build_meta(_sample_book(), ingested="2026-06-30")
    fm = ing.dump_frontmatter(m)
    assert fm.startswith("---\n") and fm.rstrip().endswith("---")
    assert 'title: "Critical Path"' in fm
    assert "type: repos" in fm
    assert "zotero_key: ABC123" in fm
    assert "held: true" in fm

def test_render_record_has_body():
    m = ing.build_meta(_sample_book(), ingested="2026-06-30")
    rec = ing.render_record(m, "## Notes\n\nbody text")
    assert rec.startswith("---\n")
    assert "# Critical Path" in rec
    assert "body text" in rec

Step 2: Run test to verify it fails

Run: python3 -m pytest zotero/test_ingest_to_wiki.py -k "build_meta or frontmatter or render" -v
Expected: FAIL — attribute errors

Step 3: Write minimal implementation

def _trim(text, limit=500):
    text = " ".join((text or "").split())
    return text[:limit].rstrip() + ("…" if len(text) > limit else "")


def build_meta(data, ingested):
    cols = set(data.get("collections", []))
    raw_type = classify(data)
    tags = [authorship_tag(data)]
    tags += [t["tag"] for t in data.get("tags", []) if t.get("tag")]
    if WISH_LIST_KEY in cols:
        tags.append("wish-list")
    summary = _trim(data.get("abstractNote", "")) or \
        f"{data.get('itemType', 'item')} from the Buckyverse Zotero collection."
    meta = {
        "title": data.get("title") or "(untitled)",
        "source": pick_source_url(data),
        "type": raw_type,
        "ingested": ingested,
        "tags": tags,
        "summary": summary,
        "zotero_key": data.get("key", ""),
        "zotero_collections": [KEY_TO_NAME[k] for k in cols if k in KEY_TO_NAME],
        "held": WISH_LIST_KEY not in cols,
        "fetch_status": "metadata-only",
    }
    if data.get("ISBN"):
        meta["isbn"] = data["ISBN"]
    if data.get("DOI"):
        meta["doi"] = data["DOI"]
    return meta


def _yaml_scalar(v):
    if isinstance(v, bool):
        return "true" if v else "false"
    s = str(v)
    if s == "" or re.search(r'[:#"\']', s) or s != s.strip():
        return '"' + s.replace('"', '\\"') + '"'
    return s


def dump_frontmatter(meta):
    order = ["title", "source", "type", "ingested", "tags", "summary",
             "zotero_key", "zotero_collections", "held", "isbn", "doi",
             "fetch_status"]
    lines = ["---"]
    for k in order:
        if k not in meta:
            continue
        v = meta[k]
        if isinstance(v, list):
            inner = ", ".join(_yaml_scalar(x) for x in v)
            lines.append(f"{k}: [{inner}]")
        elif k in ("title", "summary", "source"):
            lines.append(f'{k}: "{str(v).replace(chr(34), chr(39))}"')
        else:
            lines.append(f"{k}: {_yaml_scalar(v)}")
    lines.append("---")
    return "\n".join(lines) + "\n"


def render_record(meta, body=""):
    out = dump_frontmatter(meta) + "\n# " + meta["title"] + "\n"
    if meta.get("summary"):
        out += "\n> " + meta["summary"] + "\n"
    if body:
        out += "\n" + body.rstrip() + "\n"
    return out

Step 4: Run test to verify it passes

Run: python3 -m pytest zotero/test_ingest_to_wiki.py -k "build_meta or frontmatter or render" -v
Expected: PASS

Step 5: Commit

git add zotero/ingest_to_wiki.py zotero/test_ingest_to_wiki.py
git commit -m "feat(zotero): frontmatter + record rendering"

Task 5: Dedup index scan + `resolve_action`

Files:

Modify: zotero/ingest_to_wiki.py
Test: zotero/test_ingest_to_wiki.py
Step 1: Write the failing test

import pathlib

def test_scan_existing_indexes_keys_and_slugs(tmp_path):
    raw = tmp_path / ".wiki" / "raw" / "repos"
    raw.mkdir(parents=True)
    (raw / "critical-path.md").write_text(
        '---\ntitle: "Critical Path"\nzotero_key: OLDKEY\n---\n# Critical Path\n')
    # a repo-root book dir
    book = tmp_path / "grunch_of_giants"
    book.mkdir()
    (book / "index.md").write_text("# Grunch")
    key2path, slug2path = ing.scan_existing(str(tmp_path))
    assert key2path["OLDKEY"].endswith("repos/critical-path.md")
    assert "critical-path" in slug2path
    assert "grunch-of-giants" in slug2path

def test_resolve_action_enrich_by_key(tmp_path):
    key2path = {"K1": ".wiki/raw/repos/x.md"}
    slug2path = {}
    action, path = ing.resolve_action(
        "K1", "x", "repos", "2026-06-30", key2path, slug2path)
    assert action == "enrich" and path == ".wiki/raw/repos/x.md"

def test_resolve_action_enrich_by_slug():
    action, path = ing.resolve_action(
        "K9", "critical-path", "repos", "2026-06-30",
        {}, {"critical-path": ".wiki/raw/repos/critical-path.md"})
    assert action == "enrich"

def test_resolve_action_create_when_no_match():
    action, path = ing.resolve_action(
        "K9", "new-thing", "articles", "2026-06-30", {}, {})
    assert action == "create"
    assert path == ".wiki/raw/articles/2026-06-30-new-thing.md"

Step 2: Run test to verify it fails

Run: python3 -m pytest zotero/test_ingest_to_wiki.py -k "scan_existing or resolve_action" -v
Expected: FAIL — attribute errors

Step 3: Write minimal implementation

import pathlib

_FM_KEY_RE = re.compile(r"^zotero_key:\s*(\S+)\s*$", re.M)


def _read_frontmatter_key(path):
    try:
        text = pathlib.Path(path).read_text(encoding="utf-8", errors="ignore")
    except OSError:
        return None
    if not text.startswith("---"):
        return None
    end = text.find("\n---", 3)
    fm = text[:end] if end != -1 else text
    m = _FM_KEY_RE.search(fm)
    return m.group(1) if m else None


def scan_existing(root="."):
    """Return (zotero_key -> path, slug -> path) for existing raw records and
    repo-root book dirs, all paths relative to *root*."""
    root = pathlib.Path(root)
    key2path, slug2path = {}, {}
    raw_root = root / RAW_DIR
    if raw_root.is_dir():
        for p in raw_root.rglob("*.md"):
            if p.name == "_index.md":
                continue
            rel = str(p.relative_to(root))
            stem = p.stem
            # articles are prefixed with an ISO date; strip it for slug match
            stem = re.sub(r"^\d{4}-\d{2}-\d{2}-", "", stem)
            slug2path.setdefault(stem, rel)
            k = _read_frontmatter_key(p)
            if k:
                key2path[k] = rel
    for idx in root.glob("*/index.md"):
        if str(idx).startswith(str(raw_root)):
            continue
        slug = slugify(idx.parent.name)
        slug2path.setdefault(slug, str(idx.relative_to(root)))
    return key2path, slug2path


def resolve_action(key, slug, raw_type, ingested, key2path, slug2path):
    if key and key in key2path:
        return "enrich", key2path[key]
    if slug in slug2path:
        return "enrich", slug2path[slug]
    return "create", target_path(raw_type, slug, ingested)

Step 4: Run test to verify it passes

Run: python3 -m pytest zotero/test_ingest_to_wiki.py -k "scan_existing or resolve_action" -v
Expected: PASS

Step 5: Commit

git add zotero/ingest_to_wiki.py zotero/test_ingest_to_wiki.py
git commit -m "feat(zotero): dedup index scan + create/enrich resolution"

Task 6: `enrich_record` (inject missing provenance keys)

Files:

Modify: zotero/ingest_to_wiki.py
Test: zotero/test_ingest_to_wiki.py
Step 1: Write the failing test

def test_enrich_adds_missing_keys_only(tmp_path):
    p = tmp_path / "rec.md"
    p.write_text('---\ntitle: "X"\nsource: "s"\n---\n# X\n\nbody\n')
    changed = ing.enrich_record(str(p),
                                {"zotero_key": "K1", "isbn": "978"})
    assert changed is True
    text = p.read_text()
    assert "zotero_key: K1" in text
    assert "isbn: 978" in text
    assert text.strip().endswith("body")          # body preserved
    assert text.count("title:") == 1              # not duplicated

def test_enrich_is_idempotent(tmp_path):
    p = tmp_path / "rec.md"
    p.write_text('---\ntitle: "X"\nzotero_key: K1\n---\n# X\n')
    changed = ing.enrich_record(str(p), {"zotero_key": "K1"})
    assert changed is False
    assert p.read_text().count("zotero_key:") == 1

Step 2: Run test to verify it fails

Run: python3 -m pytest zotero/test_ingest_to_wiki.py -k "enrich" -v
Expected: FAIL — AttributeError

Step 3: Write minimal implementation

def enrich_record(path, additions):
    """Add any of *additions* whose key is absent from the file's frontmatter.
    Never overwrites existing keys or the body. Returns True if it changed."""
    p = pathlib.Path(path)
    text = p.read_text(encoding="utf-8")
    if not text.startswith("---"):
        return False
    end = text.find("\n---", 3)
    if end == -1:
        return False
    fm = text[4:end]
    rest = text[end:]                       # starts with "\n---"
    existing = set(re.findall(r"^([A-Za-z0-9_]+):", fm, re.M))
    new_lines = []
    for k, v in additions.items():
        if k in existing:
            continue
        if isinstance(v, list):
            new_lines.append(f"{k}: [{', '.join(_yaml_scalar(x) for x in v)}]")
        else:
            new_lines.append(f"{k}: {_yaml_scalar(v)}")
    if not new_lines:
        return False
    fm = fm.rstrip("\n") + "\n" + "\n".join(new_lines) + "\n"
    p.write_text("---\n" + fm + rest, encoding="utf-8")
    return True

Step 4: Run test to verify it passes

Run: python3 -m pytest zotero/test_ingest_to_wiki.py -k "enrich" -v
Expected: PASS

Step 5: Commit

git add zotero/ingest_to_wiki.py zotero/test_ingest_to_wiki.py
git commit -m "feat(zotero): enrich existing records with provenance keys"

Task 7: HTML → text extraction + PDF best-effort

Files:

Modify: zotero/ingest_to_wiki.py
Test: zotero/test_ingest_to_wiki.py
Step 1: Write the failing test

def test_extract_text_from_html_strips_tags_and_scripts():
    html = ("<html><head><style>x{}</style><script>bad()</script></head>"
            "<body><h1>Title</h1><p>Hello&nbsp;world</p></body></html>")
    out = ing.extract_text_from_html(html)
    assert "Title" in out and "Hello world" in out
    assert "bad()" not in out and "x{}" not in out

def test_extract_text_from_html_truncates():
    html = "<p>" + ("z " * 5000) + "</p>"
    assert len(ing.extract_text_from_html(html, maxchars=100)) <= 100

Step 2: Run test to verify it fails

Run: python3 -m pytest zotero/test_ingest_to_wiki.py -k "extract_text_from_html" -v
Expected: FAIL — AttributeError

Step 3: Write minimal implementation

import html as _htmllib
import subprocess
import tempfile
from html.parser import HTMLParser


class _TextExtractor(HTMLParser):
    _SKIP = {"script", "style", "head", "noscript"}

    def __init__(self):
        super().__init__()
        self._skip_depth = 0
        self.parts = []

    def handle_starttag(self, tag, attrs):
        if tag in self._SKIP:
            self._skip_depth += 1

    def handle_endtag(self, tag):
        if tag in self._SKIP and self._skip_depth:
            self._skip_depth -= 1

    def handle_data(self, data):
        if self._skip_depth == 0 and data.strip():
            self.parts.append(data.strip())


def extract_text_from_html(html_text, maxchars=20000):
    parser = _TextExtractor()
    try:
        parser.feed(html_text)
    except Exception:
        pass
    text = _htmllib.unescape(" ".join(parser.parts))
    text = re.sub(r"\s+", " ", text).strip()
    return text[:maxchars]


def pdf_to_text(pdf_bytes, maxchars=20000):
    """Best-effort PDF text via the `pdftotext` CLI; returns '' on failure."""
    try:
        with tempfile.NamedTemporaryFile(suffix=".pdf") as f:
            f.write(pdf_bytes)
            f.flush()
            out = subprocess.run(
                ["pdftotext", "-q", f.name, "-"],
                capture_output=True, timeout=60)
        text = out.stdout.decode("utf-8", errors="ignore")
        return re.sub(r"\s+", " ", text).strip()[:maxchars]
    except Exception:
        return ""

Step 4: Run test to verify it passes

Run: python3 -m pytest zotero/test_ingest_to_wiki.py -k "extract_text_from_html" -v
Expected: PASS

Step 5: Commit

git add zotero/ingest_to_wiki.py zotero/test_ingest_to_wiki.py
git commit -m "feat(zotero): html and pdf text extraction"

Task 8: `ZoteroClient` (local API) with injectable opener

Files:

Modify: zotero/ingest_to_wiki.py
Test: zotero/test_ingest_to_wiki.py
Step 1: Write the failing test

import json as _json

class _FakeResp:
    def __init__(self, body, headers=None):
        self._body = body
        self.headers = headers or {}
    def read(self):
        return self._body
    def __enter__(self):
        return self
    def __exit__(self, *a):
        return False

def _make_opener(routes):
    """routes: dict url-substring -> (body_bytes, headers)."""
    def _open(req, timeout=None):
        url = req.full_url if hasattr(req, "full_url") else req
        for frag, (body, headers) in routes.items():
            if frag in url:
                return _FakeResp(body, headers)
        raise AssertionError("unexpected url: " + url)
    return _open

def test_client_paginates_collection_items():
    page1 = _json.dumps([{"data": {"key": f"K{i}"}} for i in range(100)]).encode()
    page2 = _json.dumps([{"data": {"key": "K100"}}]).encode()
    calls = {"n": 0}
    def _open(req, timeout=None):
        calls["n"] += 1
        return _FakeResp(page1 if calls["n"] == 1 else page2,
                         {"Total-Results": "101"})
    client = ing.ZoteroClient(opener=_open)
    items = client.collection_items_top("ABC")
    assert len(items) == 101
    assert items[-1]["data"]["key"] == "K100"

def test_client_children():
    body = _json.dumps([{"data": {"key": "C1", "itemType": "attachment"}}]).encode()
    client = ing.ZoteroClient(opener=_make_opener({"/children": (body, {})}))
    kids = client.item_children("K1")
    assert kids[0]["data"]["key"] == "C1"

Step 2: Run test to verify it fails

Run: python3 -m pytest zotero/test_ingest_to_wiki.py -k "client" -v
Expected: FAIL — AttributeError: ... 'ZoteroClient'

Step 3: Write minimal implementation

import json
import urllib.request


def _default_opener(req, timeout=None):
    return urllib.request.urlopen(req, timeout=timeout)


class ZoteroClient:
    def __init__(self, base=API_BASE, group_id=GROUP_ID, opener=None, timeout=30):
        self.base = base
        self.group_id = group_id
        self.opener = opener or _default_opener
        self.timeout = timeout

    def _get(self, path, params=None):
        url = f"{self.base}/groups/{self.group_id}{path}"
        if params:
            from urllib.parse import urlencode
            url += "?" + urlencode(params)
        req = urllib.request.Request(url)
        req.add_header("Zotero-API-Version", "3")
        with self.opener(req, timeout=self.timeout) as resp:
            body = resp.read()
            total = resp.headers.get("Total-Results")
        return json.loads(body.decode("utf-8")), total

    def collection_items_top(self, collection_key):
        items, start, limit = [], 0, 100
        while True:
            batch, _ = self._get(
                f"/collections/{collection_key}/items/top",
                {"format": "json", "limit": limit, "start": start})
            if not batch:
                break
            items.extend(batch)
            if len(batch) < limit:
                break
            start += limit
        return items

    def item_children(self, item_key):
        kids, _ = self._get(f"/items/{item_key}/children", {"format": "json"})
        return kids

    def item_file(self, item_key):
        url = f"{self.base}/groups/{self.group_id}/items/{item_key}/file"
        req = urllib.request.Request(url)
        req.add_header("Zotero-API-Version", "3")
        with self.opener(req, timeout=self.timeout) as resp:
            return resp.read()

Step 4: Run test to verify it passes

Run: python3 -m pytest zotero/test_ingest_to_wiki.py -k "client" -v
Expected: PASS

Step 5: Commit

git add zotero/ingest_to_wiki.py zotero/test_ingest_to_wiki.py
git commit -m "feat(zotero): local API client with pagination"

Task 9: Fetch-content helper + body assembly

Files:

Modify: zotero/ingest_to_wiki.py
Test: zotero/test_ingest_to_wiki.py

This wraps attachment + live-URL fetching into one function that returns
(body, fetch_status) and never raises.

Step 1: Write the failing test

def test_fetch_body_attachment_html(monkeypatch):
    html = b"<html><body><p>Attached text here</p></body></html>"
    class C:
        def item_children(self, k):
            return [{"data": {"key": "C1", "itemType": "attachment",
                              "contentType": "text/html", "linkMode": "imported_url"}}]
        def item_file(self, k):
            return html
    data = {"key": "K1", "itemType": "webpage"}
    meta = {"numChildren": 1}
    body, status = ing.fetch_body(C(), data, meta, fetch=True, url_fetcher=None)
    assert "Attached text here" in body
    assert status == "ok"

def test_fetch_body_no_fetch_returns_metadata_only():
    body, status = ing.fetch_body(None, {"key": "K1"}, {"numChildren": 0},
                                  fetch=False, url_fetcher=None)
    assert body == ""
    assert status == "metadata-only"

def test_fetch_body_live_url(monkeypatch):
    data = {"key": "K1", "itemType": "webpage", "url": "http://example.com"}
    def fake_url_fetcher(url):
        return "<p>Live page body</p>"
    body, status = ing.fetch_body(None, data, {"numChildren": 0},
                                  fetch=True, url_fetcher=fake_url_fetcher)
    assert "Live page body" in body
    assert status == "ok"

def test_fetch_body_failure_is_nonfatal():
    data = {"key": "K1", "itemType": "webpage", "url": "http://example.com"}
    def boom(url):
        raise RuntimeError("network down")
    body, status = ing.fetch_body(None, data, {"numChildren": 0},
                                  fetch=True, url_fetcher=boom)
    assert status in ("failed", "metadata-only")

Step 2: Run test to verify it fails

Run: python3 -m pytest zotero/test_ingest_to_wiki.py -k "fetch_body" -v
Expected: FAIL — AttributeError

Step 3: Write minimal implementation

def default_url_fetcher(url, timeout=20):
    req = urllib.request.Request(url, headers={"User-Agent": "buckyverse-ingest/1.0"})
    with urllib.request.urlopen(req, timeout=timeout) as resp:
        ctype = resp.headers.get("Content-Type", "")
        raw = resp.read()
    if "pdf" in ctype:
        return None  # binary; handled via attachments, not live fetch
    return raw.decode("utf-8", errors="ignore")


def fetch_body(client, data, meta, fetch, url_fetcher):
    """Return (markdown_body, fetch_status). Never raises."""
    if not fetch:
        return "", "metadata-only"
    parts, got, failed = [], False, False

    # 1) attachments
    if meta.get("numChildren", 0) and client is not None:
        try:
            for child in client.item_children(data["key"]):
                cd = child.get("data", {})
                if cd.get("itemType") != "attachment":
                    continue
                ctype = cd.get("contentType", "")
                try:
                    blob = client.item_file(cd["key"])
                except Exception:
                    failed = True
                    continue
                if "pdf" in ctype:
                    txt = pdf_to_text(blob)
                elif "html" in ctype or "text" in ctype:
                    txt = extract_text_from_html(blob.decode("utf-8", "ignore"))
                else:
                    txt = ""
                if txt:
                    parts.append("## Full text (attachment)\n\n" + txt)
                    got = True
        except Exception:
            failed = True

    # 2) live URL
    if data.get("url") and url_fetcher is not None:
        try:
            html_text = url_fetcher(data["url"])
            if html_text:
                txt = extract_text_from_html(html_text)
                if txt:
                    parts.append(f"## Page content ({data['url']})\n\n" + txt)
                    got = True
        except Exception:
            failed = True

    if got:
        return "\n\n".join(parts), "ok"
    return "", ("failed" if failed else "metadata-only")

Step 4: Run test to verify it passes

Run: python3 -m pytest zotero/test_ingest_to_wiki.py -k "fetch_body" -v
Expected: PASS

Step 5: Commit

git add zotero/ingest_to_wiki.py zotero/test_ingest_to_wiki.py
git commit -m "feat(zotero): attachment + live-url body assembly"

Task 10: Index/report writers

Files:

Modify: zotero/ingest_to_wiki.py
Test: zotero/test_ingest_to_wiki.py
Step 1: Write the failing test

def test_write_report_counts(tmp_path):
    summary = {"created": 5, "enriched": 3, "skipped": 0, "failed": 1,
               "by_type": {"repos": 4, "articles": 4, "papers": 1},
               "unknown_types": ["podcast"], "date": "2026-06-30"}
    out = tmp_path / "report.md"
    ing.write_report(str(out), summary)
    text = out.read_text()
    assert "created: 5" in text or "Created: 5" in text or "**Created:** 5" in text
    assert "podcast" in text

def test_update_raw_index_stats(tmp_path):
    raw = tmp_path / ".wiki" / "raw"
    (raw / "repos").mkdir(parents=True)
    (raw / "articles").mkdir(parents=True)
    (raw / "repos" / "a.md").write_text("---\ntitle: a\n---\n")
    (raw / "articles" / "b.md").write_text("---\ntitle: b\n---\n")
    (raw / "_index.md").write_text(
        "# Raw Sources Index\n\nLast updated: 2026-01-01\n\n## Recent Changes\n\n- old\n")
    ing.update_raw_index(str(tmp_path), "2026-06-30",
                         note="Ingested Zotero collections")
    text = (raw / "_index.md").read_text()
    assert "2026-06-30" in text
    assert "Ingested Zotero collections" in text

Step 2: Run test to verify it fails

Run: python3 -m pytest zotero/test_ingest_to_wiki.py -k "report or raw_index" -v
Expected: FAIL — AttributeError

Step 3: Write minimal implementation

def write_report(path, summary):
    lines = [
        "# Zotero ingest report",
        "",
        f"Date: {summary.get('date', '')}",
        "",
        f"- **Created:** {summary.get('created', 0)}",
        f"- **Enriched:** {summary.get('enriched', 0)}",
        f"- **Skipped:** {summary.get('skipped', 0)}",
        f"- **Failed fetch:** {summary.get('failed', 0)}",
        "",
        "## By type",
        "",
    ]
    for t, n in sorted(summary.get("by_type", {}).items()):
        lines.append(f"- {t}: {n}")
    unknown = summary.get("unknown_types", [])
    if unknown:
        lines += ["", "## Unknown item types encountered", ""]
        lines += [f"- {u}" for u in sorted(set(unknown))]
    pathlib.Path(path).write_text("\n".join(lines) + "\n", encoding="utf-8")


def update_raw_index(root, date, note):
    idx = pathlib.Path(root) / RAW_DIR / "_index.md"
    if not idx.exists():
        return
    text = idx.read_text(encoding="utf-8")
    text = re.sub(r"Last updated: .*", f"Last updated: {date}", text, count=1)
    bullet = f"- {date}: {note}"
    if "## Recent Changes" in text and bullet not in text:
        text = text.replace("## Recent Changes\n\n",
                            f"## Recent Changes\n\n{bullet}\n", 1)
    idx.write_text(text, encoding="utf-8")

Step 4: Run test to verify it passes

Run: python3 -m pytest zotero/test_ingest_to_wiki.py -k "report or raw_index" -v
Expected: PASS

Step 5: Commit

git add zotero/ingest_to_wiki.py zotero/test_ingest_to_wiki.py
git commit -m "feat(zotero): index + report writers"

Task 11: `main()` orchestration + CLI

Files:

Modify: zotero/ingest_to_wiki.py
Test: zotero/test_ingest_to_wiki.py
Step 1: Write the failing test

def test_run_ingest_dry_run_plans_without_writing(tmp_path):
    # one new item, one that matches an existing slug -> enrich
    (tmp_path / ".wiki" / "raw" / "repos").mkdir(parents=True)
    (tmp_path / ".wiki" / "raw" / "articles").mkdir(parents=True)
    (tmp_path / ".wiki" / "raw" / "repos" / "critical-path.md").write_text(
        '---\ntitle: "Critical Path"\n---\n# Critical Path\n')
    items = [
        {"data": {"key": "K1", "itemType": "book", "title": "Critical Path",
                  "collections": ["YAAHI5VV"]}, "meta": {"numChildren": 0}},
        {"data": {"key": "K2", "itemType": "webpage", "title": "Fresh Page",
                  "collections": ["L84AV9YL"], "url": ""}, "meta": {"numChildren": 0}},
    ]
    summary = ing.run_ingest(items, root=str(tmp_path), today="2026-06-30",
                             fetch=False, dry_run=True)
    assert summary["created"] == 1
    assert summary["enriched"] == 1
    # nothing written in dry-run
    assert not (tmp_path / ".wiki" / "raw" / "articles"
                / "2026-06-30-fresh-page.md").exists()

def test_run_ingest_writes_and_is_idempotent(tmp_path):
    (tmp_path / ".wiki" / "raw" / "articles").mkdir(parents=True)
    items = [{"data": {"key": "K2", "itemType": "webpage", "title": "Fresh Page",
                       "collections": ["L84AV9YL"]}, "meta": {"numChildren": 0}}]
    s1 = ing.run_ingest(items, root=str(tmp_path), today="2026-06-30",
                        fetch=False, dry_run=False)
    assert s1["created"] == 1
    out = tmp_path / ".wiki" / "raw" / "articles" / "2026-06-30-fresh-page.md"
    assert out.exists()
    assert "zotero_key: K2" in out.read_text()
    # second run: matched by key -> enrich (0 created)
    s2 = ing.run_ingest(items, root=str(tmp_path), today="2026-06-30",
                        fetch=False, dry_run=False)
    assert s2["created"] == 0

Step 2: Run test to verify it fails

Run: python3 -m pytest zotero/test_ingest_to_wiki.py -k "run_ingest" -v
Expected: FAIL — AttributeError

Step 3: Write minimal implementation

import argparse
import os


def run_ingest(items, root, today, fetch, dry_run, client=None,
               url_fetcher=None, limit=None):
    key2path, slug2path = scan_existing(root)
    summary = {"created": 0, "enriched": 0, "skipped": 0, "failed": 0,
               "by_type": {}, "unknown_types": [], "date": today}
    seen = set()
    for item in items:
        data = item.get("data", item)
        meta_info = item.get("meta", {})
        key = data.get("key", "")
        if key in seen:
            continue
        seen.add(key)
        if limit is not None and (summary["created"] + summary["enriched"]) >= limit:
            break
        raw_type = classify(data)
        if is_unknown_type(data):
            summary["unknown_types"].append(data.get("itemType", "?"))
        slug = slugify(data.get("title", "") or key)
        action, path = resolve_action(key, slug, raw_type, today,
                                      key2path, slug2path)
        summary["by_type"][raw_type] = summary["by_type"].get(raw_type, 0) + 1
        if action == "enrich":
            additions = {"zotero_key": key}
            if data.get("ISBN"):
                additions["isbn"] = data["ISBN"]
            if data.get("DOI"):
                additions["doi"] = data["DOI"]
            if not dry_run:
                enrich_record(os.path.join(root, path), additions)
            summary["enriched"] += 1
            continue
        # create
        m = build_meta(data, ingested=today)
        body, status = fetch_body(client, data, meta_info, fetch=fetch,
                                  url_fetcher=url_fetcher)
        m["fetch_status"] = status
        if status == "failed":
            summary["failed"] += 1
        if not dry_run:
            full = os.path.join(root, path)
            os.makedirs(os.path.dirname(full), exist_ok=True)
            pathlib.Path(full).write_text(render_record(m, body),
                                          encoding="utf-8")
            key2path[key] = path
            slug2path.setdefault(slug, path)
        summary["created"] += 1
    return summary


def fetch_all_collections(client):
    items, seen = [], set()
    for name, key in COLLECTIONS:
        for it in client.collection_items_top(key):
            k = it["data"]["key"]
            if k in seen:
                continue
            seen.add(k)
            items.append(it)
    return items


def main(argv=None):
    ap = argparse.ArgumentParser(description="Ingest Buckyverse Zotero collections.")
    ap.add_argument("--dry-run", action="store_true")
    ap.add_argument("--no-fetch", action="store_true",
                    help="skip attachment/live-url fetching")
    ap.add_argument("--limit", type=int, default=None)
    ap.add_argument("--root", default=".")
    ap.add_argument("--today", default=None)
    args = ap.parse_args(argv)

    today = args.today or __import__("datetime").date.today().isoformat()
    client = ZoteroClient()
    try:
        items = fetch_all_collections(client)
    except Exception as e:
        print(f"ERROR: cannot reach local Zotero API at {API_BASE} "
              f"(is Zotero running with the local API enabled?): {e}")
        return 2
    print(f"Fetched {len(items)} unique items from {len(COLLECTIONS)} collections.")

    if not args.dry_run:
        pathlib.Path(os.path.join(args.root, SNAPSHOT_PATH)).write_text(
            json.dumps(items, indent=2), encoding="utf-8")

    summary = run_ingest(
        items, root=args.root, today=today,
        fetch=not args.no_fetch, dry_run=args.dry_run,
        client=client, url_fetcher=default_url_fetcher, limit=args.limit)

    if not args.dry_run:
        update_raw_index(args.root, today,
                         note=f"Ingested {summary['created']} Zotero collection "
                              f"items ({summary['enriched']} enriched) → raw")
        write_report(os.path.join(args.root, REPORT_PATH), summary)

    print(f"created={summary['created']} enriched={summary['enriched']} "
          f"failed={summary['failed']} by_type={summary['by_type']}")
    if summary["unknown_types"]:
        print("unknown types:", sorted(set(summary["unknown_types"])))
    return 0


if __name__ == "__main__":
    raise SystemExit(main())

Step 4: Run the full test suite

Run: python3 -m pytest zotero/test_ingest_to_wiki.py -v
Expected: PASS (all tests)

Step 5: Commit

git add zotero/ingest_to_wiki.py zotero/test_ingest_to_wiki.py
git commit -m "feat(zotero): main orchestration + CLI"

Task 12: Live dry-run, real run, idempotency, lint

Files:

Generated: .wiki/raw/**, zotero/buckyverse_collections.json, zotero/last-ingest-report.md
Step 1: Dry-run against the live local API

Run: python3 zotero/ingest_to_wiki.py --dry-run
Expected: prints Fetched N unique items … then created=… enriched=… ….
Sanity-check: created + enriched ≈ number of distinct curated items; created should be far less than the ~1170 group library (most By-Bucky books enrich existing repos). If "cannot reach local Zotero API", start Zotero / enable Preferences → Advanced → "Allow other applications…", then retry.

Step 2: Real run

Run: python3 zotero/ingest_to_wiki.py
Expected: writes zotero/buckyverse_collections.json, new .wiki/raw/** records, updates .wiki/raw/_index.md, writes zotero/last-ingest-report.md.

Step 3: Idempotency check (second run creates nothing)

Run: python3 zotero/ingest_to_wiki.py
Expected: created=0 (only enriched=…). This is the primary correctness signal.

Step 4: Spot-check generated records

Run:

ls .wiki/raw/repos | wc -l
ls .wiki/raw/articles | tail -5
sed -n '1,30p' "$(ls .wiki/raw/articles/2026-06-30-* | head -1)"

Expected: frontmatter has zotero_key, correct type, held, tags; People encyclopedia entries appear under articles; a Wish List item shows held: false and wish-list tag.

Step 5: Lint the wiki

Run the /wiki:lint skill.
Expected: no new broken indexes/links from the ingest. Fix any flagged issues, re-run.

Step 6: Commit generated content

git add .wiki/raw zotero/buckyverse_collections.json zotero/last-ingest-report.md
git commit -m "data(zotero): ingest Buckyverse collections into .wiki/raw"

Self-Review

Spec coverage:

Source/scope (live local API, 5 collections) → Task 8 (collection_items_top), Task 11 (COLLECTIONS, fetch_all_collections). ✓
Audit snapshot JSON → Task 11 main(). ✓
Type → raw mapping → Tasks 2, 4. ✓
Frontmatter incl. zotero_key/held/wish-list/isbn/doi → Task 4. ✓
Dedup (key then slug; enrich vs create; repo-root dirs) → Tasks 5, 6, 11. ✓
Idempotency (0 created on re-run) → Tasks 6, 11 + Task 12 step 3. ✓
Full text (attachments + live URL, non-fatal) → Tasks 7, 9. ✓
Indexes + report → Task 10. ✓
Error handling (API unreachable stop; per-item non-fatal) → Tasks 9, 11. ✓
Verification (dry-run, idempotency, spot-check, lint) → Task 12. ✓
Out-of-scope items (no group-library, no topic compile, no zotero/index.md edits, read-only) → respected; script only reads Zotero and writes raw. ✓

Placeholder scan: none — every code step contains full implementation.

Type consistency: slugify, classify, target_path, build_meta, dump_frontmatter, render_record, scan_existing, resolve_action, enrich_record, extract_text_from_html, pdf_to_text, ZoteroClient.{collection_items_top,item_children,item_file}, fetch_body, write_report, update_raw_index, run_ingest, fetch_all_collections, main — names are used consistently across tasks. fetch_body signature (client, data, meta, fetch, url_fetcher) matches its call in run_ingest. ✓

Zotero Collections Ingest Implementation Plan

Reference constants (used across tasks — keep identical)

Task 1: Module scaffold + slugify

Task 2: classify + is_unknown_type + target_path

Task 3: authorship_tag + pick_source_url

Task 4: build_meta + dump_frontmatter + render_record

Task 5: Dedup index scan + resolve_action

Task 6: enrich_record (inject missing provenance keys)

Task 7: HTML → text extraction + PDF best-effort

Task 8: ZoteroClient (local API) with injectable opener

Task 9: Fetch-content helper + body assembly

Task 10: Index/report writers

Task 11: main() orchestration + CLI

Task 12: Live dry-run, real run, idempotency, lint

Self-Review

Task 1: Module scaffold + `slugify`

Task 2: `classify` + `is_unknown_type` + `target_path`

Task 3: `authorship_tag` + `pick_source_url`

Task 4: `build_meta` + `dump_frontmatter` + `render_record`

Task 5: Dedup index scan + `resolve_action`

Task 6: `enrich_record` (inject missing provenance keys)

Task 8: `ZoteroClient` (local API) with injectable opener

Task 11: `main()` orchestration + CLI