From bbfda72bc0d8db2c2ee7141f1b102c3a3dcb76da Mon Sep 17 00:00:00 2001 From: ecki Date: Sun, 20 Sep 2026 17:21:38 +0200 Subject: [PATCH] Jambase Calls reduziert --- .env.example | 1 + README.md | 1 + backend/app/config.py | 1 + backend/app/providers/jambase.py | 223 +++++++++++++++++++++---------- backend/app/services.py | 113 +++++++++++++++- 5 files changed, 267 insertions(+), 72 deletions(-) diff --git a/.env.example b/.env.example index a2d209c..e8f0607 100644 --- a/.env.example +++ b/.env.example @@ -12,6 +12,7 @@ TICKETMASTER_API_KEY= JAMBASE_API_KEY= JAMBASE_USER_AGENT=eventlens/0.1 JAMBASE_SYNC_INTERVAL_HOURS=72 +JAMBASE_BATCH_SIZE=10 BANDSINTOWN_APP_ID= EVENTIM_ENABLED=true POLL_INTERVAL_HOURS=6 diff --git a/README.md b/README.md index fff8b9b..c01099e 100644 --- a/README.md +++ b/README.md @@ -36,6 +36,7 @@ API-Statusinfo findest du unter `http://127.0.0.1:8001/api`. - `JAMBASE_API_KEY`: JamBase Data API Bearer Token - `JAMBASE_USER_AGENT`: eigene App-Kennung fuer JamBase-Requests - `JAMBASE_SYNC_INTERVAL_HOURS`: Mindestabstand fuer JamBase-API-Laeufe, Standard `72`; `0` deaktiviert die Bremse +- `JAMBASE_BATCH_SIZE`: Anzahl Watchlist-Namen pro JamBase-Bulk-Request, Standard `10` - `BANDSINTOWN_APP_ID`: echte Bandsintown App-ID fuer Artist-Events - `EVENTIM_ENABLED`: aktiviert den Eventim-Website-Provider - `EVENTLENS_AUTH_USERNAME`, `EVENTLENS_AUTH_PASSWORD`: optionaler Passwortschutz fuer Webfrontend und API diff --git a/backend/app/config.py b/backend/app/config.py index c13934c..edff352 100644 --- a/backend/app/config.py +++ b/backend/app/config.py @@ -15,6 +15,7 @@ class Settings: jambase_api_key = os.getenv("JAMBASE_API_KEY", "") jambase_user_agent = os.getenv("JAMBASE_USER_AGENT", "eventlens/0.1") jambase_sync_interval_hours = int(os.getenv("JAMBASE_SYNC_INTERVAL_HOURS", "72")) + jambase_batch_size = int(os.getenv("JAMBASE_BATCH_SIZE", "10")) bandsintown_app_id = os.getenv("BANDSINTOWN_APP_ID", "") eventim_enabled = os.getenv("EVENTIM_ENABLED", "true").lower() == "true" poll_interval_hours = int(os.getenv("POLL_INTERVAL_HOURS", "6")) diff --git a/backend/app/providers/jambase.py b/backend/app/providers/jambase.py index 93e39a8..847fcc0 100644 --- a/backend/app/providers/jambase.py +++ b/backend/app/providers/jambase.py @@ -3,7 +3,7 @@ from datetime import datetime import requests from app.config import settings -from app.models import RegionScope, WatchType +from app.models import RegionScope, WatchItem, WatchType from app.providers.utils import normalize_search_text @@ -20,35 +20,109 @@ class JamBaseProvider: watch_type: WatchType, region_scope: RegionScope, ) -> list[dict]: - self.last_status = "ok" - self.last_message = "" + self._reset_status() if not self.is_configured(): self.last_status = "ok" self.last_message = "JamBase ist nicht konfiguriert. Bitte JAMBASE_API_KEY setzen." return [] - headers = { + items = self._fetch_items(term, watch_type, region_scope) + if items is None: + return [] + + results = self._dedupe_results( + self._event_data_for_term(item, term, watch_type, region_scope) + for item in items + ) + + self.last_match_count = len(results) + self.last_message = ( + f"JamBase lieferte {self.last_raw_count} Roh-Events fuer '{term}', " + f"davon {len(results)} passend nach Eventlens-Filtern. " + f"{self.last_api_call_count} API-Call." + ) + return results + + def search_events_bulk(self, watch_items: list[WatchItem]) -> dict[int, list[dict]]: + self._reset_status() + results_by_watch_item_id: dict[int, list[dict]] = { + watch_item.id: [] for watch_item in watch_items + } + if not self.is_configured(): + self.last_status = "ok" + self.last_message = "JamBase ist nicht konfiguriert. Bitte JAMBASE_API_KEY setzen." + return results_by_watch_item_id + + grouped_items: dict[tuple[WatchType, RegionScope], list[WatchItem]] = {} + for watch_item in watch_items: + grouped_items.setdefault( + (watch_item.watch_type, watch_item.region_scope), + [], + ).append(watch_item) + + for (watch_type, region_scope), grouped_watch_items in grouped_items.items(): + for batch in self._chunks(grouped_watch_items, settings.jambase_batch_size): + query_term = "|".join(watch_item.name for watch_item in batch) + items = self._fetch_items(query_term, watch_type, region_scope) + if items is None: + return results_by_watch_item_id + + for item in items: + for watch_item in batch: + event_data = self._event_data_for_term( + item, + watch_item.name, + watch_item.watch_type, + watch_item.region_scope, + ) + if event_data: + results_by_watch_item_id[watch_item.id].append(event_data) + + for watch_item_id, results in results_by_watch_item_id.items(): + results_by_watch_item_id[watch_item_id] = self._dedupe_results(results) + + self.last_match_count = sum(len(results) for results in results_by_watch_item_id.values()) + self.last_message = ( + f"JamBase Bulk-Sync: {self.last_api_call_count} API-Calls, " + f"{self.last_raw_count} Roh-Events, {self.last_match_count} passende Events " + f"fuer {len(watch_items)} Watchlist-Eintraege." + ) + return results_by_watch_item_id + + def _reset_status(self): + self.last_status = "ok" + self.last_message = "" + self.last_raw_count = 0 + self.last_match_count = 0 + self.last_api_call_count = 0 + + def _headers(self) -> dict: + return { "Authorization": f"Bearer {settings.jambase_api_key}", "Accept": "application/json", "User-Agent": settings.jambase_user_agent, } + def _fetch_items( + self, + term: str, + watch_type: WatchType, + region_scope: RegionScope, + ) -> list[dict] | None: payload = None - raw_count = 0 - self.last_raw_count = 0 - self.last_match_count = 0 tried_params: list[str] = [] bad_request_messages: list[str] = [] for params in self._build_param_candidates(term, watch_type, region_scope): tried_params.append(", ".join(sorted(params.keys())) or "keine") - response = requests.get(self.base_url, headers=headers, params=params, timeout=30) + self.last_api_call_count += 1 + response = requests.get(self.base_url, headers=self._headers(), params=params, timeout=30) if response.status_code == 401: self.last_status = "blocked" self.last_message = ( f"JamBase hat den API-Key fuer '{term}' nicht akzeptiert (401)." ) - return [] + return None if response.status_code == 403: self.last_status = "blocked" @@ -56,7 +130,7 @@ class JamBaseProvider: f"JamBase verweigert Zugriff fuer '{term}' (403). " "Pruefe Plan oder Berechtigungen." ) - return [] + return None if response.status_code == 400: bad_request_messages.append( @@ -73,10 +147,9 @@ class JamBaseProvider: f"JamBase lieferte keine gueltige JSON-Antwort fuer '{term}' " f"(Status {response.status_code})." ) - return [] + return None raw_items = self._extract_items(payload) - raw_count = len(raw_items) - self.last_raw_count = raw_count + self.last_raw_count += len(raw_items) if raw_items: break @@ -84,74 +157,82 @@ class JamBaseProvider: self.last_status = "error" detail = bad_request_messages[0] if bad_request_messages else "Keine Detailmeldung." self.last_message = f"JamBase akzeptierte die Anfrage fuer '{term}' nicht: {detail}" - return [] + return None - items = self._extract_items(payload) self.last_status = "ok" + return self._extract_items(payload) + def _event_data_for_term( + self, + item: dict, + term: str, + watch_type: WatchType, + region_scope: RegionScope, + ) -> dict | None: normalized_term = normalize_search_text(term) - results: list[dict] = [] + title = self._get_first(item, "name", "title", "eventName") or term + artists = self._extract_artist_names(item) + haystack = normalize_search_text(" ".join([title] + artists)) + if watch_type == WatchType.artist: + if normalized_term not in haystack: + return None + elif normalized_term not in normalize_search_text(title): + return None + + city = self._extract_city(item) + country_code = self._extract_country_code(item) + + if region_scope == RegionScope.hamburg and normalize_search_text(city) != "hamburg": + return None + if region_scope == RegionScope.germany and country_code not in { + "DE", + "DEU", + "GER", + "Germany", + "Deutschland", + }: + return None + + event_date = self._extract_event_date(item) + if event_date and event_date.date() < datetime.utcnow().date(): + return None + + external_id = str( + self._get_first(item, "id", "eventId", "identifier") + or item.get("slug") + or item.get("url") + or f"jambase:{normalize_search_text(title)}:{event_date.isoformat() if event_date else 'na'}" + ) + return { + "external_id": external_id, + "title": title, + "matched_term": term, + "venue_name": self._extract_venue_name(item), + "city": city, + "country_code": country_code, + "event_date": event_date, + "ticket_url": self._extract_ticket_url(item), + "image_url": self._extract_image_url(item), + "raw_payload": item, + } + + def _dedupe_results(self, results) -> list[dict]: + deduped: list[dict] = [] seen_ids: set[str] = set() - - for item in items: - title = self._get_first(item, "name", "title", "eventName") or term - artists = self._extract_artist_names(item) - haystack = normalize_search_text(" ".join([title] + artists)) - if watch_type == WatchType.artist: - if normalized_term not in haystack: - continue - elif normalized_term not in normalize_search_text(title): + for result in results: + if not result: continue - - city = self._extract_city(item) - country_code = self._extract_country_code(item) - - if region_scope == RegionScope.hamburg and normalize_search_text(city) != "hamburg": - continue - if region_scope == RegionScope.germany and country_code not in { - "DE", - "DEU", - "GER", - "Germany", - "Deutschland", - }: - continue - - event_date = self._extract_event_date(item) - if event_date and event_date.date() < datetime.utcnow().date(): - continue - - external_id = str( - self._get_first(item, "id", "eventId", "identifier") - or item.get("slug") - or item.get("url") - or f"jambase:{normalize_search_text(title)}:{event_date.isoformat() if event_date else 'na'}" - ) + external_id = result["external_id"] if external_id in seen_ids: continue seen_ids.add(external_id) + deduped.append(result) + return deduped - results.append( - { - "external_id": external_id, - "title": title, - "matched_term": term, - "venue_name": self._extract_venue_name(item), - "city": city, - "country_code": country_code, - "event_date": event_date, - "ticket_url": self._extract_ticket_url(item), - "image_url": self._extract_image_url(item), - "raw_payload": item, - } - ) - - self.last_match_count = len(results) - self.last_message = ( - f"JamBase lieferte {raw_count} Roh-Events fuer '{term}', " - f"davon {len(results)} passend nach Eventlens-Filtern." - ) - return results + def _chunks(self, items: list[WatchItem], size: int): + chunk_size = max(1, size) + for index in range(0, len(items), chunk_size): + yield items[index : index + chunk_size] def _build_param_candidates( self, diff --git a/backend/app/services.py b/backend/app/services.py index b80d429..4048bbd 100644 --- a/backend/app/services.py +++ b/backend/app/services.py @@ -137,6 +137,7 @@ def init_provider_sync_state(provider_name: str) -> dict: "error_terms": [], "event_count": 0, "raw_event_count": 0, + "api_call_count": 0, "last_message": "", } @@ -148,6 +149,7 @@ def record_provider_sync_state( message: str, event_count: int = 0, raw_event_count: int | None = None, + api_call_count: int | None = None, ): if PROVIDER_STATUS_PRIORITY[status] > PROVIDER_STATUS_PRIORITY[state["status"]]: state["status"] = status @@ -157,6 +159,8 @@ def record_provider_sync_state( state["event_count"] += event_count if raw_event_count is not None: state["raw_event_count"] += raw_event_count + if api_call_count is not None: + state["api_call_count"] += api_call_count elif status == ProviderStatusType.blocked: if term not in state["blocked_terms"]: state["blocked_terms"].append(term) @@ -188,6 +192,11 @@ def build_provider_status_message(state: dict) -> str: if state["raw_event_count"] else "" ) + + ( + f" {state['api_call_count']} API-Calls." + if state["api_call_count"] + else "" + ) ) @@ -394,6 +403,12 @@ def run_sync(db: Session, force_providers: bool = False) -> SyncResult: for provider in providers if provider.source_name not in skipped_provider_messages ] + bulk_providers = [ + provider for provider in providers_to_sync if hasattr(provider, "search_events_bulk") + ] + regular_providers = [ + provider for provider in providers_to_sync if provider not in bulk_providers + ] provider_states = { provider.source_name: init_provider_sync_state(provider.source_name) for provider in providers_to_sync @@ -481,7 +496,7 @@ def run_sync(db: Session, force_providers: bool = False) -> SyncResult: prune_stale_source_events(db, watch_item, source_name, seen_source_event_ids) db.commit() - for provider in providers_to_sync: + for provider in regular_providers: try: events = provider.search_events( term=watch_item.name, @@ -501,6 +516,7 @@ def run_sync(db: Session, force_providers: bool = False) -> SyncResult: message=provider_message, event_count=len(events), raw_event_count=getattr(provider, "last_raw_count", None), + api_call_count=getattr(provider, "last_api_call_count", None), ) except Exception as exc: provider_message = getattr( @@ -567,6 +583,101 @@ def run_sync(db: Session, force_providers: bool = False) -> SyncResult: notifications_skipped += 1 db.commit() + watch_items_by_id = {watch_item.id: watch_item for watch_item in active_items} + for provider in bulk_providers: + try: + events_by_watch_item_id = provider.search_events_bulk(active_items) + provider_status_name = getattr(provider, "last_status", "ok") + provider_message = getattr( + provider, + "last_message", + f"{provider.source_name} completed bulk sync.", + ) + state = provider_states[provider.source_name] + provider_status = ProviderStatusType(provider_status_name) + if PROVIDER_STATUS_PRIORITY[provider_status] > PROVIDER_STATUS_PRIORITY[state["status"]]: + state["status"] = provider_status + if provider_status == ProviderStatusType.ok: + state["ok_count"] += len(active_items) + state["event_count"] += sum( + len(events) for events in events_by_watch_item_id.values() + ) + state["raw_event_count"] += getattr(provider, "last_raw_count", 0) or 0 + state["api_call_count"] += getattr(provider, "last_api_call_count", 0) or 0 + elif provider_status == ProviderStatusType.blocked: + state["blocked_terms"].append("Bulk-Sync") + elif provider_status == ProviderStatusType.error: + state["error_terms"].append("Bulk-Sync") + state["last_message"] = provider_message + except Exception as exc: + provider_message = getattr( + provider, + "last_message", + f"Provider bulk sync failed for {provider.source_name}.", + ) + if not provider_message: + provider_message = ( + f"{provider.source_name} bulk sync failed: " + f"{type(exc).__name__}: {exc}" + ) + logger.exception( + "Provider bulk sync failed for provider=%s", + provider.source_name, + ) + db.rollback() + record_provider_sync_state( + state=provider_states[provider.source_name], + status=ProviderStatusType.error, + term="Bulk-Sync", + message=provider_message, + ) + continue + + for watch_item_id, events in events_by_watch_item_id.items(): + watch_item = watch_items_by_id.get(watch_item_id) + if watch_item is None: + continue + for event_data in events: + tracked_event, is_new = upsert_event( + db=db, + watch_item=watch_item, + provider_name=provider.source_name, + event_data=event_data, + ) + + if is_new: + new_events += 1 + else: + updated_events += 1 + + should_notify = ( + is_new + and tracked_event.discovery_notified_at is None + and not has_equivalent_existing_event(db, tracked_event) + ) + if should_notify: + status = send_email_notification( + db=db, + tracked_event=tracked_event, + notification_type=NotificationType.discovery, + subject=f"Neuer Termin fuer {watch_item.name}", + body=( + f"Es wurde ein neuer Termin fuer '{watch_item.name}' gefunden.\n\n" + f"Quelle: {tracked_event.source}\n" + f"Titel: {tracked_event.title}\n" + f"Ort: {tracked_event.venue_name or 'unbekannt'}\n" + f"Stadt: {tracked_event.city or 'unbekannt'}\n" + f"Datum: {tracked_event.event_date or 'unbekannt'}\n" + f"Tickets: {tracked_event.ticket_url or 'keine URL'}\n" + ), + ) + if status == NotificationStatus.sent: + tracked_event.discovery_notified_at = datetime.utcnow() + notifications_sent += 1 + else: + notifications_skipped += 1 + db.commit() + for provider_name, state in provider_states.items(): update_provider_status( db=db,