rixdagen/scripts/sync_talks.py

"""
Synkroniserar nya anföranden från riksdagen.se till databasen och processar dem.

Pipeline (körs dagligen via systemd timer):
  1. Ladda ned årets anföranden från riksdagen.se (ersätter tidigare nerladdning)
  2. Infoga nya anföranden i ArangoDB (hoppar över redan existerande)
  3. Tilldela debatt-ID:n till anföranden som saknar det
  4. Bygg embeddings för datum som saknar chunks
  5. Generera sammanfattningar för datum som saknar summary

Kör manuellt: python scripts/sync_talks.py
"""

import os
import sys
import logging
from datetime import datetime
from io import BytesIO
from urllib.request import urlopen
from zipfile import ZipFile

# Säkerställ att vi kör från projektroten och att lokala moduler hittas
os.chdir("/home/lasse/riksdagen")
sys.path.append("/home/lasse/riksdagen")

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s [%(levelname)s] %(message)s",
)
logger = logging.getLogger(__name__)

# Systemprompt som används av LLM:en vid sammanfattning av debatter
SYSTEM_MESSAGE = """Din uppgift är att sammanfatta debatter i Sveriges riksdag. 
Du kommer först att få enskilda tal som du ska sammanfatta var för sig, efter det ska du sammanfatta hela debatten.
Sammanfattningarna ska vara på svenska och vara koncisa och informativa. 
Det är viktigt att du förstår vad som är kärnan i varje tal och debatt, fokusera därför på de argument och sakförhållanden som framförs.
"""


def get_current_session_year() -> int:
    """
    Returnerar startåret för aktuell riksdagssession.

    Riksdagssessionen löper september–augusti, så:
      - Januari–Augusti 2026 → sessionen startade sep 2025 → returnerar 2025
      - September–December 2025 → sessionen startade sep 2025 → returnerar 2025

    Returns:
        int: Fyrsiffrigt startår för aktuell session (t.ex. 2025).
    """
    now = datetime.now()
    if now.month >= 9:
        return now.year
    else:
        return now.year - 1


def download_current_year(session_year: int) -> str:
    """
    Laddar ned och extraherar ZIP-arkivet för angiven riksdagssession,
    och ersätter eventuella tidigare nerladdade filer för det året.

    Riksdagen uppdaterar kontinuerligt samma ZIP-fil under pågående session,
    så vi måste ladda ned den på nytt varje gång för att få med nya anföranden.

    Args:
        session_year (int): Sessionens startår (t.ex. 2025 för session 2025/26).

    Returns:
        str: Sökväg till katalogen dit filerna extraherades.
    """
    second_part = str(session_year + 1)[2:]  # t.ex. "26" för 2026
    url = f"https://data.riksdagen.se/dataset/anforande/anforande-{session_year}{second_part}.json.zip"
    folder_name = f"anforande-{session_year}{second_part}"
    dir_path = os.path.join("talks", folder_name)

    logger.info(f"Downloading {url} → {dir_path}")
    os.makedirs(dir_path, exist_ok=True)

    # Rensa gamla filer så vi får en färsk kopia
    for f in os.listdir(dir_path):
        os.remove(os.path.join(dir_path, f))

    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 get_unsummarized_dates() -> list[str]:
    """
    Hämtar datum från ArangoDB som har anföranden utan sammanfattning.

    Returns:
        list[str]: Sorterad lista med datumsträngar, t.ex. ["2026-02-10", "2026-02-11"].
    """
    from arango_client import arango

    cursor = arango.db.aql.execute(
        """
        FOR doc IN talks
        FILTER doc.summary == null
        RETURN DISTINCT doc.datum
        """,
        ttl=300,
    )
    dates = sorted(list(cursor))
    logger.info(f"Found {len(dates)} dates with unsummarized talks")
    return dates


def sync() -> None:
    """
    Kör hela sync-pipelinen:
      1. Ladda ned årets anföranden
      2. Infoga nya anföranden i ArangoDB
      3. Tilldela debatt-ID:n
      4. Bygg embeddings för nya datum
      5. Generera sammanfattningar för nya datum
    """
    logger.info("=== Starting daily riksdagen sync ===")

    # --- Steg 1: Ladda ned ---
    session_year = get_current_session_year()
    logger.info(f"Current session year: {session_year}/{session_year + 1}")
    dir_path = download_current_year(session_year)

    # --- Steg 2: Infoga nya anföranden i ArangoDB ---
    # update_folder() hämtar alla befintliga _key:s från databasen och hoppar
    # över dem, så enbart nya anföranden infogas.
    logger.info("Stage 2: Inserting new talks into ArangoDB...")
    from scripts.documents_to_arango import update_folder

    new_talks = update_folder(os.path.abspath(dir_path))
    logger.info(f"Stage 2 complete: {new_talks} new talks inserted")

    # --- Steg 3: Tilldela debatt-ID:n ---
    # Anföranden som saknar fältet 'debate' grupperas i debatter baserat på
    # datum och om de är repliker eller ej.
    logger.info("Stage 3: Assigning debate IDs to talks missing them...")
    from scripts.debates import make_debate_ids

    make_debate_ids()
    logger.info("Stage 3 complete")

    # --- Steg 4: Chunk + bygg embeddings i ArangoDB ---
    # make_arango_embeddings() hittar alla anföranden som saknar chunks i
    # 'chunks'-kollektionen, chunkar texten, genererar vektorer via Ollama
    # och lagrar allt direkt i ArangoDB. ChromaDB används inte.
    logger.info("Stage 4: Chunking and embedding new talks into ArangoDB...")
    from scripts.make_arango_embeddings import make_arango_embeddings

    total_chunks = make_arango_embeddings()
    logger.info(f"Stage 4 complete: {total_chunks} chunks created")

    # --- Steg 5: Generera sammanfattningar ---
    # process_debate_date() hoppar automatiskt över anföranden som redan har
    # en sammanfattning, så det är säkert att köra igen.
    new_dates = get_unsummarized_dates()
    if new_dates:
        logger.info(f"Stage 5: Generating summaries for {len(new_dates)} dates...")
        from scripts.debates import process_debate_date

        for date in new_dates:
            process_debate_date(date, SYSTEM_MESSAGE)
        logger.info(f"Stage 5 complete: summaries generated for {len(new_dates)} dates")
    else:
        logger.info("Stage 5: No unsummarized dates, skipping")

    logger.info("=== Sync complete ===")


if __name__ == "__main__":
    sync()