Files
Orchestrator/bahn/wissensdatenbank/src/strategy_detect.py
T

83 lines
3.0 KiB
Python

"""Erkennt zu einer URL die passende Verarbeitungs-Strategie (frueh, beim Issue).
So sieht man schon beim Anlegen, ob eine URL in unsere bekannten Strategien passt
oder ob eine NEUE Strategie gebaut werden muss. Funktioniert pro URL (ein Tool kann
mehrere URLs mit je eigener Strategie haben).
"""
from __future__ import annotations
import re
KNOWN = {"confluence_page", "confluence_tree", "confluence_faq", "crawler", "sitemap", "pdf", "gitlab_md", "file"}
def detect_strategy(url: str) -> str | None:
"""Heuristik anhand der URL. None => unbekannt -> neue Strategie noetig."""
u = url.strip().lower()
if not u.startswith(("http://", "https://")):
# anderes Schema (ftp://, sharepoint://, ...) -> unbekannt
if "://" in u:
return None
# repo-relativer Pfad -> Datei(en) aus dem Repo
return "file" if u else None
if u.endswith(".pdf"):
return "pdf"
if "sitemap" in u and u.endswith((".xml", ".xml.gz")):
return "sitemap"
# GitLab-Projekt -> Markdown-Dateien
host = u.split("/")[2] if "//" in u else ""
if host.startswith("git.") or "gitlab" in host:
return "gitlab_md"
# DB-Confluence (jaas) bzw. generische Confluence-Seiten
if "arija-confluence" in u or re.search(r"/spaces/[^/]+/pages/\d+", u) or "/wiki/" in u:
return "confluence_tree"
# bekannte oeffentliche Webquelle -> Crawler (Detailseiten/Index)
if u.startswith(("http://", "https://")):
return "crawler"
return None
def validate_urls(urls: list[str]) -> list[dict]:
"""Liefert pro URL die erkannte Strategie + ob sie bekannt ist."""
out = []
for url in urls:
strat = detect_strategy(url)
out.append({
"url": url,
"strategy": strat or "UNBEKANNT",
"known": strat is not None,
"note": "" if strat else "Keine passende Strategie -> neue Strategie pruefen/bauen",
})
return out
def sniff_strategy(url: str, timeout: int = 10) -> str | None:
"""Verfeinert die Erkennung per Netzwerk (HEAD/GET).
- Content-Type application/pdf -> 'pdf'
- HTML-Seite mit >=3 PDF-Links -> 'pdf' (PDF-Sammelseite, z.B. INB/Regelwerk)
- sonst Fallback auf detect_strategy().
Netzwerk-Fehler -> Fallback auf detect_strategy().
"""
base = detect_strategy(url)
try:
import requests
except ImportError:
return base
ua = {"User-Agent": "wissensdatenbank/0.1"}
try:
h = requests.head(url, allow_redirects=True, timeout=timeout, headers=ua)
ct = h.headers.get("Content-Type", "").lower()
if "application/pdf" in ct:
return "pdf"
if "text/html" in ct or not ct:
from bs4 import BeautifulSoup
r = requests.get(url, timeout=timeout, headers=ua)
soup = BeautifulSoup(r.text, "html.parser")
pdfs = [a for a in soup.find_all("a", href=True) if ".pdf" in a["href"].lower()]
if len(pdfs) >= 3:
return "pdf"
except Exception:
pass
return base