from datetime import datetime, timedelta import logging import re from sqlalchemy import desc, select from sqlalchemy.orm import Session from app.models import ( NotificationStatus, NotificationType, ProviderStatus, ProviderStatusType, SourceStatusType, TrackedEvent, WatchItem, WatchSource, ) from app.notifications import send_email_notification from app.providers.registry import get_providers from app.schemas import SyncResult from app.source_scanner import SourceScanner from app.config import settings logger = logging.getLogger(__name__) PROVIDER_PRIORITY = { "ticketmaster": 4, "jambase": 3, "eventim": 2, "bandsintown": 1, } PROVIDER_STATUS_PRIORITY = { ProviderStatusType.error: 3, ProviderStatusType.blocked: 2, ProviderStatusType.ok: 1, } def get_provider_sync_interval_hours(provider_name: str) -> int: if provider_name == "jambase": return settings.jambase_sync_interval_hours return 0 def list_watch_items(db: Session) -> list[WatchItem]: return list(db.scalars(select(WatchItem).order_by(WatchItem.name))) def list_events(db: Session) -> list[TrackedEvent]: events = list(db.scalars(select(TrackedEvent))) deduped: list[TrackedEvent] = [] for event in events: duplicate_index = next( ( index for index, existing in enumerate(deduped) if events_are_equivalent(existing, event) ), None, ) if duplicate_index is None: deduped.append(event) continue if is_preferred_event(event, deduped[duplicate_index]): deduped[duplicate_index] = event return sorted(deduped, key=event_sort_key) def event_sort_key(event: TrackedEvent): today = datetime.utcnow().date() if event.event_date is None: return (1, datetime.max) if event.event_date.date() >= today: return (0, event.event_date) return (2, datetime.max - (event.event_date - datetime.min)) def list_notifications(db: Session): from app.models import NotificationLog return list(db.scalars(select(NotificationLog).order_by(desc(NotificationLog.created_at)))) def list_provider_statuses(db: Session) -> list[ProviderStatus]: return list(db.scalars(select(ProviderStatus).order_by(ProviderStatus.provider_name))) def list_watch_sources(db: Session, watch_item_id: int | None = None) -> list[WatchSource]: stmt = select(WatchSource).order_by(WatchSource.created_at) if watch_item_id is not None: stmt = stmt.where(WatchSource.watch_item_id == watch_item_id) return list(db.scalars(stmt)) def update_provider_status( db: Session, provider_name: str, status: ProviderStatusType, message: str, mark_success: bool = True, ): provider_status = db.scalar( select(ProviderStatus).where(ProviderStatus.provider_name == provider_name) ) now = datetime.utcnow() if provider_status is None: provider_status = ProviderStatus( provider_name=provider_name, status=status, message=message, last_checked_at=now, last_success_at=now if status == ProviderStatusType.ok else None, ) db.add(provider_status) return provider_status.status = status provider_status.message = message provider_status.last_checked_at = now if status == ProviderStatusType.ok and mark_success: provider_status.last_success_at = now def init_provider_sync_state(provider_name: str) -> dict: return { "provider_name": provider_name, "status": ProviderStatusType.ok, "ok_count": 0, "blocked_terms": [], "error_terms": [], "event_count": 0, "raw_event_count": 0, "api_call_count": 0, "last_message": "", } def record_provider_sync_state( state: dict, status: ProviderStatusType, term: str, 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 if status == ProviderStatusType.ok: state["ok_count"] += 1 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) elif status == ProviderStatusType.error: if term not in state["error_terms"]: state["error_terms"].append(term) state["last_message"] = message def build_provider_status_message(state: dict) -> str: if state["status"] == ProviderStatusType.error: if state["last_message"]: return state["last_message"] terms = ", ".join(state["error_terms"][:3]) return f"Fehler bei {state['provider_name']} fuer: {terms}." if state["status"] == ProviderStatusType.blocked: if state["last_message"]: return state["last_message"] terms = ", ".join(state["blocked_terms"][:3]) return f"{state['provider_name']} wurde geblockt fuer: {terms}." return ( f"{state['provider_name']} erfolgreich fuer {state['ok_count']} " f"Watchlist-Eintraege geprueft, {state['event_count']} passende Events." + ( f" {state['raw_event_count']} Roh-Events ausgewertet." if state["raw_event_count"] else "" ) + ( f" {state['api_call_count']} API-Calls." if state["api_call_count"] else "" ) ) def normalize_event_text(value: str | None) -> str: if not value: return "" cleaned = re.sub(r"[^a-z0-9]+", " ", value.casefold()) return " ".join(cleaned.split()) def titles_match(left: str | None, right: str | None) -> bool: normalized_left = normalize_event_text(left) normalized_right = normalize_event_text(right) if not normalized_left or not normalized_right: return False return ( normalized_left == normalized_right or normalized_left in normalized_right or normalized_right in normalized_left ) def event_has_placeholder_time(event: TrackedEvent) -> bool: if event.event_date is None: return True return ( bool(event.source and event.source.startswith("source:")) and event.event_date.minute == 0 and event.event_date.hour in (0, 12) ) def event_has_specific_time(event: TrackedEvent) -> bool: return event.event_date is not None and not event_has_placeholder_time(event) def get_event_date_key(value: datetime | None): return value.date() if value else None def events_are_equivalent(left: TrackedEvent, right: TrackedEvent) -> bool: if left.watch_item_id != right.watch_item_id: return False if get_event_date_key(left.event_date) != get_event_date_key(right.event_date): return False left_city = normalize_event_text(left.city) right_city = normalize_event_text(right.city) if left_city and right_city and left_city != right_city: return False title_matches = titles_match(left.title, right.title) venue_matches = titles_match(left.venue_name, right.venue_name) same_matched_term = ( normalize_event_text(left.matched_term) and normalize_event_text(left.matched_term) == normalize_event_text(right.matched_term) ) one_has_placeholder_time = event_has_placeholder_time(left) != event_has_placeholder_time(right) return title_matches or venue_matches or (same_matched_term and one_has_placeholder_time) def is_preferred_event(candidate: TrackedEvent, current: TrackedEvent) -> bool: candidate_score = ( 1 if candidate.is_ticket_purchased else 0, 1 if event_has_specific_time(candidate) else 0, get_provider_priority(candidate.source), 1 if candidate.ticket_url else 0, candidate.last_seen_at or datetime.min, ) current_score = ( 1 if current.is_ticket_purchased else 0, 1 if event_has_specific_time(current) else 0, get_provider_priority(current.source), 1 if current.ticket_url else 0, current.last_seen_at or datetime.min, ) return candidate_score > current_score def get_provider_priority(source: str | None) -> int: if source and source.startswith("source:"): return 5 return PROVIDER_PRIORITY.get(source or "", 0) def has_equivalent_existing_event(db: Session, tracked_event: TrackedEvent) -> bool: stmt = select(TrackedEvent).where( TrackedEvent.watch_item_id == tracked_event.watch_item_id, TrackedEvent.id != tracked_event.id, ) existing_events = list(db.scalars(stmt)) return any(events_are_equivalent(tracked_event, existing) for existing in existing_events) def upsert_event( db: Session, watch_item: WatchItem, provider_name: str, event_data: dict, ) -> tuple[TrackedEvent, bool]: stmt = select(TrackedEvent).where( TrackedEvent.watch_item_id == watch_item.id, TrackedEvent.source == provider_name, TrackedEvent.external_id == event_data["external_id"], ) tracked_event = db.scalar(stmt) is_new = tracked_event is None if tracked_event is None: tracked_event = TrackedEvent( watch_item=watch_item, source=provider_name, external_id=event_data["external_id"], title=event_data["title"], matched_term=event_data["matched_term"], venue_name=event_data.get("venue_name"), city=event_data.get("city"), country_code=event_data.get("country_code"), event_date=event_data.get("event_date"), ticket_url=event_data.get("ticket_url"), image_url=event_data.get("image_url"), raw_payload=event_data.get("raw_payload"), ) db.add(tracked_event) db.flush() else: tracked_event.title = event_data["title"] tracked_event.matched_term = event_data["matched_term"] tracked_event.venue_name = event_data.get("venue_name") tracked_event.city = event_data.get("city") tracked_event.country_code = event_data.get("country_code") tracked_event.event_date = event_data.get("event_date") tracked_event.ticket_url = event_data.get("ticket_url") tracked_event.image_url = event_data.get("image_url") tracked_event.raw_payload = event_data.get("raw_payload") tracked_event.last_seen_at = datetime.utcnow() return tracked_event, is_new def prune_stale_source_events( db: Session, watch_item: WatchItem, source_name: str, seen_external_ids: set[str], ) -> int: stmt = select(TrackedEvent).where( TrackedEvent.watch_item_id == watch_item.id, TrackedEvent.source == source_name, TrackedEvent.is_ticket_purchased.is_(False), ) stale_events = [ event for event in db.scalars(stmt) if event.external_id not in seen_external_ids ] for event in stale_events: db.delete(event) return len(stale_events) def get_provider_skip_message(db: Session, provider_name: str, force_providers: bool) -> str | None: if force_providers: return None interval_hours = get_provider_sync_interval_hours(provider_name) if interval_hours <= 0: return None provider_status = db.scalar( select(ProviderStatus).where(ProviderStatus.provider_name == provider_name) ) if provider_status is None or provider_status.last_success_at is None: return None next_allowed_at = provider_status.last_success_at + timedelta(hours=interval_hours) if datetime.utcnow() >= next_allowed_at: return None return ( f"{provider_name} uebersprungen: letzter erfolgreicher API-Lauf ist weniger " f"als {interval_hours} Stunden her. Fuer sofortigen Lauf /sync?force_providers=true nutzen." ) def run_sync(db: Session, force_providers: bool = False) -> SyncResult: providers = get_providers() source_scanner = SourceScanner() skipped_provider_messages = { provider.source_name: skip_message for provider in providers if ( skip_message := get_provider_skip_message( db=db, provider_name=provider.source_name, force_providers=force_providers, ) ) } providers_to_sync = [ provider 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 } active_items = list( db.scalars(select(WatchItem).where(WatchItem.is_active.is_(True)).order_by(WatchItem.name)) ) new_events = 0 updated_events = 0 notifications_sent = 0 notifications_skipped = 0 for watch_item in active_items: active_sources = [source for source in watch_item.sources if source.is_active] for source in active_sources: source_name = f"source:{source.id}" seen_source_event_ids: set[str] = set() try: events = source_scanner.scan(watch_item, source) source.last_status = ( SourceStatusType.ok if events else SourceStatusType.no_match ) source.last_message = source_scanner.last_message or ( f"{len(events)} passende Events gefunden." if events else "Keine passenden Events auf dieser Quelle gefunden." ) source.last_checked_at = datetime.utcnow() except Exception as exc: logger.exception( "Source scan failed for watch_item=%s source=%s", watch_item.name, source.url, ) db.rollback() source.last_status = SourceStatusType.error source.last_message = f"Scan fehlgeschlagen: {exc}" source.last_checked_at = datetime.utcnow() db.add(source) db.commit() continue for event_data in events: tracked_event, is_new = upsert_event( db=db, watch_item=watch_item, provider_name=source_name, event_data=event_data, ) seen_source_event_ids.add(tracked_event.external_id) 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: {source.label or source.url}\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.add(source) prune_stale_source_events(db, watch_item, source_name, seen_source_event_ids) db.commit() for provider in regular_providers: try: events = provider.search_events( term=watch_item.name, watch_type=watch_item.watch_type, region_scope=watch_item.region_scope, ) provider_status_name = getattr(provider, "last_status", "ok") provider_message = getattr( provider, "last_message", f"{provider.source_name} completed successfully for '{watch_item.name}'.", ) record_provider_sync_state( state=provider_states[provider.source_name], status=ProviderStatusType(provider_status_name), term=watch_item.name, 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( provider, "last_message", f"Provider sync failed for '{watch_item.name}'.", ) if not provider_message: provider_message = ( f"{provider.source_name} failed for '{watch_item.name}': " f"{type(exc).__name__}: {exc}" ) logger.exception( "Provider sync failed for provider=%s watch_item=%s", provider.source_name, watch_item.name, ) db.rollback() record_provider_sync_state( state=provider_states[provider.source_name], status=ProviderStatusType.error, term=watch_item.name, message=provider_message, ) 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() 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, provider_name=provider_name, status=state["status"], message=build_provider_status_message(state), ) for provider_name, message in skipped_provider_messages.items(): update_provider_status( db=db, provider_name=provider_name, status=ProviderStatusType.ok, message=message, mark_success=False, ) db.commit() return SyncResult( scanned_watch_items=len(active_items), new_events=new_events, updated_events=updated_events, notifications_sent=notifications_sent, notifications_skipped=notifications_skipped, ) def send_reminders(db: Session) -> int: now = datetime.utcnow() start = now + timedelta(days=6) end = now + timedelta(days=7, hours=12) stmt = select(TrackedEvent).where( TrackedEvent.is_ticket_purchased.is_(True), TrackedEvent.event_date.is_not(None), TrackedEvent.reminder_notified_at.is_(None), ) candidates = list(db.scalars(stmt)) sent_count = 0 for tracked_event in candidates: if tracked_event.event_date is None: continue if not (start <= tracked_event.event_date <= end): continue status = send_email_notification( db=db, tracked_event=tracked_event, notification_type=NotificationType.reminder, subject=f"Erinnerung: {tracked_event.title} in einer Woche", body=( f"Du hast Karten fuer '{tracked_event.title}' markiert.\n\n" f"Veranstaltung: {tracked_event.title}\n" f"Datum: {tracked_event.event_date}\n" f"Venue: {tracked_event.venue_name or 'unbekannt'}\n" f"Stadt: {tracked_event.city or 'unbekannt'}\n" f"Tickets: {tracked_event.ticket_url or 'keine URL'}\n" ), ) if status == NotificationStatus.sent: tracked_event.reminder_notified_at = datetime.utcnow() sent_count += 1 db.commit() return sent_count