from __future__ import annotations import os import sys from pathlib import Path from typing import List import pandas as pd import plotly.express as px import streamlit as st ROOT = Path(__file__).resolve().parent SRC = ROOT / "src" if str(SRC) not in sys.path: sys.path.insert(0, str(SRC)) from dip_client import ( # noqa: E402 KB_COLUMNS, RESOURCE_TYPES, build_knowledge_base, build_query_params, empty_knowledge_base, save_knowledge_base, ) st.set_page_config( page_title="German Promise Tracker · DIP Knowledge Base", page_icon="🗳️", layout="wide", ) DATA_DIR = ROOT / "data" KB_PATH = DATA_DIR / "dip_knowledge_base.csv" PROMISE_TEMPLATE = DATA_DIR / "manual_promise_tracker_template.csv" def get_secret_or_env(name: str) -> str: try: value = st.secrets.get(name, "") except Exception: value = "" return value or os.environ.get(name, "") @st.cache_data(show_spinner=False) def load_kb() -> pd.DataFrame: if KB_PATH.exists(): return pd.read_csv(KB_PATH, dtype=str).fillna("") return empty_knowledge_base() def save_uploaded_tracker(uploaded_file) -> pd.DataFrame: if uploaded_file is None: if PROMISE_TEMPLATE.exists(): return pd.read_csv(PROMISE_TEMPLATE, dtype=str).fillna("") return pd.DataFrame() return pd.read_csv(uploaded_file, dtype=str).fillna("") def keyword_filter(df: pd.DataFrame, query: str) -> pd.DataFrame: if not query.strip() or df.empty: return df terms = [t.strip().lower() for t in query.replace(";", ",").split(",") if t.strip()] if not terms: return df search_cols = [ "title", "abstract", "text_excerpt", "subject_area", "descriptors", "initiative", "consultation_status", "procedure_type", "procedure_position", "document_number", ] combined = df[[c for c in search_cols if c in df.columns]].astype(str).agg(" ".join, axis=1).str.lower() mask = combined.apply(lambda text: all(term in text for term in terms)) return df[mask] def render_record_cards(df: pd.DataFrame, max_cards: int = 30) -> None: if df.empty: st.info("No matching DIP records in the current knowledge base.") return for _, row in df.head(max_cards).iterrows(): with st.container(border=True): st.markdown(f"### {row.get('title', '')}") st.markdown( f"**DIP type:** `{row.get('resource_type', '')}` · " f"**DIP ID:** `{row.get('dip_id', '')}` · " f"**Date:** {row.get('date', '') or '—'} · " f"**Updated:** {row.get('updated', '') or '—'}" ) meta_bits = [] for label, col in [ ("Election period", "election_period"), ("Document", "document_number"), ("Document type", "document_type"), ("Procedure type", "procedure_type"), ("Consultation status", "consultation_status"), ("Subject area", "subject_area"), ]: val = row.get(col, "") if val: meta_bits.append(f"**{label}:** {val}") if meta_bits: st.markdown(" · ".join(meta_bits)) abstract = row.get("abstract", "") or row.get("text_excerpt", "") if abstract: st.markdown(abstract) links = [] if row.get("api_url"): links.append(f"[DIP API record]({row.get('api_url')})") if row.get("pdf_url"): links.append(f"[PDF]({row.get('pdf_url')})") if links: st.markdown(" · ".join(links)) st.title("🗳️ German Promise Tracker — Bundestag DIP Knowledge Base") st.caption( "This version builds the project knowledge base from the Bundestag DIP API only. " "The app displays legislative/procedural evidence; it does not automatically invent promise fulfilment statuses." ) with st.expander("What this dashboard does and does not claim", expanded=False): st.markdown( """ **DIP-derived evidence:** procedures, documents, plenary protocols, activities and person records fetched from the Bundestag DIP API. **Not automatically inferred:** whether a politician's promise is completed, broken, or in progress. The DIP field `beratungsstand` is shown as the official legislative/procedural status where available, but a promise-status judgement still requires a separate reviewed row. **Recommended workflow:** collect the relevant DIP evidence here, then manually link it to a promise in the tracker tab. """ ) api_key_default = get_secret_or_env("DIP_API_KEY") build_tab, explorer_tab, tracker_tab, methodology_tab = st.tabs( ["1 · Build / Refresh DIP KB", "2 · Knowledge Base Explorer", "3 · Promise Evidence Tracker", "Methodology"] ) with build_tab: st.subheader("Build the knowledge base from the Bundestag DIP API") st.markdown( "Set `DIP_API_KEY` as a Hugging Face Space secret for deployment. For local testing, enter the key below." ) col_a, col_b = st.columns([0.55, 0.45]) with col_a: api_key_input = st.text_input( "DIP API key", value=api_key_default, type="password", help="The key is sent only to the Bundestag DIP API. It is not written to disk.", ) resources: List[str] = st.multiselect( "DIP resources to fetch", list(RESOURCE_TYPES), default=["vorgang", "vorgangsposition", "drucksache"], help="Text endpoints can be larger. Start with metadata endpoints, then add text endpoints if needed.", ) max_pages = st.slider( "Max cursor pages per resource", min_value=1, max_value=20, value=2, help="The API returns up to 100 metadata records per page and usually fewer for full-text endpoints.", ) with col_b: wahlperiode = st.number_input("Wahlperiode", min_value=1, max_value=99, value=21, step=1) date_start = st.text_input("Document date start: f.datum.start", value="") date_end = st.text_input("Document date end: f.datum.end", value="") updated_start = st.text_input("Updated start: f.aktualisiert.start", value="") updated_end = st.text_input("Updated end: f.aktualisiert.end", value="") zuordnung = st.selectbox("Zuordnung", ["", "BT", "BR", "BV", "EK"], index=0) params = build_query_params( wahlperiode=int(wahlperiode) if wahlperiode else None, date_start=date_start or None, date_end=date_end or None, updated_start=updated_start or None, updated_end=updated_end or None, zuordnung=zuordnung or None, ) st.code(params, language="json") if st.button("Fetch from DIP API and rebuild KB", type="primary"): if not api_key_input.strip(): st.error("Please provide a DIP API key or set the Hugging Face secret `DIP_API_KEY`.") elif not resources: st.error("Select at least one DIP resource.") else: with st.spinner("Fetching DIP records and normalising the knowledge base..."): try: df, raw_docs, metadata = build_knowledge_base( api_key=api_key_input, resources=resources, params=params, max_pages_per_resource=max_pages, ) save_knowledge_base(df, raw_docs, metadata, DATA_DIR) load_kb.clear() st.success(f"Knowledge base rebuilt with {len(df)} unique DIP records.") st.json(metadata) except Exception as exc: st.error(f"DIP fetch failed: {exc}") current_df = load_kb() st.info(f"Current local KB size: {len(current_df)} records.") with explorer_tab: st.subheader("DIP Knowledge Base Explorer") df = load_kb() if df.empty: st.warning("The local knowledge base is empty. Use the build tab to fetch DIP records first.") else: f1, f2, f3, f4 = st.columns([0.25, 0.25, 0.25, 0.25]) with f1: selected_resources = st.multiselect( "Resource type", sorted(df["resource_type"].unique()), default=sorted(df["resource_type"].unique()), ) with f2: selected_periods = st.multiselect( "Wahlperiode", sorted([x for x in df["election_period"].unique() if x]), default=sorted([x for x in df["election_period"].unique() if x]), ) with f3: selected_status = st.multiselect( "DIP consultation status", sorted([x for x in df["consultation_status"].unique() if x]), default=sorted([x for x in df["consultation_status"].unique() if x]), ) with f4: selected_doc_type = st.multiselect( "Document type", sorted([x for x in df["document_type"].unique() if x]), default=sorted([x for x in df["document_type"].unique() if x]), ) search = st.text_input("Search inside fetched API records", "") filtered = df.copy() if selected_resources: filtered = filtered[filtered["resource_type"].isin(selected_resources)] if selected_periods: filtered = filtered[filtered["election_period"].isin(selected_periods)] if selected_status: filtered = filtered[filtered["consultation_status"].isin(selected_status)] if selected_doc_type: filtered = filtered[filtered["document_type"].isin(selected_doc_type)] filtered = keyword_filter(filtered, search) k1, k2, k3, k4 = st.columns(4) k1.metric("Records", len(filtered)) k2.metric("Resource types", filtered["resource_type"].nunique()) k3.metric("With PDF", int((filtered["pdf_url"].astype(str) != "").sum())) k4.metric("With status", int((filtered["consultation_status"].astype(str) != "").sum())) c1, c2 = st.columns(2) if not filtered.empty: counts = filtered["resource_type"].value_counts().reset_index() counts.columns = ["resource_type", "count"] c1.plotly_chart(px.bar(counts, x="resource_type", y="count", text="count", title="Records by DIP resource"), use_container_width=True) status_counts = filtered["consultation_status"].replace("", "No status").value_counts().reset_index() status_counts.columns = ["consultation_status", "count"] c2.plotly_chart(px.bar(status_counts, x="consultation_status", y="count", text="count", title="DIP consultation status"), use_container_width=True) table_cols = [ "resource_type", "dip_id", "title", "date", "updated", "election_period", "document_number", "document_type", "procedure_type", "consultation_status", "subject_area", "initiative", "api_url", "pdf_url", ] st.dataframe( filtered[[c for c in table_cols if c in filtered.columns]], use_container_width=True, hide_index=True, column_config={ "api_url": st.column_config.LinkColumn("DIP API"), "pdf_url": st.column_config.LinkColumn("PDF"), "title": st.column_config.TextColumn("Title", width="large"), }, ) st.download_button( "Download filtered DIP KB as CSV", filtered.to_csv(index=False).encode("utf-8"), file_name="filtered_dip_knowledge_base.csv", mime="text/csv", ) st.markdown("### Evidence cards") render_record_cards(filtered) with tracker_tab: st.subheader("Promise Evidence Tracker") st.markdown( "Upload or edit a reviewed promise tracker CSV, then search the DIP knowledge base for evidence. " "The dashboard does not assign promise status automatically." ) uploaded = st.file_uploader("Optional: upload reviewed promise tracker CSV", type=["csv"]) tracker = save_uploaded_tracker(uploaded) required_tracker_cols = [ "promise_id", "promise_text", "promise_source", "promise_date", "category", "actor", "reviewed_status", "review_notes", "linked_dip_ids", "linked_evidence_urls", "last_reviewed", ] if tracker.empty: tracker = pd.DataFrame(columns=required_tracker_cols) for col in required_tracker_cols: if col not in tracker.columns: tracker[col] = "" st.markdown("#### Reviewed promise rows") st.data_editor( tracker[required_tracker_cols], use_container_width=True, hide_index=True, num_rows="dynamic", column_config={ "promise_text": st.column_config.TextColumn("Promise text", width="large"), "linked_evidence_urls": st.column_config.TextColumn("Linked evidence URLs", width="large"), }, ) st.download_button( "Download blank / edited tracker template", tracker[required_tracker_cols].to_csv(index=False).encode("utf-8"), file_name="manual_promise_tracker_template.csv", mime="text/csv", ) st.markdown("#### Search the DIP KB for evidence") df = load_kb() evidence_query = st.text_input("Search terms, comma-separated", placeholder="e.g. Mietpreisbremse, Wohnungsbau, DigitalPakt") evidence = keyword_filter(df, evidence_query) if not df.empty else df evidence_cols = [ "resource_type", "dip_id", "title", "date", "document_number", "procedure_type", "consultation_status", "subject_area", "api_url", "pdf_url", ] if evidence_query.strip(): st.caption(f"Matches: {len(evidence)}") st.dataframe( evidence[[c for c in evidence_cols if c in evidence.columns]], use_container_width=True, hide_index=True, column_config={ "api_url": st.column_config.LinkColumn("DIP API"), "pdf_url": st.column_config.LinkColumn("PDF"), }, ) if not evidence.empty: export = evidence[[c for c in evidence_cols if c in evidence.columns]].copy() export["reviewed_status"] = "Needs human review" export["review_notes"] = "" st.download_button( "Download evidence candidates for manual review", export.to_csv(index=False).encode("utf-8"), file_name="dip_evidence_candidates.csv", mime="text/csv", ) with methodology_tab: st.subheader("Methodology") st.markdown( """ ### Knowledge-base rule Every record in `data/dip_knowledge_base.csv` must come from one of the Bundestag DIP API endpoints. The app stores: - the original DIP resource type and ID, - title, abstract or text excerpt, - date and last-updated metadata, - legislative/procedural fields such as `beratungsstand`, `vorgangstyp`, `sachgebiet`, `initiative`, and document number where returned by the API, - PDF and API links where returned or constructible from the API endpoint, - raw API JSON in `data/dip_raw_documents.jsonl` for auditability. ### Promise-status rule A Bundestag record is evidence, not a political promise-status judgement. The tracker can show `beratungsstand` from DIP, but it should not automatically label a promise as completed or broken without human review. ### Recommended status labels for the manually reviewed tracker - `Completed`: the evidence directly shows that the promised legal or administrative action was completed. - `In progress`: formal steps exist, but implementation is not complete. - `Not started`: no relevant evidence has been found in the KB or other reviewed sources. - `Broken`: evidence shows the promise was reversed, abandoned, or the stated deadline was missed. - `Needs human review`: the evidence is relevant but not enough for a status decision. """ )