| 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 ( |
| 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. |
| """ |
| ) |
|
|
