ecki
635 lines
22 KiB
Python
635 lines
22 KiB
Python
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,
|
|
"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,
|
|
):
|
|
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
|
|
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 ""
|
|
)
|
|
)
|
|
|
|
|
|
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
|
|
]
|
|
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 providers_to_sync:
|
|
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),
|
|
)
|
|
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()
|
|
|
|
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
|