From e2b71c64b163004487d58b4965157c7429270661 Mon Sep 17 00:00:00 2001 From: Lasse Server Date: Mon, 3 Aug 2026 08:20:39 +0200 Subject: [PATCH] Add the ingest pipeline and CLI, replacing the ad-hoc scripts MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `python -m ingest.cli fetch|load|sync` now exists — README, PORTING.md, the systemd example and pyproject's console script have all been referencing it. This replaces six scripts that the rename had left genuinely broken: the speech loader's INSERT listed `year, year` as two columns, and both loaders read the source JSON using plenum's column names rather than the Riksdag's field names, so a sync would have written empty rows without erroring. Rewriting on top of the adapter was safer than repairing them, because the adapter makes the source-name/column-name boundary explicit rather than leaving it to be remembered. Two source quirks found by running it over real archives rather than reasoning about it: a few records carry a bare string where a child element belongs, which took the whole load down on an AttributeError; and party codes appear in both cases, so `parties` held {S,s} and any filter on 'S' would have silently missed half the documents. Both are handled in the adapter, where source messiness belongs. Also adds doc_type to schema.sql. It was only in the migration, so a fresh install lacked a column the loader writes. Verified end to end against real archives: 400 speeches and 400 documents ingested into a clean database, producing 3,742 authors and 2,229 proposals, with search vectors built by the triggers and full-text queries returning hits. Post-load steps (debate ids, embeddings, summaries) still run separately via scripts/debates.py and scripts/make_embeddings.py; wiring them into `sync` is tracked. Co-Authored-By: Claude Opus 5 --- _postgres/schema.sql | 6 + bootstrap.py | 2 +- ingest/adapters/riksdagen.py | 16 +- ingest/cli.py | 78 +++++++++ ingest/pipeline.py | 209 ++++++++++++++++++++++++ recovery.sh | 6 +- scripts/backfill_motions.sh | 29 ---- scripts/documents_to_postgres.py | 167 ------------------- scripts/download_motions.py | 91 ----------- scripts/download_talks.py | 63 -------- scripts/fix_replik_from_json.py | 182 --------------------- scripts/motions_progress.sh | 29 ---- scripts/motions_to_postgres.py | 268 ------------------------------- scripts/sync_motions.py | 62 ------- scripts/sync_talks.py | 130 --------------- 15 files changed, 309 insertions(+), 1029 deletions(-) create mode 100644 ingest/cli.py create mode 100644 ingest/pipeline.py delete mode 100755 scripts/backfill_motions.sh delete mode 100644 scripts/documents_to_postgres.py delete mode 100644 scripts/download_motions.py delete mode 100644 scripts/download_talks.py delete mode 100644 scripts/fix_replik_from_json.py delete mode 100755 scripts/motions_progress.sh delete mode 100644 scripts/motions_to_postgres.py delete mode 100644 scripts/sync_motions.py delete mode 100644 scripts/sync_talks.py 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("

", "\n") - text = re.sub(r"\n+", "\n", text) - text = text.strip() - text = re.sub(r"<.*?>", "", text) - return text - - -def clean_speaker_name(name: str) -> str: - if name is None: - return "" - name = name.strip() - name = re.sub(r"\s*\(.*?\)\s*$", "", name) - return name.strip() - - -def _parse_date(s: str) -> str | None: - if not s: - return None - return str(s).split(" ")[0][:10] or None - - -def process_folder(folder_path: str, already_processed: set[str] = frozenset()) -> list[dict]: - """Parse JSON files in folder_path, return list of talk dicts (skipping known IDs).""" - docs = [] - for file in os.listdir(folder_path): - if not file.endswith(".json"): - continue - try: - with open(os.path.join(folder_path, file), "r", encoding="utf-8-sig") as f: - data = json.load(f) - doc = data["anforande"] - source_doc_id = doc.get("dok_id", "") - sequence = doc.get("sequence", "") - speech_id = f"{source_doc_id}-{sequence}" # unique per speech, e.g. "GH09116-16" - if speech_id in already_processed: - continue - doc["year"] = int(doc.get("dok_rm", "0000")[:4]) - doc.pop("dok_rm", None) - doc["text"] = clean_text(doc.get("text", "")) - doc["speaker_name"] = clean_speaker_name(doc.get("speaker_name", "")) - doc["id"] = speech_id - doc["dok_id"] = source_doc_id - doc["source_speech_id"] = doc.get("source_speech_id", "") # UUID, kept for reference - doc["date"] = _parse_date(doc.get("source_datetime", "")) - doc["source_datetime"] = doc.get("source_datetime", "") - doc["title"] = doc.get("dok_titel", "") - doc.pop("dok_titel", None) - doc["sequence"] = int(sequence) if sequence else 0 - doc["source_record_id"] = doc.get("dok_hangar_id", "") - doc.pop("dok_hangar_id", None) - doc["is_reply"] = doc.get("is_reply", "N") == "Y" - doc.pop("source_updated_at", None) - doc.pop("underrubrik", None) - year = doc.get("year") or (int(doc["date"][:4]) if doc.get("date") else None) - doc["year"] = year - docs.append(doc) - except (json.JSONDecodeError, KeyError, ValueError) as e: - logging.warning(f"Skipping {file}: {e}") - return docs - - -_UPSERT_SQL = """ -INSERT INTO speeches ( - id, source_speech_id, source_doc_id, - text, section_title, - sequence, activity_type, - speaker_name, party, person_id, - date, source_datetime, year, year, - related_doc_id, source_doc_number, source_record_id, title, - is_reply -) VALUES %s -ON CONFLICT (id) DO NOTHING -""" - - -def _doc_to_row(doc: dict) -> tuple: - return ( - doc.get("id"), # source_speech_id UUID — primary key - doc.get("source_speech_id"), # same value, kept for reference - doc.get("dok_id"), # debate/protocol document id - doc.get("text"), - doc.get("section_title"), - doc.get("sequence"), - doc.get("activity_type"), - doc.get("speaker_name"), - doc.get("party"), - doc.get("person_id"), - doc.get("date"), - doc.get("source_datetime"), - doc.get("year"), - doc.get("year"), - doc.get("related_doc_id"), - doc.get("source_doc_number"), - doc.get("source_record_id"), - doc.get("title"), - doc.get("is_reply", False), - ) - - -def insert_docs(docs: list[dict]) -> None: - if not docs: - return - rows = [_doc_to_row(d) for d in docs if d.get("id")] - if rows: - pg.execute_values(_UPSERT_SQL, rows) - - -def update_folder(path: str, already_processed: set[str] = None) -> int: - """ - Upsert talk documents from JSON files in path into PostgreSQL. - Returns the number of new speeches inserted. - """ - if already_processed is None: - rows = pg.execute("SELECT id FROM speeches") - already_processed = {row["id"] for row in rows} - - docs = process_folder(path, already_processed) - insert_docs(docs) - return len(docs) - - -if __name__ == "__main__": - # Load all folders in speeches/ into PostgreSQL - existing = {row["id"] for row in pg.execute("SELECT id FROM speeches")} - total = 0 - for folder in sorted(os.listdir("speeches")): - path = str(bootstrap.DATA_DIR / 'speeches' / folder) - if not os.path.isdir(path): - continue - print(f"Processing {folder} …", end=" ", flush=True) - docs = process_folder(path, already_processed=existing) - insert_docs(docs) - new_ids = {d["id"] for d in docs if d.get("id")} - existing |= new_ids - total += len(new_ids) - print(f"{len(new_ids)} inserted") - print(f"\nTotal: {total} new speeches inserted") diff --git a/scripts/download_motions.py b/scripts/download_motions.py deleted file mode 100644 index 7f1a21f..0000000 --- a/scripts/download_motions.py +++ /dev/null @@ -1,91 +0,0 @@ -""" -Laddar ned motioner från riksdagens öppna data som ZIP-arkiv med JSON-filer. - -Arkiven är uppdelade i fyraårsperioder (förankrade vid 1998; allt före ligger -i 1990-1997) och uppdateras dagligen: - - https://data.riksdagen.se/dataset/dokument/mot-{range}.json.zip - -Varje arkiv innehåller en JSON-fil per motion (dokumentstatus-kuvert med -metadata + fulltext som HTML). Extraheras till motioner/mot-{range}/. - -Backfill (alla perioder 1990→): python scripts/download_motions.py -Används av sync_motions.py för daglig uppdatering av aktuell year. -""" -from pathlib import Path - -import logging -import os -import sys -from io import BytesIO -from urllib.request import urlopen -from zipfile import ZipFile - -sys.path.insert(0, str(Path(__file__).resolve().parents[1])) -import bootstrap # noqa: E402,F401 — sets cwd and sys.path to the project root - -logging.basicConfig( - level=logging.INFO, - format="%(asctime)s [%(levelname)s] %(message)s", -) -logger = logging.getLogger(__name__) - -RANGES = [ - "1990-1997", - "1998-2001", - "2002-2005", - "2006-2009", - "2010-2013", - "2014-2017", - "2018-2021", - "2022-2025", -] -URL_TEMPLATE = "https://data.riksdagen.se/dataset/dokument/mot-{r}.json.zip" -MOTIONS_DIR = "motioner" - - -def get_current_range(year: int) -> str: - """Fyraårsperiod förankrad vid 1998; allt före 1998 ligger i 1990-1997.""" - if year < 1998: - return "1990-1997" - start = year - ((year - 1998) % 4) - return f"{start}-{start + 3}" - - -def download_range(r: str, force: bool = False) -> str: - """ - Laddar ned och extraherar arkivet för en year till motioner/mot-{r}/. - Hoppar över om mappen redan är ifylld, om inte force=True (töms först). - """ - dir_path = os.path.join(MOTIONS_DIR, f"mot-{r}") - - if os.path.exists(dir_path) and os.listdir(dir_path): - if not force: - logger.info(f"Skipping {dir_path}, already exists and is not empty.") - return dir_path - for f in os.listdir(dir_path): - os.remove(os.path.join(dir_path, f)) - - os.makedirs(dir_path, exist_ok=True) - url = URL_TEMPLATE.format(r=r) - logger.info(f"Downloading {url} → {dir_path}") - - with urlopen(url) as resp: - with ZipFile(BytesIO(resp.read())) as zf: - zf.extractall(dir_path) - - count = len(os.listdir(dir_path)) - logger.info(f"Extracted {count} files to {dir_path}") - return dir_path - - -def download_all(force: bool = False) -> None: - for r in RANGES: - download_range(r, force=force) - - -if __name__ == "__main__": - if len(sys.argv) > 1: - download_range(sys.argv[1], force="--force" in sys.argv) - else: - download_all() diff --git a/scripts/download_talks.py b/scripts/download_talks.py deleted file mode 100644 index 1a197fb..0000000 --- a/scripts/download_talks.py +++ /dev/null @@ -1,63 +0,0 @@ -import requests -from bs4 import BeautifulSoup -from io import BytesIO -from urllib.request import urlopen -from zipfile import ZipFile -import os -from time import sleep - - -def download(all=False, year=None): - if all: - for year in range(1999, 2026): - first_part = str(year) - second_part = str(year + 1)[2:] - if first_part == '1999': - url = 'https://data.riksdagen.se/dataset/anforande/anforande-19992000.json.zip' - else: - url = f"https://data.riksdagen.se/dataset/anforande/anforande-{first_part}{second_part}.json.zip" - print(url) - - # Ensure the 'speeches' directory exists - talks_dir = "speeches" - os.makedirs(talks_dir, exist_ok=True) - - # Create a subdirectory for the current year range - dir_name = os.path.join(talks_dir, f"anforande-20{first_part}{second_part}") - if os.path.exists(dir_name) and os.listdir(dir_name): - print(f"Skipping {dir_name}, already exists and is not empty.") - continue - - os.makedirs(dir_name, exist_ok=True) - - # Download and extract the zip file directly into the subdirectory - with urlopen(url) as zipresp: - with ZipFile(BytesIO(zipresp.read())) as zfile: - zfile.extractall(dir_name) - elif year: - first_part = str(year) - second_part = str(year + 1)[2:] - url = f"https://data.riksdagen.se/dataset/anforande/anforande-{first_part}{second_part}.json.zip" - print(url) - - # Ensure the 'speeches' directory exists - talks_dir = "speeches" - os.makedirs(talks_dir, exist_ok=True) - - # Create a subdirectory for the current year range - dir_name = os.path.join(talks_dir, f"anforande-20{first_part}{second_part}") - if os.path.exists(dir_name) and os.listdir(dir_name): - print(f"Skipping {dir_name}, already exists and is not empty.") - return - - os.makedirs(dir_name, exist_ok=True) - - # Download and extract the zip file directly into the subdirectory - with urlopen(url) as zipresp: - with ZipFile(BytesIO(zipresp.read())) as zfile: - zfile.extractall(dir_name) - -if __name__ == "__main__": - while True: - new_files = download() - diff --git a/scripts/fix_replik_from_json.py b/scripts/fix_replik_from_json.py deleted file mode 100644 index e58dad6..0000000 --- a/scripts/fix_replik_from_json.py +++ /dev/null @@ -1,182 +0,0 @@ -""" -Fix speeches.is_reply by reading the ground-truth value from the JSON source files. - -Background: - documents_to_postgres.py uses ON CONFLICT DO NOTHING, so speeches ingested before - is_reply was handled correctly were never updated. This script does the corrective - pass: reads every JSON file, finds speeches where the stored is_reply differs from the - file, updates the DB, then re-assigns debate IDs for the affected dates so that - proper multi-speaker debates are created. - -Steps: - 1. Scan all speeches/ subfolders and build {speech_id → replik_bool} from JSON files. - 2. Query DB for speeches where is_reply is currently False. - 3. Update any talk whose JSON says True. - 4. For each affected date, clear speeches.debate and re-run make_debate_ids(). - 5. Delete stale single-talk debate rows whose IDs no longer exist in speeches.debate. - -Usage: - python scripts/fix_replik_from_json.py -""" -from pathlib import Path - -import logging -import os -import sys -from collections import defaultdict - -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 -from scripts.documents_to_postgres import process_folder -from scripts.debates import assign_debate_ids - -logging.basicConfig( - level=logging.INFO, - format="%(asctime)s [%(levelname)s] %(message)s", -) -logger = logging.getLogger(__name__) - -TALKS_DIR = str(bootstrap.DATA_DIR / "speeches") -BATCH_SIZE = 500 - - -def build_replik_map() -> dict[str, bool]: - """Return {speech_id: is_reply} for every JSON file in speeches/.""" - replik_map: dict[str, bool] = {} - folders = sorted(os.listdir(TALKS_DIR)) - for i, folder in enumerate(folders): - path = os.path.join(TALKS_DIR, folder) - if not os.path.isdir(path): - continue - docs = process_folder(path) - for doc in docs: - tid = doc.get("id") - if tid: - replik_map[tid] = bool(doc.get("is_reply", False)) - if (i + 1) % 5 == 0 or (i + 1) == len(folders): - logger.info(f" Scanned {i+1}/{len(folders)} folders ({len(replik_map):,} speeches)") - return replik_map - - -def fix_replik(replik_map: dict[str, bool]) -> set[str]: - """ - Update speeches where DB is_reply=False but JSON says True. - Returns the set of affected talk IDs. - """ - # Only need to check speeches where DB currently says False - db_false = pg.execute("SELECT id FROM speeches WHERE is_reply = false") - db_false_ids = {row["id"] for row in db_false} - logger.info(f"Talks with is_reply=False in DB: {len(db_false_ids):,}") - - to_fix = [tid for tid in db_false_ids if replik_map.get(tid) is True] - logger.info(f"Talks to correct (JSON says True, DB says False): {len(to_fix):,}") - - if not to_fix: - return set() - - for i in range(0, len(to_fix), BATCH_SIZE): - batch = to_fix[i : i + BATCH_SIZE] - pg.execute_many( - "UPDATE speeches SET is_reply = true WHERE id = %s", - [(tid,) for tid in batch], - ) - if (i + BATCH_SIZE) % 5000 == 0 or i + BATCH_SIZE >= len(to_fix): - logger.info(f" Updated {min(i+BATCH_SIZE, len(to_fix)):,}/{len(to_fix):,}") - - return set(to_fix) - - -def get_affected_dates(fixed_ids: set[str]) -> list[str]: - """Return distinct dates for the speeches whose is_reply was corrected.""" - if not fixed_ids: - return [] - rows = pg.execute( - "SELECT DISTINCT date::text AS date FROM speeches WHERE id = ANY(%s::text[])", - (list(fixed_ids),), - ) - return [row["date"] for row in rows if row.get("date")] - - -def reassign_debate_ids(dates: list[str]) -> None: - """ - For each date, clear speeches.debate and re-assign using assign_debate_ids(). - This is the same logic as make_debate_ids() but scoped to the given dates. - """ - logger.info(f"Re-assigning debate IDs for {len(dates):,} dates …") - for i, date in enumerate(sorted(dates)): - # Clear existing debate assignments for this date - pg.execute_void( - "UPDATE speeches SET debate = NULL WHERE date = %s::date", - (date,), - ) - - speeches = pg.execute( - """ - SELECT id, is_reply - FROM speeches - WHERE date = %s::date - ORDER BY sequence ASC - """, - (date,), - ) - if not speeches: - continue - - updated = assign_debate_ids(list(speeches), date) - pg.execute_many( - "UPDATE speeches SET debate = %s WHERE id = %s", - [(doc["debate"], doc["id"]) for doc in updated], - ) - - if (i + 1) % 100 == 0 or (i + 1) == len(dates): - logger.info(f" {i+1}/{len(dates)} dates processed") - - -def remove_stale_debate_rows() -> int: - """ - Delete debate rows whose debate ID no longer appears in speeches.debate. - This cleans up the old one-per-talk debate rows that became invalid after - debate IDs were re-assigned. - """ - result = pg.execute( - """ - DELETE FROM debates - WHERE debate NOT IN (SELECT DISTINCT debate FROM speeches WHERE debate IS NOT NULL) - RETURNING debate - """ - ) - count = len(result) if result else 0 - logger.info(f"Removed {count:,} stale debate rows") - return count - - -def main() -> None: - logger.info("=== Step 1: Build is_reply map from JSON files ===") - replik_map = build_replik_map() - logger.info(f"Total speeches in JSON files: {len(replik_map):,}") - - logger.info("=== Step 2: Fix is_reply in DB ===") - fixed_ids = fix_replik(replik_map) - if not fixed_ids: - logger.info("Nothing to fix — is_reply is already correct.") - return - - logger.info("=== Step 3: Find affected dates ===") - dates = get_affected_dates(fixed_ids) - logger.info(f"Affected dates: {len(dates):,}") - - logger.info("=== Step 4: Re-assign debate IDs ===") - reassign_debate_ids(dates) - - logger.info("=== Step 5: Remove stale debate rows ===") - remove_stale_debate_rows() - - logger.info("=== Done! ===") - logger.info("Next: run 'python scripts/debates.py' (or deploy_debate_embeddings.sh)") - logger.info("to generate summaries for the newly grouped multi-talk debates.") - - -if __name__ == "__main__": - main() diff --git a/scripts/motions_progress.sh b/scripts/motions_progress.sh deleted file mode 100755 index efaca5d..0000000 --- a/scripts/motions_progress.sh +++ /dev/null @@ -1,29 +0,0 @@ -#!/bin/bash -# Progress of the motion embedding backfills. Run anytime: -# bash scripts/motions_progress.sh -set -e -cd "$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)" -export PGPASSWORD=$(grep '^PG_PASSWORD=' .env | cut -d= -f2) -PGDB=$(grep '^PG_DB=' .env | cut -d= -f2) -PGUSER=$(grep '^PG_USER=' .env | cut -d= -f2) -PGHOST=$(grep '^PG_HOST=' .env | cut -d= -f2) - -echo "=== Motion embedding coverage ($(date '+%H:%M:%S')) ===" -psql -h "$PGHOST" -U "$PGUSER" -d "$PGDB" -tA -F $'\t' <<'SQL' | column -t -s $'\t' -SELECT 'full-text chunks (motions)' AS what, - count(*) FILTER (WHERE c.id IS NOT NULL) AS done, - count(*) AS total, - round(100.0*count(*) FILTER (WHERE c.id IS NOT NULL)/nullif(count(*),0),1)||'%' AS pct -FROM (SELECT dok_id FROM motions WHERE has_text) m -LEFT JOIN LATERAL (SELECT 1 AS id FROM motion_chunks c WHERE c.motion_id = m.dok_id LIMIT 1) c ON true -UNION ALL -SELECT 'yrkande embeddings', - count(*) FILTER (WHERE embedding IS NOT NULL), - count(*), - round(100.0*count(*) FILTER (WHERE embedding IS NOT NULL)/nullif(count(*),0),1)||'%' -FROM motion_yrkanden; -SQL - -echo -echo "=== Backfill processes ===" -pgrep -af "make_embeddings.py|backfill_motions.sh" | grep -v pgrep || echo " (none running — backfills finished)" diff --git a/scripts/motions_to_postgres.py b/scripts/motions_to_postgres.py deleted file mode 100644 index 2b7dd86..0000000 --- a/scripts/motions_to_postgres.py +++ /dev/null @@ -1,268 +0,0 @@ -""" -Läser in motioner från JSON-filer (dokumentstatus-kuvert) till PostgreSQL. - -Används av sync_motions.py (update_folder) och kan köras direkt för att -(om)ladda alla mappar i motioner/: - - python scripts/motions_to_postgres.py -""" -from pathlib import Path - -import json -import logging -import os -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 bs4 import BeautifulSoup -from postgres_client import pg -from scripts.documents_to_postgres import _parse_date, clean_speaker_name - -logging.basicConfig(level=logging.WARNING, format="%(asctime)s - %(levelname)s - %(message)s") - -# Rader kortare än så räknas som textlösa (inskannad PDF med stubb-HTML) -MIN_TEXT_LEN = 200 - - -def _as_list(v) -> list: - """Riksdagens XML→JSON: ett barn blir dict, flera blir list, inget blir None.""" - if v is None: - return [] - return v if isinstance(v, list) else [v] - - -def _clean(v): - """API:et serialiserar null som strängen 'None'.""" - if v is None or v == "None": - return None - return v - - -def html_to_text(html: str) -> str: - """Extraherar ren text ur dokumentets html-fält (inleds med stort