irrlicht/swp-01-ingest
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
ab20fa5d15ece02e264970aa5cecb47c3fee7377
swp-01-ingest/poller.py
T
irrlichtandClaude Opus 5 ab20fa5d15 Fehler einer Quelle nicht mehr die Transaktion mitreissen lassen 
Der Handler in durchlauf() fing den Fehler einer Quelle ab, damit sie nicht
den ganzen Durchlauf killt, liess die abgebrochene Postgres-Transaktion aber
stehen. Das notiere() unmittelbar danach lief in genau diese Transaktion und
warf InFailedSqlTransaction - ungefangen, also war der Prozess tot.

Doppelt aergerlich: der eigentliche Fehler stand nur in `fehler` und wurde
weder protokolliert noch ausgegeben, weil beides erst nach dem notiere()
kommt. Im Journal blieb allein die Folgewirkung sichtbar.

Beide Senken bekommen rollback(), der Fehlerpfad ruft es auf, und notiere()
selbst laeuft in einem eigenen try - ein fehlgeschlagenes Protokoll ist ein
Grund fuer eine Meldung, nicht fuer einen Abbruch.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_013vHZCwcCJnz4LKNgvT4bxv
2026-09-06 12:51:13 +02:00

667 lines
26 KiB
Python
Executable File
Raw Blame History

#!/usr/bin/env python3
"""Ingest-Poller fuer den deutschen Ereignisdatensatz.
Liest die Quellen aus sources.toml, holt alle 15 Minuten die Feeds und legt
jedes neue Item roh ab. Bewusst ohne Kodierung: was hier landet, soll sich
beliebig oft neu kodieren lassen.
Zwei Senken:
--sink postgres://... Produktivbetrieb, die eigentliche Datenbank
--sink jsonl:/pfad/dir Schatten-Ingest als gzip-JSONL pro Slot
Aufrufe:
poller.py --sink "$DATABASE_URL" # Dauerbetrieb, 15-Minuten-Takt
poller.py --sink "$DATABASE_URL" --once # ein Durchlauf
poller.py --check-robots # Lizenzstatus nachpruefen
"""
import argparse
import datetime as dt
import gzip
import hashlib
import html
import json
import os
import pathlib
import re
import signal
import sys
import time
import tomllib
import sqlite3
import urllib.error
import urllib.parse
import urllib.request
import urllib.robotparser
import xml.etree.ElementTree as ET
from email.utils import parsedate_to_datetime
HIER = pathlib.Path(__file__).parent
USER_AGENT = os.environ.get(
"INGEST_USER_AGENT",
"wurzelwerk-ingest/0.1 (+https://stinkwurzpresse.de/wurzelwerk; kontakt@stinkwurzpresse.de)",
)
TAKT = 15 * 60 # Slotlaenge in Sekunden
VERSATZ = 30 # Sekunden nach der Slotgrenze, nicht exakt darauf
NS = {
"a": "http://www.w3.org/2005/Atom",
"dc": "http://purl.org/dc/elements/1.1/",
"content": "http://purl.org/rss/1.0/modules/content/",
"media": "http://search.yahoo.com/mrss/",
}
# Header, die wir zum Abruf mitschreiben. Kosten nichts und beantworten
# spaeter Fragen zur Zwischenspeicherung, die man sonst nicht mehr stellen kann.
HEADER_MERKEN = ("Date", "Age", "Cache-Control", "Content-Type",
"Content-Length", "Server", "ETag", "Last-Modified")
_lauf = True
def _stop(signum, frame):
global _lauf
_lauf = False
print(f"[{_jetzt():%H:%M:%S}] Signal {signum}, beende nach diesem Durchlauf",
flush=True)
# --------------------------------------------------------------------------
# Hilfsfunktionen
# --------------------------------------------------------------------------
def _jetzt():
return dt.datetime.now(dt.timezone.utc)
def slot_von(ts):
"""Auf TAKT-Sekunden seit Epoch abrunden (bei 15 min auf :00/:15/:30/:45)."""
sekunden = int(ts.timestamp())
return dt.datetime.fromtimestamp(
(sekunden // TAKT) * TAKT, tz=dt.timezone.utc)
def norm_titel(t):
"""Wie build_site.norm_title: Basis fuer das Trigramm-Clustering."""
t = (t or "").lower().split(" - ")[0].split(" | ")[0]
# \w haelt Umlaute und ss, ohne die es im Deutschen nicht geht
return re.sub(r"\W+", " ", t, flags=re.UNICODE).strip()
_TAG_RE = re.compile(r"<[^>]+>")
def klartext(s):
"""HTML aus Teasern entfernen. Feeds kodieren teils doppelt."""
if not s:
return None
s = html.unescape(html.unescape(s))
s = _TAG_RE.sub(" ", s)
return re.sub(r"\s+", " ", s).strip() or None
def datum(s):
s = (s or "").strip()
if not s:
return None
try:
d = parsedate_to_datetime(s)
except Exception:
try:
d = dt.datetime.fromisoformat(s.replace("Z", "+00:00"))
except Exception:
return None
if d.tzinfo is None:
d = d.replace(tzinfo=dt.timezone.utc)
return d.astimezone(dt.timezone.utc)
def inhalt_hash(titel, teaser, url):
"""Fingerabdruck des Item-Inhalts.
Identisch zur SQL-Formel in der Migration, damit Bestandsdaten und neue
Zeilen denselben Hash tragen. md5 reicht - es geht um Gleichheit, nicht
um Faelschungssicherheit.
"""
roh = "\x1f".join((titel or "", teaser or "", url or ""))
return hashlib.md5(roh.encode("utf-8"), usedforsecurity=False).hexdigest()
def _text(el):
return (el.text or "").strip() if el is not None and el.text else None
def _erst(item, pfade):
for p in pfade:
el = item.find(p, NS) if ":" in p else item.find(p)
v = _text(el)
if v:
return v
return None
# --------------------------------------------------------------------------
# Feed-Abruf und -Parsing
# --------------------------------------------------------------------------
def hole(url, etag=None, last_modified=None, timeout=30):
"""Bedingter GET.
Gibt (status, bytes|None, etag, last_modified, header) zurueck. Die
Rohbytes werden bewusst durchgereicht und nicht nur geparst: der Parser
wird sich aendern, die Historie soll dann rueckwirkend davon profitieren.
"""
kopf = {"User-Agent": USER_AGENT, "Accept-Encoding": "gzip"}
if etag:
kopf["If-None-Match"] = etag
if last_modified:
kopf["If-Modified-Since"] = last_modified
req = urllib.request.Request(url, headers=kopf)
try:
with urllib.request.urlopen(req, timeout=timeout) as r:
roh = r.read()
if r.headers.get("Content-Encoding") == "gzip":
roh = gzip.decompress(roh)
header = {k: r.headers.get(k) for k in HEADER_MERKEN
if r.headers.get(k)}
return (r.status, roh, r.headers.get("ETag"),
r.headers.get("Last-Modified"), header)
except urllib.error.HTTPError as err:
if err.code == 304:
return 304, None, etag, last_modified, {}
raise
_HREF_RE = re.compile(r'href=["\']([^"\']+)["\']', re.I)
def _kategorien(it):
"""<category>, dc:subject, Atom-term. Vom Verlag selbst vergebene Label."""
aus = []
for el in it.findall("category") + it.findall("dc:subject", NS):
if _text(el):
aus.append(_text(el))
for el in it.findall("a:category", NS):
if el.get("term"):
aus.append(el.get("term"))
# Reihenfolge erhalten, Dubletten raus
return list(dict.fromkeys(aus))
def _medien(it):
"""enclosure und Media RSS. Acht von zwoelf Quellen liefern Bilder."""
aus = []
for el in it.findall("enclosure"):
if el.get("url"):
aus.append({"url": el.get("url"), "typ": el.get("type"),
"bytes": el.get("length"), "rolle": "enclosure"})
for tag, rolle in (("media:content", "content"),
("media:thumbnail", "thumbnail")):
for el in it.findall(tag, NS):
if el.get("url"):
aus.append({"url": el.get("url"), "typ": el.get("type"),
"bytes": el.get("fileSize"), "rolle": rolle})
return aus
def _links(*html_stuecke):
"""Verlinkungen aus dem Teaser-HTML, bevor klartext() die Tags entfernt.
GDELT haelt das Gegenstueck als PAGE_LINKS im GKG: welche Meldung auf
welche verweist, ist ein eigenes Signal.
"""
aus = []
for st in html_stuecke:
if st:
aus += _HREF_RE.findall(html.unescape(st))
return list(dict.fromkeys(aus))
def parse_feed(roh):
"""RSS 2.0, RDF und Atom auf eine gemeinsame Form bringen.
Gibt (items, uebersprungen) zurueck. Uebersprungen wird nur, was keinen
Titel hat - das wird gezaehlt statt stillschweigend verworfen.
"""
wurzel = ET.fromstring(roh)
eintraege = wurzel.findall(".//item") or wurzel.findall(".//a:entry", NS)
aus, uebersprungen = [], 0
for pos, it in enumerate(eintraege):
titel = _erst(it, ["title", "a:title"])
if not titel:
uebersprungen += 1
continue
link = _erst(it, ["link", "a:id"])
if not link:
for le in it.findall("a:link", NS):
if le.get("rel") in (None, "alternate") and le.get("href"):
link = le.get("href")
break
guid = _erst(it, ["guid", "a:id"]) or link
if not guid:
guid = hashlib.sha256(titel.encode()).hexdigest()[:32]
# Kurzfassung und Volltext getrennt halten: frueher gewann die kurze
# description und content:encoded fiel unter den Tisch.
teaser_html = _erst(it, ["description", "a:summary"])
voll_html = _erst(it, ["content:encoded", "a:content"])
if not teaser_html:
teaser_html = voll_html
eintrag = {
"titel": klartext(titel),
"url": link,
"guid": guid,
"position": pos, # Reihenfolge im Feed = oft die Gewichtung
"teaser": klartext(teaser_html),
"volltext": klartext(voll_html) if voll_html is not teaser_html else None,
"autor": _erst(it, ["author", "dc:creator", "a:author/a:name"]),
"kategorien": _kategorien(it),
"medien": _medien(it),
"links": _links(teaser_html, voll_html),
"pubdate": datum(_erst(it, ["pubDate", "dc:date", "a:published"])),
"aktualisiert": datum(_erst(it, ["a:updated", "dc:modified"])),
}
if eintrag["pubdate"] is None:
eintrag["pubdate"] = eintrag["aktualisiert"]
# Der Hash bleibt bewusst auf Titel/Teaser/URL beschraenkt. Naehme er
# die neuen Felder auf, gaelte beim ersten Lauf nach dem Update jedes
# Bestandsitem als revidiert.
eintrag["inhalt_hash"] = inhalt_hash(
eintrag["titel"], eintrag["teaser"], eintrag["url"])
aus.append(eintrag)
return aus, uebersprungen
# --------------------------------------------------------------------------
# Senken
# --------------------------------------------------------------------------
class JsonlSenke:
"""Schatten-Ingest fuer den Pi: gzip-JSONL, ein Verzeichnis pro Tag.
Kein Schema, keine Abfragen - nur die Garantie, dass die Rohdaten noch
da sind, wenn der Produktiv-Poller einen Tag lang stillstand. Die SQLite daneben haelt
nur die gesehenen Schluessel, damit nicht bei jedem Slot der komplette
Feed erneut geschrieben wird.
"""
def __init__(self, ziel):
self.ziel = pathlib.Path(ziel)
self.ziel.mkdir(parents=True, exist_ok=True)
self.db = sqlite3.connect(self.ziel / "gesehen.sqlite")
self.db.execute("create table if not exists gesehen ("
"quelle text, guid text, hash text, "
"primary key (quelle, guid))")
self.db.commit()
def sync_quellen(self, quellen):
(self.ziel / "sources.snapshot.json").write_text(
json.dumps(quellen, indent=2, ensure_ascii=False, default=str))
def zustand(self, qid):
return None, None
def rollback(self):
self.db.rollback()
def schreibe(self, quelle, items, slot):
frisch, neu, geaendert = [], 0, 0
for it in items:
row = self.db.execute(
"select hash from gesehen where quelle = ? and guid = ?",
(quelle["id"], it["guid"])).fetchone()
if row is None:
self.db.execute("insert into gesehen (quelle, guid, hash) "
"values (?, ?, ?)",
(quelle["id"], it["guid"], it["inhalt_hash"]))
neu += 1
elif row[0] != it["inhalt_hash"]:
self.db.execute("update gesehen set hash = ? "
"where quelle = ? and guid = ?",
(it["inhalt_hash"], quelle["id"], it["guid"]))
geaendert += 1
else:
continue
frisch.append(it)
self.db.commit()
if not frisch:
return 0, 0, 0
tag = self.ziel / slot.strftime("%Y%m%d")
tag.mkdir(exist_ok=True)
pfad = tag / (slot.strftime("%Y%m%d%H%M%S") + ".jsonl.gz")
# gzip-Member lassen sich anhaengen, das Ergebnis bleibt gueltiges gzip
with gzip.open(pfad, "at", encoding="utf-8") as f:
for it in frisch:
f.write(json.dumps(
{"quelle": quelle["id"], "slot": slot.isoformat(),
"abgerufen_am": _jetzt().isoformat(), **it},
ensure_ascii=False, default=str) + "\n")
return neu, geaendert, 0
def speichere_feed(self, quelle, slot, status, header, koerper):
"""Rohes Feed-XML gezippt ablegen, dedupliziert ueber den Hash."""
if not koerper:
return False
h = hashlib.sha256(koerper).hexdigest()
tag = self.ziel / slot.strftime("%Y%m%d") / "feeds"
tag.mkdir(parents=True, exist_ok=True)
pfad = tag / f"{quelle['id']}-{h[:16]}.xml.gz"
if pfad.exists():
return False
pfad.write_bytes(gzip.compress(koerper))
return True
def notiere(self, *a, **kw):
pass
def close(self):
self.db.close()
class PostgresSenke:
def __init__(self, dsn):
import psycopg
self.conn = psycopg.connect(dsn, autocommit=False)
def rollback(self):
self.conn.rollback()
def sync_quellen(self, quellen):
with self.conn.cursor() as cur:
for q in quellen:
cur.execute("""
insert into quellen
(id, name, typ, url, land, traeger, ressort,
lizenz_status, robots_geprueft_am, aktiv, notiz)
values (%(id)s, %(name)s, %(typ)s, %(url)s, %(land)s,
%(traeger)s, %(ressort)s, %(lizenz_status)s,
%(robots_geprueft_am)s, %(aktiv)s, %(notiz)s)
on conflict (id) do update set
name = excluded.name, typ = excluded.typ,
url = excluded.url, land = excluded.land,
traeger = excluded.traeger, ressort = excluded.ressort,
lizenz_status = excluded.lizenz_status,
robots_geprueft_am = excluded.robots_geprueft_am,
aktiv = excluded.aktiv, notiz = excluded.notiz
""", {k: q.get(k) for k in
("id", "name", "typ", "url", "land", "traeger", "ressort",
"lizenz_status", "robots_geprueft_am", "aktiv", "notiz")})
self.conn.commit()
def zustand(self, qid):
with self.conn.cursor() as cur:
cur.execute("select etag, last_modified from quellen where id = %s",
(qid,))
r = cur.fetchone()
return (r[0], r[1]) if r else (None, None)
def schreibe(self, quelle, items, slot):
"""Neue Items einfuegen, geaenderte fortschreiben.
Redaktionen aendern Ueberschriften nach der Veroeffentlichung. Frueher
lief das in ein `do nothing` und war unsichtbar. Jetzt schreibt der
Upsert den aktuellen Stand fort und legt jede Fassung zusaetzlich in
raw_item_versionen ab - auch die erste, damit die Historie vollstaendig
ist. Bleibt der Hash gleich, liefert das WHERE keine Zeile zurueck und
es passiert nichts.
"""
neu = geaendert = nachgetragen = 0
with self.conn.cursor() as cur:
for it in items:
cur.execute("""
insert into raw_items
(quelle, item_key, url, titel, titel_norm, teaser,
autor, pubdate, slot, inhalt_hash, roh,
position, volltext, kategorien, medien, links,
aktualisiert)
values (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s,
%s, %s, %s, %s, %s, %s)
on conflict (quelle, item_key) do update set
url = excluded.url,
titel = excluded.titel,
titel_norm = excluded.titel_norm,
teaser = excluded.teaser,
autor = excluded.autor,
pubdate = excluded.pubdate,
position = coalesce(raw_items.position,
excluded.position),
volltext = excluded.volltext,
kategorien = excluded.kategorien,
medien = excluded.medien,
links = excluded.links,
aktualisiert = excluded.aktualisiert,
inhalt_hash = excluded.inhalt_hash,
roh = excluded.roh
where raw_items.inhalt_hash is distinct from excluded.inhalt_hash
or raw_items.kategorien is null
returning id, (xmax = 0) as ist_neu
""", (quelle["id"], it["guid"], it["url"], it["titel"],
norm_titel(it["titel"]), it["teaser"], it["autor"],
it["pubdate"], slot, it["inhalt_hash"],
json.dumps(it, ensure_ascii=False, default=str),
it["position"], it["volltext"], it["kategorien"],
json.dumps(it["medien"], ensure_ascii=False),
it["links"], it["aktualisiert"]))
zeile = cur.fetchone()
if not zeile:
continue # unveraendert, nichts zu tun
rid, ist_neu = zeile
# Ob der Inhalt wirklich neu ist, sagt der Unique-Index der
# Fassungstabelle - zuverlaessiger als ein Vergleich in
# RETURNING, wo die Zeile schon fortgeschrieben ist.
cur.execute("""
insert into raw_item_versionen
(raw_item_id, slot, titel, teaser, url, inhalt_hash)
values (%s, %s, %s, %s, %s, %s)
on conflict (raw_item_id, inhalt_hash) do nothing
returning id
""", (rid, slot, it["titel"], it["teaser"], it["url"],
it["inhalt_hash"]))
inhalt_neu = cur.fetchone() is not None
if ist_neu:
neu += 1
elif inhalt_neu:
geaendert += 1
cur.execute("""
update raw_items
set revisionen = revisionen + 1,
zuletzt_geaendert = now()
where id = %s
""", (rid,))
else:
nachgetragen += 1 # nur die neuen Felder gefuellt
self.conn.commit()
return neu, geaendert, nachgetragen
def merke_zustand(self, qid, etag, last_modified, geaendert):
with self.conn.cursor() as cur:
cur.execute("""
update quellen set etag = %s, last_modified = %s,
zuletzt_geprueft = now(),
zuletzt_geaendert = case when %s then now()
else zuletzt_geaendert end
where id = %s
""", (etag, last_modified, geaendert, qid))
self.conn.commit()
def speichere_feed(self, quelle, slot, status, header, koerper):
"""Rohes Feed-XML gezippt ablegen, dedupliziert ueber den Hash.
Der Parser wird sich aendern. Nur mit den Rohbytes laesst sich eine
spaetere Verbesserung rueckwirkend auf die Historie anwenden - dieselbe
Ueberlegung, aus der raw_items und codings getrennt sind.
"""
if not koerper:
return False
h = hashlib.sha256(koerper).hexdigest()
gz = gzip.compress(koerper)
with self.conn.cursor() as cur:
cur.execute("""
insert into feed_abrufe
(quelle, slot, http_status, header, koerper_hash,
koerper, bytes_roh, bytes_gz)
values (%s, %s, %s, %s, %s, %s, %s, %s)
on conflict (quelle, koerper_hash) do nothing
returning id
""", (quelle["id"], slot, status,
json.dumps(header, ensure_ascii=False), h,
gz, len(koerper), len(gz)))
neu = cur.fetchone() is not None
self.conn.commit()
return neu
def notiere(self, quelle, slot, dauer_ms, status, gesehen, neu,
geaendert, uebersprungen, bytes_roh, fehler):
with self.conn.cursor() as cur:
cur.execute("""
insert into poll_laeufe
(quelle, slot, dauer_ms, http_status, items_gesehen,
items_neu, items_geaendert, items_uebersprungen,
bytes_roh, fehler)
values (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s)
""", (quelle, slot, dauer_ms, status, gesehen, neu, geaendert,
uebersprungen, bytes_roh, fehler))
self.conn.commit()
def close(self):
self.conn.close()
# --------------------------------------------------------------------------
# Durchlauf
# --------------------------------------------------------------------------
def lade_quellen(pfad):
with open(pfad, "rb") as f:
return tomllib.load(f)["quelle"]
def durchlauf(quellen, senke, slot):
ges_neu = ges_gesehen = ges_geaendert = 0
for q in quellen:
if q["typ"] != "rss":
continue
t0 = time.monotonic()
etag, lm = senke.zustand(q["id"])
status = gesehen = neu = geaendert = uebersprungen = bytes_roh = 0
nachgetragen = 0
fehler = None
try:
status, roh, n_etag, n_lm, header = hole(q["url"], etag, lm)
if status == 304:
pass
else:
bytes_roh = len(roh)
senke.speichere_feed(q, slot, status, header, roh)
items, uebersprungen = parse_feed(roh)
gesehen = len(items)
neu, geaendert, nachgetragen = senke.schreibe(q, items, slot)
if hasattr(senke, "merke_zustand"):
senke.merke_zustand(q["id"], n_etag, n_lm, neu > 0)
except Exception as e: # eine Quelle darf nicht
fehler = f"{type(e).__name__}: {e}" # den Durchlauf killen
status = getattr(e, "code", None)
# Ohne Rollback bleibt eine abgebrochene Transaktion stehen, und
# das notiere() gleich darunter scheitert an ihr statt den Fehler
# festzuhalten - der eigentliche Grund waere dann verloren.
senke.rollback()
dauer = int((time.monotonic() - t0) * 1000)
try:
senke.notiere(q["id"], slot, dauer, status, gesehen, neu, geaendert,
uebersprungen, bytes_roh, fehler)
except Exception as e: # das Protokoll darf den Lauf nicht killen
senke.rollback()
print(f" ! {q['id']:14} Protokoll fehlgeschlagen: "
f"{type(e).__name__}: {e}", flush=True)
ges_neu += neu
ges_geaendert += geaendert
ges_gesehen += gesehen
marke = "!" if fehler else (" " if status != 304 else ".")
print(f" {marke} {q['id']:14} {str(status):>4} "
f"{gesehen:4} gesehen {neu:4} neu {geaendert:3} rev "
f"{uebersprungen:2} uebrg"
+ (f" {nachgetragen:4} nachgetr" if nachgetragen else "")
+ f" {dauer:5} ms"
+ (f" {fehler}" if fehler else ""), flush=True)
return ges_gesehen, ges_neu, ges_geaendert
def check_robots(quellen):
"""Lizenzstatus gegen die aktuelle robots.txt nachpruefen."""
print(f"{'quelle':14} {'im Register':12} {'robots.txt':12} Abweichung")
abweichungen = 0
for q in quellen:
if q["typ"] != "rss":
continue
p = urllib.parse.urlparse(q["url"])
rp = urllib.robotparser.RobotFileParser()
rp.set_url(f"{p.scheme}://{p.netloc}/robots.txt")
try:
rp.read()
ist = "offen" if rp.can_fetch(USER_AGENT, q["url"]) else "gesperrt"
except Exception as e:
ist = "fehler"
soll = q["lizenz_status"]
# 'ungeprueft' im Register ist keine Abweichung, sondern eine Aufgabe
weicht = soll != "ungeprueft" and ist != soll
abweichungen += weicht
print(f"{q['id']:14} {soll:12} {ist:12} {'<<< PRUEFEN' if weicht else ''}")
print(f"\n{abweichungen} Abweichung(en). "
f"Bei Treffern sources.toml anpassen, nicht den Code.")
return abweichungen
def main():
ap = argparse.ArgumentParser(description=__doc__,
formatter_class=argparse.RawDescriptionHelpFormatter)
ap.add_argument("--sources", default=str(HIER / "sources.toml"))
ap.add_argument("--sink", default=os.environ.get("DATABASE_URL"),
help="postgres://... oder jsonl:/pfad/zum/verzeichnis")
ap.add_argument("--once", action="store_true", help="ein Durchlauf, dann Ende")
ap.add_argument("--check-robots", action="store_true",
help="Lizenzstatus nachpruefen, nichts abrufen")
args = ap.parse_args()
alle = lade_quellen(args.sources)
if args.check_robots:
return 1 if check_robots(alle) else 0
if not args.sink:
ap.error("--sink oder DATABASE_URL noetig")
aktiv = [q for q in alle if q.get("aktiv")]
gesperrt = [q for q in alle if q.get("lizenz_status") == "gesperrt"]
if any(q.get("aktiv") for q in gesperrt):
sys.exit("FEHLER: gesperrte Quelle ist aktiv geschaltet")
if args.sink.startswith("jsonl:"):
senke = JsonlSenke(args.sink[len("jsonl:"):])
else:
senke = PostgresSenke(args.sink)
senke.sync_quellen(alle)
print(f"{len(aktiv)} aktive Quellen, {len(gesperrt)} gesperrt, "
f"Senke: {args.sink.split('@')[-1]}", flush=True)
signal.signal(signal.SIGTERM, _stop)
signal.signal(signal.SIGINT, _stop)
try:
while _lauf:
slot = slot_von(_jetzt())
print(f"[{slot:%Y-%m-%d %H:%M}] Slot", flush=True)
gesehen, neu, geaendert = durchlauf(aktiv, senke, slot)
print(f" = {gesehen} gesehen, {neu} neu, "
f"{geaendert} revidiert", flush=True)
if args.once:
break
ziel = slot + dt.timedelta(seconds=TAKT + VERSATZ)
while _lauf and _jetzt() < ziel:
time.sleep(min(5, (ziel - _jetzt()).total_seconds()))
finally:
senke.close()
return 0
if __name__ == "__main__":
sys.exit(main())
Reference in New Issue View Git Blame Copy Permalink