The project could only run from /home/lasse/riksdagen: 22 files hardcoded that path, mostly as an `os.chdir(...)` + `sys.path.append(...)` prelude. Those now derive the root from the file's own location via bootstrap.py, and bulk data resolves against PLENUM_DATA_DIR. requirements.txt was from March 2023 and listed Streamlit, altair and pydeck while omitting fastapi, uvicorn, psycopg2 and pgvector. Replaced with a pyproject.toml derived from the actual import graph, plus ruff and pytest config. Postgres opened its connection pool in __init__, so `import backend.app` failed outright without a reachable database — breaking test collection and any tooling that merely imports the app. The pool is now opened on first use behind a lock. systemd units and an nginx site move to deploy/examples/ with __PROJECT_ROOT__ and __DOMAIN__ placeholders. Real values belong in the deployment's own deploy/prod/, which is a path upstream never writes to, so the two cannot conflict on merge. Removed: scripts/migrate_keys.py (an already-executed Arango migration referenced nowhere) and a dead `import talks2db` in download_talks.py that pointed at a file excluded from this repo. Verified: `import backend.app` now succeeds with neither a database nor any network, and all rewritten scripts import from a foreign working directory with their module docstrings intact. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>main
parent
843db25211
commit
d36ace941b
31 changed files with 330 additions and 492 deletions
@ -1,23 +1,33 @@ |
||||
""" |
||||
Ensures the project root is in sys.path and sets the working directory to the project root. |
||||
"""Locate the project root and make it importable. |
||||
|
||||
Scripts under ``scripts/`` are run directly (``python scripts/sync_talks.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. |
||||
|
||||
Import this module at the top of your scripts to make project modules (like utils.py) importable from anywhere. |
||||
The predecessor hardcoded ``/home/lasse/riksdagen`` in 22 files, which is why the |
||||
project could only ever run on one machine. |
||||
""" |
||||
from __future__ import annotations |
||||
|
||||
import os |
||||
import sys |
||||
from pathlib import Path |
||||
|
||||
PROJECT_ROOT = Path(__file__).resolve().parent |
||||
|
||||
# Bulk source data — downloaded corpora, not repository content. Kept outside the |
||||
# tree by default so a checkout stays small; point PLENUM_DATA_DIR at an existing |
||||
# download to reuse one. |
||||
DATA_DIR = Path(os.environ.get("PLENUM_DATA_DIR") or PROJECT_ROOT / "data") |
||||
|
||||
PROJECT_ROOT = "/home/lasse/riksdagen" |
||||
|
||||
def set_project_root() -> None: |
||||
""" |
||||
Sets the working directory to the project root and ensures it's in sys.path. |
||||
|
||||
Returns: |
||||
None |
||||
""" |
||||
os.chdir(PROJECT_ROOT) |
||||
if PROJECT_ROOT not in sys.path: |
||||
sys.path.insert(0, PROJECT_ROOT) |
||||
"""Put the project root on ``sys.path`` and make it the working directory.""" |
||||
root = str(PROJECT_ROOT) |
||||
if root not in sys.path: |
||||
sys.path.insert(0, root) |
||||
os.chdir(root) |
||||
|
||||
|
||||
set_project_root() |
||||
|
||||
@ -0,0 +1,57 @@ |
||||
# nginx site for Plenum: static frontend + API reverse proxy. |
||||
# |
||||
# Substitute __DOMAIN__ and __PROJECT_ROOT__, then place in |
||||
# /etc/nginx/sites-available/ and symlink into sites-enabled/. |
||||
# |
||||
# Obtain the certificate first (certbot --nginx -d __DOMAIN__); the TLS paths |
||||
# below assume Let's Encrypt's default layout. |
||||
|
||||
server { |
||||
listen 80; |
||||
listen [::]:80; |
||||
server_name __DOMAIN__; |
||||
|
||||
# ACME challenges must stay on port 80. |
||||
location /.well-known/acme-challenge/ { root /var/www/html; } |
||||
location / { return 301 https://$host$request_uri; } |
||||
} |
||||
|
||||
server { |
||||
listen 443 ssl http2; |
||||
listen [::]:443 ssl http2; |
||||
server_name __DOMAIN__; |
||||
|
||||
ssl_certificate /etc/letsencrypt/live/__DOMAIN__/fullchain.pem; |
||||
ssl_certificate_key /etc/letsencrypt/live/__DOMAIN__/privkey.pem; |
||||
|
||||
# Chat and research stream over SSE. Buffering would hold events until the |
||||
# response completed, which looks exactly like a hung request to the user. |
||||
location /api/ { |
||||
proxy_pass http://127.0.0.1:8000; |
||||
proxy_http_version 1.1; |
||||
proxy_set_header Host $host; |
||||
proxy_set_header X-Real-IP $remote_addr; |
||||
proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for; |
||||
proxy_set_header X-Forwarded-Proto $scheme; |
||||
proxy_buffering off; |
||||
proxy_cache off; |
||||
# A deep-research turn can legitimately run for minutes. |
||||
proxy_read_timeout 600s; |
||||
} |
||||
|
||||
root __PROJECT_ROOT__/frontend/dist; |
||||
index index.html; |
||||
|
||||
# Vite emits content-hashed filenames, so assets are safe to cache forever. |
||||
location /assets/ { |
||||
expires 1y; |
||||
add_header Cache-Control "public, immutable"; |
||||
} |
||||
|
||||
# Single-page app: unknown paths are client-side routes, not 404s. |
||||
location / { try_files $uri $uri/ /index.html; } |
||||
|
||||
gzip on; |
||||
gzip_types text/css application/javascript application/json image/svg+xml; |
||||
gzip_min_length 1024; |
||||
} |
||||
@ -0,0 +1,45 @@ |
||||
# Plenum API (FastAPI behind nginx). |
||||
# |
||||
# Copy to /etc/systemd/system/plenum-api.service and substitute: |
||||
# __PROJECT_ROOT__ absolute path to the checkout, e.g. /srv/plenum |
||||
# __USER__ the account that owns the checkout |
||||
# |
||||
# sed -e "s|__PROJECT_ROOT__|$PWD|g" -e "s|__USER__|$USER|g" \ |
||||
# deploy/examples/plenum-api.service.example \ |
||||
# | sudo tee /etc/systemd/system/plenum-api.service |
||||
# sudo systemctl daemon-reload && sudo systemctl enable --now plenum-api |
||||
|
||||
[Unit] |
||||
Description=Plenum API |
||||
Wants=network-online.target |
||||
After=network-online.target postgresql.service |
||||
# Don't hammer a genuinely broken deploy — give up after 5 failures in 5 minutes |
||||
# so the unit lands in `failed`, where it is visible, instead of looping forever. |
||||
StartLimitIntervalSec=300 |
||||
StartLimitBurst=5 |
||||
|
||||
[Service] |
||||
Type=exec |
||||
User=__USER__ |
||||
WorkingDirectory=__PROJECT_ROOT__ |
||||
EnvironmentFile=__PROJECT_ROOT__/.env |
||||
ExecStart=__PROJECT_ROOT__/.venv/bin/uvicorn backend.app:app --host 127.0.0.1 --port 8000 |
||||
|
||||
# Research jobs run as detached children (start_new_session=True in |
||||
# backend/services/research/jobs.py) so a long dig survives an API restart. The |
||||
# default KillMode=control-group would kill the whole cgroup and take those jobs |
||||
# with it, silently defeating that design. With KillMode=process only uvicorn is |
||||
# signalled; orphans are reaped by reap_stale_jobs() and capped by MAX_JOB_RUNTIME_SECS. |
||||
KillMode=process |
||||
TimeoutStopSec=30 |
||||
|
||||
Restart=always |
||||
RestartSec=5 |
||||
|
||||
# journalctl -u plenum-api -f |
||||
StandardOutput=journal |
||||
StandardError=journal |
||||
SyslogIdentifier=plenum-api |
||||
|
||||
[Install] |
||||
WantedBy=multi-user.target |
||||
@ -0,0 +1,26 @@ |
||||
# Daily incremental ingest. Paired with plenum-sync.timer. |
||||
# |
||||
# Copy to /etc/systemd/system/plenum-sync.service and substitute __PROJECT_ROOT__ |
||||
# and __USER__ as described in plenum-api.service.example. |
||||
# |
||||
# `--source all` runs every source declared under `sources:` in parliament.yaml. |
||||
# Use `--source speeches` / `--source documents` to split them across timers. |
||||
|
||||
[Unit] |
||||
Description=Plenum daily source sync |
||||
After=network-online.target |
||||
Wants=network-online.target |
||||
|
||||
[Service] |
||||
Type=oneshot |
||||
User=__USER__ |
||||
WorkingDirectory=__PROJECT_ROOT__ |
||||
EnvironmentFile=__PROJECT_ROOT__/.env |
||||
ExecStart=__PROJECT_ROOT__/.venv/bin/python -m ingest.cli sync --source all |
||||
|
||||
# journalctl -u plenum-sync |
||||
StandardOutput=journal |
||||
StandardError=journal |
||||
|
||||
[Install] |
||||
WantedBy=multi-user.target |
||||
@ -0,0 +1,17 @@ |
||||
# Runs plenum-sync.service once a day. |
||||
# |
||||
# Pick a time shortly after the source publishes its daily update. Sources that |
||||
# are heavy on the embedding host are best staggered rather than run together. |
||||
|
||||
[Unit] |
||||
Description=Run the Plenum source sync daily |
||||
|
||||
[Timer] |
||||
OnCalendar=*-*-* 06:00:00 |
||||
# If the machine was off at the scheduled time, run once it comes back up. |
||||
Persistent=true |
||||
# Avoid every deployment hitting the upstream open-data server at the same second. |
||||
RandomizedDelaySec=900 |
||||
|
||||
[Install] |
||||
WantedBy=timers.target |
||||
@ -1,41 +0,0 @@ |
||||
[Unit] |
||||
Description=Riksdagen API (FastAPI, serves rixdagen.se/api via nginx) |
||||
Documentation=https://rixdagen.se |
||||
# Postgres runs in the riksdagen-pg container, and env_manager pulls part of the |
||||
# environment over the network at import time — so both must be up first. |
||||
Wants=network-online.target |
||||
After=network-online.target docker.service |
||||
# Don't hammer a genuinely broken deploy — give up after 5 failures in 5 min so |
||||
# the unit lands in `failed` where it is visible, instead of looping forever. |
||||
StartLimitIntervalSec=300 |
||||
StartLimitBurst=5 |
||||
|
||||
[Service] |
||||
Type=exec |
||||
User=lasse |
||||
Group=lasse |
||||
WorkingDirectory=/home/lasse/riksdagen |
||||
# No EnvironmentFile: the app loads .env itself (config.py -> env_manager.set_env()) |
||||
# and that file contains keys systemd's parser would read differently, e.g. |
||||
# "LLM_MODEL =" with a trailing space. WorkingDirectory is what makes it resolve. |
||||
ExecStart=/home/lasse/riksdagen/.venv/bin/uvicorn backend.app:app --host 127.0.0.1 --port 8000 |
||||
|
||||
# Research jobs run as detached child processes (start_new_session=True in |
||||
# backend/services/research/jobs.py) precisely so a long dig survives an API |
||||
# restart. The default KillMode=control-group would kill the whole cgroup on |
||||
# restart and take those jobs with it, silently breaking that design. With |
||||
# KillMode=process only uvicorn is signalled; any orphaned job is cleaned up by |
||||
# reap_stale_jobs() and capped by MAX_JOB_RUNTIME_SECS. |
||||
KillMode=process |
||||
TimeoutStopSec=30 |
||||
|
||||
Restart=always |
||||
RestartSec=5 |
||||
|
||||
# journalctl -u riksdagen-api -f |
||||
StandardOutput=journal |
||||
StandardError=journal |
||||
SyslogIdentifier=riksdagen-api |
||||
|
||||
[Install] |
||||
WantedBy=multi-user.target |
||||
@ -1,18 +0,0 @@ |
||||
[Unit] |
||||
Description=Riksdagen daily motions sync |
||||
# Wait for network before starting |
||||
After=network-online.target |
||||
Wants=network-online.target |
||||
|
||||
[Service] |
||||
Type=oneshot |
||||
User=lasse |
||||
WorkingDirectory=/home/lasse/riksdagen |
||||
EnvironmentFile=/home/lasse/riksdagen/.env |
||||
ExecStart=/home/lasse/riksdagen/.venv/bin/python /home/lasse/riksdagen/scripts/sync_motions.py |
||||
# Log stdout/stderr to the systemd journal (view with: journalctl -u riksdagen-motions-sync) |
||||
StandardOutput=journal |
||||
StandardError=journal |
||||
|
||||
[Install] |
||||
WantedBy=multi-user.target |
||||
@ -1,12 +0,0 @@ |
||||
[Unit] |
||||
Description=Run riksdagen daily motions sync at 06:30 |
||||
|
||||
[Timer] |
||||
# 06:30 — offset from the 06:00 talks sync so they don't compete for the |
||||
# embedding host and the Postgres pool |
||||
OnCalendar=*-*-* 06:30:00 |
||||
# If the server was off at 06:30, run the job as soon as it comes back up |
||||
Persistent=true |
||||
|
||||
[Install] |
||||
WantedBy=timers.target |
||||
@ -1,19 +0,0 @@ |
||||
[Unit] |
||||
Description=Riksdagen daily talk sync |
||||
# Wait for network before starting |
||||
After=network-online.target |
||||
Wants=network-online.target |
||||
|
||||
[Service] |
||||
Type=oneshot |
||||
User=lasse |
||||
WorkingDirectory=/home/lasse/riksdagen |
||||
# Loads ARANGO_PWD and other env vars from the project .env file |
||||
EnvironmentFile=/home/lasse/riksdagen/.env |
||||
ExecStart=/home/lasse/riksdagen/.venv/bin/python /home/lasse/riksdagen/scripts/sync_talks.py |
||||
# Log stdout/stderr to the systemd journal (view with: journalctl -u riksdagen-sync) |
||||
StandardOutput=journal |
||||
StandardError=journal |
||||
|
||||
[Install] |
||||
WantedBy=multi-user.target |
||||
@ -1,11 +0,0 @@ |
||||
[Unit] |
||||
Description=Run riksdagen daily talk sync at 06:00 |
||||
|
||||
[Timer] |
||||
# Run every day at 06:00 |
||||
OnCalendar=*-*-* 06:00:00 |
||||
# If the server was off at 06:00, run the job as soon as it comes back up |
||||
Persistent=true |
||||
|
||||
[Install] |
||||
WantedBy=timers.target |
||||
@ -0,0 +1,85 @@ |
||||
[build-system] |
||||
requires = ["setuptools>=68"] |
||||
build-backend = "setuptools.build_meta" |
||||
|
||||
[project] |
||||
name = "plenum" |
||||
version = "0.1.0" |
||||
description = "Search, chat and research over parliamentary speeches and documents" |
||||
readme = "README.md" |
||||
requires-python = ">=3.10" |
||||
license = { text = "AGPL-3.0-or-later" } |
||||
authors = [{ name = "Lasse Edfast", email = "lasse@edfast.se" }] |
||||
keywords = ["parliament", "open-data", "rag", "civic-tech", "full-text-search"] |
||||
classifiers = [ |
||||
"License :: OSI Approved :: GNU Affero General Public License v3 or later (AGPLv3+)", |
||||
"Programming Language :: Python :: 3", |
||||
"Topic :: Text Processing :: Indexing", |
||||
] |
||||
|
||||
# Derived from the actual import graph, not from a frozen environment, so the list |
||||
# stays readable and reviewable. Run `python -m scripts.check_deps` to re-verify. |
||||
dependencies = [ |
||||
"fastapi>=0.110", |
||||
"uvicorn[standard]>=0.27", |
||||
"pydantic>=2.6", |
||||
"psycopg2-binary>=2.9", |
||||
"pgvector>=0.2.5", |
||||
"openai>=1.30", |
||||
"python-dotenv>=1.0", |
||||
"PyYAML>=6.0", |
||||
"requests>=2.31", |
||||
"httpx>=0.27", |
||||
"beautifulsoup4>=4.12", |
||||
"bcrypt>=4.1", |
||||
"cryptography>=42.0", |
||||
"psutil>=5.9", |
||||
] |
||||
|
||||
[project.optional-dependencies] |
||||
# The MCP server is optional: the web app runs without it. |
||||
mcp = ["fastmcp>=0.2"] |
||||
dev = [ |
||||
"pytest>=8.0", |
||||
"ruff>=0.4", |
||||
] |
||||
|
||||
[project.urls] |
||||
Homepage = "https://github.com/lasseedfast/plenum" |
||||
Source = "https://git.edfast.se/plenum/plenum" |
||||
|
||||
[project.scripts] |
||||
plenum-ingest = "ingest.cli:main" |
||||
|
||||
[tool.setuptools] |
||||
# Root-level modules kept for import compatibility with existing call sites. |
||||
py-modules = ["postgres_client", "utils", "bootstrap", "parliament"] |
||||
|
||||
[tool.setuptools.packages.find] |
||||
include = ["backend*", "packages*", "_postgres*", "ingest*"] |
||||
exclude = ["tests*", "scripts*", "docs*", "frontend*"] |
||||
|
||||
[tool.setuptools.package-data] |
||||
"_postgres" = ["*.sql", "migrations/*.sql"] |
||||
|
||||
[tool.ruff] |
||||
line-length = 100 |
||||
target-version = "py310" |
||||
|
||||
[tool.ruff.lint] |
||||
select = ["E", "F", "W", "I", "UP", "B"] |
||||
ignore = [ |
||||
"E501", # line length is handled by the formatter |
||||
"B008", # FastAPI's Depends() in defaults is idiomatic |
||||
] |
||||
|
||||
[tool.ruff.lint.per-file-ignores] |
||||
# Star-imports of the console helpers predate this config; tracked for cleanup. |
||||
"backend/services/chat.py" = ["F403", "F405"] |
||||
"backend/services/llm_tools.py" = ["F403", "F405"] |
||||
"backend/services/mp_chat.py" = ["F403", "F405"] |
||||
"scripts/debates.py" = ["F403", "F405"] |
||||
|
||||
[tool.pytest.ini_options] |
||||
testpaths = ["tests"] |
||||
pythonpath = ["."] |
||||
@ -1,324 +0,0 @@ |
||||
""" |
||||
One-time migration: swap talks._key from anforande_id (UUID) to id (dok_id). |
||||
|
||||
After this migration: |
||||
- talks._key = old "id" field (e.g. "H90982") |
||||
- talks.anforande_id = preserved old UUID for reference |
||||
- chunks.parent_id = "talks/H90982" (was "talks/<uuid>") |
||||
- debates.talk_ids = ["talks/H90982", ...] |
||||
|
||||
Execution order matters: |
||||
Step 1 — build mapping from talks (while old docs exist) |
||||
Step 2 — update chunks.parent_id (single AQL; DOCUMENT lookup needs old talks) |
||||
Step 3 — update debates.talk_ids (single AQL; DOCUMENT lookup needs old talks) |
||||
Step 4 — re-key talks (insert new + delete old) |
||||
|
||||
Usage: |
||||
python scripts/migrate_keys.py # live run |
||||
python scripts/migrate_keys.py --dry-run # count only, no changes |
||||
""" |
||||
|
||||
import os |
||||
import sys |
||||
import argparse |
||||
import logging |
||||
import time |
||||
|
||||
os.chdir("/home/lasse/riksdagen") |
||||
sys.path.append("/home/lasse/riksdagen") |
||||
|
||||
# Connect directly to localhost with no read timeout. |
||||
# The default 60s client timeout is too short for full-collection scans. |
||||
from arango import ArangoClient |
||||
from arango.http import DefaultHTTPClient |
||||
from dotenv import load_dotenv |
||||
load_dotenv() |
||||
_client = ArangoClient(hosts="http://localhost:8529", http_client=DefaultHTTPClient(request_timeout=None)) |
||||
db = _client.db("riksdagen", username="riksdagen", password=os.environ["ARANGO_PWD"]) |
||||
|
||||
logging.basicConfig( |
||||
level=logging.INFO, |
||||
format="%(asctime)s [%(levelname)s] %(message)s", |
||||
) |
||||
logger = logging.getLogger(__name__) |
||||
|
||||
BATCH_SIZE = 500 |
||||
|
||||
|
||||
def safe_batch_execute(func, *args, max_retries=5, **kwargs): |
||||
"""Executes an ArangoDB batch operation with exponential backoff for lock timeouts.""" |
||||
for attempt in range(max_retries): |
||||
try: |
||||
return func(*args, **kwargs) |
||||
except Exception as e: |
||||
# Check if the error is a lock timeout (ERR 1200) |
||||
if "1200" in str(e) or "timeout waiting to lock key" in str(e).lower(): |
||||
if attempt < max_retries - 1: |
||||
time.sleep(0.5 * (2 ** attempt)) # 0.5s, 1.0s, 2.0s, 4.0s |
||||
continue |
||||
# Raise immediately if it's a different error or we're out of retries |
||||
raise e |
||||
|
||||
|
||||
def build_mapping() -> dict[str, str]: |
||||
"""Return {old_uuid: new_id} for every talk that still has a UUID _key.""" |
||||
logger.info("Building old→new key mapping from talks collection...") |
||||
cursor = db.aql.execute( |
||||
""" |
||||
FOR t IN talks |
||||
FILTER t.id != null AND t.id != "" |
||||
FILTER t._key != t.id |
||||
RETURN {old: t._key, new: t.id} |
||||
""", |
||||
batch_size=5000, |
||||
ttl=600, |
||||
) |
||||
mapping = {row["old"]: row["new"] for row in cursor} |
||||
logger.info(f" {len(mapping)} talks need re-keying") |
||||
return mapping |
||||
|
||||
|
||||
def migrate_chunks(mapping: dict[str, str], dry_run: bool) -> None: |
||||
""" |
||||
Update chunks.parent_id from 'talks/{uuid}' → 'talks/{id}'. |
||||
|
||||
Read phase: stream chunk _key + parent_id in Python (read-only, no locks). |
||||
Write phase: batch UPDATE by _key (point locks, no cross-collection join). |
||||
Must run BEFORE talk re-keying so the mapping is still valid. |
||||
""" |
||||
if not db.has_collection("chunks"): |
||||
logger.info("No chunks collection — skipping.") |
||||
return |
||||
|
||||
logger.info("Step 2: Scanning chunks for stale parent_ids...") |
||||
cursor = db.aql.execute( |
||||
"FOR c IN chunks RETURN {k: c._key, p: c.parent_id}", |
||||
batch_size=10000, |
||||
ttl=3600, |
||||
stream=True, |
||||
) |
||||
updates = [] |
||||
for row in cursor: |
||||
parent = row["p"] or "" |
||||
if not parent.startswith("talks/"): |
||||
continue |
||||
old_talk_key = parent[len("talks/"):] |
||||
new_talk_key = mapping.get(old_talk_key) |
||||
if new_talk_key: |
||||
updates.append({"k": row["k"], "p": f"talks/{new_talk_key}"}) |
||||
|
||||
# Deduplicate updates to prevent batch deadlocks on live data |
||||
unique_updates = {u["k"]: u["p"] for u in updates} |
||||
updates = [{"k": k, "p": p} for k, p in unique_updates.items()] |
||||
|
||||
logger.info(f" {len(updates)} unique chunks need updating") |
||||
if dry_run: |
||||
logger.info(" [dry-run] skipping writes") |
||||
return |
||||
|
||||
chunks_col = db.collection("chunks") |
||||
total = 0 |
||||
for i in range(0, len(updates), 100): |
||||
batch = [{"_key": u["k"], "parent_id": u["p"]} for u in updates[i : i + 100]] |
||||
safe_batch_execute(chunks_col.update_many, batch, silent=True) |
||||
total += len(batch) |
||||
print(f" chunks updated: {total}/{len(updates)}", end="\r") |
||||
print() |
||||
logger.info(f"Step 2 done: {total} chunks updated") |
||||
|
||||
|
||||
def migrate_debates(mapping: dict[str, str], dry_run: bool) -> None: |
||||
""" |
||||
Update debates.talk_ids from ["talks/{uuid}", ...] → ["talks/{id}", ...]. |
||||
|
||||
Read phase: stream debate _key + talk_ids in Python (read-only, no locks). |
||||
Write phase: batch UPDATE by _key (point locks, no cross-collection join). |
||||
Must run BEFORE talk re-keying so the mapping is still valid. |
||||
""" |
||||
if not db.has_collection("debates"): |
||||
logger.info("No debates collection — skipping.") |
||||
return |
||||
|
||||
logger.info("Step 3: Scanning debates for stale talk_ids...") |
||||
cursor = db.aql.execute( |
||||
"FOR d IN debates FILTER d.talk_ids != null RETURN {k: d._key, ids: d.talk_ids}", |
||||
batch_size=5000, |
||||
ttl=300, |
||||
) |
||||
updates = [] |
||||
for row in cursor: |
||||
new_ids = [] |
||||
changed = False |
||||
for tid in row["ids"]: |
||||
if tid.startswith("talks/"): |
||||
old_key = tid[len("talks/"):] |
||||
new_key = mapping.get(old_key) |
||||
if new_key: |
||||
new_ids.append(f"talks/{new_key}") |
||||
changed = True |
||||
continue |
||||
new_ids.append(tid) |
||||
if changed: |
||||
updates.append({"k": row["k"], "ids": new_ids}) |
||||
|
||||
# Deduplicate updates to prevent batch deadlocks on live data |
||||
unique_updates = {u["k"]: u["ids"] for u in updates} |
||||
updates = [{"k": k, "ids": ids} for k, ids in unique_updates.items()] |
||||
|
||||
logger.info(f" {len(updates)} unique debates need updating") |
||||
if dry_run: |
||||
logger.info(" [dry-run] skipping writes") |
||||
return |
||||
|
||||
debates_col = db.collection("debates") |
||||
total = 0 |
||||
for i in range(0, len(updates), 100): |
||||
batch = [{"_key": u["k"], "talk_ids": u["ids"]} for u in updates[i : i + 100]] |
||||
safe_batch_execute(debates_col.update_many, batch, silent=True) |
||||
total += len(batch) |
||||
print(f" debates updated: {total}/{len(updates)}", end="\r") |
||||
print() |
||||
logger.info(f"Step 3 done: {total} debates updated") |
||||
|
||||
|
||||
def migrate_talks(mapping: dict[str, str], dry_run: bool) -> None: |
||||
"""Re-insert talks with new _key, preserve old UUID as anforande_id field.""" |
||||
talks_col = db.collection("talks") |
||||
old_keys = list(mapping.keys()) |
||||
total = len(old_keys) |
||||
done = 0 |
||||
|
||||
logger.info(f"Step 4: Re-keying {total} talks (batch_size={BATCH_SIZE})...") |
||||
|
||||
for i in range(0, total, BATCH_SIZE): |
||||
batch_old_keys = old_keys[i : i + BATCH_SIZE] |
||||
|
||||
docs = list(db.aql.execute( |
||||
"FOR k IN @keys RETURN DOCUMENT(CONCAT('talks/', k))", |
||||
bind_vars={"keys": batch_old_keys}, |
||||
batch_size=BATCH_SIZE, |
||||
)) |
||||
|
||||
new_docs = [] |
||||
for doc in docs: |
||||
if doc is None: |
||||
continue |
||||
old_key = doc["_key"] |
||||
new_key = mapping.get(old_key) |
||||
if not new_key: |
||||
continue |
||||
new_doc = {k: v for k, v in doc.items() if k not in ("_key", "_id", "_rev")} |
||||
new_doc["_key"] = new_key |
||||
new_doc["anforande_id"] = old_key # preserve UUID |
||||
new_docs.append(new_doc) |
||||
|
||||
done += len(new_docs) |
||||
print(f" talks {done}/{total}", end="\r") |
||||
|
||||
if dry_run: |
||||
continue |
||||
|
||||
if new_docs: |
||||
safe_batch_execute(talks_col.insert_many, new_docs, overwrite=True) |
||||
|
||||
safe_batch_execute(talks_col.delete_many, [{"_key": k} for k in batch_old_keys], silent=True) |
||||
|
||||
print() |
||||
if dry_run: |
||||
logger.info(f" [dry-run] {done} talks would be re-keyed") |
||||
else: |
||||
logger.info(f"Step 4 done: {done} talks re-keyed") |
||||
|
||||
|
||||
def rekey_chunks(dry_run: bool) -> None: |
||||
""" |
||||
Re-key chunks from '{uuid}:{idx}' to '{short_id}:{idx}'. |
||||
|
||||
Talks are already re-keyed; use talks.anforande_id (preserved old UUID) for the mapping. |
||||
Runs server-side via batched AQL INSERT+REMOVE to avoid transferring embedding vectors. |
||||
Requires a persistent index on talks.anforande_id for performance. |
||||
""" |
||||
if not db.has_collection("chunks"): |
||||
logger.info("No chunks collection — skipping chunk re-key.") |
||||
return |
||||
|
||||
# Ensure index exists so the FILTER tt.anforande_id == old_uuid lookup is fast. |
||||
talks_col = db.collection("talks") |
||||
existing_fields = {f for idx in talks_col.indexes() for f in idx.get("fields", [])} |
||||
if "anforande_id" not in existing_fields: |
||||
logger.info(" Creating persistent index on talks.anforande_id...") |
||||
talks_col.add_persistent_index(fields=["anforande_id"], unique=False, sparse=True) |
||||
|
||||
if dry_run: |
||||
result = list(db.aql.execute( |
||||
'FOR c IN chunks FILTER CONTAINS(c._key, "-") COLLECT WITH COUNT INTO n RETURN n', |
||||
ttl=120, |
||||
)) |
||||
logger.info(f" [dry-run] {result[0] if result else 0} chunks would be re-keyed") |
||||
return |
||||
|
||||
logger.info("Step 5: Re-keying chunks from UUID to short-id format...") |
||||
total = 0 |
||||
while True: |
||||
result = list(db.aql.execute( |
||||
""" |
||||
FOR c IN chunks |
||||
FILTER CONTAINS(c._key, "-") |
||||
LET parts = SPLIT(c._key, ":") |
||||
LET old_uuid = parts[0] |
||||
LET idx = parts[1] |
||||
LET t = FIRST(FOR tt IN talks FILTER tt.anforande_id == old_uuid LIMIT 1 RETURN tt) |
||||
FILTER t != null |
||||
LET new_key = CONCAT(t._key, ":", idx) |
||||
LET new_doc = MERGE(UNSET(c, "_key", "_id", "_rev"), |
||||
{_key: new_key, parent_id: CONCAT("talks/", t._key)}) |
||||
INSERT new_doc INTO chunks OPTIONS {overwriteMode: "ignore"} |
||||
REMOVE c IN chunks |
||||
LIMIT 200 |
||||
RETURN 1 |
||||
""", |
||||
ttl=300, |
||||
)) |
||||
batch_count = len(result) |
||||
total += batch_count |
||||
print(f" chunks re-keyed: {total}", end="\r") |
||||
if batch_count == 0: |
||||
break |
||||
print() |
||||
logger.info(f"Step 5 done: {total} chunks re-keyed") |
||||
|
||||
|
||||
def main(): |
||||
parser = argparse.ArgumentParser(description="Migrate talks _key from UUID to id (dok_id)") |
||||
parser.add_argument("--dry-run", action="store_true", help="Print counts only, make no changes") |
||||
parser.add_argument("--rekey-chunks", action="store_true", help="Also re-key chunks to short-id format (step 5)") |
||||
args = parser.parse_args() |
||||
|
||||
if args.dry_run: |
||||
logger.info("=== DRY RUN — no data will be modified ===") |
||||
else: |
||||
logger.info("=== LIVE RUN — data will be modified ===") |
||||
|
||||
mapping = build_mapping() |
||||
if not mapping: |
||||
logger.info("Nothing to migrate — all talks already use short keys.") |
||||
if args.rekey_chunks: |
||||
rekey_chunks(args.dry_run) |
||||
return |
||||
|
||||
# Steps 2 and 3 must run before step 4 (talk re-keying) — mapping uses old UUIDs as keys. |
||||
migrate_chunks(mapping, args.dry_run) |
||||
migrate_debates(mapping, args.dry_run) |
||||
migrate_talks(mapping, args.dry_run) |
||||
|
||||
if args.rekey_chunks: |
||||
rekey_chunks(args.dry_run) |
||||
|
||||
if args.dry_run: |
||||
logger.info("=== Dry run complete. Re-run without --dry-run to apply. ===") |
||||
else: |
||||
logger.info("=== Migration complete. ===") |
||||
|
||||
|
||||
if __name__ == "__main__": |
||||
main() |
||||
Loading…
Reference in new issue