diff --git a/_postgres/schema.sql b/_postgres/schema.sql
index 115faab..6bf1081 100644
--- a/_postgres/schema.sql
+++ b/_postgres/schema.sql
@@ -177,6 +177,12 @@ CREATE INDEX IF NOT EXISTS debates_summary_embedding_idx ON debates
CREATE TABLE IF NOT EXISTS documents (
-- Primary key: doc_id (e.g. "HD02846")
doc_id TEXT PRIMARY KEY,
+
+ -- Which kind of member-submitted document this is. Everything is 'motion'
+ -- today; the column exists so bills, written questions and committee reports
+ -- can share this table rather than each needing their own.
+ doc_type TEXT NOT NULL DEFAULT 'motion',
+
source_record_id TEXT,
-- Identity / classification
diff --git a/bootstrap.py b/bootstrap.py
index 22b1f17..c8420da 100644
--- a/bootstrap.py
+++ b/bootstrap.py
@@ -1,6 +1,6 @@
"""Locate the project root and make it importable.
-Scripts under ``scripts/`` are run directly (``python scripts/sync_talks.py``), so
+Scripts under ``scripts/`` are run directly (``python scripts/make_embeddings.py``), so
the repository root is not on ``sys.path`` and relative paths would resolve against
whatever directory the caller happened to be in. Importing this module fixes both,
deriving the root from this file's own location.
diff --git a/ingest/adapters/riksdagen.py b/ingest/adapters/riksdagen.py
index 0de305a..382f152 100644
--- a/ingest/adapters/riksdagen.py
+++ b/ingest/adapters/riksdagen.py
@@ -22,11 +22,17 @@ from typing import Any, Iterable, Iterator, Optional
# ── source quirks ─────────────────────────────────────────────────────────────
-def _as_list(value: Any) -> list:
- """One child arrives as an object, several as a list, none as null."""
+def _as_list(value: Any) -> list[dict]:
+ """One child arrives as an object, several as a list, none as null.
+
+ Non-object entries are dropped rather than raising: a handful of records in
+ the archive carry a bare string where a child element is expected, and one
+ malformed row should not abort a multi-hour load.
+ """
if value is None:
return []
- return value if isinstance(value, list) else [value]
+ items = value if isinstance(value, list) else [value]
+ return [i for i in items if isinstance(i, dict)]
def _clean(value: Any) -> Any:
@@ -166,7 +172,9 @@ def adapt_document(payload: dict) -> Optional[dict]:
"ordinal": i,
"person_id": _clean(person.get("intressent_id")),
"name": _clean(person.get("namn")),
- "party": _clean(person.get("partibet")),
+ # The archive carries both "S" and "s" for the same party. Left as-is,
+ # a filter on 'S' silently misses half the documents.
+ "party": (_clean(person.get("partibet")) or "").upper() or None,
"role": _clean(person.get("roll")),
})
diff --git a/ingest/cli.py b/ingest/cli.py
new file mode 100644
index 0000000..b509f65
--- /dev/null
+++ b/ingest/cli.py
@@ -0,0 +1,78 @@
+"""Command-line entry point for ingesting a parliament's open data.
+
+ python -m ingest.cli fetch --source documents --range 2022-2025
+ python -m ingest.cli load --source documents
+ python -m ingest.cli sync --source all
+
+`fetch` downloads, `load` adapts and upserts what is on disk, `sync` does both.
+Everything is resumable: archives already unpacked are skipped, and inserts use
+ON CONFLICT DO NOTHING, so re-running after an interruption is safe.
+"""
+from __future__ import annotations
+
+import argparse
+import sys
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
+
+from ingest import pipeline # noqa: E402
+from ingest.adapters import load_adapter # noqa: E402
+from parliament import PARLIAMENT # noqa: E402
+
+SOURCES = ("speeches", "documents", "people")
+
+
+def _adapter():
+ module_path = PARLIAMENT.sources.get("adapter")
+ if not module_path:
+ raise SystemExit("parliament.yaml has no `sources.adapter`. See docs/PORTING.md.")
+ return load_adapter(module_path)
+
+
+def _resolve(source: str) -> list[str]:
+ if source == "all":
+ return [s for s in SOURCES if s in PARLIAMENT.sources]
+ if source not in SOURCES:
+ raise SystemExit(f"Unknown source {source!r}; expected one of {SOURCES + ('all',)}")
+ return [source]
+
+
+def main(argv: list[str] | None = None) -> int:
+ parser = argparse.ArgumentParser(prog="ingest", description=__doc__)
+ parser.add_argument("command", choices=("fetch", "load", "sync"))
+ parser.add_argument("--source", default="all",
+ help=f"one of {', '.join(SOURCES)}, or 'all' (default)")
+ parser.add_argument("--range", dest="ranges", action="append",
+ help="archive range to fetch, e.g. 2022-2025. Repeatable. "
+ "Defaults to the ranges in parliament.yaml.")
+ parser.add_argument("--limit", type=int,
+ help="stop after this many records — useful for a smoke test")
+ args = parser.parse_args(argv)
+
+ adapter = _adapter()
+ print(f"parliament: {PARLIAMENT.meta.get('name')} adapter: {PARLIAMENT.sources['adapter']}")
+
+ exit_code = 0
+ for source in _resolve(args.source):
+ print(f"\n── {source} ──")
+ try:
+ if args.command in ("fetch", "sync"):
+ pipeline.fetch(source, args.ranges)
+ if args.command in ("load", "sync"):
+ counts = pipeline.load(source, adapter.adapt, limit=args.limit)
+ print(f" read {counts['read']}, wrote {counts['written']}, "
+ f"skipped {counts['skipped']}")
+ except Exception as exc:
+ # One failing source should not abandon the others; a daily sync that
+ # aborts because one archive moved is worse than a partial one.
+ print(f" FAILED: {type(exc).__name__}: {exc}")
+ exit_code = 1
+
+ if args.command in ("load", "sync"):
+ print("\nNext: python scripts/make_embeddings.py")
+ return exit_code
+
+
+if __name__ == "__main__":
+ sys.exit(main())
diff --git a/ingest/pipeline.py b/ingest/pipeline.py
new file mode 100644
index 0000000..4cb6b61
--- /dev/null
+++ b/ingest/pipeline.py
@@ -0,0 +1,209 @@
+"""Generic fetch → adapt → upsert pipeline.
+
+Country-neutral: every source-specific decision comes from `sources:` in
+parliament.yaml and from the adapter named there.
+"""
+from __future__ import annotations
+
+import io
+import json
+import zipfile
+from pathlib import Path
+from typing import Callable, Iterator, Optional
+
+import requests
+
+from bootstrap import DATA_DIR
+from parliament import PARLIAMENT
+from postgres_client import pg
+
+# Rows per INSERT. Large enough that round-trips are not the bottleneck, small
+# enough that one bad batch is cheap to retry.
+BATCH = 500
+
+
+# ── fetching ──────────────────────────────────────────────────────────────────
+
+
+def source_config(name: str) -> dict:
+ sources = PARLIAMENT.sources
+ if name not in sources:
+ available = [k for k in sources if k != "adapter"]
+ raise ValueError(f"No source {name!r} in parliament.yaml; have {available}")
+ return sources[name]
+
+
+def dest_dir(name: str) -> Path:
+ return DATA_DIR / source_config(name).get("dest_dir", name)
+
+
+def fetch(name: str, ranges: Optional[list[str]] = None) -> list[Path]:
+ """Download a source's bulk archives and unpack them under DATA_DIR.
+
+ Archives already unpacked are skipped, so this is safe to re-run and cheap to
+ resume after an interruption — which matters when a full download is tens of GB.
+ """
+ cfg = source_config(name)
+ kind = cfg.get("kind", "zip-dataset")
+ target = dest_dir(name)
+ target.mkdir(parents=True, exist_ok=True)
+
+ if kind == "json":
+ out = target / f"{name}.json"
+ out.write_bytes(requests.get(cfg["url"], timeout=120).content)
+ print(f" fetched {out.relative_to(DATA_DIR)}")
+ return [out]
+
+ if kind != "zip-dataset":
+ raise ValueError(f"Unsupported source kind {kind!r} for {name}")
+
+ wanted = ranges or cfg.get("ranges")
+ if not wanted:
+ raise ValueError(
+ f"Source {name!r} has no `ranges:` in parliament.yaml and none was given. "
+ f"Pass --range, or list them in the config."
+ )
+
+ written = []
+ for rng in wanted:
+ folder = target / rng
+ if folder.exists() and any(folder.iterdir()):
+ print(f" {rng}: already present, skipping")
+ continue
+ url = cfg["url_template"].format(range=rng)
+ print(f" {rng}: downloading {url}")
+ resp = requests.get(url, timeout=1800)
+ if resp.status_code != 200:
+ print(f" {rng}: HTTP {resp.status_code}, skipping")
+ continue
+ folder.mkdir(parents=True, exist_ok=True)
+ with zipfile.ZipFile(io.BytesIO(resp.content)) as zf:
+ zf.extractall(folder)
+ written.append(folder)
+ print(f" {rng}: unpacked {len(list(folder.rglob('*.json')))} files")
+ return written
+
+
+def read_records(name: str) -> Iterator[dict]:
+ """Yield every source record on disk for a source."""
+ root = dest_dir(name)
+ if not root.exists():
+ raise FileNotFoundError(f"No data at {root}. Run `ingest.cli fetch --source {name}` first.")
+ for path in sorted(root.rglob("*.json")):
+ try:
+ # utf-8-sig: the archives carry a byte-order mark.
+ with open(path, encoding="utf-8-sig") as fh:
+ payload = json.load(fh)
+ except (json.JSONDecodeError, OSError) as exc:
+ print(f" skipping {path.name}: {exc}")
+ continue
+ # A people listing is one file holding many records.
+ if isinstance(payload, dict) and "personlista" in payload:
+ for person in payload["personlista"].get("person", []):
+ yield person
+ else:
+ yield payload
+
+
+# ── loading ───────────────────────────────────────────────────────────────────
+
+
+def _insert(table: str, columns: list[str], rows: list[tuple], conflict: str) -> int:
+ if not rows:
+ return 0
+ sql = (
+ f"INSERT INTO {table} ({', '.join(columns)}) VALUES %s "
+ f"ON CONFLICT ({conflict}) DO NOTHING"
+ )
+ pg.execute_values(sql, rows)
+ return len(rows)
+
+
+def _flush(table: str, columns: list[str], conflict: str, buffer: list[dict]) -> int:
+ """Insert and clear a buffer. Argument order matches `_target()` unpacking."""
+ rows = [tuple(r.get(c) for c in columns) for r in buffer]
+ n = _insert(table, columns, rows, conflict)
+ buffer.clear()
+ return n
+
+
+SPEECH_COLUMNS = [
+ "id", "source_speech_id", "source_doc_id", "text", "section_title", "sequence",
+ "activity_type", "speaker_name", "party", "person_id", "date", "source_datetime",
+ "year", "session_year", "related_doc_id", "source_doc_number", "source_record_id",
+ "title", "is_reply",
+]
+DOCUMENT_COLUMNS = [
+ "doc_id", "doc_type", "source_record_id", "session_label", "designation", "subtype",
+ "committee", "status", "date", "source_updated_at", "published_at", "session_year",
+ "title", "subtitle", "text", "proposals_text", "has_text", "url_text", "url_html",
+ "url_pdf", "parties", "author_names", "proposals_raw", "attachments", "num_proposals",
+]
+AUTHOR_COLUMNS = ["doc_id", "ordinal", "person_id", "name", "party", "role"]
+PROPOSAL_COLUMNS = [
+ "id", "doc_id", "ordinal", "number", "text",
+ "committee_recommendation", "chamber_decision", "handled_in",
+]
+PERSON_COLUMNS = [
+ "person_id", "source_record_id", "source_record_guid", "source_id", "birth_year",
+ "gender", "last_name", "first_name", "sort_name", "home_town", "party",
+ "constituency", "status", "source_url", "image_url_small", "image_url_medium",
+ "image_url_large", "assignments", "contact_details", "name", "active",
+]
+
+
+def _json_columns(rows: list[dict], columns: Iterator[str]) -> None:
+ """psycopg2 cannot adapt dicts/lists destined for JSONB; serialise them."""
+ for row in rows:
+ for col in columns:
+ if isinstance(row.get(col), (dict, list)):
+ row[col] = json.dumps(row[col], ensure_ascii=False)
+
+
+def load(name: str, adapt: Callable[[str, dict], Optional[dict]], limit: Optional[int] = None) -> dict:
+ """Adapt every record on disk for a source and upsert it."""
+ counts = {"read": 0, "written": 0, "skipped": 0}
+ buf: list[dict] = []
+ authors: list[dict] = []
+ proposals: list[dict] = []
+
+ for payload in read_records(name):
+ if limit and counts["read"] >= limit:
+ break
+ counts["read"] += 1
+ row = adapt(name, payload)
+ if row is None:
+ counts["skipped"] += 1
+ continue
+
+ if name == "documents":
+ _json_columns([row["document"]], ("proposals_raw", "attachments"))
+ buf.append(row["document"])
+ authors.extend(row["authors"])
+ proposals.extend(row["proposals"])
+ else:
+ if name == "people":
+ _json_columns([row], ("assignments", "contact_details"))
+ buf.append(row)
+
+ if len(buf) >= BATCH:
+ counts["written"] += _flush(*_target(name), buf)
+ if authors:
+ _flush("document_authors", AUTHOR_COLUMNS, "doc_id, ordinal", authors)
+ if proposals:
+ _flush("document_proposals", PROPOSAL_COLUMNS, "id", proposals)
+
+ counts["written"] += _flush(*_target(name), buf)
+ if authors:
+ _flush("document_authors", AUTHOR_COLUMNS, "doc_id, ordinal", authors)
+ if proposals:
+ _flush("document_proposals", PROPOSAL_COLUMNS, "id", proposals)
+ return counts
+
+
+def _target(name: str) -> tuple[str, list[str], str]:
+ return {
+ "speeches": ("speeches", SPEECH_COLUMNS, "id"),
+ "documents": ("documents", DOCUMENT_COLUMNS, "doc_id"),
+ "people": ("people", PERSON_COLUMNS, "person_id"),
+ }[name]
diff --git a/recovery.sh b/recovery.sh
index 3c0e329..2dfbcb9 100755
--- a/recovery.sh
+++ b/recovery.sh
@@ -7,7 +7,7 @@
# bash recovery.sh 2>&1 | tee logs/recovery.log
#
# Steps (ordered, each depends on the previous):
-# 1. Load ~450k raw talks from disk JSON files (documents_to_postgres.py)
+# 1. Load raw speeches from disk JSON files (ingest.cli load)
# 2. Enrich with summaries+tags from talks_training in ArangoDB
# 3. Assign debate IDs to talks that don't have one
# 4. Migrate debates from ArangoDB
@@ -48,9 +48,9 @@ ok "Pre-flight passed"
# =============================================================================
# STEP 1 — Load all talks from disk JSON files
# =============================================================================
-step "Step 1 — Load talks from disk (documents_to_postgres.py)"
+step "Step 1 — Load speeches from disk (ingest.cli load)"
-$PYTHON scripts/documents_to_postgres.py \
+$PYTHON -m ingest.cli load --source speeches \
2>&1 | tee "$LOG_DIR/step1_documents_to_postgres.log"
ok "Step 1 done"
diff --git a/scripts/backfill_motions.sh b/scripts/backfill_motions.sh
deleted file mode 100755
index abd757f..0000000
--- a/scripts/backfill_motions.sh
+++ /dev/null
@@ -1,29 +0,0 @@
-#!/bin/bash
-# One-time backfill of motioner 1990→idag.
-#
-# Kör i tre steg (alla resumérbara — avbrott är ofarligt, kör bara om):
-# 1. Ladda ned alla mot-arkiv som saknas (hoppar över ifyllda mappar)
-# 2. Parsa alla mappar i motioner/ till Postgres (ON CONFLICT DO NOTHING)
-# 3. Bygg embeddings år för år, nyast först
-#
-# Starta frikopplat: nohup scripts/backfill_motions.sh > backfill_motions.log 2>&1 &
-set -e
-cd "$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)"
-PY=.venv/bin/python
-
-echo "=== Step 1: download all ranges ==="
-$PY scripts/download_motions.py
-
-echo "=== Step 2: parse all folders into Postgres ==="
-$PY scripts/motions_to_postgres.py
-
-echo "=== Step 3: full-text embeddings, newest year first ==="
-for year in $(seq 2025 -1 1990); do
- echo "--- embeddings for year $year ---"
- $PY scripts/make_embeddings.py motions "$year"
-done
-
-echo "=== Step 4: yrkande (condensed proposal) embeddings ==="
-$PY scripts/make_embeddings.py yrkanden
-
-echo "=== Backfill complete ==="
diff --git a/scripts/documents_to_postgres.py b/scripts/documents_to_postgres.py
deleted file mode 100644
index 6671044..0000000
--- a/scripts/documents_to_postgres.py
+++ /dev/null
@@ -1,167 +0,0 @@
-"""
-Läser in anföranden från JSON-filer till PostgreSQL.
-
-Ersätter scripts/documents_to_arango.py.
-
-Används av sync_talks.py (update_folder) och kan köras direkt för att
-(om)ladda alla mappar i speeches/:
-
- python scripts/documents_to_postgres.py
-"""
-from pathlib import Path
-
-import json
-import logging
-import os
-import re
-import sys
-
-sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
-import bootstrap # noqa: E402,F401 — sets cwd and sys.path to the project root
-
-from postgres_client import pg
-
-logging.basicConfig(level=logging.WARNING, format="%(asctime)s - %(levelname)s - %(message)s")
-
-
-def clean_text(text: str) -> str:
- if text is None:
- return ""
- text = text.replace("
", "\n").replace("
", "\n").replace("
", "\n")
- text = text.replace("