507 lines
21 KiB
Python
Executable File
507 lines
21 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
# /// script
|
|
# dependencies = [
|
|
# "beautifulsoup4>=4.12.0",
|
|
# "requests>=2.31.0",
|
|
# ]
|
|
# ///
|
|
"""
|
|
Regensburg Ratsinformationssystem (ALLRIS net) Client & CLI.
|
|
Provides search, calendar, session agenda, Vorlagen extraction, and Obsidian wiki compilation.
|
|
"""
|
|
|
|
import argparse
|
|
import datetime
|
|
import json
|
|
import os
|
|
import re
|
|
import shutil
|
|
import subprocess
|
|
import sys
|
|
import urllib.parse
|
|
from xml.etree import ElementTree as ET
|
|
|
|
def ensure_dependencies_and_run():
|
|
"""If beautifulsoup4 or requests is not installed in current Python, re-exec via uv run."""
|
|
try:
|
|
import bs4 # noqa: F401
|
|
import requests # noqa: F401
|
|
except ImportError:
|
|
uv_path = shutil.which("uv")
|
|
if uv_path and os.environ.get("_REGENSBURG_RIS_REEXEC") != "1":
|
|
env = os.environ.copy()
|
|
env["_REGENSBURG_RIS_REEXEC"] = "1"
|
|
cmd = [uv_path, "run", os.path.abspath(__file__), *sys.argv[1:]]
|
|
try:
|
|
res = subprocess.run(cmd, env=env)
|
|
sys.exit(res.returncode)
|
|
except Exception as e:
|
|
sys.stderr.write(f"Failed to re-execute via uv: {e}\n")
|
|
sys.stderr.write("Error: 'beautifulsoup4' or 'requests' not found. Please run with:\n")
|
|
sys.stderr.write(f" uv run {sys.argv[0]} ...\n")
|
|
sys.exit(1)
|
|
|
|
ensure_dependencies_and_run()
|
|
|
|
from bs4 import BeautifulSoup
|
|
import requests
|
|
|
|
BASE_URL = "https://srv19.regensburg.de/bi/"
|
|
DEFAULT_HEADERS = {
|
|
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36"
|
|
}
|
|
|
|
MARKITDOWN_SCRIPT = os.path.abspath(
|
|
os.path.join(os.path.dirname(__file__), "..", "..", "markitdown", "scripts", "convert_to_markdown.py")
|
|
)
|
|
|
|
class AllrisClient:
|
|
def __init__(self, base_url: str = BASE_URL):
|
|
self.base_url = base_url.rstrip("/") + "/"
|
|
self.session = requests.Session()
|
|
self.session.headers.update(DEFAULT_HEADERS)
|
|
|
|
def search(self, query: str, limit: int = 15) -> list[dict]:
|
|
"""
|
|
Query ALLRIS full-text search engine (yw020.asp).
|
|
Uses ISO-8859-1 (Latin-1) URL encoding required by the ASP backend.
|
|
"""
|
|
search_url = urllib.parse.urljoin(self.base_url, "yw020.asp")
|
|
post_data = {
|
|
"searchForm": "true",
|
|
"SetRECHERCHE": "1",
|
|
"REG": "A",
|
|
"SUCHWORT": query,
|
|
"searchName": "true",
|
|
"searchCont": "true",
|
|
"searchVo": "true",
|
|
"searchSi": "true",
|
|
"searchNaz": "true",
|
|
"TRDATFIL": "(alle)",
|
|
"filtdatum": "filter",
|
|
"Suchen": "Suchen",
|
|
}
|
|
encoded_data = urllib.parse.urlencode(post_data, encoding="iso-8859-1").encode("iso-8859-1")
|
|
headers = dict(DEFAULT_HEADERS)
|
|
headers["Content-Type"] = "application/x-www-form-urlencoded"
|
|
|
|
resp = self.session.post(search_url, data=encoded_data, headers=headers, timeout=20)
|
|
resp.raise_for_status()
|
|
|
|
html = resp.content.decode("iso-8859-1", errors="replace")
|
|
soup = BeautifulSoup(html, "html.parser")
|
|
|
|
results = []
|
|
for a in soup.find_all("a", href=re.compile(r"(vo020|to010|to020)\.asp")):
|
|
href = a.get("href", "")
|
|
clean_href = href.split("#")[0]
|
|
full_url = urllib.parse.urljoin(self.base_url, clean_href)
|
|
title = a.get_text(" ", strip=True)
|
|
if not title:
|
|
continue
|
|
|
|
item_type = "unknown"
|
|
item_id = ""
|
|
code = ""
|
|
|
|
if "vo020.asp" in clean_href:
|
|
item_type = "vorlage"
|
|
m_id = re.search(r"VOLFDNR=(\d+)", clean_href)
|
|
item_id = m_id.group(1) if m_id else ""
|
|
m_code = re.search(r"VO/\d+/\d+(?:/[A-Za-z0-9]+)?", title)
|
|
if m_code:
|
|
code = m_code.group(0)
|
|
elif "to010.asp" in clean_href:
|
|
item_type = "session"
|
|
m_id = re.search(r"SILFDNR=(\d+)", clean_href)
|
|
item_id = m_id.group(1) if m_id else ""
|
|
elif "to020.asp" in clean_href:
|
|
item_type = "top"
|
|
m_id = re.search(r"TOLFDNR=(\d+)", clean_href)
|
|
item_id = m_id.group(1) if m_id else ""
|
|
|
|
# De-duplicate by full_url
|
|
if not any(r["url"] == full_url for r in results):
|
|
results.append({
|
|
"type": item_type,
|
|
"id": item_id,
|
|
"code": code,
|
|
"title": title,
|
|
"url": full_url,
|
|
})
|
|
|
|
if len(results) >= limit:
|
|
break
|
|
|
|
return results
|
|
|
|
def get_calendar(self, days: int = 30, from_date: str = None, to_date: str = None) -> list[dict]:
|
|
"""
|
|
Fetch calendar meetings using native ALLRIS RSS/XML export (si010.asp).
|
|
"""
|
|
now = datetime.datetime.now()
|
|
if not from_date:
|
|
from_date = now.strftime("%d.%m.%Y")
|
|
if not to_date:
|
|
to_date = (now + datetime.timedelta(days=days)).strftime("%d.%m.%Y")
|
|
|
|
date_range = f"{from_date}-{to_date}"
|
|
cal_url = urllib.parse.urljoin(
|
|
self.base_url, f"si010.asp?PALFDNR=1&selfaction=ws&template=rss&kaldatvonbis={date_range}"
|
|
)
|
|
|
|
resp = self.session.get(cal_url, timeout=20)
|
|
resp.raise_for_status()
|
|
|
|
root = ET.fromstring(resp.content)
|
|
sessions = []
|
|
for elem in root.findall(".//list/element"):
|
|
silfdnr = elem.findtext("silfdnr", "").strip()
|
|
siname = elem.findtext("siname", "").strip()
|
|
sitext = elem.findtext("sitext", "").strip()
|
|
grname = elem.findtext("grname", "").strip()
|
|
raname = elem.findtext("raname", "").strip()
|
|
raort = elem.findtext("raort", "").strip()
|
|
sisbvcs = elem.findtext("sisbvcs", "").strip()
|
|
sisb = elem.findtext("sisb", "").strip()
|
|
link_to = elem.findtext("link_to", "").strip()
|
|
|
|
full_link = urllib.parse.urljoin(self.base_url, link_to) if link_to else f"{self.base_url}to010.asp?SILFDNR={silfdnr}"
|
|
|
|
sessions.append({
|
|
"id": silfdnr,
|
|
"code": siname,
|
|
"title": sitext,
|
|
"committee": grname,
|
|
"time": sisb,
|
|
"iso_time": sisbvcs,
|
|
"room": f"{raname}, {raort}" if raort else raname,
|
|
"url": full_link,
|
|
})
|
|
|
|
return sessions
|
|
|
|
def get_session(self, silfdnr: str) -> dict:
|
|
"""
|
|
Fetch session agenda items (TOPs) using ALLRIS RSS/XML web service (to010.asp).
|
|
"""
|
|
session_url = urllib.parse.urljoin(
|
|
self.base_url, f"to010.asp?SILFDNR={silfdnr}&selfaction=ws&template=rss"
|
|
)
|
|
resp = self.session.get(session_url, timeout=20)
|
|
resp.raise_for_status()
|
|
|
|
root = ET.fromstring(resp.content)
|
|
paname = root.findtext(".//paname", "").strip()
|
|
grname = root.findtext(".//grname", "").strip()
|
|
grtext = root.findtext(".//grtext", "").strip()
|
|
sisb = root.findtext(".//sisb", "").strip()
|
|
|
|
agenda_items = []
|
|
for elem in root.findall(".//list/element"):
|
|
tolfdnr = elem.findtext("tolfdnr", "").strip()
|
|
tofnum = elem.findtext("tofnum", "").strip()
|
|
totext1 = elem.findtext("totext1", "").strip()
|
|
tovonr = elem.findtext("tovonr", "").strip()
|
|
link_vo = elem.findtext("link_vo", "").strip()
|
|
toinfo = elem.findtext("toinfo", "").strip()
|
|
|
|
full_vo_link = urllib.parse.urljoin(self.base_url, link_vo) if link_vo else ""
|
|
|
|
agenda_items.append({
|
|
"top_id": tolfdnr,
|
|
"top_num": tofnum,
|
|
"subject": totext1,
|
|
"vorlage_id": tovonr if tovonr != "0" else None,
|
|
"vorlage_url": full_vo_link,
|
|
"info": toinfo,
|
|
})
|
|
|
|
return {
|
|
"session_id": silfdnr,
|
|
"institution": paname,
|
|
"committee": grname,
|
|
"title": grtext,
|
|
"time": sisb,
|
|
"url": f"{self.base_url}to010.asp?SILFDNR={silfdnr}",
|
|
"agenda": agenda_items,
|
|
}
|
|
|
|
def get_vorlage(self, volfdnr: str, convert_attachments: bool = False) -> dict:
|
|
"""
|
|
Fetch Vorlage details, metadata, verbatim resolution text, and attachments (vo020.asp).
|
|
"""
|
|
vo_url = urllib.parse.urljoin(self.base_url, f"vo020.asp?VOLFDNR={volfdnr}")
|
|
resp = self.session.get(vo_url, timeout=20)
|
|
resp.raise_for_status()
|
|
|
|
html = resp.content.decode("iso-8859-1", errors="replace")
|
|
soup = BeautifulSoup(html, "html.parser")
|
|
|
|
# Extract metadata table
|
|
meta = {}
|
|
for tr in soup.find_all("tr"):
|
|
tds = tr.find_all("td")
|
|
if len(tds) >= 2 and "kb1" in tds[0].get("class", []):
|
|
k = tds[0].get_text(strip=True).rstrip(":")
|
|
v = tds[1].get_text(" ", strip=True)
|
|
meta[k] = v
|
|
|
|
# Extract Beratungsfolge (consultations)
|
|
consultations = []
|
|
for tr in soup.find_all("tr", class_=["zl11", "zl12"]):
|
|
tds = [td.get_text(strip=True) for td in tr.find_all("td") if td.get_text(strip=True)]
|
|
if len(tds) >= 2:
|
|
consultations.append(" | ".join(tds))
|
|
|
|
# Extract Resolution text ("Der Stadtrat / Ausschuss beschließt:")
|
|
resolution_text = ""
|
|
sachverhalt_text = ""
|
|
paragraphs = [p.get_text(strip=True) for p in soup.find_all("p") if p.get_text(strip=True)]
|
|
|
|
recording_res = False
|
|
res_lines = []
|
|
for p in paragraphs:
|
|
if "beschließt:" in p.lower() or "beschliesst:" in p.lower() or "beschluss:" in p.lower():
|
|
recording_res = True
|
|
if recording_res:
|
|
res_lines.append(p)
|
|
if "anlagen" in p.lower() or len(res_lines) > 20:
|
|
break
|
|
if res_lines:
|
|
resolution_text = "\n".join(res_lines)
|
|
|
|
# Extract attachments
|
|
attachments = []
|
|
for a in soup.find_all("a", href=re.compile(r"___tmp/tmp/[^\"]+\.pdf", re.I)):
|
|
href = a.get("href", "")
|
|
full_pdf_url = urllib.parse.urljoin(self.base_url, href)
|
|
name = a.get_text(strip=True)
|
|
size_match = re.search(r"\((\d+[\.,]?\d*\s*[kMG]B)\)", name)
|
|
size_str = size_match.group(1) if size_match else ""
|
|
|
|
att_info = {
|
|
"name": name,
|
|
"size": size_str,
|
|
"url": full_pdf_url,
|
|
}
|
|
|
|
if convert_attachments:
|
|
att_info["markdown"] = self._convert_pdf(full_pdf_url)
|
|
|
|
attachments.append(att_info)
|
|
|
|
return {
|
|
"id": volfdnr,
|
|
"url": vo_url,
|
|
"title": meta.get("Betreff", ""),
|
|
"status": meta.get("Status", "öffentlich"),
|
|
"type": meta.get("Vorlage-Art", ""),
|
|
"department": meta.get("Federführend", ""),
|
|
"referent": meta.get("Berichterstatter/in", ""),
|
|
"consultations": consultations,
|
|
"resolution": resolution_text,
|
|
"attachments": attachments,
|
|
}
|
|
|
|
def _convert_pdf(self, pdf_url: str) -> str:
|
|
"""Helper to invoke markitdown script on a PDF URL."""
|
|
if os.path.exists(MARKITDOWN_SCRIPT):
|
|
try:
|
|
cmd = [sys.executable, MARKITDOWN_SCRIPT, "--url", pdf_url]
|
|
res = subprocess.run(cmd, capture_output=True, text=True, timeout=90)
|
|
if res.returncode == 0:
|
|
return res.stdout.strip()
|
|
except Exception as e:
|
|
return f"[Fehler bei PDF-Konvertierung: {e}]"
|
|
return "[MarkItDown Script nicht gefunden]"
|
|
|
|
def compile_topic(self, query: str, vault_path: str = None) -> str:
|
|
"""
|
|
Compile an end-to-end Obsidian-compatible markdown dossier for a given search topic.
|
|
Optionally saves directly to knowledge-wiki/concepts/ and sources/.
|
|
"""
|
|
search_results = self.search(query, limit=12)
|
|
vorlagen_data = []
|
|
|
|
for r in search_results:
|
|
if r["type"] == "vorlage":
|
|
vo = self.get_vorlage(r["id"], convert_attachments=False)
|
|
vorlagen_data.append(vo)
|
|
|
|
slug = re.sub(r"[^a-zA-Z0-9\-_]", "-", query.lower().replace("ä", "ae").replace("ö", "oe").replace("ü", "ue").replace("ß", "ss"))
|
|
today = datetime.datetime.now().strftime("%Y-%m-%d")
|
|
|
|
md = []
|
|
md.append("---")
|
|
md.append(f"id: regensburg-{slug}")
|
|
md.append(f'title: "Stadtrat Regensburg: {query}"')
|
|
md.append("type: concept")
|
|
md.append("tags:")
|
|
md.append(" - stadtpolitik")
|
|
md.append(" - regensburg")
|
|
md.append(f" - {slug}")
|
|
md.append("sources:")
|
|
md.append(f" - sources/regensburg-stadtrat/{slug}-beschluesse.md")
|
|
md.append(f'updatedAt: "{today}"')
|
|
md.append("---\n")
|
|
|
|
md.append(f"# 🏛️ Stadtrat Regensburg: {query}\n")
|
|
md.append(f"> Automatisch kompiliertes Dossier aus dem Ratsinformationssystem (ALLRIS net) vom {today}.\n")
|
|
|
|
md.append("## 📌 Relevante Vorlagen & Beschlüsse\n")
|
|
for vo in vorlagen_data:
|
|
md.append(f"### [{vo['title']}]({vo['url']})")
|
|
md.append(f"- **ID**: {vo['id']} | **Status**: {vo['status']} | **Art**: {vo['type']}")
|
|
if vo["department"]:
|
|
md.append(f"- **Federführend**: {vo['department']}")
|
|
if vo["referent"]:
|
|
md.append(f"- **Berichterstatter**: {vo['referent']}")
|
|
if vo["consultations"]:
|
|
md.append(f"- **Beratungsfolge**: {' → '.join(vo['consultations'])}")
|
|
if vo["resolution"]:
|
|
md.append(f"\n**Beschlussantrag / Beschlusstext:**\n```text\n{vo['resolution']}\n```\n")
|
|
if vo["attachments"]:
|
|
md.append("**Anlagen (PDFs):**")
|
|
for att in vo["attachments"]:
|
|
md.append(f"- [{att['name']}]({att['url']}) ({att['size']})")
|
|
md.append("\n---\n")
|
|
|
|
md.append("## 🔗 Querverweise im Wiki")
|
|
md.append("- [[entities/stadtrat-regensburg|Stadtrat der Stadt Regensburg]]")
|
|
md.append("- [[entities/gertrud-maltz-schwarzfischer|Oberbürgermeisterin Gertrud Maltz-Schwarzfischer]]\n")
|
|
|
|
compiled_text = "\n".join(md)
|
|
|
|
if vault_path and os.path.exists(vault_path):
|
|
# Write concept note
|
|
concept_dir = os.path.join(vault_path, "concepts")
|
|
os.makedirs(concept_dir, exist_ok=True)
|
|
concept_file = os.path.join(concept_dir, f"{slug}.md")
|
|
with open(concept_file, "w", encoding="utf-8") as f:
|
|
f.write(compiled_text)
|
|
|
|
# Write raw source note
|
|
sources_dir = os.path.join(vault_path, "sources", "regensburg-stadtrat")
|
|
os.makedirs(sources_dir, exist_ok=True)
|
|
source_file = os.path.join(sources_dir, f"{slug}-beschluesse.md")
|
|
with open(source_file, "w", encoding="utf-8") as f:
|
|
f.write(f"# Raw Source: {query}\n\nSearch Results from ALLRIS:\n\n" + json.dumps(search_results, indent=2, ensure_ascii=False))
|
|
|
|
# Append to log.md
|
|
log_file = os.path.join(vault_path, "log.md")
|
|
if os.path.exists(log_file):
|
|
log_entry = f"\n## [{today}] ingest | Stadtrat Regensburg: {query}\n"
|
|
log_entry += f"- Ingested ALLRIS search results to `sources/regensburg-stadtrat/{slug}-beschluesse.md`\n"
|
|
log_entry += f"- Compiled concept dossier to `concepts/{slug}.md`\n"
|
|
log_entry += f"- Found {len(vorlagen_data)} council resolutions / Vorlagen\n"
|
|
with open(log_file, "a", encoding="utf-8") as f:
|
|
f.write(log_entry)
|
|
|
|
print(f"Filed to vault: concepts/{slug}.md and sources/regensburg-stadtrat/{slug}-beschluesse.md")
|
|
|
|
return compiled_text
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser(description="Regensburger Sitzungsdienst (ALLRIS net) CLI")
|
|
subparsers = parser.add_subparsers(dest="command", required=True)
|
|
|
|
# Search
|
|
search_p = subparsers.add_parser("search", help="Full-text search in ALLRIS")
|
|
search_p.add_argument("query", help="Search keyword (e.g. 'Kaufhof', 'Stadtbahn')")
|
|
search_p.add_argument("--limit", type=int, default=15, help="Max results (default: 15)")
|
|
search_p.add_argument("--json", action="store_true", help="Output raw JSON")
|
|
|
|
# Calendar
|
|
cal_p = subparsers.add_parser("calendar", help="View upcoming/recent meetings")
|
|
cal_p.add_argument("--days", type=int, default=30, help="Days ahead to fetch (default: 30)")
|
|
cal_p.add_argument("--from-date", help="Start date (DD.MM.YYYY)")
|
|
cal_p.add_argument("--to-date", help="End date (DD.MM.YYYY)")
|
|
cal_p.add_argument("--json", action="store_true", help="Output raw JSON")
|
|
|
|
# Session
|
|
sess_p = subparsers.add_parser("session", help="View agenda for a meeting (SILFDNR)")
|
|
sess_p.add_argument("silfdnr", help="Session ID (SILFDNR, e.g. 2665)")
|
|
sess_p.add_argument("--json", action="store_true", help="Output raw JSON")
|
|
|
|
# Vorlage
|
|
vo_p = subparsers.add_parser("vorlage", help="View Vorlage details & resolutions (VOLFDNR)")
|
|
vo_p.add_argument("volfdnr", help="Vorlage ID (VOLFDNR, e.g. 21362)")
|
|
vo_p.add_argument("--convert-attachments", action="store_true", help="Convert PDF attachments to markdown using markitdown")
|
|
vo_p.add_argument("--json", action="store_true", help="Output raw JSON")
|
|
|
|
# Compile Topic
|
|
comp_p = subparsers.add_parser("compile-topic", help="Search and compile Obsidian concept note")
|
|
comp_p.add_argument("query", help="Topic to compile (e.g. 'Kaufhof')")
|
|
comp_p.add_argument("--vault", help="Path to knowledge-wiki vault (optional)")
|
|
|
|
args = parser.parse_args()
|
|
client = AllrisClient()
|
|
|
|
if args.command == "search":
|
|
res = client.search(args.query, limit=args.limit)
|
|
if args.json:
|
|
print(json.dumps(res, indent=2, ensure_ascii=False))
|
|
else:
|
|
print(f"Found {len(res)} results for '{args.query}':\n")
|
|
for r in res:
|
|
print(f"- [{r['type'].upper()}] {r['title']}")
|
|
print(f" ID: {r['id']} | Code: {r.get('code', '')} | URL: {r['url']}\n")
|
|
|
|
elif args.command == "calendar":
|
|
res = client.get_calendar(days=args.days, from_date=args.from_date, to_date=args.to_date)
|
|
if args.json:
|
|
print(json.dumps(res, indent=2, ensure_ascii=False))
|
|
else:
|
|
print(f"Found {len(res)} meetings:\n")
|
|
for s in res:
|
|
print(f"📅 {s['time']} | {s['committee']}")
|
|
print(f" {s['title']}")
|
|
print(f" Ort: {s['room']} | ID: {s['id']} | URL: {s['url']}\n")
|
|
|
|
elif args.command == "session":
|
|
res = client.get_session(args.silfdnr)
|
|
if args.json:
|
|
print(json.dumps(res, indent=2, ensure_ascii=False))
|
|
else:
|
|
print(f"🏛️ {res['committee']}: {res['title']} ({res['time']})")
|
|
print(f"URL: {res['url']}\n")
|
|
print("Tagesordnung (TOPs):")
|
|
for item in res["agenda"]:
|
|
vo_str = f" [Vorlage: {item['vorlage_id']}]" if item["vorlage_id"] else ""
|
|
print(f" TOP {item['top_num']}: {item['subject']}{vo_str}")
|
|
if item["info"]:
|
|
print(f" Info: {item['info']}")
|
|
|
|
elif args.command == "vorlage":
|
|
res = client.get_vorlage(args.volfdnr, convert_attachments=args.convert_attachments)
|
|
if args.json:
|
|
print(json.dumps(res, indent=2, ensure_ascii=False))
|
|
else:
|
|
print(f"📄 Vorlage {res['id']}: {res['title']}")
|
|
print(f"Typ: {res['type']} | Status: {res['status']} | Federführend: {res['department']}")
|
|
if res["referent"]:
|
|
print(f"Berichterstatter: {res['referent']}")
|
|
print(f"URL: {res['url']}\n")
|
|
if res["consultations"]:
|
|
print("Beratungsfolge:")
|
|
for c in res["consultations"]:
|
|
print(f" • {c}")
|
|
print()
|
|
if res["resolution"]:
|
|
print(f"Beschlusstext:\n{res['resolution']}\n")
|
|
if res["attachments"]:
|
|
print("Anlagen:")
|
|
for a in res["attachments"]:
|
|
print(f" 📎 {a['name']} ({a['size']}) -> {a['url']}")
|
|
if "markdown" in a:
|
|
print(f"\n--- Extrahierter Markdown-Inhalt ({a['name']}) ---\n{a['markdown'][:500]}...\n")
|
|
|
|
elif args.command == "compile-topic":
|
|
res = client.compile_topic(args.query, vault_path=args.vault)
|
|
if not args.vault:
|
|
print(res)
|
|
|
|
if __name__ == "__main__":
|
|
main()
|