julia/osdb-1-claude-opusplan
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: import existing publications from XML (story #05)

Browse Source 
scripts/import_publications.py imports all osw=1 publications from
03-data/osws.xml — upserts pub rows, loads cover images, scans
03-data/osdb/{pubId}/ for fruit images and PDFs, links matched fruits
by osdb_number. Idempotent re-runs. 17 unit tests. Makefile updated
to include import_publications_test in make test. Added
scripts/README_import.md with run order. Added spec-first rule to
.claude/CLAUDE.md.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
julia
2026-06-18 11:47:45 +02:00
co-authored by Claude Sonnet 4.6
parent dfa3d91ab9
commit e8d48d4fcb
6 changed files with 501 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.claude/CLAUDE.md
+11
View File
@@ -0,0 +1,11 @@
## Taiga
- TAIGA_URL: https://taiga.db-extern.de
- TAIGA_PROJECT_SLUG: obstsortendatenbank
## Stories
Before implementing a story: read its spec file in docs/planning/specs/ first.
## Branching
Stories developed sequentially; prior story branch may not be merged to main yet.
Before creating feature branch: run `git branch -r | grep feature/` — if predecessor
story branch exists on remote and is unmerged, base new branch on it, not main.
Makefile
+1 -1
View File
@@ -34,7 +34,7 @@ run-frontend:
test:
cd backend && go test ./...
cd frontend && npm run test -- --run
cd scripts && python3 -m unittest import_fruits_test -v
cd scripts && python3 -m unittest import_fruits_test import_publications_test -v
## fmt: format all code
fmt:
README.md
+1 -1
View File
@@ -6,7 +6,7 @@ A full-stack fruit-variety database (Go + Vue 3).
- Users can view a "Hello, OSDB!" landing page that confirms the backend is reachable via the Vite proxy.
- Users can create, view, edit, and delete fruit varieties, including managing synonyms and uploading images.
- Administrators can bulk-import the legacy fruit database from XML using `scripts/import_fruits.py`.
- Administrators can bulk-import the legacy fruit database from XML using `scripts/import_fruits.py`, and then import all publications (cover images, linked fruits, PDFs, fruit images) using `scripts/import_publications.py`.
- Users can manage publications (books, catalogues) with cover images, linked fruits, per-fruit PDF descriptions, and fruit images; fruit detail pages show publication descriptions and images.
## Quick Start
scripts/README_import.md
+37
View File
@@ -0,0 +1,37 @@
# Import Scripts
## Prerequisites
- Postgres running and migrated (`make migrate-up`)
- `DATABASE_URL` set, e.g.:
```
export DATABASE_URL=postgres://user:pass@localhost:5432/osdb
```
- `03-data/` directory present at repo root
- `psycopg2` installed (`pip install psycopg2-binary`)
---
## Run Order
Scripts must be run in order — publications import depends on fruits being present.
### 1. Import fruits
```bash
DATABASE_URL=... python3 scripts/import_fruits.py
```
Imports fruits, synonyms, and fruit images from `03-data/obstsorten.xml`.
Idempotent: upserts fruits by `osdb_number`.
### 2. Import publications
```bash
DATABASE_URL=... python3 scripts/import_publications.py
```
Imports publications from `03-data/osws.xml` (only `<obj>` elements with `<osw>=1`).
For each publication: upserts the row, loads cover image, links fruits found by
filesystem scan of `03-data/osdb/{pubId}/`, and imports PDFs and fruit images.
Idempotent: clears and re-inserts linked data on re-run.
scripts/import_publications.py
+215
View File
@@ -0,0 +1,215 @@
#!/usr/bin/env python3
"""
Import publications from 03-data/osws.xml into the OSDB database.
Usage:
DATABASE_URL=postgres://... python3 scripts/import_publications.py
Idempotent: upserts publication by osdb_pub_id; deletes and re-inserts
linked fruits, descriptions, and fruit images per publication on re-run.
"""
import os
import re
import sys
import xml.etree.ElementTree as ET
from pathlib import Path
import psycopg2
DATA_DIR = os.path.join(os.path.dirname(os.path.abspath(__file__)), "..", "03-data")
def parse_publications(xml_path: str) -> list:
"""
Parse osws.xml and return list of publication dicts for <obj> elements with <osw>=1.
Each dict: id, title, author (None if "."), img_path (relative to data root, or None).
"""
tree = ET.parse(xml_path)
root = tree.getroot()
pubs = []
for obj in root.findall("obj"):
if (obj.findtext("osw") or "").strip() != "1":
continue
author_raw = (obj.findtext("author") or "").strip()
img = obj.findtext("img")
pubs.append({
"id": (obj.findtext("id") or "").strip(),
"title": (obj.findtext("name") or "").strip(),
"author": author_raw if author_raw and author_raw != "." else None,
"img_path": img.strip() if img else None,
})
return pubs
def scan_pub_dir(pub_dir: str, pub_id: str) -> dict:
"""
Scan pub_dir for files matching {osdb_number}_{pub_id}_s0.jpg and {osdb_number}_{pub_id}.pdf.
Returns dict mapping osdb_number → {img_path?: str, pdf_path?: str}.
Ignores _tn.jpg thumbnails and any other files.
"""
d = Path(pub_dir)
if not d.is_dir():
return {}
result = {}
img_re = re.compile(rf"^(.+)_{re.escape(pub_id)}_s0\.jpg$")
pdf_re = re.compile(rf"^(.+)_{re.escape(pub_id)}\.pdf$")
for f in d.iterdir():
m = img_re.match(f.name)
if m:
result.setdefault(m.group(1), {})["img_path"] = str(f)
continue
m = pdf_re.match(f.name)
if m:
result.setdefault(m.group(1), {})["pdf_path"] = str(f)
return result
def import_publications(conn, data_dir: str = None) -> dict:
"""
Import all publications from osws.xml into the database.
Returns counts: publications, covers, fruits_linked, pdfs, images, skipped_fruits.
"""
if data_dir is None:
data_dir = DATA_DIR
data_path = Path(data_dir)
pubs = parse_publications(str(data_path / "osws.xml"))
counts = {
"publications": 0,
"covers": 0,
"fruits_linked": 0,
"pdfs": 0,
"images": 0,
"skipped_fruits": 0,
}
with conn.cursor() as cur:
for pub in pubs:
pub_id = pub["id"]
# Load cover image bytes (skip silently if file missing or unreadable)
cover_data = None
if pub["img_path"]:
cover_path = data_path / pub["img_path"]
try:
cover_data = cover_path.read_bytes()
counts["covers"] += 1
except OSError:
pass
# Upsert publication row
cur.execute(
"""
INSERT INTO publications (title, author, osdb_pub_id, image_data, updated_at)
VALUES (%s, %s, %s, %s, NOW())
ON CONFLICT (osdb_pub_id) DO UPDATE
SET title = EXCLUDED.title,
author = EXCLUDED.author,
image_data = EXCLUDED.image_data,
updated_at = NOW()
RETURNING id
""",
(
pub["title"],
pub["author"],
pub_id,
psycopg2.Binary(cover_data) if cover_data else None,
),
)
db_pub_id = cur.fetchone()[0]
counts["publications"] += 1
# Scan filesystem for linked fruits
pub_dir = str(data_path / "osdb" / pub_id)
fruit_files = scan_pub_dir(pub_dir, pub_id)
# Resolve osdb_numbers → DB fruit IDs; warn and skip unknowns
linked = {}
for osdb_number, files in fruit_files.items():
cur.execute("SELECT id FROM fruits WHERE osdb_number = %s", (osdb_number,))
row = cur.fetchone()
if row is None:
print(
f" WARNING: osdb_number not in fruits table, skipping: {osdb_number}",
file=sys.stderr,
)
counts["skipped_fruits"] += 1
continue
linked[osdb_number] = {"db_id": row[0], **files}
# Idempotent: clear previous linked data for this publication
cur.execute("DELETE FROM publication_fruit_images WHERE publication_id = %s", (db_pub_id,))
cur.execute("DELETE FROM publication_descriptions WHERE publication_id = %s", (db_pub_id,))
cur.execute("DELETE FROM publication_fruits WHERE publication_id = %s", (db_pub_id,))
# Re-insert linked fruits, PDFs, and images
for osdb_number, info in linked.items():
fruit_db_id = info["db_id"]
cur.execute(
"INSERT INTO publication_fruits (publication_id, fruit_id) VALUES (%s, %s)",
(db_pub_id, fruit_db_id),
)
counts["fruits_linked"] += 1
if "pdf_path" in info:
pdf_data = Path(info["pdf_path"]).read_bytes()
cur.execute(
"""
INSERT INTO publication_descriptions (publication_id, fruit_id, pdf_data)
VALUES (%s, %s, %s)
""",
(db_pub_id, fruit_db_id, psycopg2.Binary(pdf_data)),
)
counts["pdfs"] += 1
if "img_path" in info:
img_data = Path(info["img_path"]).read_bytes()
filename = Path(info["img_path"]).name
cur.execute(
"""
INSERT INTO publication_fruit_images
(publication_id, fruit_id, filename, data, image_type)
VALUES (%s, %s, %s, %s, 'fruit')
""",
(db_pub_id, fruit_db_id, filename, psycopg2.Binary(img_data)),
)
counts["images"] += 1
conn.commit()
return counts
def main():
database_url = os.environ.get("DATABASE_URL")
if not database_url:
print("Error: DATABASE_URL environment variable not set", file=sys.stderr)
sys.exit(1)
print("Connecting to database...")
conn = psycopg2.connect(database_url)
try:
print("Importing publications...")
counts = import_publications(conn)
print("\nDone.")
print(f" Publications upserted: {counts['publications']}")
print(f" Cover images loaded: {counts['covers']}")
print(f" Fruits linked: {counts['fruits_linked']}")
print(f" PDFs imported: {counts['pdfs']}")
print(f" Fruit images imported: {counts['images']}")
print(f" Fruits skipped: {counts['skipped_fruits']}")
except Exception as e:
print(f"Error: {e}", file=sys.stderr)
conn.rollback()
sys.exit(1)
finally:
conn.close()
if __name__ == "__main__":
main()
scripts/import_publications_test.py
+236
View File
@@ -0,0 +1,236 @@
import os
import tempfile
import unittest
from pathlib import Path
from unittest.mock import MagicMock, call, patch
from import_publications import parse_publications, scan_pub_dir, import_publications
MINIMAL_XML_STR = """<?xml version="1.0" encoding="iso-8859-1" ?>
<root>
<obj>
<id>skip_me</id>
<name>Not a publication</name>
<author>Nobody</author>
<osw>0</osw>
</obj>
<obj>
<id>ber</id>
<name>Bernisches Stammregister</name>
<author>.</author>
<img>osdb/ber/ber_s0.jpg</img>
<osw>1</osw>
</obj>
<obj>
<id>cal</id>
<name>Obst- und Beeren</name>
<author>Calwer</author>
<img>osdb/cal/cal_s0.jpg</img>
<osw>1</osw>
</obj>
<obj>
<id>noc</id>
<name>No Cover</name>
<author>Someone</author>
<osw>1</osw>
</obj>
</root>
"""
MINIMAL_XML = MINIMAL_XML_STR.encode("iso-8859-1")
def _write_xml(tmp_dir, content=MINIMAL_XML):
p = Path(tmp_dir) / "osws.xml"
p.write_bytes(content)
return str(p)
class TestParsePublications(unittest.TestCase):
def setUp(self):
self.tmp = tempfile.mkdtemp()
self.xml_path = _write_xml(self.tmp)
def test_excludes_osw_not_1(self):
pubs = parse_publications(self.xml_path)
ids = [p["id"] for p in pubs]
self.assertNotIn("skip_me", ids)
def test_includes_osw_1(self):
pubs = parse_publications(self.xml_path)
ids = [p["id"] for p in pubs]
self.assertIn("ber", ids)
self.assertIn("cal", ids)
def test_author_dot_becomes_none(self):
pubs = parse_publications(self.xml_path)
ber = next(p for p in pubs if p["id"] == "ber")
self.assertIsNone(ber["author"])
def test_author_string_preserved(self):
pubs = parse_publications(self.xml_path)
cal = next(p for p in pubs if p["id"] == "cal")
self.assertEqual(cal["author"], "Calwer")
def test_img_path_present(self):
pubs = parse_publications(self.xml_path)
cal = next(p for p in pubs if p["id"] == "cal")
self.assertEqual(cal["img_path"], "osdb/cal/cal_s0.jpg")
def test_img_path_absent_is_none(self):
pubs = parse_publications(self.xml_path)
noc = next(p for p in pubs if p["id"] == "noc")
self.assertIsNone(noc["img_path"])
def test_title_mapped(self):
pubs = parse_publications(self.xml_path)
cal = next(p for p in pubs if p["id"] == "cal")
self.assertEqual(cal["title"], "Obst- und Beeren")
class TestScanPubDir(unittest.TestCase):
def setUp(self):
self.tmp = tempfile.mkdtemp()
def test_returns_empty_for_missing_directory(self):
result = scan_pub_dir(os.path.join(self.tmp, "nonexistent"), "xyz")
self.assertEqual(result, {})
def test_extracts_osdb_number_from_image(self):
d = Path(self.tmp)
(d / "apfel_ber_s0.jpg").write_bytes(b"img")
result = scan_pub_dir(str(d), "ber")
self.assertIn("apfel", result)
self.assertEqual(result["apfel"]["img_path"], str(d / "apfel_ber_s0.jpg"))
def test_extracts_osdb_number_from_pdf(self):
d = Path(self.tmp)
(d / "birne_cal.pdf").write_bytes(b"pdf")
result = scan_pub_dir(str(d), "cal")
self.assertIn("birne", result)
self.assertEqual(result["birne"]["pdf_path"], str(d / "birne_cal.pdf"))
def test_combines_img_and_pdf_for_same_fruit(self):
d = Path(self.tmp)
(d / "apfel_ber_s0.jpg").write_bytes(b"img")
(d / "apfel_ber.pdf").write_bytes(b"pdf")
result = scan_pub_dir(str(d), "ber")
self.assertIn("img_path", result["apfel"])
self.assertIn("pdf_path", result["apfel"])
def test_ignores_thumbnail_files(self):
d = Path(self.tmp)
(d / "apfel_ber_tn.jpg").write_bytes(b"tn")
result = scan_pub_dir(str(d), "ber")
self.assertEqual(result, {})
def test_multi_segment_osdb_number(self):
d = Path(self.tmp)
(d / "berner_grauechapfel_ber_s0.jpg").write_bytes(b"img")
result = scan_pub_dir(str(d), "ber")
self.assertIn("berner_grauechapfel", result)
class TestImportPublications(unittest.TestCase):
def _make_data_dir(self, pubs):
"""Build a temp data dir with osws.xml and osdb/<id>/ dirs."""
tmp = tempfile.mkdtemp()
lines = ['<?xml version="1.0" encoding="iso-8859-1" ?>', "<root>"]
for p in pubs:
lines.append(" <obj>")
lines.append(f" <id>{p['id']}</id>")
lines.append(f" <name>{p['name']}</name>")
author = p.get("author", "Test")
lines.append(f" <author>{author}</author>")
if p.get("img"):
lines.append(f" <img>{p['img']}</img>")
lines.append(" <osw>1</osw>")
lines.append(" </obj>")
lines.append("</root>")
Path(tmp, "osws.xml").write_bytes("\n".join(lines).encode("iso-8859-1"))
for p in pubs:
pub_dir = Path(tmp, "osdb", p["id"])
pub_dir.mkdir(parents=True, exist_ok=True)
for fruit in p.get("fruits", []):
if fruit.get("img"):
(pub_dir / f"{fruit['osdb_number']}_{p['id']}_s0.jpg").write_bytes(b"\x89PNG")
if fruit.get("pdf"):
(pub_dir / f"{fruit['osdb_number']}_{p['id']}.pdf").write_bytes(b"%PDF")
if p.get("cover_bytes"):
cover_rel = p["img"]
cover_path = Path(tmp, cover_rel)
cover_path.parent.mkdir(parents=True, exist_ok=True)
cover_path.write_bytes(p["cover_bytes"])
return tmp
def _make_conn(self, pub_db_id=42, fruit_db_ids=None):
"""Return a mock psycopg2 connection."""
conn = MagicMock()
cur = MagicMock()
conn.cursor.return_value.__enter__.return_value = cur
# fetchone side_effect: first call = pub id; subsequent = fruit lookups.
# None in fruit_db_ids → bare None (simulates no row found); int → (int,) tuple.
fruit_db_ids = fruit_db_ids or []
side_effects = [(pub_db_id,)] + [((fid,) if fid is not None else None) for fid in fruit_db_ids]
cur.fetchone.side_effect = side_effects
return conn, cur
def test_missing_cover_image_does_not_abort_import(self):
data_dir = self._make_data_dir([
{"id": "ber", "name": "Test", "img": "osdb/ber/ber_s0.jpg"},
])
# cover file NOT created → missing
conn, cur = self._make_conn(pub_db_id=1)
counts = import_publications(conn, data_dir)
self.assertEqual(counts["publications"], 1)
self.assertEqual(counts["covers"], 0)
def test_fruit_not_in_db_is_skipped_with_warning(self):
data_dir = self._make_data_dir([
{"id": "cal", "name": "Cal", "fruits": [
{"osdb_number": "unknown_fruit", "img": True},
]},
])
conn, cur = self._make_conn(pub_db_id=5, fruit_db_ids=[None])
counts = import_publications(conn, data_dir)
self.assertEqual(counts["skipped_fruits"], 1)
self.assertEqual(counts["fruits_linked"], 0)
def test_happy_path_upserts_pub_links_fruits_imports_files(self):
data_dir = self._make_data_dir([
{
"id": "pom",
"name": "Pomologie",
"author": "Diel",
"img": "osdb/pom/pom_s0.jpg",
"cover_bytes": b"\x89PNG",
"fruits": [
{"osdb_number": "apfelsorte_alpha", "img": True, "pdf": True},
],
}
])
conn, cur = self._make_conn(pub_db_id=7, fruit_db_ids=[99])
counts = import_publications(conn, data_dir)
self.assertEqual(counts["publications"], 1)
self.assertEqual(counts["covers"], 1)
self.assertEqual(counts["fruits_linked"], 1)
self.assertEqual(counts["pdfs"], 1)
self.assertEqual(counts["images"], 1)
self.assertEqual(counts["skipped_fruits"], 0)
def test_osw_not_1_never_imported(self):
tmp = tempfile.mkdtemp()
xml = b"""<?xml version="1.0" encoding="iso-8859-1"?>
<root>
<obj><id>skip</id><name>Skip</name><author>X</author><osw>0</osw></obj>
</root>"""
Path(tmp, "osws.xml").write_bytes(xml)
conn, cur = self._make_conn()
counts = import_publications(conn, tmp)
self.assertEqual(counts["publications"], 0)
cur.execute.assert_not_called()
if __name__ == "__main__":
unittest.main()