seepuls/.claude/skills/seepuls-curate-events.md

name	description	argument-hint	allowed-tools
seepuls-curate-events	Crawlt Events einer Venue von ihrer offiziellen Programm-Seite und schreibt sie in seepuls.events. Pipeline Plan→Recherche (WebFetch+AI)→Design→Validate→Publish mit Reviewer-Stops und Dedupe via external_id_hash.	<venue-slug> [--source-url <url>]	Read Write Edit WebSearch WebFetch Bash(ls *) Bash(mkdir *) Bash(grep *) Bash(cat *) Bash(curl *) Bash(jq *) Bash(docker exec *) Bash(ssh *) Bash(python3 *) Bash(openssl *) AskUserQuestion

/seepuls-curate-events — Event-Crawl für eine Venue

Crawlt eine Event-Listing-Seite, extrahiert strukturierte Events, und schreibt sie nach Reviewer-Stop in seepuls.events auf prod-DB.

DB ist SOT für Events (kurzlebig, hoher Volumen). Audit-Trail in ~/Documents/seepuls-drafts/<venue-slug>-events-<YYYY-MM-DD>/.

Schwester-Skills: /seepuls-curate-venue, /seepuls-validate-venue.

Pflicht-Lektüre

• seepuls/STATUS.md + seepuls/docs/CURATOR.md
• mana/docs/AGGREGATOR_POLICY.md — Hard-Rules (robots, Attribution, ≤200 chars Description, Hot-Link Bilder)
• seepuls/apps/api/src/db/schema/events.ts — Schema mit external_id_hash

Inputs

1. <venue-slug> (Pflicht) — muss in seepuls.venues existieren.
2. --source-url <url> (optional) — Programm-/Event-Listing-URL. Wenn fehlt: vom Skill aus venue.websiteUrl + WebSearch abgeleitet.

Workspace

~/Documents/seepuls-drafts/<venue-slug>-events-<YYYY-MM-DD>/
├─ plan.md           Stufe 1
├─ research/
│  └─ raw.md         WebFetch-Roh-Output
├─ design/
│  └─ events.jsonl   Strukturierte Kandidaten
├─ validate/
│  └─ report.md
└─ publish/
   ├─ insert.sql     Generiertes SQL
   └─ run.log        Output Insert

Pipeline (5 Stufen)

Stufe 1 — Plan

1. Venue-Existenz:

   ssh mana-server "docker exec seepuls-postgres psql -U seepuls -d seepuls -tAc \"SELECT id, name, website_url, source_domain FROM seepuls.venues WHERE slug='<venue-slug>'\""
   

   Treffer-Miss → stoppen, /seepuls-curate-venue <slug> zuerst.
2. Listing-URL bestimmen:
   • Wenn --source-url gegeben → die nehmen.
   • Sonst: venue.websiteUrl, plus mit WebSearch nach "<venue-name>" programm OR veranstaltungen OR kalender schauen.
3. AGGREGATOR_POLICY-Pre-Check (robots.txt der Source-Domain):

   curl -sf -H "User-Agent: seepuls/0.0.1 (+https://seepuls.mana.how; kontakt@mana.how)" \
     --max-time 5 "https://<domain>/robots.txt" | head -40
   

   Disallow: / für * oder seepuls? → STOPP, blocked_domains eintragen statt crawlen.
4. plan.md:
   • Venue-Slug + Name + venue-ID (aus Stufe 1)
   • Listing-URL
   • Erwartete Event-Zahl (Range, z.B. 3–20)
   • Streitfälle (z.B. „Vorverkauf-Daten gemischt", „Wiederholungs- Termine derselben Aufführung")
5. Reviewer-Stop: User sieht Plan + Listing-URL, sagt go.

Stufe 2 — Recherche (Feed-First, dann WebFetch)

Goldstandard-Pfad (Stage 2.0 — iCal/RSS-Feed): BEVOR WebFetch versucht wird, prüfe TEC-WordPress-Standard-Endpoints. iCal-Daten sind strukturiert (RFC 5545) — kein LLM-Halluzinations-Risiko, kein Jahres-Drift, exakte Zeiten in DTSTART, kein Bot-Protection.

Probe-Liste (mit Mozilla-User-Agent, 4–8 Pfade pro Domain):

for path in "/?ical=1" "/events/?ical=1" "/event/?ical=1" \
            "/veranstaltungen/?ical=1" "/?post_type=tribe_events&ical=1" \
            "/?post_type=event&ical=1"; do
  curl -s -H "User-Agent: Mozilla/5.0" --max-time 4 \
    "https://<domain>$path" | head -c 300
done

Feed gefunden (BEGIN:VCALENDAR + ≥1 BEGIN:VEVENT)? → direkt zu Stage 3 mit Pure-Python-iCal-Parser:

import re
def parse_ical(text):
    text = re.sub(r'\r?\n[ \t]', '', text)  # Line-folding auflösen
    events = []
    for block in re.findall(r'BEGIN:VEVENT(.*?)END:VEVENT', text, re.DOTALL):
        ev = {}
        for line in block.strip().split('\n'):
            line = line.strip()
            if ':' not in line: continue
            kr, val = line.split(':', 1)
            key = kr.split(';')[0]  # Param wie TZID;VALUE=DATE strippen
            ev[key] = val.replace('\\n', ' ').replace('\\,', ',').replace('\\;', ';')
        events.append(ev)
    return events

# DTSTART:20260520T090000 → ISO-8601 + CEST
def parse_dtstart(s):
    if 'T' in s:
        dt = datetime.datetime.strptime(s, '%Y%m%dT%H%M%S')
        return dt.strftime('%Y-%m-%dT%H:%M:%S+02:00'), dt.strftime('%H%M')
    dt = datetime.datetime.strptime(s, '%Y%m%d')
    return dt.strftime('%Y-%m-%dT20:00:00+02:00'), '2000'

# Bei Slug-Kollision (gleicher Event-Titel, gleicher Tag, 2 Termine):
# Anhang HH:MM aus DTSTART
seen = set()
for e in events:
    base = f"{starts_at[:10]}-{slugify(summary)}"
    slug = base
    while slug in seen:
        slug = f"{base}-{hhmm[:2]}{hhmm[2:]}"
    seen.add(slug)

Bekannt-funktionierende iCal-Endpoints (Stand 2026-05-15):

Domain	Endpoint	Events
apollokreuzlingen.ch	/?ical=1 oder /events/?ical=1	14 (mit exakten Zeiten)

Weitere Sites mit RSS aber leeren Event-Feeds: theaterandergrenze.ch, zebra-kino.de, kult-x.ch (die haben TEC nicht aktiviert oder events sind custom-post-type ohne feed). Bei denen weiter WebFetch oder Aggregator.

Standard-Pfad (Stage 2.1 — WebFetch + AI-Extraktion): wenn kein iCal-Feed da. WebFetch auf die Listing-URL mit dem Prompt unten. Wenn das ≥3 Events mit starts_at != null liefert, ist Stage 2.1 fertig.

Multi-Step-Pfad (Stage 2.2 — wenn Listing keine Daten liefert):

1. Stage 2a: WebFetch der Listing-URL → bekommt Event-Stubs mit source_url- Detail-Links, aber starts_at: null.
2. Stage 2b: pro Detail-URL ein eigener WebFetch mit kürzerem Prompt (siehe „Detail-Page-Prompt" unten). 5–10 Detail-Fetches parallel sind OK — alle laufen über dieselbe Domain → robots-Politeness gilt trotzdem (max 1 req/s). Bei großen Listings: in Batches à 5 mit 1.5s Pause zwischen Batches.
3. Stage 2c: Detail-Ergebnisse mit Listing-Stubs joinen (über source_url).

Fallback-Pfad (JS-heavy Sites, Listing rendert leer):

• Aggregator-Sites anzapfen: konstanz-info.com, bodensee.de, party-news.de, thurgaukultur.ch, kreuzlinger.net, bodensee-kultur.info. Diese listen oft Events vieler Venues konsolidiert.
• Pro Aggregator: WebFetch der venue-spezifischen Subseite (z.B. konstanz-info.com/veranstaltungen/<venue-slug>).
• Attribution: source_url ist der Aggregator-Listing-Pfad, im description-Feld bleibt die Original-Venue.
• Achtung: nicht alle Aggregator-Listings sind aktuell. Cross-Check mit Original-Venue-Site stichprobenartig.

WebFetch-Prompt (Listing-Variante):

Extrahiere alle anstehenden Veranstaltungen aus dieser Programm-Seite
als JSON-Array. Pro Event:
  - title (string, Originalsprache, ohne Zusätze wie "ABO 1" oder Wochentag-Präfixe)
  - starts_at (ISO-8601 wenn möglich: YYYY-MM-DDTHH:mm:ss+02:00; fallback YYYY-MM-DD; null wenn unklar)
  - ends_at (ISO-8601 wenn auffindbar, sonst null)
  - description (≤200 Zeichen, Originalsprache, KEIN Re-Phrasing)
  - ticket_url (Direkt-Link, wenn vorhanden, sonst null)
  - image_url (absolute URL, wenn auf Seite eingebunden, sonst null)
  - source_url (Detail-Link auf das Event wenn vorhanden, sonst die Listing-URL)

Regeln:
- Nur kommende Events (in der Zukunft).
- Maximal 50 Events.
- Wenn Datum NICHT eindeutig ist (z.B. "demnächst"): starts_at = null setzen
  (NICHT droppen — Multi-Step-Pfad kann Datum aus Detail-Page nachholen).
- Wenn keine Events auf der Seite: leeres Array [].

Antworte AUSSCHLIESSLICH mit dem JSON-Array, kein Markdown-Wrap.

WebFetch-Prompt (Detail-Page-Variante) — wenn Listing nur Stubs hatte:

Extrahiere für diese eine Veranstaltung:
- title
- starts_at (ISO-8601 oder YYYY-MM-DD)
- ends_at (oder null)
- description (≤200 chars, Original)
- ticket_url (oder null)
- image_url (oder null)

Antworte als compact JSON object, kein Wrap.

Output → research/raw.json (Listing-Ergebnis) + research/details/<n>.json (pro Detail-Page). Anzahl prüfen vs Plan-Erwartung. Wenn 0: explizit flag im plan.md was die Quelle ist (JS-rendered, Login, fail) und dem User vorschlagen, Aggregator-Fallback oder anderen Listing-Pfad zu versuchen.

Stufe 3 — Design (Datum parsen + Hash + Slug)

Lese das JSON-Array, kanonisiere jedes Event. Ready-to-copy Python-Helpers (Unicode-sicher dank NFKD-Normalisierung):

import json, hashlib, datetime, re, secrets, unicodedata

def parse_date(s):
    if not s: return None
    # 1. Strict ISO-8601 mit Time + TZ
    if re.match(r'^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}', s):
        try: return datetime.datetime.fromisoformat(s.replace('Z', '+00:00'))
        except: pass
    # 2. YYYY-MM-DD: 20:00 CEST als Konzert-Default
    if re.match(r'^\d{4}-\d{2}-\d{2}$', s):
        return datetime.datetime.fromisoformat(f"{s}T20:00:00+02:00")
    # 3. DD.MM.YYYY (deutsches Format)
    m = re.match(r'^(\d{2})\.(\d{2})\.(\d{4})$', s)
    if m:
        d, mo, y = m.groups()
        return datetime.datetime.fromisoformat(f"{y}-{mo}-{d}T20:00:00+02:00")
    # KEIN free-form-Parsing — "Samstag 15. Juni" → None (Event droppen)
    return None

def event_hash(title, starts_at_iso, venue_id):
    norm = re.sub(r'\s+', ' ', title.lower()).strip()
    iso_min = starts_at_iso[:16]
    return hashlib.sha256(f"{norm}|{iso_min}|{venue_id}".encode()).hexdigest()[:32]

def slugify(text, max_len=48):
    """Unicode-sicher: Á→a, č→c, ß→ss, é→e. Nutzt NFKD-Normalisierung."""
    if not text: return ''
    # 1. NFKD trennt Akzente vom Basis-Buchstaben
    t = unicodedata.normalize('NFKD', text)
    # 2. Mb-Combining-Chars filtern
    t = ''.join(c for c in t if not unicodedata.combining(c))
    # 3. Deutsche Sonder-Replaces (ß ist nicht NFKD-zerlegbar)
    t = t.lower().replace('ß', 'ss')
    # 4. Alles außer [a-z0-9-] zu Bindestrich, Trim Bindestriche
    t = re.sub(r'[^a-z0-9]+', '-', t).strip('-')
    return t[:max_len].rstrip('-')

def event_slug(title, dt):
    date_part = dt.strftime('%Y-%m-%d')
    title_part = slugify(title)
    return f"{date_part}-{title_part}" if title_part else date_part

def truncate_at_sentence(s, max_len=220):
    """Beschreibung am Satzende kürzen statt mid-word."""
    if not s or len(s) <= max_len: return s
    cut = s[:max_len]
    # Suche letztes Satzende
    for sep in ['. ', '! ', '? ', '… ']:
        i = cut.rfind(sep)
        if i > max_len * 0.6:
            return cut[:i+1]
    # Fallback: letztes Leerzeichen
    i = cut.rfind(' ')
    return (cut[:i] + '…') if i > 0 else cut

Wichtig: slugify ist NFKD-basiert — kein Manual-ä→ae/á→a- Mapping mehr nötig. Funktioniert für alle europäischen Sprachen (CZ, PL, FR, ES, IT). Beispiel: "CALLE MÁLAGA" → calle-malaga (statt früher calle-m-laga).

Pro Kandidat: {title, slug, starts_at, ends_at, description, ticket_url, image_url, source_url, external_id_hash} in design/events.jsonl.

Filter:

• starts_at < now - 12h → droppen (vergangen)
• description > 220 chars → trimmen am Satzende
• external_id_hash collision innerhalb der Charge → erster bleibt

Reviewer-Stop: User sieht Markdown-Tabelle (Datum · Titel · Quelle) mit 10 Stichproben + Gesamtzahl + Streitfälle (events ohne ticket_url o.Ä.). Sagt go.

Stufe 4 — Validate

Schreibe validate/report.md mit:

• venue-exists ✓ (aus Stufe 1)
• datum-future: alle starts_at ≥ now − 12h
• datum-parsed: alle starts_at sind gültige ISO-8601-Datums
• dedupe-existing gegen DB:

  ssh mana-server "docker exec seepuls-postgres psql -U seepuls -d seepuls -tAc \
    \"SELECT external_id_hash FROM seepuls.events WHERE venue_id='<vid>'\""
  

  Existing-Hashes-Set bilden. Pro Kandidat: schon da? → wird zu UPDATE statt INSERT (Stufe 5).
• description-length: alle ≤ 220 Zeichen
• source-coverage: alle haben source_url nicht null
• image-https: image_url ist https:// oder null (kein data: oder relative)

Bei rotem Finding: nicht eigenmächtig fixen — User informieren.

Stufe 5 — Publish (SQL gegen prod-DB)

Generiere publish/insert.sql:

INSERT INTO seepuls.events
  (id, slug, venue_id, event_source_id, title, description, starts_at,
   ends_at, source_url, ticket_url, image_url, external_id_hash,
   first_seen_at, last_seen_at, raw_payload)
VALUES
  ('ev_<random12>', '<slug>', '<vid>', NULL, '<title>', '<desc>',
   '<starts_at_iso>'::timestamptz, NULL, '<src>', '<ticket>', '<img>',
   '<hash>', NOW(), NOW(), '<json>'::jsonb)
ON CONFLICT (external_id_hash) DO UPDATE
SET title = EXCLUDED.title,
    description = EXCLUDED.description,
    starts_at = EXCLUDED.starts_at,
    ends_at = EXCLUDED.ends_at,
    ticket_url = EXCLUDED.ticket_url,
    image_url = EXCLUDED.image_url,
    source_url = EXCLUDED.source_url,
    last_seen_at = NOW(),
    soft_deleted_at = NULL,
    raw_payload = EXCLUDED.raw_payload,
    updated_at = NOW();

Strings escapen (Python-Heredoc bevorzugt). Pro Event eine INSERT-Anweisung.

Ausführen:

ssh mana-server "docker exec -i seepuls-postgres psql -U seepuls -d seepuls" \
  < ~/Documents/seepuls-drafts/<…>/publish/insert.sql 2>&1 \
  | tee publish/run.log

Plus crawl_jobs-Audit-Eintrag als Compliance-Beweis:

INSERT INTO seepuls.crawl_jobs
  (id, target_url, target_domain, job_kind, source_id, user_agent_used,
   robots_check_passed, respected_robots_delay, status, http_status,
   new_items, updated_items, started_at, finished_at)
VALUES
  ('cj_<random16>', '<listing-url>', '<domain>', 'event-source-crawl',
   NULL, 'seepuls/0.0.1 (+…)', true, true, 'ok', 200,
   <new>, <updated>, NOW() - interval '5 seconds', NOW());

Anti-Halluzinations-Regeln

• Niemals Datum raten. Wenn aus der Seite nicht eindeutig: droppen.
• Niemals Description LLM-paraphrasieren. Direkt aus Quelle, ≤ 220 Zeichen, am Satzende kürzen.
• Niemals Bilder lokal cachen. Hot-Link via image_url (Policy §3).
• Niemals Pre-Verkauf-Daten als Event. „Tickets ab 10.5." ist kein Event-Datum.
• Niemals außerhalb der Venue. Wenn das Listing Events einer anderen Venue enthält („zu Gast bei …"), nicht übernehmen.

Reviewer-Stops sind Pflicht

Nach Stufe 1 (Plan + Listing-URL) und Stufe 3 (Vorschau-Liste). Auch bei „mach einfach". Stops retten den Korpus vor Falsch-Daten.

Was NICHT der Skill macht

• Crawl-Pipeline-Scheduler aktivieren. Der scheduler.ts läuft via SEEPULS_SCHEDULER=on und braucht eine event_source-Tabelle — außerhalb dieses Skills.
• Take-Down-Handling. /api/v1/takedown + Mod-Workflow.
• Bilder cachen. Hot-Link Default laut Policy.
• Verlängern auf 100+ Events ohne Reviewer-Sampling. Bei großen Listings (Saison-Programm 200 Events): Sampling-Stop in Stufe 3.

Known-Patterns pro Domain (Lessons aus Demo-Runs)

Pattern	Domains (Beispiele)	Skill-Strategie
Statisches HTML mit Events im Body	zebra-kino.de, bodensee-kultur.info	Single-Step Listing-Fetch reicht
WordPress mit Event-Detail-Pages	apollokreuzlingen.ch, museumrosenegg.ch	Multi-Step: Listing → Detail-Pages
JS-Single-Page-App (Splash ohne Daten)	kantine-kn.de, cinestar.de, theaterkonstanz.de, k9-kulturzentrum.de	Aggregator-Fallback nutzen
Aggregator-Sites mit konsolidierten Events	konstanz-info.com, bodensee.de, party-news.de, kreuzlinger.net, thurgaukultur.ch	Venue-Subseite des Aggregators fetchen
iCal-Feed verlinkt	manche Wordpress-Sites	Direct-Fetch der .ics-URL (Python icalendar-Lib)
Ticket-Plattform-Subdomain	seetickets.com, eventfrog.ch, ticketino.com	Manchmal pro Venue separate Subdomain mit konkreten Daten

Anti-Pattern: nicht versuchen, JS-Rendering durch Wiederholtes WebFetchen anderer Subseiten zu kompensieren. Wenn 2 Versuche leer: direkt auf Aggregator switchen.

Venue-ID-Stability (wichtig)

Venue-IDs sind nicht stable über Slug-Form abzuleiten. Die ersten 5 Venues kamen via direktem SQL-Insert (IDs v_seed_*), die späteren 33 via YAML-Importer (IDs v_yaml_*). Beim Insert-SQL NIE die ID aus dem Slug raten — immer aus DB lesen:

import subprocess, json
def venue_id_for_slug(slug):
    out = subprocess.check_output([
        'ssh', 'mana-server',
        f"docker exec seepuls-postgres psql -U seepuls -d seepuls -tAc "
        f"\"SELECT id FROM seepuls.venues WHERE slug='{slug}'\""
    ], text=True).strip()
    return out or None

Bei FK-Constraint-Violation (Key (venue_id)=(…) is not present in table "venues") → IDs in DB checken, Mapping korrigieren.

DB-Migration-Quirk

db:push legt uniqueIndex aus dem Drizzle-Schema NICHT immer an (prüfbar mit \\di seepuls.events*). Wenn der erste echte Insert einer neuen Venue mit ON CONFLICT (external_id_hash) failt mit „no unique or exclusion constraint matching":

echo "CREATE UNIQUE INDEX IF NOT EXISTS events_dedupe_uq ON seepuls.events(external_id_hash);
CREATE UNIQUE INDEX IF NOT EXISTS events_slug_uq ON seepuls.events(slug);
CREATE INDEX IF NOT EXISTS events_venue_idx ON seepuls.events(venue_id);
CREATE INDEX IF NOT EXISTS events_starts_at_idx ON seepuls.events(starts_at);" \
  | ssh mana-server "bash -lc 'docker exec -i seepuls-postgres psql -U seepuls -d seepuls'"

Idempotent — kann jederzeit re-run. Saubere Lösung: drizzle-kit generate + migrate statt push. V2.

Dependencies

• WebFetch (Claude-Skill-Tool, AI-Extraktion ohne mana-llm/research)
• ssh + docker exec auf mana-server für SQL
• Sub-Agent mana-compliance für Veto bei sensiblen Themen
• Python 3 mit unicodedata, hashlib, secrets, re, datetime (alle stdlib)

Referenzen

• Schwester-Skill: /seepuls-curate-venue
• Schema: apps/api/src/db/schema/events.ts
• Crawl-Policy: ../mana/docs/AGGREGATOR_POLICY.md
• DB-Pattern: gleiche Hash-Heuristik wie apps/api/src/services/event-extraction.ts (Server-Side)