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

---
name: seepuls-curate-events
description: 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.
argument-hint: '<venue-slug> [--source-url <url>]'
allowed-tools:
  - 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-curate-venue.md),
[`/seepuls-validate-venue`](./seepuls-validate-venue.md).

## 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:
   ```bash
   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):
   ```bash
   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):
```bash
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:

```python
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):

```python
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:
  ```bash
  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`:

```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:
```bash
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:
```sql
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:

```python
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":

```bash
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)