Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
254 changes: 254 additions & 0 deletions integrations/aew.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,254 @@
from typing import List
from datetime import datetime, timedelta
from zoneinfo import ZoneInfo
import re
import logging

import requests
from bs4 import BeautifulSoup
from fastapi import HTTPException

from base import CalendarBase, Event, IntegrationBase

logger = logging.getLogger(__name__)


AEW_EVENTS_URL = "https://www.allelitewrestling.com/aew-events"

# Timezone mappings for AEW events
TIMEZONE_MAP = {
"PT": "America/Los_Angeles",
"MT": "America/Denver",
"CT": "America/Chicago",
"ET": "America/New_York",
"EST": "America/New_York",
"AEDT": "Australia/Sydney",
"AEST": "Australia/Sydney",
"GMT": "Europe/London",
"BST": "Europe/London",
}

# Month pattern for date matching
MONTHS = ["JANUARY", "FEBRUARY", "MARCH", "APRIL", "MAY", "JUNE",
"JULY", "AUGUST", "SEPTEMBER", "OCTOBER", "NOVEMBER", "DECEMBER"]


def parse_aew_datetime(date_str: str, time_str: str) -> datetime | None:
"""
Parse AEW date and time strings into a UTC datetime.

Args:
date_str: e.g., "FEBRUARY 11, 2026"
time_str: e.g., "4:30pm PT" or "TBA"

Returns:
datetime in UTC, or None if parsing fails
"""
try:
# Parse the date
date_str = date_str.strip().upper()
date = datetime.strptime(date_str, "%B %d, %Y")

# Handle TBA times - default to 8 PM ET
if not time_str or time_str.upper() == "TBA":
time_str = "8:00pm ET"

time_str = time_str.strip()

# Extract timezone abbreviation from end of time string
parts = time_str.split()
if len(parts) >= 2:
tz_abbr = parts[-1].upper()
time_part = " ".join(parts[:-1])
else:
tz_abbr = "ET"
time_part = time_str

# Parse time (e.g., "4:30pm" or "8:00pm")
time_part = time_part.lower().replace(".", "")
if ":" in time_part:
time_obj = datetime.strptime(time_part, "%I:%M%p")
else:
time_obj = datetime.strptime(time_part, "%I%p")

# Combine date and time
local_dt = datetime(
date.year, date.month, date.day,
time_obj.hour, time_obj.minute, 0
)

# Convert to UTC
tz_name = TIMEZONE_MAP.get(tz_abbr, "America/New_York")
local_tz = ZoneInfo(tz_name)
local_dt = local_dt.replace(tzinfo=local_tz)
utc_dt = local_dt.astimezone(ZoneInfo("UTC")).replace(tzinfo=None)

return utc_dt

except Exception:
return None


def get_aew_events() -> List[dict]:
"""
Fetch all events from AEW.com events page.

The AEW website is Wix-based and JavaScript-rendered.
We fetch the page and parse the text content which follows a pattern:

DATE
EVENT NAME
City
EVENT DETAILS >
LOCATION
Venue
VENUE NAME
Time
TIME VALUE
BUY TICKETS
"""
try:
response = requests.get(AEW_EVENTS_URL, timeout=30, headers={
"User-Agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36"
})
if response.status_code != 200:
return []

soup = BeautifulSoup(response.text, "html.parser")

# Get all text content
text = soup.get_text(separator="\n")
lines = [line.strip() for line in text.split("\n") if line.strip()]

events = []
seen_uids = set()

i = 0
while i < len(lines):
line = lines[i]

# Look for date pattern: MONTH DAY, YEAR
date_match = re.match(r"^(" + "|".join(MONTHS) + r")\s+\d{1,2},\s+\d{4}$", line.upper())

if date_match:
date_str = line
event_name = None
city = None
venue = None
time_str = None

# Look ahead for event details (up to 15 lines or next date)
j = i + 1
while j < min(i + 15, len(lines)):
next_line = lines[j]

# Stop if we hit another date
if re.match(r"^(" + "|".join(MONTHS) + r")\s+\d{1,2},\s+\d{4}$", next_line.upper()):
break

# Skip navigation/label lines
if next_line in ["City", "Venue", "Time", "EVENT DETAILS >", "BUY TICKETS",
"ON SALE: TBA", "Empty heading", "UPCOMING EVENTS"]:
j += 1
continue

# Event name contains "AEW"
if not event_name and "AEW" in next_line:
event_name = next_line

# City/location pattern: "City, STATE" or "City, Country"
elif not city and re.match(r"^[A-Za-z\s]+,\s*[A-Za-z\s]+$", next_line) and len(next_line) < 50:
city = next_line

# Venue names typically contain these words
elif not venue and any(v in next_line for v in
["Arena", "Center", "Centre", "Stadium", "Auditorium", "Ballroom",
"Civic", "House", "Entertainment", "Bank"]):
venue = next_line

# Time pattern: "H:MMpm TZ" or "H:MMam TZ"
elif not time_str and re.match(r"^\d{1,2}:\d{2}[ap]m\s+[A-Z]+$", next_line, re.I):
time_str = next_line

j += 1

# Create event if we have minimum data
if event_name and date_str:
# Generate UID from event name and date
slug = re.sub(r"[^a-z0-9]+", "-", event_name.lower()).strip("-")
date_slug = re.sub(r"[^a-z0-9]+", "-", date_str.lower()).strip("-")
uid = f"aew-{slug}-{date_slug}"

if uid not in seen_uids:
seen_uids.add(uid)

# Parse datetime
start = parse_aew_datetime(date_str, time_str or "TBA")
if start:
location = ""
if venue and city:
location = f"{venue}, {city}"
elif city:
location = city
elif venue:
location = venue

events.append({
"uid": uid,
"title": event_name,
"start": start,
"location": location,
"description": "",
})

i = j # Skip to where we left off
else:
i += 1

return events

except Exception as e:
logger.warning(f"AEW scraper error: {e}")
return []


class AewCalendar(CalendarBase):
def fetch_events(self) -> List[Event]:
"""Fetch AEW events from allelitewrestling.com."""
try:
event_data = get_aew_events()
events: List[Event] = []

for data in event_data:
# AEW shows typically last 2-3 hours, PPVs 4+ hours
is_ppv = any(x in data["title"].lower() for x in
["revolution", "dynasty", "double or nothing", "forbidden door",
"all in", "all out", "full gear", "worlds end", "grand slam"])
duration = timedelta(hours=4) if is_ppv else timedelta(hours=3)

events.append(
Event(
uid=data["uid"],
title=data["title"],
start=data["start"],
end=data["start"] + duration,
all_day=False,
description=data["description"],
location=data["location"],
)
)

# Sort by start time
events.sort(key=lambda e: e.start)
self.events = events
return events

except HTTPException:
raise
except Exception as e:
raise HTTPException(status_code=500, detail=str(e)) from e


class AewIntegration(IntegrationBase):
def fetch_calendars(self, *args, **kwargs):
return None
72 changes: 59 additions & 13 deletions integrations/ufc.py
Original file line number Diff line number Diff line change
Expand Up @@ -120,20 +120,66 @@ def get_event_details(event_url: str) -> dict | None:
location = " ".join(venue_el.get_text().split())
break

# Build description with fight card
# Build description with fight card and broadcast info
description_parts = []

# Main card fights
main_card = soup.select_one("#main-card")
if main_card:
fights = main_card.select(".c-listing-fight")
if fights:
description_parts.append("Main Card:")
for fight in fights[:6]: # Limit to 6 fights
red_corner = fight.select_one(".c-listing-fight__corner--red .c-listing-fight__corner-name")
blue_corner = fight.select_one(".c-listing-fight__corner--blue .c-listing-fight__corner-name")
if red_corner and blue_corner:
description_parts.append(f" {red_corner.get_text(strip=True)} vs {blue_corner.get_text(strip=True)}")
# Find fight card sections (Main Card, Prelims, etc.)
broadcaster_containers = soup.select(".c-event-fight-card-broadcaster__container")

for container in broadcaster_containers:
# Get card title (Main Card, Prelims, etc.)
card_title_el = container.select_one(".c-event-fight-card-broadcaster__card-title strong")
card_title = card_title_el.get_text(strip=True) if card_title_el else "Fight Card"

# Get broadcaster/streaming info
broadcaster_link = container.select_one(".c-event-fight-card-broadcaster__link a")
broadcaster = ""
if broadcaster_link:
broadcaster = broadcaster_link.get_text(strip=True)

# Get the fights for this card section
# The fights follow the broadcaster container in the next <section>
next_section = container.find_next_sibling("section", class_="l-listing--stacked--full-width")
if not next_section:
# Try finding it as next element
next_section = container.find_next("section", class_="l-listing--stacked--full-width")

fights_text = []
if next_section:
fights = next_section.select(".c-listing-fight")
for fight in fights[:5]: # Limit to 5 fights per card
red_name = fight.select_one(".c-listing-fight__corner-name--red")
blue_name = fight.select_one(".c-listing-fight__corner-name--blue")
weight_class = fight.select_one(".c-listing-fight__class-text")

if red_name and blue_name:
# Extract first and last names with proper spacing
red_given = red_name.select_one(".c-listing-fight__corner-given-name")
red_family = red_name.select_one(".c-listing-fight__corner-family-name")
blue_given = blue_name.select_one(".c-listing-fight__corner-given-name")
blue_family = blue_name.select_one(".c-listing-fight__corner-family-name")

if red_given and red_family and blue_given and blue_family:
red = f"{red_given.get_text(strip=True)} {red_family.get_text(strip=True)}"
blue = f"{blue_given.get_text(strip=True)} {blue_family.get_text(strip=True)}"
else:
# Fallback to full text with space separator
red = " ".join(red_name.get_text().split())
blue = " ".join(blue_name.get_text().split())

fight_str = f"• {red} vs {blue}"
if weight_class:
weight = weight_class.get_text(strip=True).replace(" Bout", "")
fight_str += f" ({weight})"
fights_text.append(fight_str)

if fights_text:
header = card_title
if broadcaster:
header += f" - {broadcaster}"
description_parts.append(header)
description_parts.extend(fights_text)
description_parts.append("") # Empty line between sections

# Generate a unique ID from the URL
event_slug = event_url.split("/event/")[-1].split("?")[0]
Expand All @@ -144,7 +190,7 @@ def get_event_details(event_url: str) -> dict | None:
"title": title,
"start": start_time,
"location": location,
"description": "\n".join(description_parts) if description_parts else "",
"description": "\n".join(description_parts).strip() if description_parts else "",
}

except Exception:
Expand Down
Loading