From bccee9ebc1b36a140103bf3b715e91b8cb1b7def Mon Sep 17 00:00:00 2001 From: SergeantPanda Date: Tue, 23 Jun 2026 11:50:52 -0500 Subject: [PATCH] refactor(epg): extract EPG generation logic into dedicated module - Moved all XMLTV output logic from `apps/output/views.py` to `apps/output/epg.py`, streamlining the codebase and maintaining the existing HTTP endpoint functionality. - Updated related tests and references to ensure proper integration with the new module structure. --- CHANGELOG.md | 2 + apps/output/epg.py | 1745 +++++++++++++++++ ...hunk_cache.py => streaming_chunk_cache.py} | 36 +- ...cache.py => test_streaming_chunk_cache.py} | 30 +- apps/output/tests.py | 11 +- apps/output/views.py | 1714 +--------------- 6 files changed, 1790 insertions(+), 1748 deletions(-) create mode 100644 apps/output/epg.py rename apps/output/{epg_chunk_cache.py => streaming_chunk_cache.py} (81%) rename apps/output/{test_epg_chunk_cache.py => test_streaming_chunk_cache.py} (84%) diff --git a/CHANGELOG.md b/CHANGELOG.md index 3315456b..840cb349 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -15,6 +15,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Changed +- **EPG generation extracted into `apps/output/epg.py`.** All XMLTV output logic (`generate_epg`, `generate_dummy_programs`, `generate_custom_dummy_programs`, `generate_dummy_epg`, and supporting helpers) moved from `apps/output/views.py` into a dedicated module. `views.py` retains the thin HTTP endpoint wrappers and auth checks; `epg.py` handles all content generation. No behavior change. + - **Channel list with nested streams loads faster.** `GET /api/channels/channels/?include_streams=true` (Channels UI and single-channel fetch) now builds nested stream payloads from the prefetched `channelstream_set` instead of issuing one extra streams M2M query per channel. ### Performance diff --git a/apps/output/epg.py b/apps/output/epg.py new file mode 100644 index 00000000..354d6e5b --- /dev/null +++ b/apps/output/epg.py @@ -0,0 +1,1745 @@ +"""XMLTV (EPG) output generation. + +Consolidates the EPG export logic that backs the `/epg` endpoint and the XC +XMLTV endpoint: real programme streaming, dummy/custom dummy program +generation, and the streaming XMLTV builder. HTTP endpoints live in views.py +and call into this module; Redis chunk caching lives in streaming_chunk_cache.py. +""" + +import html +import logging +from datetime import datetime, timedelta + +import regex + +from django.db.models import Prefetch +from django.http import Http404 +from django.urls import reverse +from django.utils import timezone as django_timezone + +from apps.channels.models import Channel, ChannelProfile, Stream +from apps.channels.utils import format_channel_number +from apps.epg.models import ProgramData +from apps.output.streaming_chunk_cache import stream_cached_response +from core.utils import build_absolute_uri_with_port, log_system_event + +logger = logging.getLogger(__name__) + +_EPG_CHANNEL_XML_BATCH_SIZE = 200 +_EPG_PROGRAM_YIELD_BATCH_SIZE = 1000 +_EPG_PROGRAM_DB_CHUNK_SIZE = 20000 + + +def _programme_overlaps_export_window(start_time, end_time, lookback_cutoff, cutoff_date): + if end_time < lookback_cutoff: + return False + if cutoff_date is not None and start_time >= cutoff_date: + return False + return True + + +def _ceil_to_half_hour(dt): + """Round a datetime up to the next :00 or :30 boundary.""" + dt = dt.replace(second=0, microsecond=0) + remainder = dt.minute % 30 + if remainder == 0: + return dt + return dt + timedelta(minutes=30 - remainder) + + +def _epg_export_teardown(): + from django.db import close_old_connections + + from core.utils import ( + _is_gevent_monkey_patched, + cleanup_memory, + trim_c_allocator_heap, + ) + + close_old_connections() + + def _run(): + cleanup_memory(force_collection=True) + trim_c_allocator_heap() + + if _is_gevent_monkey_patched(): + import gevent + + gevent.spawn(_run) + else: + _run() + + +def _ordered_channel_streams(channel): + """Return a channel's streams ordered by channelstream join order.""" + prefetched = getattr(channel, '_prefetched_objects_cache', {}).get('streams') + if prefetched is not None: + return list(prefetched) + return list(channel.streams.all().order_by('channelstream__order')) + + +def _pattern_match_name_from_custom_props(channel, effective_name, custom_props): + """Name used for custom dummy EPG regex matching (channel or stream title). + + Returns (name, stream_lookup_failed). stream_lookup_failed is True only when + name_source is 'stream' but the configured index is missing or out of range. + """ + if custom_props.get('name_source') != 'stream': + return effective_name, False + stream_index = custom_props.get('stream_index', 1) - 1 + streams = _ordered_channel_streams(channel) + if 0 <= stream_index < len(streams): + return streams[stream_index].name, False + return effective_name, True + + +def generate_fallback_programs(channel_id, channel_name, now, num_days, program_length_hours, fallback_title, fallback_description): + """ + Generate dummy programs using custom fallback templates when patterns don't match. + + Args: + channel_id: Channel ID for the programs + channel_name: Channel name to use as fallback in templates + now: Current datetime (in UTC) + num_days: Number of days to generate programs for + program_length_hours: Length of each program in hours + fallback_title: Custom fallback title template (empty string if not provided) + fallback_description: Custom fallback description template (empty string if not provided) + + Returns: + List of program dictionaries + """ + programs = [] + + # Use custom fallback title or channel name as default + title = fallback_title if fallback_title else channel_name + + # Use custom fallback description or a simple default message + if fallback_description: + description = fallback_description + else: + description = f"EPG information is currently unavailable for {channel_name}" + + # Create programs for each day + for day in range(num_days): + day_start = now + timedelta(days=day) + + # Create programs with specified length throughout the day + for hour_offset in range(0, 24, program_length_hours): + # Calculate program start and end times + start_time = day_start + timedelta(hours=hour_offset) + end_time = start_time + timedelta(hours=program_length_hours) + + programs.append({ + "channel_id": channel_id, + "start_time": start_time, + "end_time": end_time, + "title": title, + "description": description, + }) + + return programs + + +def generate_dummy_programs( + channel_id, + channel_name, + num_days=1, + program_length_hours=4, + epg_source=None, + export_lookback=None, + export_cutoff=None, +): + """ + Generate dummy EPG programs for channels. + + If epg_source is provided and it's a custom dummy EPG with patterns, + use those patterns to generate programs from the channel title. + Otherwise, generate default dummy programs. + + Args: + channel_id: Channel ID for the programs + channel_name: Channel title/name + num_days: Number of days to generate programs for + program_length_hours: Length of each program in hours + epg_source: Optional EPGSource for custom dummy EPG with patterns + + Returns: + List of program dictionaries + """ + # Get current time rounded to hour + now = django_timezone.now() + now = now.replace(minute=0, second=0, microsecond=0) + + # Check if this is a custom dummy EPG with regex patterns + if epg_source and epg_source.source_type == 'dummy' and epg_source.custom_properties: + custom_programs = generate_custom_dummy_programs( + channel_id, channel_name, now, num_days, + epg_source.custom_properties, + export_lookback=export_lookback, + export_cutoff=export_cutoff, + ) + if custom_programs is not None: + return custom_programs + + logger.info(f"Custom pattern didn't match for '{channel_name}', checking for custom fallback templates") + + custom_props = epg_source.custom_properties + fallback_title = custom_props.get('fallback_title_template', '').strip() + fallback_description = custom_props.get('fallback_description_template', '').strip() + + if fallback_title or fallback_description: + logger.info(f"Using custom fallback templates for '{channel_name}'") + return generate_fallback_programs( + channel_id, channel_name, now, num_days, + program_length_hours, fallback_title, fallback_description + ) + logger.info(f"No custom fallback templates found, using default dummy EPG") + + # Default humorous program descriptions based on time of day + time_descriptions = { + (0, 4): [ + f"Late Night with {channel_name} - Where insomniacs unite!", + f"The 'Why Am I Still Awake?' Show on {channel_name}", + f"Counting Sheep - A {channel_name} production for the sleepless", + ], + (4, 8): [ + f"Dawn Patrol - Rise and shine with {channel_name}!", + f"Early Bird Special - Coffee not included", + f"Morning Zombies - Before coffee viewing on {channel_name}", + ], + (8, 12): [ + f"Mid-Morning Meetings - Pretend you're paying attention while watching {channel_name}", + f"The 'I Should Be Working' Hour on {channel_name}", + f"Productivity Killer - {channel_name}'s daytime programming", + ], + (12, 16): [ + f"Lunchtime Laziness with {channel_name}", + f"The Afternoon Slump - Brought to you by {channel_name}", + f"Post-Lunch Food Coma Theater on {channel_name}", + ], + (16, 20): [ + f"Rush Hour - {channel_name}'s alternative to traffic", + f"The 'What's For Dinner?' Debate on {channel_name}", + f"Evening Escapism - {channel_name}'s remedy for reality", + ], + (20, 24): [ + f"Prime Time Placeholder - {channel_name}'s finest not-programming", + f"The 'Netflix Was Too Complicated' Show on {channel_name}", + f"Family Argument Avoider - Courtesy of {channel_name}", + ], + } + + programs = [] + + # Create programs for each day + for day in range(num_days): + day_start = now + timedelta(days=day) + + # Create programs with specified length throughout the day + for hour_offset in range(0, 24, program_length_hours): + # Calculate program start and end times + start_time = day_start + timedelta(hours=hour_offset) + end_time = start_time + timedelta(hours=program_length_hours) + + # Get the hour for selecting a description + hour = start_time.hour + + # Find the appropriate time slot for description + for time_range, descriptions in time_descriptions.items(): + start_range, end_range = time_range + if start_range <= hour < end_range: + # Pick a description using the sum of the hour and day as seed + # This makes it somewhat random but consistent for the same timeslot + description = descriptions[(hour + day) % len(descriptions)] + break + else: + # Fallback description if somehow no range matches + description = f"Placeholder program for {channel_name} - EPG data went on vacation" + + programs.append({ + "channel_id": channel_id, + "start_time": start_time, + "end_time": end_time, + "title": channel_name, + "description": description, + }) + + return programs + + +def generate_custom_dummy_programs( + channel_id, + channel_name, + now, + num_days, + custom_properties, + export_lookback=None, + export_cutoff=None, +): + """ + Generate programs using custom dummy EPG regex patterns. + + Extracts information from channel title using regex patterns and generates + programs based on the extracted data. + + TIMEZONE HANDLING: + ------------------ + The timezone parameter specifies the timezone of the event times in your channel + titles using standard timezone names (e.g., 'US/Eastern', 'US/Pacific', 'Europe/London'). + DST (Daylight Saving Time) is handled automatically by pytz. + + Examples: + - Channel: "NHL 01: Bruins VS Maple Leafs @ 8:00PM ET" + - Set timezone = "US/Eastern" + - In October (DST): 8:00PM EDT → 12:00AM UTC (automatically uses UTC-4) + - In January (no DST): 8:00PM EST → 1:00AM UTC (automatically uses UTC-5) + + Args: + channel_id: Channel ID for the programs + channel_name: Channel title to parse + now: Current datetime (in UTC) + num_days: Number of days to generate programs for + custom_properties: Dict with title_pattern, time_pattern, templates, etc. + - timezone: Timezone name (e.g., 'US/Eastern') + + Returns: + List of program dictionaries with start_time/end_time in UTC + """ + import pytz + + logger.info(f"Generating custom dummy programs for channel: {channel_name}") + + # Extract patterns from custom properties + title_pattern = custom_properties.get('title_pattern', '') + time_pattern = custom_properties.get('time_pattern', '') + date_pattern = custom_properties.get('date_pattern', '') + + # Get timezone name (e.g., 'US/Eastern', 'US/Pacific', 'Europe/London') + timezone_value = custom_properties.get('timezone', 'UTC') + output_timezone_value = custom_properties.get('output_timezone', '') # Optional: display times in different timezone + program_duration = custom_properties.get('program_duration', 180) # Minutes + title_template = custom_properties.get('title_template', '') + subtitle_template = custom_properties.get('subtitle_template', '') + description_template = custom_properties.get('description_template', '') + + # Templates for upcoming/ended programs + upcoming_title_template = custom_properties.get('upcoming_title_template', '') + upcoming_description_template = custom_properties.get('upcoming_description_template', '') + ended_title_template = custom_properties.get('ended_title_template', '') + ended_description_template = custom_properties.get('ended_description_template', '') + + # Image URL templates + channel_logo_url_template = custom_properties.get('channel_logo_url', '') + program_poster_url_template = custom_properties.get('program_poster_url', '') + + # EPG metadata options + category_string = custom_properties.get('category', '') + # Split comma-separated categories and strip whitespace, filter out empty strings + categories = [cat.strip() for cat in category_string.split(',') if cat.strip()] if category_string else [] + include_date = custom_properties.get('include_date', True) + include_live = custom_properties.get('include_live', False) + include_new = custom_properties.get('include_new', False) + + # Parse timezone name + try: + source_tz = pytz.timezone(timezone_value) + logger.debug(f"Using timezone: {timezone_value} (DST will be handled automatically)") + except pytz.exceptions.UnknownTimeZoneError: + logger.warning(f"Unknown timezone: {timezone_value}, defaulting to UTC") + source_tz = pytz.utc + + # Parse output timezone if provided (for display purposes) + output_tz = None + if output_timezone_value: + try: + output_tz = pytz.timezone(output_timezone_value) + logger.debug(f"Using output timezone for display: {output_timezone_value}") + except pytz.exceptions.UnknownTimeZoneError: + logger.warning(f"Unknown output timezone: {output_timezone_value}, will use source timezone") + output_tz = None + + if not title_pattern: + logger.warning(f"No title_pattern in custom_properties, falling back to default") + return None + + logger.debug(f"Title pattern from DB: {repr(title_pattern)}") + + # Convert PCRE/JavaScript named groups (?) to Python format (?P) + # This handles patterns created with JavaScript regex syntax + # Use negative lookahead to avoid matching lookbehind (?<=) and negative lookbehind (?]+)>', r'(?P<\1>', title_pattern) + logger.debug(f"Converted title pattern: {repr(title_pattern)}") + + # Compile regex patterns using the enhanced regex module + # (supports variable-width lookbehinds like JavaScript) + try: + title_regex = regex.compile(title_pattern) + except Exception as e: + logger.error(f"Invalid title regex pattern after conversion: {e}") + logger.error(f"Pattern was: {repr(title_pattern)}") + return None + + time_regex = None + if time_pattern: + # Convert PCRE/JavaScript named groups to Python format + # Use negative lookahead to avoid matching lookbehind (?<=) and negative lookbehind (?]+)>', r'(?P<\1>', time_pattern) + logger.debug(f"Converted time pattern: {repr(time_pattern)}") + try: + time_regex = regex.compile(time_pattern) + except Exception as e: + logger.warning(f"Invalid time regex pattern after conversion: {e}") + logger.warning(f"Pattern was: {repr(time_pattern)}") + + # Compile date regex if provided + date_regex = None + if date_pattern: + # Convert PCRE/JavaScript named groups to Python format + # Use negative lookahead to avoid matching lookbehind (?<=) and negative lookbehind (?]+)>', r'(?P<\1>', date_pattern) + logger.debug(f"Converted date pattern: {repr(date_pattern)}") + try: + date_regex = regex.compile(date_pattern) + except Exception as e: + logger.warning(f"Invalid date regex pattern after conversion: {e}") + logger.warning(f"Pattern was: {repr(date_pattern)}") + + # Try to match the channel name with the title pattern + # Use search() instead of match() to match JavaScript behavior where .match() searches anywhere in the string + title_match = title_regex.search(channel_name) + if not title_match: + logger.debug(f"Channel name '{channel_name}' doesn't match title pattern") + return None + + groups = title_match.groupdict() + logger.debug(f"Title pattern matched. Groups: {groups}") + + # Helper function to format template with matched groups + def format_template(template, groups, url_encode=False): + """Replace {groupname} placeholders with matched group values + + Args: + template: Template string with {groupname} placeholders + groups: Dict of group names to values + url_encode: If True, URL encode the group values for safe use in URLs + """ + if not template: + return '' + result = template + for key, value in groups.items(): + if url_encode and value: + # URL encode the value to handle spaces and special characters + from urllib.parse import quote + encoded_value = quote(str(value), safe='') + result = result.replace(f'{{{key}}}', encoded_value) + else: + result = result.replace(f'{{{key}}}', str(value) if value else '') + return result + + # Extract time from title if time pattern exists + time_info = None + time_groups = {} + if time_regex: + time_match = time_regex.search(channel_name) + if time_match: + time_groups = time_match.groupdict() + try: + hour = int(time_groups.get('hour')) + # Handle optional minute group - could be None if not captured + minute_value = time_groups.get('minute') + minute = int(minute_value) if minute_value is not None else 0 + ampm = time_groups.get('ampm') + ampm = ampm.lower() if ampm else None + + # Determine if this is 12-hour or 24-hour format + if ampm in ('am', 'pm'): + # 12-hour format: convert to 24-hour + if ampm == 'pm' and hour != 12: + hour += 12 + elif ampm == 'am' and hour == 12: + hour = 0 + logger.debug(f"Extracted time (12-hour): {hour}:{minute:02d} {ampm}") + else: + # 24-hour format: hour is already in 24-hour format + # Validate that it's actually a 24-hour time (0-23) + if hour > 23: + logger.warning(f"Invalid 24-hour time: {hour}. Must be 0-23.") + hour = hour % 24 # Wrap around just in case + logger.debug(f"Extracted time (24-hour): {hour}:{minute:02d}") + + time_info = {'hour': hour, 'minute': minute} + except (ValueError, TypeError) as e: + logger.warning(f"Error parsing time: {e}") + + # Extract date from title if date pattern exists + date_info = None + date_groups = {} + if date_regex: + date_match = date_regex.search(channel_name) + if date_match: + date_groups = date_match.groupdict() + try: + # Support various date group names: month, day, year + month_str = date_groups.get('month', '') + day_str = date_groups.get('day', '') + year_str = date_groups.get('year', '') + + # Parse day - default to current day if empty or invalid + day = int(day_str) if day_str else now.day + + # Parse year - default to current year if empty or invalid (matches frontend behavior) + year = int(year_str) if year_str else now.year + + # Parse month - can be numeric (1-12) or text (Jan, January, etc.) + month = None + if month_str: + if month_str.isdigit(): + month = int(month_str) + else: + # Try to parse text month names + import calendar + month_str_lower = month_str.lower() + # Check full month names + for i, month_name in enumerate(calendar.month_name): + if month_name.lower() == month_str_lower: + month = i + break + # Check abbreviated month names if not found + if month is None: + for i, month_abbr in enumerate(calendar.month_abbr): + if month_abbr.lower() == month_str_lower: + month = i + break + + # Default to current month if not extracted or invalid + if month is None: + month = now.month + + if month and 1 <= month <= 12 and 1 <= day <= 31: + date_info = {'year': year, 'month': month, 'day': day} + logger.debug(f"Extracted date: {year}-{month:02d}-{day:02d}") + else: + logger.warning(f"Invalid date values: month={month}, day={day}, year={year}") + except (ValueError, TypeError) as e: + logger.warning(f"Error parsing date: {e}") + + # Merge title groups, time groups, and date groups for template formatting + all_groups = {**groups, **time_groups, **date_groups} + + # Add normalized versions of all groups for cleaner URLs + # These remove all non-alphanumeric characters and convert to lowercase + for key, value in list(all_groups.items()): + if value: + # Remove all non-alphanumeric characters (except spaces temporarily) + # then replace spaces with nothing, and convert to lowercase + normalized = regex.sub(r'[^a-zA-Z0-9\s]', '', str(value)) + normalized = regex.sub(r'\s+', '', normalized).lower() + all_groups[f'{key}_normalize'] = normalized + + # Format channel logo URL if template provided (with URL encoding) + channel_logo_url = None + if channel_logo_url_template: + channel_logo_url = format_template(channel_logo_url_template, all_groups, url_encode=True) + logger.debug(f"Formatted channel logo URL: {channel_logo_url}") + + # Format program poster URL if template provided (with URL encoding) + program_poster_url = None + if program_poster_url_template: + program_poster_url = format_template(program_poster_url_template, all_groups, url_encode=True) + logger.debug(f"Formatted program poster URL: {program_poster_url}") + + # Add formatted time strings for better display (handles minutes intelligently) + if time_info: + hour_24 = time_info['hour'] + minute = time_info['minute'] + + # Determine the base date to use for placeholders + # If date was extracted, use it; otherwise use current date + if date_info: + base_date = datetime(date_info['year'], date_info['month'], date_info['day']) + else: + base_date = datetime.now() + + # If output_timezone is specified, convert the display time to that timezone + if output_tz: + # Create a datetime in the source timezone using the base date + temp_date = source_tz.localize(base_date.replace(hour=hour_24, minute=minute, second=0, microsecond=0)) + # Convert to output timezone + temp_date_output = temp_date.astimezone(output_tz) + # Extract converted hour and minute for display + hour_24 = temp_date_output.hour + minute = temp_date_output.minute + logger.debug(f"Converted display time from {source_tz} to {output_tz}: {hour_24}:{minute:02d}") + + # Add date placeholders based on the OUTPUT timezone + # This ensures {date}, {month}, {day}, {year} reflect the converted timezone + all_groups['date'] = temp_date_output.strftime('%Y-%m-%d') + all_groups['month'] = str(temp_date_output.month) + all_groups['day'] = str(temp_date_output.day) + all_groups['year'] = str(temp_date_output.year) + logger.debug(f"Converted date placeholders to {output_tz}: {all_groups['date']}") + else: + # No output timezone conversion - use source timezone for date + # Create temp date to get proper date in source timezone using the base date + temp_date_source = source_tz.localize(base_date.replace(hour=hour_24, minute=minute, second=0, microsecond=0)) + all_groups['date'] = temp_date_source.strftime('%Y-%m-%d') + all_groups['month'] = str(temp_date_source.month) + all_groups['day'] = str(temp_date_source.day) + all_groups['year'] = str(temp_date_source.year) + + # Format 24-hour start time string - only include minutes if non-zero + if minute > 0: + all_groups['starttime24'] = f"{hour_24}:{minute:02d}" + else: + all_groups['starttime24'] = f"{hour_24:02d}:00" + + # Convert 24-hour to 12-hour format for {starttime} placeholder + # Note: hour_24 is ALWAYS in 24-hour format at this point (converted earlier if needed) + ampm = 'AM' if hour_24 < 12 else 'PM' + hour_12 = hour_24 + if hour_24 == 0: + hour_12 = 12 + elif hour_24 > 12: + hour_12 = hour_24 - 12 + + # Format 12-hour start time string - only include minutes if non-zero + if minute > 0: + all_groups['starttime'] = f"{hour_12}:{minute:02d} {ampm}" + else: + all_groups['starttime'] = f"{hour_12} {ampm}" + + # Format long version that always includes minutes (e.g., "9:00 PM" instead of "9 PM") + all_groups['starttime_long'] = f"{hour_12}:{minute:02d} {ampm}" + + # Calculate end time based on program duration + # Create a datetime for calculations + temp_start = datetime.now(source_tz).replace(hour=hour_24, minute=minute, second=0, microsecond=0) + temp_end = temp_start + timedelta(minutes=program_duration) + + # Extract end time components (already in correct timezone if output_tz was applied above) + end_hour_24 = temp_end.hour + end_minute = temp_end.minute + + # Format 24-hour end time string - only include minutes if non-zero + if end_minute > 0: + all_groups['endtime24'] = f"{end_hour_24}:{end_minute:02d}" + else: + all_groups['endtime24'] = f"{end_hour_24:02d}:00" + + # Convert 24-hour to 12-hour format for {endtime} placeholder + end_ampm = 'AM' if end_hour_24 < 12 else 'PM' + end_hour_12 = end_hour_24 + if end_hour_24 == 0: + end_hour_12 = 12 + elif end_hour_24 > 12: + end_hour_12 = end_hour_24 - 12 + + # Format 12-hour end time string - only include minutes if non-zero + if end_minute > 0: + all_groups['endtime'] = f"{end_hour_12}:{end_minute:02d} {end_ampm}" + else: + all_groups['endtime'] = f"{end_hour_12} {end_ampm}" + + # Format long version that always includes minutes (e.g., "9:00 PM" instead of "9 PM") + all_groups['endtime_long'] = f"{end_hour_12}:{end_minute:02d} {end_ampm}" + + # Generate programs + programs = [] + + # If we have extracted time AND date, the event happens on a SPECIFIC date + # If we have time but NO date, generate for multiple days (existing behavior) + # All other days and times show "Upcoming" before or "Ended" after + event_happened = False + + # Determine how many iterations we need + if date_info and time_info: + # Specific date extracted - only generate for that one date + iterations = 1 + logger.debug(f"Date extracted, generating single event for specific date") + else: + # No specific date - use num_days (existing behavior) + iterations = num_days + + for day in range(iterations): + event_overlaps_window = True + if date_info and time_info: + current_date = datetime( + date_info['year'], + date_info['month'], + date_info['day'], + ).date() + event_start_naive = datetime.combine( + current_date, + datetime.min.time().replace( + hour=time_info['hour'], + minute=time_info['minute'], + ), + ) + try: + event_start_utc = source_tz.localize(event_start_naive).astimezone(pytz.utc) + except Exception as e: + logger.error(f"Error localizing time to {source_tz}: {e}") + event_start_utc = django_timezone.make_aware(event_start_naive, pytz.utc) + event_end_utc = event_start_utc + timedelta(minutes=program_duration) + + lookback = export_lookback if export_lookback is not None else now + event_overlaps_window = _programme_overlaps_export_window( + event_start_utc, event_end_utc, lookback, export_cutoff + ) + if not event_overlaps_window: + logger.debug( + "Custom dummy event outside export window; filling window only: %s", + channel_name, + ) + event_happened = event_end_utc < lookback + day_start = _ceil_to_half_hour(lookback) + if export_cutoff is not None: + day_end = export_cutoff + else: + day_end = now + timedelta(days=num_days if num_days > 0 else 3) + else: + day_start = source_tz.localize( + datetime.combine(current_date, datetime.min.time()) + ).astimezone(pytz.utc) + day_end = day_start + timedelta(days=1) + if export_lookback is not None: + day_start = max(day_start, export_lookback) + if export_cutoff is not None: + day_end = min(day_end, export_cutoff) + else: + day_start = now + timedelta(days=day) + day_end = day_start + timedelta(days=1) + if export_lookback is not None: + day_start = max(day_start, export_lookback) + if export_cutoff is not None: + day_end = min(day_end, export_cutoff) + + if day_start >= day_end: + continue + + if time_info: + if not date_info: + now_in_source_tz = now.astimezone(source_tz) + current_date = (now_in_source_tz + timedelta(days=day)).date() + logger.debug(f"No date extracted, using day offset in {source_tz}: {current_date}") + + event_start_naive = datetime.combine( + current_date, + datetime.min.time().replace( + hour=time_info['hour'], + minute=time_info['minute'], + ), + ) + try: + event_start_local = source_tz.localize(event_start_naive) + event_start_utc = event_start_local.astimezone(pytz.utc) + logger.debug(f"Converted {event_start_local} to UTC: {event_start_utc}") + except Exception as e: + logger.error(f"Error localizing time to {source_tz}: {e}") + event_start_utc = django_timezone.make_aware(event_start_naive, pytz.utc) + + event_end_utc = event_start_utc + timedelta(minutes=program_duration) + + lookback = export_lookback if export_lookback is not None else now + if not _programme_overlaps_export_window( + event_start_utc, event_end_utc, lookback, export_cutoff + ): + continue + else: + logger.debug(f"Using extracted date: {current_date}") + + # Pre-generate the main event title and description for reuse + if title_template: + main_event_title = format_template(title_template, all_groups) + else: + title_parts = [] + if 'league' in all_groups and all_groups['league']: + title_parts.append(all_groups['league']) + if 'team1' in all_groups and 'team2' in all_groups: + title_parts.append(f"{all_groups['team1']} vs {all_groups['team2']}") + elif 'title' in all_groups and all_groups['title']: + title_parts.append(all_groups['title']) + main_event_title = ' - '.join(title_parts) if title_parts else channel_name + + if subtitle_template: + main_event_subtitle = format_template(subtitle_template, all_groups) + else: + main_event_subtitle = None + + if description_template: + main_event_description = format_template(description_template, all_groups) + else: + main_event_description = main_event_title + + + + # Determine if this day is before, during, or after the event + # Event only happens on day 0 (first day) when it falls inside the window + is_event_day = (day == 0) and event_overlaps_window + + if is_event_day and not event_happened: + current_time = day_start + + while current_time < event_start_utc and current_time < day_end: + program_start_utc = current_time + program_end_utc = min(current_time + timedelta(minutes=program_duration), event_start_utc) + + # Use custom upcoming templates if provided, otherwise use defaults + if upcoming_title_template: + upcoming_title = format_template(upcoming_title_template, all_groups) + else: + upcoming_title = main_event_title + + if upcoming_description_template: + upcoming_description = format_template(upcoming_description_template, all_groups) + else: + upcoming_description = f"Upcoming: {main_event_description}" + + # Build custom_properties for upcoming programs (only date, no category/live) + program_custom_properties = {} + + # Add date if requested (YYYY-MM-DD format from start time in event timezone) + if include_date: + # Convert UTC time to event timezone for date calculation + local_time = program_start_utc.astimezone(source_tz) + date_str = local_time.strftime('%Y-%m-%d') + program_custom_properties['date'] = date_str + + # Add program poster URL if provided + if program_poster_url: + program_custom_properties['icon'] = program_poster_url + + programs.append({ + "channel_id": channel_id, + "start_time": program_start_utc, + "end_time": program_end_utc, + "title": upcoming_title, + "sub_title": None, # No subtitle for filler programs + "description": upcoming_description, + "custom_properties": program_custom_properties, + "channel_logo_url": channel_logo_url, # Pass channel logo for EPG generation + }) + + current_time += timedelta(minutes=program_duration) + + # Add the MAIN EVENT at the extracted time + # Build custom_properties for main event (includes category and live) + main_event_custom_properties = {} + + # Add categories if provided + if categories: + main_event_custom_properties['categories'] = categories + + # Add date if requested (YYYY-MM-DD format from start time in event timezone) + if include_date: + # Convert UTC time to event timezone for date calculation + local_time = event_start_utc.astimezone(source_tz) + date_str = local_time.strftime('%Y-%m-%d') + main_event_custom_properties['date'] = date_str + + # Add live flag if requested + if include_live: + main_event_custom_properties['live'] = True + + # Add new flag if requested + if include_new: + main_event_custom_properties['new'] = True + + # Add program poster URL if provided + if program_poster_url: + main_event_custom_properties['icon'] = program_poster_url + + programs.append({ + "channel_id": channel_id, + "start_time": event_start_utc, + "end_time": event_end_utc, + "title": main_event_title, + "sub_title": main_event_subtitle, + "description": main_event_description, + "custom_properties": main_event_custom_properties, + "channel_logo_url": channel_logo_url, # Pass channel logo for EPG generation + }) + + event_happened = True + + # Fill programs AFTER the event until end of export day window + current_time = max(event_end_utc, day_start) + + while current_time < day_end: + program_start_utc = current_time + program_end_utc = min(current_time + timedelta(minutes=program_duration), day_end) + + # Use custom ended templates if provided, otherwise use defaults + if ended_title_template: + ended_title = format_template(ended_title_template, all_groups) + else: + ended_title = main_event_title + + if ended_description_template: + ended_description = format_template(ended_description_template, all_groups) + else: + ended_description = f"Ended: {main_event_description}" + + # Build custom_properties for ended programs (only date, no category/live) + program_custom_properties = {} + + # Add date if requested (YYYY-MM-DD format from start time in event timezone) + if include_date: + # Convert UTC time to event timezone for date calculation + local_time = program_start_utc.astimezone(source_tz) + date_str = local_time.strftime('%Y-%m-%d') + program_custom_properties['date'] = date_str + + # Add program poster URL if provided + if program_poster_url: + program_custom_properties['icon'] = program_poster_url + + programs.append({ + "channel_id": channel_id, + "start_time": program_start_utc, + "end_time": program_end_utc, + "title": ended_title, + "sub_title": None, # No subtitle for filler programs + "description": ended_description, + "custom_properties": program_custom_properties, + "channel_logo_url": channel_logo_url, # Pass channel logo for EPG generation + }) + + current_time += timedelta(minutes=program_duration) + else: + # This day is either before the event (future days) or after the event happened + # Fill entire day with appropriate message + current_time = day_start + + # If event already happened, all programs show "Ended" + # If event hasn't happened yet (shouldn't occur with day 0 logic), show "Upcoming" + is_ended = event_happened + + while current_time < day_end: + program_start_utc = current_time + program_end_utc = min(current_time + timedelta(minutes=program_duration), day_end) + + # Use custom templates based on whether event has ended or is upcoming + if is_ended: + if ended_title_template: + program_title = format_template(ended_title_template, all_groups) + else: + program_title = main_event_title + + if ended_description_template: + program_description = format_template(ended_description_template, all_groups) + else: + program_description = f"Ended: {main_event_description}" + else: + if upcoming_title_template: + program_title = format_template(upcoming_title_template, all_groups) + else: + program_title = main_event_title + + if upcoming_description_template: + program_description = format_template(upcoming_description_template, all_groups) + else: + program_description = f"Upcoming: {main_event_description}" + + # Build custom_properties (only date for upcoming/ended filler programs) + program_custom_properties = {} + + # Add date if requested (YYYY-MM-DD format from start time in event timezone) + if include_date: + # Convert UTC time to event timezone for date calculation + local_time = program_start_utc.astimezone(source_tz) + date_str = local_time.strftime('%Y-%m-%d') + program_custom_properties['date'] = date_str + + # Add program poster URL if provided + if program_poster_url: + program_custom_properties['icon'] = program_poster_url + + programs.append({ + "channel_id": channel_id, + "start_time": program_start_utc, + "end_time": program_end_utc, + "title": program_title, + "sub_title": None, # No subtitle for filler programs + "description": program_description, + "custom_properties": program_custom_properties, + "channel_logo_url": channel_logo_url, + }) + + current_time += timedelta(minutes=program_duration) + else: + # No extracted time - fill entire day with regular intervals + # day_start and day_end are already in UTC, so no conversion needed + programs_per_day = max(1, int(24 / (program_duration / 60))) + + for program_num in range(programs_per_day): + program_start_utc = day_start + timedelta(minutes=program_num * program_duration) + program_end_utc = program_start_utc + timedelta(minutes=program_duration) + + if title_template: + title = format_template(title_template, all_groups) + else: + title_parts = [] + if 'league' in all_groups and all_groups['league']: + title_parts.append(all_groups['league']) + if 'team1' in all_groups and 'team2' in all_groups: + title_parts.append(f"{all_groups['team1']} vs {all_groups['team2']}") + elif 'title' in all_groups and all_groups['title']: + title_parts.append(all_groups['title']) + title = ' - '.join(title_parts) if title_parts else channel_name + + if subtitle_template: + subtitle = format_template(subtitle_template, all_groups) + else: + subtitle = None + + if description_template: + description = format_template(description_template, all_groups) + else: + description = title + + # Build custom_properties for this program + program_custom_properties = {} + + # Add categories if provided + if categories: + program_custom_properties['categories'] = categories + + # Add date if requested (YYYY-MM-DD format from start time in event timezone) + if include_date: + # Convert UTC time to event timezone for date calculation + local_time = program_start_utc.astimezone(source_tz) + date_str = local_time.strftime('%Y-%m-%d') + program_custom_properties['date'] = date_str + + # Add live flag if requested + if include_live: + program_custom_properties['live'] = True + + # Add new flag if requested + if include_new: + program_custom_properties['new'] = True + + # Add program poster URL if provided + if program_poster_url: + program_custom_properties['icon'] = program_poster_url + + programs.append({ + "channel_id": channel_id, + "start_time": program_start_utc, + "end_time": program_end_utc, + "title": title, + "sub_title": subtitle, + "description": description, + "custom_properties": program_custom_properties, + "channel_logo_url": channel_logo_url, # Pass channel logo for EPG generation + }) + + logger.info(f"Generated {len(programs)} custom dummy programs for {channel_name}") + return programs + + +def generate_dummy_epg( + channel_id, channel_name, xml_lines=None, num_days=1, program_length_hours=4 +): + """ + Generate dummy EPG programs for channels without EPG data. + Creates program blocks for a specified number of days. + + Args: + channel_id: The channel ID to use in the program entries + channel_name: The name of the channel to use in program titles + xml_lines: Optional list to append lines to, otherwise returns new list + num_days: Number of days to generate EPG data for (default: 1) + program_length_hours: Length of each program block in hours (default: 4) + + Returns: + List of XML lines for the dummy EPG entries + """ + if xml_lines is None: + xml_lines = [] + + for program in generate_dummy_programs(channel_id, channel_name, num_days=1, program_length_hours=4): + # Format times in XMLTV format + start_str = program['start_time'].strftime("%Y%m%d%H%M%S %z") + stop_str = program['end_time'].strftime("%Y%m%d%H%M%S %z") + + # Create program entry with escaped channel name + xml_lines.append( + f' ' + ) + xml_lines.append(f" {html.escape(program['title'])}") + + # Add subtitle if available + if program.get('sub_title'): + xml_lines.append(f" {html.escape(program['sub_title'])}") + + xml_lines.append(f" {html.escape(program['description'])}") + + # Add custom_properties if present + custom_data = program.get('custom_properties', {}) + + # Categories + if 'categories' in custom_data: + for cat in custom_data['categories']: + xml_lines.append(f" {html.escape(cat)}") + + # Date tag + if 'date' in custom_data: + xml_lines.append(f" {html.escape(custom_data['date'])}") + + # Live tag + if custom_data.get('live', False): + xml_lines.append(f" ") + + # New tag + if custom_data.get('new', False): + xml_lines.append(f" ") + + xml_lines.append(f" ") + + return xml_lines + + +def generate_epg(request, profile_name=None, user=None): + """ + Dynamically generate an XMLTV (EPG) file using a streaming response. + Since the EPG data is stored independently of Channels, we group programmes + by their associated EPGData record. + """ + user_custom = (user.custom_properties or {}) if user else {} + try: + num_days = int(request.GET.get('days', user_custom.get('epg_days', 0))) + num_days = max(0, min(num_days, 365)) + except (ValueError, TypeError): + num_days = 0 + try: + prev_days = int(request.GET.get('prev_days', user_custom.get('epg_prev_days', 0))) + prev_days = max(0, min(prev_days, 30)) + except (ValueError, TypeError): + prev_days = 0 + use_cached_logos = request.GET.get('cachedlogos', 'true').lower() != 'false' + tvg_id_source = request.GET.get('tvg_id_source', 'channel_number').lower() + cache_params = ( + f"{profile_name or 'all'}:{user.username if user else 'anonymous'}" + f":d={num_days}:p={prev_days}:logos={use_cached_logos}:tvgid={tvg_id_source}" + ) + content_cache_key = f"epg_content:{cache_params}" + + def epg_generator(): + """Generator function that yields EPG data with keep-alives during processing.""" + + yield '\n' + yield ( + '\n' + ) + + # Get channels based on user/profile + if user is not None: + if user.user_level < 10: + user_profile_count = user.channel_profiles.count() + + # If user has ALL profiles or NO profiles, give unrestricted access + if user_profile_count == 0: + # No profile filtering - user sees all channels based on user_level + filters = {"user_level__lte": user.user_level} + # Hide adult content if user preference is set + if (user.custom_properties or {}).get('hide_adult_content', False): + filters["is_adult"] = False + base_qs = Channel.objects.filter(**filters).select_related('logo', 'epg_data__epg_source') + else: + # User has specific limited profiles assigned + filters = { + "channelprofilemembership__enabled": True, + "user_level__lte": user.user_level, + "channelprofilemembership__channel_profile__in": user.channel_profiles.all() + } + # Hide adult content if user preference is set + if (user.custom_properties or {}).get('hide_adult_content', False): + filters["is_adult"] = False + base_qs = Channel.objects.filter(**filters).select_related('logo', 'epg_data__epg_source').distinct() + else: + base_qs = Channel.objects.filter(user_level__lte=user.user_level).select_related('logo', 'epg_data__epg_source') + else: + if profile_name is not None: + try: + channel_profile = ChannelProfile.objects.get(name=profile_name) + except ChannelProfile.DoesNotExist: + logger.warning("Requested channel profile (%s) during epg generation does not exist", profile_name) + raise Http404(f"Channel profile '{profile_name}' not found") + base_qs = Channel.objects.filter( + channelprofilemembership__channel_profile=channel_profile, + channelprofilemembership__enabled=True, + ).select_related('logo', 'epg_data__epg_source') + else: + base_qs = Channel.objects.all().select_related('logo', 'epg_data__epg_source') + + # Resolve effective values at SQL level and exclude hidden channels + # so output ordering/display honors user overrides. + from apps.channels.managers import with_effective_values + channels = list( + with_effective_values(base_qs, select_related_fks=True) + .exclude(hidden_from_output=True) + .order_by("effective_channel_number") + # programme_index is a multi-MB JSON byte-offset index that EPG + # generation never reads; defer it so it isn't fetched and JSON-parsed + # once per channel (was ~13s of the request on large guides). + .defer("epg_data__epg_source__programme_index") + .prefetch_related( + Prefetch('streams', queryset=Stream.objects.order_by('channelstream__order')) + ) + ) + channel_count = len(channels) + + # For dummy EPG, use either the specified value or default to 3 days + dummy_days = num_days if num_days > 0 else 3 + + # Calculate cutoff dates for EPG data filtering + now = django_timezone.now() + cutoff_date = now + timedelta(days=num_days) if num_days > 0 else None + lookback_cutoff = now - timedelta(days=prev_days) + + # Build collision-free channel number mapping for XC clients (if user is authenticated) + # XC clients require integer channel numbers, so we need to ensure no conflicts + channel_num_map = {} + if user is not None: + # This is an XC client - build collision-free mapping + used_numbers = set() + + # First pass: assign integers for channels that already have integer numbers + for channel in channels: + effective_num = channel.effective_channel_number + if effective_num is not None and effective_num == int(effective_num): + num = int(effective_num) + channel_num_map[channel.id] = num + used_numbers.add(num) + + # Second pass: assign integers for channels with float numbers + for channel in channels: + effective_num = channel.effective_channel_number + if effective_num is not None and effective_num != int(effective_num): + candidate = int(effective_num) + while candidate in used_numbers: + candidate += 1 + channel_num_map[channel.id] = candidate + used_numbers.add(candidate) + + # Host/port/scheme are constant per request; precompute logo URL prefix once. + _base_url = build_absolute_uri_with_port(request, "") + _sample_logo_path = reverse("api:channels:logo-cache", args=[0]) + _logo_prefix_raw, _, _logo_suffix_raw = _sample_logo_path.partition("/0/") + _logo_url_prefix = _base_url + _logo_prefix_raw + "/" + _logo_url_suffix = "/" + _logo_suffix_raw + + dummy_program_list = [] + real_epg_map = {} + channel_xml_batch = [] + + for channel in channels: + effective_name = channel.effective_name + effective_epg_data = channel.effective_epg_data_obj + effective_epg_data_id = channel.effective_epg_data_id + effective_logo = channel.effective_logo_obj + effective_number = channel.effective_channel_number + + # user is set only for XC clients, which require integer channel numbers + if user is not None: + formatted_channel_number = channel_num_map[channel.id] + else: + formatted_channel_number = format_channel_number(effective_number) + + # Determine the channel ID based on the selected source + if tvg_id_source == 'tvg_id' and channel.effective_tvg_id: + channel_id = channel.effective_tvg_id + elif tvg_id_source == 'gracenote' and channel.effective_tvc_guide_stationid: + channel_id = channel.effective_tvc_guide_stationid + else: + channel_id = str(formatted_channel_number) if formatted_channel_number != "" else str(channel.id) + + tvg_logo = "" + + # Check if this is a custom dummy EPG with channel logo URL template + if effective_epg_data and effective_epg_data.epg_source and effective_epg_data.epg_source.source_type == 'dummy': + epg_source = effective_epg_data.epg_source + if epg_source.custom_properties: + custom_props = epg_source.custom_properties + channel_logo_url_template = custom_props.get('channel_logo_url', '') + + if channel_logo_url_template: + pattern_match_name, _ = _pattern_match_name_from_custom_props( + channel, effective_name, custom_props + ) + + # Try to extract groups from the channel/stream name and build the logo URL + title_pattern = custom_props.get('title_pattern', '') + if title_pattern: + try: + # Convert PCRE/JavaScript named groups to Python format + title_pattern = regex.sub(r'\(\?<(?![=!])([^>]+)>', r'(?P<\1>', title_pattern) + title_regex = regex.compile(title_pattern) + title_match = title_regex.search(pattern_match_name) + + if title_match: + groups = title_match.groupdict() + + # Add normalized versions of all groups for cleaner URLs + for key, value in list(groups.items()): + if value: + # Remove all non-alphanumeric characters and convert to lowercase + normalized = regex.sub(r'[^a-zA-Z0-9\s]', '', str(value)) + normalized = regex.sub(r'\s+', '', normalized).lower() + groups[f'{key}_normalize'] = normalized + + # Format the logo URL template with the matched groups (with URL encoding) + from urllib.parse import quote + for key, value in groups.items(): + if value: + encoded_value = quote(str(value), safe='') + channel_logo_url_template = channel_logo_url_template.replace(f'{{{key}}}', encoded_value) + else: + channel_logo_url_template = channel_logo_url_template.replace(f'{{{key}}}', '') + tvg_logo = channel_logo_url_template + logger.debug(f"Built channel logo URL from template: {tvg_logo}") + except Exception as e: + logger.warning(f"Failed to build channel logo URL for {effective_name}: {e}") + + # If no custom dummy logo, use regular logo logic + if not tvg_logo and effective_logo: + if use_cached_logos: + tvg_logo = f"{_logo_url_prefix}{effective_logo.id}{_logo_url_suffix}" + else: + # Use direct URL if available, otherwise fall back to cached version + direct_logo = effective_logo.url if effective_logo.url.startswith(('http://', 'https://')) else None + if direct_logo: + tvg_logo = direct_logo + else: + tvg_logo = f"{_logo_url_prefix}{effective_logo.id}{_logo_url_suffix}" + channel_xml_batch.append(f' ') + channel_xml_batch.append(f' {html.escape(effective_name)}') + channel_xml_batch.append(f' ') + channel_xml_batch.append(" ") + + if len(channel_xml_batch) >= _EPG_CHANNEL_XML_BATCH_SIZE * 4: + yield '\n'.join(channel_xml_batch) + '\n' + channel_xml_batch = [] + + pattern_match_name = effective_name + if effective_epg_data and effective_epg_data.epg_source: + epg_source = effective_epg_data.epg_source + if epg_source.custom_properties: + custom_props = epg_source.custom_properties + pattern_match_name, stream_lookup_failed = _pattern_match_name_from_custom_props( + channel, effective_name, custom_props + ) + if ( + custom_props.get('name_source') == 'stream' + and not stream_lookup_failed + and pattern_match_name != effective_name + ): + stream_index = custom_props.get('stream_index', 1) - 1 + logger.debug( + f"Using stream name for parsing: {pattern_match_name} " + f"(stream index: {stream_index})" + ) + elif stream_lookup_failed: + stream_index = custom_props.get('stream_index', 1) - 1 + logger.warning( + f"Stream index {stream_index} not found for channel " + f"{effective_name}, falling back to channel name" + ) + + if not effective_epg_data: + dummy_program_list.append((channel_id, pattern_match_name, None)) + elif effective_epg_data.epg_source and effective_epg_data.epg_source.source_type == 'dummy': + dummy_program_list.append((channel_id, pattern_match_name, effective_epg_data.epg_source)) + else: + real_epg_map.setdefault(effective_epg_data_id, []).append(channel_id) + + if channel_xml_batch: + yield '\n'.join(channel_xml_batch) + '\n' + + del channels + del channel_num_map + + batch_size = _EPG_PROGRAM_YIELD_BATCH_SIZE + + all_epg_ids = list(real_epg_map.keys()) + if all_epg_ids: + if num_days > 0: + programs_qs = ProgramData.objects.filter( + epg_id__in=all_epg_ids, + end_time__gte=lookback_cutoff, + start_time__lt=cutoff_date, + ) + else: + programs_qs = ProgramData.objects.filter( + epg_id__in=all_epg_ids, + end_time__gte=lookback_cutoff, + ) + + programs_base_qs = programs_qs.order_by('epg_id', 'id').values( + 'id', 'epg_id', 'start_time', 'end_time', 'title', 'sub_title', + 'description', 'custom_properties', + ) + + current_epg_id = None + channel_ids_for_epg = None + pending = [] + program_batch = [] + chunk_size = _EPG_PROGRAM_DB_CHUNK_SIZE + last_epg_id = 0 + last_id = 0 + _poster_url_base = build_absolute_uri_with_port(request, "/api/epg/programs/") + + def flush_pending(): + nonlocal program_batch, pending + if not pending: + return + pending.sort(key=lambda row: (row[0], row[1])) + escaped_primary = ( + html.escape(channel_ids_for_epg[0]) + if len(channel_ids_for_epg) > 1 else None + ) + for _, _, xml_text in pending: + program_batch.append(xml_text) + if escaped_primary: + for cid in channel_ids_for_epg[1:]: + program_batch.append(xml_text.replace( + f'channel="{escaped_primary}"', + f'channel="{html.escape(cid)}"', + 1, + )) + if len(program_batch) >= batch_size: + yield '\n'.join(program_batch) + '\n' + program_batch = [] + pending.clear() + + while True: + program_chunk = list( + programs_base_qs.filter(epg_id__gte=last_epg_id) + .exclude(epg_id=last_epg_id, id__lte=last_id)[:chunk_size] + ) + if not program_chunk: + break + + last_row = program_chunk[-1] + last_epg_id = last_row['epg_id'] + last_id = last_row['id'] + + for prog in program_chunk: + epg_id = prog['epg_id'] + + if epg_id != current_epg_id: + yield from flush_pending() + current_epg_id = epg_id + channel_ids_for_epg = real_epg_map[epg_id] + + primary_cid = channel_ids_for_epg[0] + # DB datetimes are UTC (USE_TZ=True, TIME_ZONE=UTC); format + # directly instead of strftime("%Y%m%d%H%M%S %z"), which is + # ~10x slower and dominates XML build over 750k rows. + st = prog['start_time'] + et = prog['end_time'] + start_str = f"{st.year:04d}{st.month:02d}{st.day:02d}{st.hour:02d}{st.minute:02d}{st.second:02d} +0000" + stop_str = f"{et.year:04d}{et.month:02d}{et.day:02d}{et.hour:02d}{et.minute:02d}{et.second:02d} +0000" + + program_xml = [f' '] + program_xml.append(f' {html.escape(prog["title"])}') + + if prog['sub_title']: + program_xml.append(f" {html.escape(prog['sub_title'])}") + + if prog['description']: + program_xml.append(f" {html.escape(prog['description'])}") + + custom_data = prog['custom_properties'] or {} + if custom_data: + + if "categories" in custom_data and custom_data["categories"]: + for category in custom_data["categories"]: + program_xml.append(f" {html.escape(category)}") + + if "keywords" in custom_data and custom_data["keywords"]: + for keyword in custom_data["keywords"]: + program_xml.append(f" {html.escape(keyword)}") + + # onscreen_episode takes priority over episode for the onscreen system + if "onscreen_episode" in custom_data: + program_xml.append(f' {html.escape(custom_data["onscreen_episode"])}') + elif "episode" in custom_data: + program_xml.append(f' E{custom_data["episode"]}') + + # Handle dd_progid format + if 'dd_progid' in custom_data: + program_xml.append(f' {html.escape(custom_data["dd_progid"])}') + + # Handle external database IDs + for system in ['thetvdb.com', 'themoviedb.org', 'imdb.com']: + if f'{system}_id' in custom_data: + program_xml.append(f' {html.escape(custom_data[f"{system}_id"])}') + + # Add season and episode numbers in xmltv_ns format if available + if "season" in custom_data and "episode" in custom_data: + season = ( + int(custom_data["season"]) - 1 + if str(custom_data["season"]).isdigit() + else 0 + ) + episode = ( + int(custom_data["episode"]) - 1 + if str(custom_data["episode"]).isdigit() + else 0 + ) + program_xml.append(f' {season}.{episode}.') + + if "language" in custom_data: + program_xml.append(f' {html.escape(custom_data["language"])}') + + if "original_language" in custom_data: + program_xml.append(f' {html.escape(custom_data["original_language"])}') + + if "length" in custom_data and isinstance(custom_data["length"], dict): + length_value = custom_data["length"].get("value", "") + length_units = custom_data["length"].get("units", "minutes") + program_xml.append(f' {html.escape(str(length_value))}') + + if "video" in custom_data and isinstance(custom_data["video"], dict): + program_xml.append(" ") + + if "audio" in custom_data and isinstance(custom_data["audio"], dict): + program_xml.append(" ") + + if "subtitles" in custom_data and isinstance(custom_data["subtitles"], list): + for subtitle in custom_data["subtitles"]: + if isinstance(subtitle, dict): + subtitle_type = subtitle.get("type", "") + type_attr = f' type="{html.escape(subtitle_type)}"' if subtitle_type else "" + program_xml.append(f" ") + if "language" in subtitle: + program_xml.append(f" {html.escape(subtitle['language'])}") + program_xml.append(" ") + + if "rating" in custom_data: + rating_system = custom_data.get("rating_system", "TV Parental Guidelines") + program_xml.append(f' ') + program_xml.append(f' {html.escape(custom_data["rating"])}') + program_xml.append(f" ") + + if "star_ratings" in custom_data and isinstance(custom_data["star_ratings"], list): + for star_rating in custom_data["star_ratings"]: + if isinstance(star_rating, dict) and "value" in star_rating: + system_attr = f' system="{html.escape(star_rating["system"])}"' if "system" in star_rating else "" + program_xml.append(f" ") + program_xml.append(f" {html.escape(star_rating['value'])}") + program_xml.append(" ") + + if "reviews" in custom_data and isinstance(custom_data["reviews"], list): + for review in custom_data["reviews"]: + if isinstance(review, dict) and "content" in review: + review_type = review.get("type", "text") + attrs = [f'type="{html.escape(review_type)}"'] + if "source" in review: + attrs.append(f'source="{html.escape(review["source"])}"') + if "reviewer" in review: + attrs.append(f'reviewer="{html.escape(review["reviewer"])}"') + attr_str = " ".join(attrs) + program_xml.append(f' {html.escape(review["content"])}') + + if "images" in custom_data and isinstance(custom_data["images"], list): + for image in custom_data["images"]: + if isinstance(image, dict) and "url" in image: + attrs = [] + for attr in ['type', 'size', 'orient', 'system']: + if attr in image: + attrs.append(f'{attr}="{html.escape(image[attr])}"') + attr_str = " " + " ".join(attrs) if attrs else "" + program_xml.append(f' {html.escape(image["url"])}') + + # Add enhanced credits handling + if "credits" in custom_data: + program_xml.append(" ") + credits = custom_data["credits"] + + for role in ['director', 'writer', 'adapter', 'producer', 'composer', 'editor', 'presenter', 'commentator', 'guest']: + if role in credits: + people = credits[role] + if isinstance(people, list): + for person in people: + program_xml.append(f" <{role}>{html.escape(person)}") + else: + program_xml.append(f" <{role}>{html.escape(people)}") + + # Handle actors separately to include role and guest attributes + if "actor" in credits: + actors = credits["actor"] + if isinstance(actors, list): + for actor in actors: + if isinstance(actor, dict): + name = actor.get("name", "") + role_attr = f' role="{html.escape(actor["role"])}"' if "role" in actor else "" + guest_attr = ' guest="yes"' if actor.get("guest") else "" + program_xml.append(f" {html.escape(name)}") + else: + program_xml.append(f" {html.escape(actor)}") + else: + program_xml.append(f" {html.escape(actors)}") + + program_xml.append(" ") + + if "date" in custom_data: + program_xml.append(f' {html.escape(custom_data["date"])}') + + if "country" in custom_data: + program_xml.append(f' {html.escape(custom_data["country"])}') + + if "icon" in custom_data: + program_xml.append(f' ') + elif "sd_icon" in custom_data: + program_xml.append(f' ') + + # Add special flags as proper tags with enhanced handling + if custom_data.get("previously_shown", False): + prev_shown_details = custom_data.get("previously_shown_details", {}) + attrs = [] + if "start" in prev_shown_details: + attrs.append(f'start="{html.escape(prev_shown_details["start"])}"') + if "channel" in prev_shown_details: + attrs.append(f'channel="{html.escape(prev_shown_details["channel"])}"') + attr_str = " " + " ".join(attrs) if attrs else "" + program_xml.append(f" ") + + if custom_data.get("premiere", False): + premiere_text = custom_data.get("premiere_text", "") + if premiere_text: + program_xml.append(f" {html.escape(premiere_text)}") + else: + program_xml.append(" ") + + if custom_data.get("last_chance", False): + last_chance_text = custom_data.get("last_chance_text", "") + if last_chance_text: + program_xml.append(f" {html.escape(last_chance_text)}") + else: + program_xml.append(" ") + + if custom_data.get("new", False): + program_xml.append(" ") + + if custom_data.get('live', False): + program_xml.append(' ') + + program_xml.append(" ") + + xml_text = '\n'.join(program_xml) + pending.append((prog['start_time'], prog['id'], xml_text)) + + del program_chunk + + yield from flush_pending() + + if program_batch: + yield '\n'.join(program_batch) + '\n' + + del real_epg_map + + for channel_id, pattern_match_name, epg_source in dummy_program_list: + program_length_hours = 4 + dummy_programs = generate_dummy_programs( + channel_id, pattern_match_name, + num_days=dummy_days, + program_length_hours=program_length_hours, + epg_source=epg_source, + export_lookback=lookback_cutoff, + export_cutoff=cutoff_date, + ) + if not dummy_programs: + continue + dummy_batch = [] + for program in dummy_programs: + start_str = program['start_time'].strftime("%Y%m%d%H%M%S %z") + stop_str = program['end_time'].strftime("%Y%m%d%H%M%S %z") + lines = [ + f' ', + f" {html.escape(program['title'])}", + ] + if program.get('sub_title'): + lines.append(f" {html.escape(program['sub_title'])}") + lines.append(f" {html.escape(program['description'])}") + custom_data = program.get('custom_properties', {}) + if 'categories' in custom_data: + for cat in custom_data['categories']: + lines.append(f" {html.escape(cat)}") + if 'date' in custom_data: + lines.append(f" {html.escape(custom_data['date'])}") + if custom_data.get('live', False): + lines.append(" ") + if custom_data.get('new', False): + lines.append(" ") + if 'icon' in custom_data: + lines.append(f' ') + lines.append(" ") + dummy_batch.append('\n'.join(lines)) + if len(dummy_batch) >= batch_size: + yield '\n'.join(dummy_batch) + '\n' + dummy_batch = [] + del dummy_programs + if dummy_batch: + yield '\n'.join(dummy_batch) + '\n' + + del dummy_program_list + + yield "\n" + + from apps.output.views import get_client_identifier + + client_id, client_ip, user_agent = get_client_identifier(request) + event_cache_key = f"epg_download:{user.username if user else 'anonymous'}:{profile_name or 'all'}:{client_id}" + + def _log_epg_download(): + from django.core.cache import cache as event_cache + + if not event_cache.get(event_cache_key): + log_system_event( + event_type='epg_download', + profile=profile_name or 'all', + user=user.username if user else 'anonymous', + channels=channel_count, + client_ip=client_ip, + user_agent=user_agent, + ) + event_cache.set(event_cache_key, True, 2) + + try: + from core.utils import _is_gevent_monkey_patched + + if _is_gevent_monkey_patched(): + import gevent + + gevent.spawn(_log_epg_download) + else: + _log_epg_download() + except Exception: + _log_epg_download() + + def build_epg_stream(): + try: + yield from epg_generator() + finally: + _epg_export_teardown() + + return stream_cached_response( + content_cache_key, + build_epg_stream, + content_type="application/xml", + filename="Dispatcharr.xml", + ) diff --git a/apps/output/epg_chunk_cache.py b/apps/output/streaming_chunk_cache.py similarity index 81% rename from apps/output/epg_chunk_cache.py rename to apps/output/streaming_chunk_cache.py index 92ea6eeb..f86bb060 100644 --- a/apps/output/epg_chunk_cache.py +++ b/apps/output/streaming_chunk_cache.py @@ -1,4 +1,4 @@ -"""Single-flight Redis chunk cache for XMLTV EPG streaming responses.""" +"""Single-flight Redis chunk cache for large streaming HTTP responses.""" import logging import time @@ -111,7 +111,7 @@ def _stream_build(redis, base_key, source, cache_ttl, lock_ttl): try: from django.core.cache import cache as django_cache - django_cache.delete(base_key) # legacy monolithic entry + django_cache.delete(base_key) # clear any non-chunked entry under this key redis.delete(chunks_key, _ready_key(base_key)) redis.set(status_key, STATUS_BUILDING, ex=lock_ttl) for chunk in source(): @@ -123,9 +123,9 @@ def _stream_build(redis, base_key, source, cache_ttl, lock_ttl): redis.expire(chunks_key, cache_ttl) redis.expire(status_key, cache_ttl) redis.expire(_ready_key(base_key), cache_ttl) - logger.debug("Cached EPG in %s chunks", redis.llen(chunks_key)) + logger.debug("Cached response in %s chunks", redis.llen(chunks_key)) except Exception: - logger.exception("EPG cache build failed for %s", base_key) + logger.exception("Chunk cache build failed for %s", base_key) redis.delete(chunks_key) redis.set(status_key, STATUS_ERROR, ex=60) raise @@ -158,14 +158,14 @@ def _stream_follow(redis, base_key, source, cache_ttl, lock_ttl, poll_interval, if offset == 0 and _try_acquire_lock(redis, base_key, lock_ttl): yield from _stream_build(redis, base_key, source, cache_ttl, lock_ttl) return - raise RuntimeError("EPG cache build failed") + raise RuntimeError("Chunk cache build failed") if time.monotonic() >= deadline: if offset == 0 and _try_acquire_lock(redis, base_key, lock_ttl): - logger.warning("EPG cache follower timed out; rebuilding %s", base_key) + logger.warning("Chunk cache follower timed out; rebuilding %s", base_key) yield from _stream_build(redis, base_key, source, cache_ttl, lock_ttl) return - logger.warning("EPG cache follower timed out after partial read for %s", base_key) + logger.warning("Chunk cache follower timed out after partial read for %s", base_key) break lock_active = bool(redis.exists(lock_key)) @@ -173,7 +173,7 @@ def _stream_follow(redis, base_key, source, cache_ttl, lock_ttl, poll_interval, idle_polls += 1 if offset == 0 and idle_polls >= max(1, int(1.0 / poll_interval)): if _try_acquire_lock(redis, base_key, lock_ttl): - logger.warning("EPG cache leader lost; rebuilding %s", base_key) + logger.warning("Chunk cache leader lost; rebuilding %s", base_key) yield from _stream_build(redis, base_key, source, cache_ttl, lock_ttl) return else: @@ -182,10 +182,12 @@ def _stream_follow(redis, base_key, source, cache_ttl, lock_ttl, poll_interval, _poll_wait(poll_interval) -def stream_epg_response( +def stream_cached_response( cache_key, source, *, + content_type="application/xml", + filename=None, cache_ttl=DEFAULT_CACHE_TTL, lock_ttl=DEFAULT_LOCK_TTL, poll_interval=DEFAULT_POLL_INTERVAL, @@ -193,16 +195,17 @@ def stream_epg_response( redis=None, ): """ - Stream XMLTV EPG output with single-flight Redis chunk caching. + Stream a large response with single-flight Redis chunk caching. ``source`` must be a callable returning a chunk iterator. Only the leader - invokes it; followers read chunks already written to Redis. + invokes it; concurrent followers replay chunks already written to Redis, so + the expensive ``source`` runs at most once per ``cache_key``. """ if redis is None: redis = _get_redis() if redis.get(_ready_key(cache_key)): - logger.debug("Serving EPG from chunk cache") + logger.debug("Serving response from chunk cache") stream = _stream_ready(redis, cache_key) else: status = _get_status(redis, cache_key) @@ -210,10 +213,10 @@ def stream_epg_response( _clear_build_keys(redis, cache_key) if _try_acquire_lock(redis, cache_key, lock_ttl): - logger.debug("Building EPG (cache leader)") + logger.debug("Building response (cache leader)") stream = _stream_build(redis, cache_key, source, cache_ttl, lock_ttl) else: - logger.debug("Following in-flight EPG build") + logger.debug("Following in-flight cache build") stream = _stream_follow( redis, cache_key, @@ -224,7 +227,8 @@ def stream_epg_response( max_follower_wait, ) - response = StreamingHttpResponse(stream, content_type="application/xml") - response["Content-Disposition"] = 'attachment; filename="Dispatcharr.xml"' + response = StreamingHttpResponse(stream, content_type=content_type) + if filename: + response["Content-Disposition"] = f'attachment; filename="{filename}"' response["Cache-Control"] = "no-cache" return response diff --git a/apps/output/test_epg_chunk_cache.py b/apps/output/test_streaming_chunk_cache.py similarity index 84% rename from apps/output/test_epg_chunk_cache.py rename to apps/output/test_streaming_chunk_cache.py index 6ca217af..599501b7 100644 --- a/apps/output/test_epg_chunk_cache.py +++ b/apps/output/test_streaming_chunk_cache.py @@ -2,14 +2,14 @@ import threading import time from unittest import TestCase -from apps.output.epg_chunk_cache import ( +from apps.output.streaming_chunk_cache import ( STATUS_BUILDING, STATUS_READY, _chunks_key, _lock_key, _ready_key, _status_key, - stream_epg_response, + stream_cached_response, ) @@ -74,7 +74,7 @@ def _consume(response): return b"".join(response.streaming_content).decode("utf-8") -class EPGChunkCacheTests(TestCase): +class StreamingChunkCacheTests(TestCase): def test_leader_caches_chunks_and_sets_ready(self): redis = FakeRedis() calls = [] @@ -84,14 +84,14 @@ class EPGChunkCacheTests(TestCase): yield "" yield "" - body = _consume(stream_epg_response("epg:test", source, redis=redis)) + body = _consume(stream_cached_response("cache:test", source, redis=redis)) self.assertEqual(body, "") self.assertEqual(calls, [1]) - self.assertEqual(redis.get(_ready_key("epg:test")), "1") - self.assertEqual(redis.get(_status_key("epg:test")), STATUS_READY) - self.assertEqual(redis.llen(_chunks_key("epg:test")), 2) - self.assertFalse(redis.exists(_lock_key("epg:test"))) + self.assertEqual(redis.get(_ready_key("cache:test")), "1") + self.assertEqual(redis.get(_status_key("cache:test")), STATUS_READY) + self.assertEqual(redis.llen(_chunks_key("cache:test")), 2) + self.assertFalse(redis.exists(_lock_key("cache:test"))) def test_cache_hit_skips_source(self): redis = FakeRedis() @@ -102,16 +102,16 @@ class EPGChunkCacheTests(TestCase): yield "" yield "" - _consume(stream_epg_response("epg:test", source, redis=redis)) + _consume(stream_cached_response("cache:test", source, redis=redis)) calls.clear() - body = _consume(stream_epg_response("epg:test", source, redis=redis)) + body = _consume(stream_cached_response("cache:test", source, redis=redis)) self.assertEqual(body, "") self.assertEqual(calls, []) def test_follower_reads_leader_chunks_without_rebuilding(self): redis = FakeRedis() - base = "epg:follow" + base = "cache:follow" leader_started = threading.Event() rebuild_calls = [] @@ -128,7 +128,7 @@ class EPGChunkCacheTests(TestCase): def leader(): _consume( - stream_epg_response( + stream_cached_response( base, slow_source, redis=redis, @@ -140,7 +140,7 @@ class EPGChunkCacheTests(TestCase): leader_thread.start() leader_started.wait(timeout=5) follower_body = _consume( - stream_epg_response( + stream_cached_response( base, forbidden_source, redis=redis, @@ -165,8 +165,8 @@ class EPGChunkCacheTests(TestCase): def worker(): barrier.wait() results[threading.current_thread().name] = _consume( - stream_epg_response( - "epg:race", + stream_cached_response( + "cache:race", source, redis=redis, poll_interval=0.01, diff --git a/apps/output/tests.py b/apps/output/tests.py index 6e2ab395..5f4fd5f3 100644 --- a/apps/output/tests.py +++ b/apps/output/tests.py @@ -34,16 +34,18 @@ class OutputEndpointTestMixin: "apps.output.views.network_access_allowed", return_value=True, ) - self._epg_teardown_patch = patch("apps.output.views._epg_export_teardown") + self._epg_teardown_patch = patch("apps.output.epg._epg_export_teardown") self._log_event_patch = patch("apps.output.views.log_system_event") + self._epg_log_event_patch = patch("apps.output.epg.log_system_event") self._close_db_patch = patch("django.db.close_old_connections") self._epg_cache_patch = patch( - "apps.output.views.stream_epg_response", + "apps.output.epg.stream_cached_response", side_effect=_epg_response_without_redis, ) self._network_patch.start() self._epg_teardown_patch.start() self._log_event_patch.start() + self._epg_log_event_patch.start() self._close_db_patch.start() self._epg_cache_patch.start() @@ -53,6 +55,7 @@ class OutputEndpointTestMixin: cache.clear() self._epg_cache_patch.stop() self._close_db_patch.stop() + self._epg_log_event_patch.stop() self._log_event_patch.stop() self._epg_teardown_patch.stop() self._network_patch.stop() @@ -339,14 +342,14 @@ class OutputEPGCustomDummyTest(TestCase): class OutputEPGHelperTest(SimpleTestCase): def test_ceil_to_half_hour_on_boundary(self): from django.utils import timezone - from apps.output.views import _ceil_to_half_hour + from apps.output.epg import _ceil_to_half_hour dt = timezone.now().replace(minute=30, second=0, microsecond=0) self.assertEqual(_ceil_to_half_hour(dt), dt) def test_ceil_to_half_hour_rounds_up(self): from django.utils import timezone - from apps.output.views import _ceil_to_half_hour + from apps.output.epg import _ceil_to_half_hour dt = timezone.now().replace(minute=17, second=42, microsecond=123456) aligned = _ceil_to_half_hour(dt) diff --git a/apps/output/views.py b/apps/output/views.py index a0875c7f..4224fb35 100644 --- a/apps/output/views.py +++ b/apps/output/views.py @@ -25,54 +25,10 @@ from apps.proxy.utils import get_user_active_connections import regex from core.utils import log_system_event, build_absolute_uri_with_port import hashlib -from apps.output.epg_chunk_cache import stream_epg_response +from apps.output.epg import generate_epg, generate_dummy_programs logger = logging.getLogger(__name__) -_EPG_CHANNEL_XML_BATCH_SIZE = 200 -_EPG_PROGRAM_YIELD_BATCH_SIZE = 1000 -_EPG_PROGRAM_DB_CHUNK_SIZE = 20000 - - -def _programme_overlaps_export_window(start_time, end_time, lookback_cutoff, cutoff_date): - if end_time < lookback_cutoff: - return False - if cutoff_date is not None and start_time >= cutoff_date: - return False - return True - - -def _ceil_to_half_hour(dt): - """Round a datetime up to the next :00 or :30 boundary.""" - dt = dt.replace(second=0, microsecond=0) - remainder = dt.minute % 30 - if remainder == 0: - return dt - return dt + timedelta(minutes=30 - remainder) - - -def _epg_export_teardown(): - from django.db import close_old_connections - - from core.utils import ( - _is_gevent_monkey_patched, - cleanup_memory, - trim_c_allocator_heap, - ) - - close_old_connections() - - def _run(): - cleanup_memory(force_collection=True) - trim_c_allocator_heap() - - if _is_gevent_monkey_patched(): - import gevent - - gevent.spawn(_run) - else: - _run() - def get_client_identifier(request): """Get client information including IP, user agent, and a unique hash identifier @@ -384,1674 +340,6 @@ def generate_m3u(request, profile_name=None, user=None): return response -def _ordered_channel_streams(channel): - """Return a channel's streams ordered by channelstream join order.""" - prefetched = getattr(channel, '_prefetched_objects_cache', {}).get('streams') - if prefetched is not None: - return list(prefetched) - return list(channel.streams.all().order_by('channelstream__order')) - - -def _pattern_match_name_from_custom_props(channel, effective_name, custom_props): - """Name used for custom dummy EPG regex matching (channel or stream title). - - Returns (name, stream_lookup_failed). stream_lookup_failed is True only when - name_source is 'stream' but the configured index is missing or out of range. - """ - if custom_props.get('name_source') != 'stream': - return effective_name, False - stream_index = custom_props.get('stream_index', 1) - 1 - streams = _ordered_channel_streams(channel) - if 0 <= stream_index < len(streams): - return streams[stream_index].name, False - return effective_name, True - - -def generate_fallback_programs(channel_id, channel_name, now, num_days, program_length_hours, fallback_title, fallback_description): - """ - Generate dummy programs using custom fallback templates when patterns don't match. - - Args: - channel_id: Channel ID for the programs - channel_name: Channel name to use as fallback in templates - now: Current datetime (in UTC) - num_days: Number of days to generate programs for - program_length_hours: Length of each program in hours - fallback_title: Custom fallback title template (empty string if not provided) - fallback_description: Custom fallback description template (empty string if not provided) - - Returns: - List of program dictionaries - """ - programs = [] - - # Use custom fallback title or channel name as default - title = fallback_title if fallback_title else channel_name - - # Use custom fallback description or a simple default message - if fallback_description: - description = fallback_description - else: - description = f"EPG information is currently unavailable for {channel_name}" - - # Create programs for each day - for day in range(num_days): - day_start = now + timedelta(days=day) - - # Create programs with specified length throughout the day - for hour_offset in range(0, 24, program_length_hours): - # Calculate program start and end times - start_time = day_start + timedelta(hours=hour_offset) - end_time = start_time + timedelta(hours=program_length_hours) - - programs.append({ - "channel_id": channel_id, - "start_time": start_time, - "end_time": end_time, - "title": title, - "description": description, - }) - - return programs - - -def generate_dummy_programs( - channel_id, - channel_name, - num_days=1, - program_length_hours=4, - epg_source=None, - export_lookback=None, - export_cutoff=None, -): - """ - Generate dummy EPG programs for channels. - - If epg_source is provided and it's a custom dummy EPG with patterns, - use those patterns to generate programs from the channel title. - Otherwise, generate default dummy programs. - - Args: - channel_id: Channel ID for the programs - channel_name: Channel title/name - num_days: Number of days to generate programs for - program_length_hours: Length of each program in hours - epg_source: Optional EPGSource for custom dummy EPG with patterns - - Returns: - List of program dictionaries - """ - # Get current time rounded to hour - now = django_timezone.now() - now = now.replace(minute=0, second=0, microsecond=0) - - # Check if this is a custom dummy EPG with regex patterns - if epg_source and epg_source.source_type == 'dummy' and epg_source.custom_properties: - custom_programs = generate_custom_dummy_programs( - channel_id, channel_name, now, num_days, - epg_source.custom_properties, - export_lookback=export_lookback, - export_cutoff=export_cutoff, - ) - if custom_programs is not None: - return custom_programs - - logger.info(f"Custom pattern didn't match for '{channel_name}', checking for custom fallback templates") - - custom_props = epg_source.custom_properties - fallback_title = custom_props.get('fallback_title_template', '').strip() - fallback_description = custom_props.get('fallback_description_template', '').strip() - - if fallback_title or fallback_description: - logger.info(f"Using custom fallback templates for '{channel_name}'") - return generate_fallback_programs( - channel_id, channel_name, now, num_days, - program_length_hours, fallback_title, fallback_description - ) - logger.info(f"No custom fallback templates found, using default dummy EPG") - - # Default humorous program descriptions based on time of day - time_descriptions = { - (0, 4): [ - f"Late Night with {channel_name} - Where insomniacs unite!", - f"The 'Why Am I Still Awake?' Show on {channel_name}", - f"Counting Sheep - A {channel_name} production for the sleepless", - ], - (4, 8): [ - f"Dawn Patrol - Rise and shine with {channel_name}!", - f"Early Bird Special - Coffee not included", - f"Morning Zombies - Before coffee viewing on {channel_name}", - ], - (8, 12): [ - f"Mid-Morning Meetings - Pretend you're paying attention while watching {channel_name}", - f"The 'I Should Be Working' Hour on {channel_name}", - f"Productivity Killer - {channel_name}'s daytime programming", - ], - (12, 16): [ - f"Lunchtime Laziness with {channel_name}", - f"The Afternoon Slump - Brought to you by {channel_name}", - f"Post-Lunch Food Coma Theater on {channel_name}", - ], - (16, 20): [ - f"Rush Hour - {channel_name}'s alternative to traffic", - f"The 'What's For Dinner?' Debate on {channel_name}", - f"Evening Escapism - {channel_name}'s remedy for reality", - ], - (20, 24): [ - f"Prime Time Placeholder - {channel_name}'s finest not-programming", - f"The 'Netflix Was Too Complicated' Show on {channel_name}", - f"Family Argument Avoider - Courtesy of {channel_name}", - ], - } - - programs = [] - - # Create programs for each day - for day in range(num_days): - day_start = now + timedelta(days=day) - - # Create programs with specified length throughout the day - for hour_offset in range(0, 24, program_length_hours): - # Calculate program start and end times - start_time = day_start + timedelta(hours=hour_offset) - end_time = start_time + timedelta(hours=program_length_hours) - - # Get the hour for selecting a description - hour = start_time.hour - - # Find the appropriate time slot for description - for time_range, descriptions in time_descriptions.items(): - start_range, end_range = time_range - if start_range <= hour < end_range: - # Pick a description using the sum of the hour and day as seed - # This makes it somewhat random but consistent for the same timeslot - description = descriptions[(hour + day) % len(descriptions)] - break - else: - # Fallback description if somehow no range matches - description = f"Placeholder program for {channel_name} - EPG data went on vacation" - - programs.append({ - "channel_id": channel_id, - "start_time": start_time, - "end_time": end_time, - "title": channel_name, - "description": description, - }) - - return programs - - -def generate_custom_dummy_programs( - channel_id, - channel_name, - now, - num_days, - custom_properties, - export_lookback=None, - export_cutoff=None, -): - """ - Generate programs using custom dummy EPG regex patterns. - - Extracts information from channel title using regex patterns and generates - programs based on the extracted data. - - TIMEZONE HANDLING: - ------------------ - The timezone parameter specifies the timezone of the event times in your channel - titles using standard timezone names (e.g., 'US/Eastern', 'US/Pacific', 'Europe/London'). - DST (Daylight Saving Time) is handled automatically by pytz. - - Examples: - - Channel: "NHL 01: Bruins VS Maple Leafs @ 8:00PM ET" - - Set timezone = "US/Eastern" - - In October (DST): 8:00PM EDT → 12:00AM UTC (automatically uses UTC-4) - - In January (no DST): 8:00PM EST → 1:00AM UTC (automatically uses UTC-5) - - Args: - channel_id: Channel ID for the programs - channel_name: Channel title to parse - now: Current datetime (in UTC) - num_days: Number of days to generate programs for - custom_properties: Dict with title_pattern, time_pattern, templates, etc. - - timezone: Timezone name (e.g., 'US/Eastern') - - Returns: - List of program dictionaries with start_time/end_time in UTC - """ - import pytz - - logger.info(f"Generating custom dummy programs for channel: {channel_name}") - - # Extract patterns from custom properties - title_pattern = custom_properties.get('title_pattern', '') - time_pattern = custom_properties.get('time_pattern', '') - date_pattern = custom_properties.get('date_pattern', '') - - # Get timezone name (e.g., 'US/Eastern', 'US/Pacific', 'Europe/London') - timezone_value = custom_properties.get('timezone', 'UTC') - output_timezone_value = custom_properties.get('output_timezone', '') # Optional: display times in different timezone - program_duration = custom_properties.get('program_duration', 180) # Minutes - title_template = custom_properties.get('title_template', '') - subtitle_template = custom_properties.get('subtitle_template', '') - description_template = custom_properties.get('description_template', '') - - # Templates for upcoming/ended programs - upcoming_title_template = custom_properties.get('upcoming_title_template', '') - upcoming_description_template = custom_properties.get('upcoming_description_template', '') - ended_title_template = custom_properties.get('ended_title_template', '') - ended_description_template = custom_properties.get('ended_description_template', '') - - # Image URL templates - channel_logo_url_template = custom_properties.get('channel_logo_url', '') - program_poster_url_template = custom_properties.get('program_poster_url', '') - - # EPG metadata options - category_string = custom_properties.get('category', '') - # Split comma-separated categories and strip whitespace, filter out empty strings - categories = [cat.strip() for cat in category_string.split(',') if cat.strip()] if category_string else [] - include_date = custom_properties.get('include_date', True) - include_live = custom_properties.get('include_live', False) - include_new = custom_properties.get('include_new', False) - - # Parse timezone name - try: - source_tz = pytz.timezone(timezone_value) - logger.debug(f"Using timezone: {timezone_value} (DST will be handled automatically)") - except pytz.exceptions.UnknownTimeZoneError: - logger.warning(f"Unknown timezone: {timezone_value}, defaulting to UTC") - source_tz = pytz.utc - - # Parse output timezone if provided (for display purposes) - output_tz = None - if output_timezone_value: - try: - output_tz = pytz.timezone(output_timezone_value) - logger.debug(f"Using output timezone for display: {output_timezone_value}") - except pytz.exceptions.UnknownTimeZoneError: - logger.warning(f"Unknown output timezone: {output_timezone_value}, will use source timezone") - output_tz = None - - if not title_pattern: - logger.warning(f"No title_pattern in custom_properties, falling back to default") - return None - - logger.debug(f"Title pattern from DB: {repr(title_pattern)}") - - # Convert PCRE/JavaScript named groups (?) to Python format (?P) - # This handles patterns created with JavaScript regex syntax - # Use negative lookahead to avoid matching lookbehind (?<=) and negative lookbehind (?]+)>', r'(?P<\1>', title_pattern) - logger.debug(f"Converted title pattern: {repr(title_pattern)}") - - # Compile regex patterns using the enhanced regex module - # (supports variable-width lookbehinds like JavaScript) - try: - title_regex = regex.compile(title_pattern) - except Exception as e: - logger.error(f"Invalid title regex pattern after conversion: {e}") - logger.error(f"Pattern was: {repr(title_pattern)}") - return None - - time_regex = None - if time_pattern: - # Convert PCRE/JavaScript named groups to Python format - # Use negative lookahead to avoid matching lookbehind (?<=) and negative lookbehind (?]+)>', r'(?P<\1>', time_pattern) - logger.debug(f"Converted time pattern: {repr(time_pattern)}") - try: - time_regex = regex.compile(time_pattern) - except Exception as e: - logger.warning(f"Invalid time regex pattern after conversion: {e}") - logger.warning(f"Pattern was: {repr(time_pattern)}") - - # Compile date regex if provided - date_regex = None - if date_pattern: - # Convert PCRE/JavaScript named groups to Python format - # Use negative lookahead to avoid matching lookbehind (?<=) and negative lookbehind (?]+)>', r'(?P<\1>', date_pattern) - logger.debug(f"Converted date pattern: {repr(date_pattern)}") - try: - date_regex = regex.compile(date_pattern) - except Exception as e: - logger.warning(f"Invalid date regex pattern after conversion: {e}") - logger.warning(f"Pattern was: {repr(date_pattern)}") - - # Try to match the channel name with the title pattern - # Use search() instead of match() to match JavaScript behavior where .match() searches anywhere in the string - title_match = title_regex.search(channel_name) - if not title_match: - logger.debug(f"Channel name '{channel_name}' doesn't match title pattern") - return None - - groups = title_match.groupdict() - logger.debug(f"Title pattern matched. Groups: {groups}") - - # Helper function to format template with matched groups - def format_template(template, groups, url_encode=False): - """Replace {groupname} placeholders with matched group values - - Args: - template: Template string with {groupname} placeholders - groups: Dict of group names to values - url_encode: If True, URL encode the group values for safe use in URLs - """ - if not template: - return '' - result = template - for key, value in groups.items(): - if url_encode and value: - # URL encode the value to handle spaces and special characters - from urllib.parse import quote - encoded_value = quote(str(value), safe='') - result = result.replace(f'{{{key}}}', encoded_value) - else: - result = result.replace(f'{{{key}}}', str(value) if value else '') - return result - - # Extract time from title if time pattern exists - time_info = None - time_groups = {} - if time_regex: - time_match = time_regex.search(channel_name) - if time_match: - time_groups = time_match.groupdict() - try: - hour = int(time_groups.get('hour')) - # Handle optional minute group - could be None if not captured - minute_value = time_groups.get('minute') - minute = int(minute_value) if minute_value is not None else 0 - ampm = time_groups.get('ampm') - ampm = ampm.lower() if ampm else None - - # Determine if this is 12-hour or 24-hour format - if ampm in ('am', 'pm'): - # 12-hour format: convert to 24-hour - if ampm == 'pm' and hour != 12: - hour += 12 - elif ampm == 'am' and hour == 12: - hour = 0 - logger.debug(f"Extracted time (12-hour): {hour}:{minute:02d} {ampm}") - else: - # 24-hour format: hour is already in 24-hour format - # Validate that it's actually a 24-hour time (0-23) - if hour > 23: - logger.warning(f"Invalid 24-hour time: {hour}. Must be 0-23.") - hour = hour % 24 # Wrap around just in case - logger.debug(f"Extracted time (24-hour): {hour}:{minute:02d}") - - time_info = {'hour': hour, 'minute': minute} - except (ValueError, TypeError) as e: - logger.warning(f"Error parsing time: {e}") - - # Extract date from title if date pattern exists - date_info = None - date_groups = {} - if date_regex: - date_match = date_regex.search(channel_name) - if date_match: - date_groups = date_match.groupdict() - try: - # Support various date group names: month, day, year - month_str = date_groups.get('month', '') - day_str = date_groups.get('day', '') - year_str = date_groups.get('year', '') - - # Parse day - default to current day if empty or invalid - day = int(day_str) if day_str else now.day - - # Parse year - default to current year if empty or invalid (matches frontend behavior) - year = int(year_str) if year_str else now.year - - # Parse month - can be numeric (1-12) or text (Jan, January, etc.) - month = None - if month_str: - if month_str.isdigit(): - month = int(month_str) - else: - # Try to parse text month names - import calendar - month_str_lower = month_str.lower() - # Check full month names - for i, month_name in enumerate(calendar.month_name): - if month_name.lower() == month_str_lower: - month = i - break - # Check abbreviated month names if not found - if month is None: - for i, month_abbr in enumerate(calendar.month_abbr): - if month_abbr.lower() == month_str_lower: - month = i - break - - # Default to current month if not extracted or invalid - if month is None: - month = now.month - - if month and 1 <= month <= 12 and 1 <= day <= 31: - date_info = {'year': year, 'month': month, 'day': day} - logger.debug(f"Extracted date: {year}-{month:02d}-{day:02d}") - else: - logger.warning(f"Invalid date values: month={month}, day={day}, year={year}") - except (ValueError, TypeError) as e: - logger.warning(f"Error parsing date: {e}") - - # Merge title groups, time groups, and date groups for template formatting - all_groups = {**groups, **time_groups, **date_groups} - - # Add normalized versions of all groups for cleaner URLs - # These remove all non-alphanumeric characters and convert to lowercase - for key, value in list(all_groups.items()): - if value: - # Remove all non-alphanumeric characters (except spaces temporarily) - # then replace spaces with nothing, and convert to lowercase - normalized = regex.sub(r'[^a-zA-Z0-9\s]', '', str(value)) - normalized = regex.sub(r'\s+', '', normalized).lower() - all_groups[f'{key}_normalize'] = normalized - - # Format channel logo URL if template provided (with URL encoding) - channel_logo_url = None - if channel_logo_url_template: - channel_logo_url = format_template(channel_logo_url_template, all_groups, url_encode=True) - logger.debug(f"Formatted channel logo URL: {channel_logo_url}") - - # Format program poster URL if template provided (with URL encoding) - program_poster_url = None - if program_poster_url_template: - program_poster_url = format_template(program_poster_url_template, all_groups, url_encode=True) - logger.debug(f"Formatted program poster URL: {program_poster_url}") - - # Add formatted time strings for better display (handles minutes intelligently) - if time_info: - hour_24 = time_info['hour'] - minute = time_info['minute'] - - # Determine the base date to use for placeholders - # If date was extracted, use it; otherwise use current date - if date_info: - base_date = datetime(date_info['year'], date_info['month'], date_info['day']) - else: - base_date = datetime.now() - - # If output_timezone is specified, convert the display time to that timezone - if output_tz: - # Create a datetime in the source timezone using the base date - temp_date = source_tz.localize(base_date.replace(hour=hour_24, minute=minute, second=0, microsecond=0)) - # Convert to output timezone - temp_date_output = temp_date.astimezone(output_tz) - # Extract converted hour and minute for display - hour_24 = temp_date_output.hour - minute = temp_date_output.minute - logger.debug(f"Converted display time from {source_tz} to {output_tz}: {hour_24}:{minute:02d}") - - # Add date placeholders based on the OUTPUT timezone - # This ensures {date}, {month}, {day}, {year} reflect the converted timezone - all_groups['date'] = temp_date_output.strftime('%Y-%m-%d') - all_groups['month'] = str(temp_date_output.month) - all_groups['day'] = str(temp_date_output.day) - all_groups['year'] = str(temp_date_output.year) - logger.debug(f"Converted date placeholders to {output_tz}: {all_groups['date']}") - else: - # No output timezone conversion - use source timezone for date - # Create temp date to get proper date in source timezone using the base date - temp_date_source = source_tz.localize(base_date.replace(hour=hour_24, minute=minute, second=0, microsecond=0)) - all_groups['date'] = temp_date_source.strftime('%Y-%m-%d') - all_groups['month'] = str(temp_date_source.month) - all_groups['day'] = str(temp_date_source.day) - all_groups['year'] = str(temp_date_source.year) - - # Format 24-hour start time string - only include minutes if non-zero - if minute > 0: - all_groups['starttime24'] = f"{hour_24}:{minute:02d}" - else: - all_groups['starttime24'] = f"{hour_24:02d}:00" - - # Convert 24-hour to 12-hour format for {starttime} placeholder - # Note: hour_24 is ALWAYS in 24-hour format at this point (converted earlier if needed) - ampm = 'AM' if hour_24 < 12 else 'PM' - hour_12 = hour_24 - if hour_24 == 0: - hour_12 = 12 - elif hour_24 > 12: - hour_12 = hour_24 - 12 - - # Format 12-hour start time string - only include minutes if non-zero - if minute > 0: - all_groups['starttime'] = f"{hour_12}:{minute:02d} {ampm}" - else: - all_groups['starttime'] = f"{hour_12} {ampm}" - - # Format long version that always includes minutes (e.g., "9:00 PM" instead of "9 PM") - all_groups['starttime_long'] = f"{hour_12}:{minute:02d} {ampm}" - - # Calculate end time based on program duration - # Create a datetime for calculations - temp_start = datetime.now(source_tz).replace(hour=hour_24, minute=minute, second=0, microsecond=0) - temp_end = temp_start + timedelta(minutes=program_duration) - - # Extract end time components (already in correct timezone if output_tz was applied above) - end_hour_24 = temp_end.hour - end_minute = temp_end.minute - - # Format 24-hour end time string - only include minutes if non-zero - if end_minute > 0: - all_groups['endtime24'] = f"{end_hour_24}:{end_minute:02d}" - else: - all_groups['endtime24'] = f"{end_hour_24:02d}:00" - - # Convert 24-hour to 12-hour format for {endtime} placeholder - end_ampm = 'AM' if end_hour_24 < 12 else 'PM' - end_hour_12 = end_hour_24 - if end_hour_24 == 0: - end_hour_12 = 12 - elif end_hour_24 > 12: - end_hour_12 = end_hour_24 - 12 - - # Format 12-hour end time string - only include minutes if non-zero - if end_minute > 0: - all_groups['endtime'] = f"{end_hour_12}:{end_minute:02d} {end_ampm}" - else: - all_groups['endtime'] = f"{end_hour_12} {end_ampm}" - - # Format long version that always includes minutes (e.g., "9:00 PM" instead of "9 PM") - all_groups['endtime_long'] = f"{end_hour_12}:{end_minute:02d} {end_ampm}" - - # Generate programs - programs = [] - - # If we have extracted time AND date, the event happens on a SPECIFIC date - # If we have time but NO date, generate for multiple days (existing behavior) - # All other days and times show "Upcoming" before or "Ended" after - event_happened = False - - # Determine how many iterations we need - if date_info and time_info: - # Specific date extracted - only generate for that one date - iterations = 1 - logger.debug(f"Date extracted, generating single event for specific date") - else: - # No specific date - use num_days (existing behavior) - iterations = num_days - - for day in range(iterations): - event_overlaps_window = True - if date_info and time_info: - current_date = datetime( - date_info['year'], - date_info['month'], - date_info['day'], - ).date() - event_start_naive = datetime.combine( - current_date, - datetime.min.time().replace( - hour=time_info['hour'], - minute=time_info['minute'], - ), - ) - try: - event_start_utc = source_tz.localize(event_start_naive).astimezone(pytz.utc) - except Exception as e: - logger.error(f"Error localizing time to {source_tz}: {e}") - event_start_utc = django_timezone.make_aware(event_start_naive, pytz.utc) - event_end_utc = event_start_utc + timedelta(minutes=program_duration) - - lookback = export_lookback if export_lookback is not None else now - event_overlaps_window = _programme_overlaps_export_window( - event_start_utc, event_end_utc, lookback, export_cutoff - ) - if not event_overlaps_window: - logger.debug( - "Custom dummy event outside export window; filling window only: %s", - channel_name, - ) - event_happened = event_end_utc < lookback - day_start = _ceil_to_half_hour(lookback) - if export_cutoff is not None: - day_end = export_cutoff - else: - day_end = now + timedelta(days=num_days if num_days > 0 else 3) - else: - day_start = source_tz.localize( - datetime.combine(current_date, datetime.min.time()) - ).astimezone(pytz.utc) - day_end = day_start + timedelta(days=1) - if export_lookback is not None: - day_start = max(day_start, export_lookback) - if export_cutoff is not None: - day_end = min(day_end, export_cutoff) - else: - day_start = now + timedelta(days=day) - day_end = day_start + timedelta(days=1) - if export_lookback is not None: - day_start = max(day_start, export_lookback) - if export_cutoff is not None: - day_end = min(day_end, export_cutoff) - - if day_start >= day_end: - continue - - if time_info: - if not date_info: - now_in_source_tz = now.astimezone(source_tz) - current_date = (now_in_source_tz + timedelta(days=day)).date() - logger.debug(f"No date extracted, using day offset in {source_tz}: {current_date}") - - event_start_naive = datetime.combine( - current_date, - datetime.min.time().replace( - hour=time_info['hour'], - minute=time_info['minute'], - ), - ) - try: - event_start_local = source_tz.localize(event_start_naive) - event_start_utc = event_start_local.astimezone(pytz.utc) - logger.debug(f"Converted {event_start_local} to UTC: {event_start_utc}") - except Exception as e: - logger.error(f"Error localizing time to {source_tz}: {e}") - event_start_utc = django_timezone.make_aware(event_start_naive, pytz.utc) - - event_end_utc = event_start_utc + timedelta(minutes=program_duration) - - lookback = export_lookback if export_lookback is not None else now - if not _programme_overlaps_export_window( - event_start_utc, event_end_utc, lookback, export_cutoff - ): - continue - else: - logger.debug(f"Using extracted date: {current_date}") - - # Pre-generate the main event title and description for reuse - if title_template: - main_event_title = format_template(title_template, all_groups) - else: - title_parts = [] - if 'league' in all_groups and all_groups['league']: - title_parts.append(all_groups['league']) - if 'team1' in all_groups and 'team2' in all_groups: - title_parts.append(f"{all_groups['team1']} vs {all_groups['team2']}") - elif 'title' in all_groups and all_groups['title']: - title_parts.append(all_groups['title']) - main_event_title = ' - '.join(title_parts) if title_parts else channel_name - - if subtitle_template: - main_event_subtitle = format_template(subtitle_template, all_groups) - else: - main_event_subtitle = None - - if description_template: - main_event_description = format_template(description_template, all_groups) - else: - main_event_description = main_event_title - - - - # Determine if this day is before, during, or after the event - # Event only happens on day 0 (first day) when it falls inside the window - is_event_day = (day == 0) and event_overlaps_window - - if is_event_day and not event_happened: - current_time = day_start - - while current_time < event_start_utc and current_time < day_end: - program_start_utc = current_time - program_end_utc = min(current_time + timedelta(minutes=program_duration), event_start_utc) - - # Use custom upcoming templates if provided, otherwise use defaults - if upcoming_title_template: - upcoming_title = format_template(upcoming_title_template, all_groups) - else: - upcoming_title = main_event_title - - if upcoming_description_template: - upcoming_description = format_template(upcoming_description_template, all_groups) - else: - upcoming_description = f"Upcoming: {main_event_description}" - - # Build custom_properties for upcoming programs (only date, no category/live) - program_custom_properties = {} - - # Add date if requested (YYYY-MM-DD format from start time in event timezone) - if include_date: - # Convert UTC time to event timezone for date calculation - local_time = program_start_utc.astimezone(source_tz) - date_str = local_time.strftime('%Y-%m-%d') - program_custom_properties['date'] = date_str - - # Add program poster URL if provided - if program_poster_url: - program_custom_properties['icon'] = program_poster_url - - programs.append({ - "channel_id": channel_id, - "start_time": program_start_utc, - "end_time": program_end_utc, - "title": upcoming_title, - "sub_title": None, # No subtitle for filler programs - "description": upcoming_description, - "custom_properties": program_custom_properties, - "channel_logo_url": channel_logo_url, # Pass channel logo for EPG generation - }) - - current_time += timedelta(minutes=program_duration) - - # Add the MAIN EVENT at the extracted time - # Build custom_properties for main event (includes category and live) - main_event_custom_properties = {} - - # Add categories if provided - if categories: - main_event_custom_properties['categories'] = categories - - # Add date if requested (YYYY-MM-DD format from start time in event timezone) - if include_date: - # Convert UTC time to event timezone for date calculation - local_time = event_start_utc.astimezone(source_tz) - date_str = local_time.strftime('%Y-%m-%d') - main_event_custom_properties['date'] = date_str - - # Add live flag if requested - if include_live: - main_event_custom_properties['live'] = True - - # Add new flag if requested - if include_new: - main_event_custom_properties['new'] = True - - # Add program poster URL if provided - if program_poster_url: - main_event_custom_properties['icon'] = program_poster_url - - programs.append({ - "channel_id": channel_id, - "start_time": event_start_utc, - "end_time": event_end_utc, - "title": main_event_title, - "sub_title": main_event_subtitle, - "description": main_event_description, - "custom_properties": main_event_custom_properties, - "channel_logo_url": channel_logo_url, # Pass channel logo for EPG generation - }) - - event_happened = True - - # Fill programs AFTER the event until end of export day window - current_time = max(event_end_utc, day_start) - - while current_time < day_end: - program_start_utc = current_time - program_end_utc = min(current_time + timedelta(minutes=program_duration), day_end) - - # Use custom ended templates if provided, otherwise use defaults - if ended_title_template: - ended_title = format_template(ended_title_template, all_groups) - else: - ended_title = main_event_title - - if ended_description_template: - ended_description = format_template(ended_description_template, all_groups) - else: - ended_description = f"Ended: {main_event_description}" - - # Build custom_properties for ended programs (only date, no category/live) - program_custom_properties = {} - - # Add date if requested (YYYY-MM-DD format from start time in event timezone) - if include_date: - # Convert UTC time to event timezone for date calculation - local_time = program_start_utc.astimezone(source_tz) - date_str = local_time.strftime('%Y-%m-%d') - program_custom_properties['date'] = date_str - - # Add program poster URL if provided - if program_poster_url: - program_custom_properties['icon'] = program_poster_url - - programs.append({ - "channel_id": channel_id, - "start_time": program_start_utc, - "end_time": program_end_utc, - "title": ended_title, - "sub_title": None, # No subtitle for filler programs - "description": ended_description, - "custom_properties": program_custom_properties, - "channel_logo_url": channel_logo_url, # Pass channel logo for EPG generation - }) - - current_time += timedelta(minutes=program_duration) - else: - # This day is either before the event (future days) or after the event happened - # Fill entire day with appropriate message - current_time = day_start - - # If event already happened, all programs show "Ended" - # If event hasn't happened yet (shouldn't occur with day 0 logic), show "Upcoming" - is_ended = event_happened - - while current_time < day_end: - program_start_utc = current_time - program_end_utc = min(current_time + timedelta(minutes=program_duration), day_end) - - # Use custom templates based on whether event has ended or is upcoming - if is_ended: - if ended_title_template: - program_title = format_template(ended_title_template, all_groups) - else: - program_title = main_event_title - - if ended_description_template: - program_description = format_template(ended_description_template, all_groups) - else: - program_description = f"Ended: {main_event_description}" - else: - if upcoming_title_template: - program_title = format_template(upcoming_title_template, all_groups) - else: - program_title = main_event_title - - if upcoming_description_template: - program_description = format_template(upcoming_description_template, all_groups) - else: - program_description = f"Upcoming: {main_event_description}" - - # Build custom_properties (only date for upcoming/ended filler programs) - program_custom_properties = {} - - # Add date if requested (YYYY-MM-DD format from start time in event timezone) - if include_date: - # Convert UTC time to event timezone for date calculation - local_time = program_start_utc.astimezone(source_tz) - date_str = local_time.strftime('%Y-%m-%d') - program_custom_properties['date'] = date_str - - # Add program poster URL if provided - if program_poster_url: - program_custom_properties['icon'] = program_poster_url - - programs.append({ - "channel_id": channel_id, - "start_time": program_start_utc, - "end_time": program_end_utc, - "title": program_title, - "sub_title": None, # No subtitle for filler programs - "description": program_description, - "custom_properties": program_custom_properties, - "channel_logo_url": channel_logo_url, - }) - - current_time += timedelta(minutes=program_duration) - else: - # No extracted time - fill entire day with regular intervals - # day_start and day_end are already in UTC, so no conversion needed - programs_per_day = max(1, int(24 / (program_duration / 60))) - - for program_num in range(programs_per_day): - program_start_utc = day_start + timedelta(minutes=program_num * program_duration) - program_end_utc = program_start_utc + timedelta(minutes=program_duration) - - if title_template: - title = format_template(title_template, all_groups) - else: - title_parts = [] - if 'league' in all_groups and all_groups['league']: - title_parts.append(all_groups['league']) - if 'team1' in all_groups and 'team2' in all_groups: - title_parts.append(f"{all_groups['team1']} vs {all_groups['team2']}") - elif 'title' in all_groups and all_groups['title']: - title_parts.append(all_groups['title']) - title = ' - '.join(title_parts) if title_parts else channel_name - - if subtitle_template: - subtitle = format_template(subtitle_template, all_groups) - else: - subtitle = None - - if description_template: - description = format_template(description_template, all_groups) - else: - description = title - - # Build custom_properties for this program - program_custom_properties = {} - - # Add categories if provided - if categories: - program_custom_properties['categories'] = categories - - # Add date if requested (YYYY-MM-DD format from start time in event timezone) - if include_date: - # Convert UTC time to event timezone for date calculation - local_time = program_start_utc.astimezone(source_tz) - date_str = local_time.strftime('%Y-%m-%d') - program_custom_properties['date'] = date_str - - # Add live flag if requested - if include_live: - program_custom_properties['live'] = True - - # Add new flag if requested - if include_new: - program_custom_properties['new'] = True - - # Add program poster URL if provided - if program_poster_url: - program_custom_properties['icon'] = program_poster_url - - programs.append({ - "channel_id": channel_id, - "start_time": program_start_utc, - "end_time": program_end_utc, - "title": title, - "sub_title": subtitle, - "description": description, - "custom_properties": program_custom_properties, - "channel_logo_url": channel_logo_url, # Pass channel logo for EPG generation - }) - - logger.info(f"Generated {len(programs)} custom dummy programs for {channel_name}") - return programs - - -def generate_dummy_epg( - channel_id, channel_name, xml_lines=None, num_days=1, program_length_hours=4 -): - """ - Generate dummy EPG programs for channels without EPG data. - Creates program blocks for a specified number of days. - - Args: - channel_id: The channel ID to use in the program entries - channel_name: The name of the channel to use in program titles - xml_lines: Optional list to append lines to, otherwise returns new list - num_days: Number of days to generate EPG data for (default: 1) - program_length_hours: Length of each program block in hours (default: 4) - - Returns: - List of XML lines for the dummy EPG entries - """ - if xml_lines is None: - xml_lines = [] - - for program in generate_dummy_programs(channel_id, channel_name, num_days=1, program_length_hours=4): - # Format times in XMLTV format - start_str = program['start_time'].strftime("%Y%m%d%H%M%S %z") - stop_str = program['end_time'].strftime("%Y%m%d%H%M%S %z") - - # Create program entry with escaped channel name - xml_lines.append( - f' ' - ) - xml_lines.append(f" {html.escape(program['title'])}") - - # Add subtitle if available - if program.get('sub_title'): - xml_lines.append(f" {html.escape(program['sub_title'])}") - - xml_lines.append(f" {html.escape(program['description'])}") - - # Add custom_properties if present - custom_data = program.get('custom_properties', {}) - - # Categories - if 'categories' in custom_data: - for cat in custom_data['categories']: - xml_lines.append(f" {html.escape(cat)}") - - # Date tag - if 'date' in custom_data: - xml_lines.append(f" {html.escape(custom_data['date'])}") - - # Live tag - if custom_data.get('live', False): - xml_lines.append(f" ") - - # New tag - if custom_data.get('new', False): - xml_lines.append(f" ") - - xml_lines.append(f" ") - - return xml_lines - - -def generate_epg(request, profile_name=None, user=None): - """ - Dynamically generate an XMLTV (EPG) file using a streaming response. - Since the EPG data is stored independently of Channels, we group programmes - by their associated EPGData record. - """ - user_custom = (user.custom_properties or {}) if user else {} - try: - num_days = int(request.GET.get('days', user_custom.get('epg_days', 0))) - num_days = max(0, min(num_days, 365)) - except (ValueError, TypeError): - num_days = 0 - try: - prev_days = int(request.GET.get('prev_days', user_custom.get('epg_prev_days', 0))) - prev_days = max(0, min(prev_days, 30)) - except (ValueError, TypeError): - prev_days = 0 - use_cached_logos = request.GET.get('cachedlogos', 'true').lower() != 'false' - tvg_id_source = request.GET.get('tvg_id_source', 'channel_number').lower() - cache_params = ( - f"{profile_name or 'all'}:{user.username if user else 'anonymous'}" - f":d={num_days}:p={prev_days}:logos={use_cached_logos}:tvgid={tvg_id_source}" - ) - content_cache_key = f"epg_content:{cache_params}" - - def epg_generator(): - """Generator function that yields EPG data with keep-alives during processing.""" - - yield '\n' - yield ( - '\n' - ) - - # Get channels based on user/profile - if user is not None: - if user.user_level < 10: - user_profile_count = user.channel_profiles.count() - - # If user has ALL profiles or NO profiles, give unrestricted access - if user_profile_count == 0: - # No profile filtering - user sees all channels based on user_level - filters = {"user_level__lte": user.user_level} - # Hide adult content if user preference is set - if (user.custom_properties or {}).get('hide_adult_content', False): - filters["is_adult"] = False - base_qs = Channel.objects.filter(**filters).select_related('logo', 'epg_data__epg_source') - else: - # User has specific limited profiles assigned - filters = { - "channelprofilemembership__enabled": True, - "user_level__lte": user.user_level, - "channelprofilemembership__channel_profile__in": user.channel_profiles.all() - } - # Hide adult content if user preference is set - if (user.custom_properties or {}).get('hide_adult_content', False): - filters["is_adult"] = False - base_qs = Channel.objects.filter(**filters).select_related('logo', 'epg_data__epg_source').distinct() - else: - base_qs = Channel.objects.filter(user_level__lte=user.user_level).select_related('logo', 'epg_data__epg_source') - else: - if profile_name is not None: - try: - channel_profile = ChannelProfile.objects.get(name=profile_name) - except ChannelProfile.DoesNotExist: - logger.warning("Requested channel profile (%s) during epg generation does not exist", profile_name) - raise Http404(f"Channel profile '{profile_name}' not found") - base_qs = Channel.objects.filter( - channelprofilemembership__channel_profile=channel_profile, - channelprofilemembership__enabled=True, - ).select_related('logo', 'epg_data__epg_source') - else: - base_qs = Channel.objects.all().select_related('logo', 'epg_data__epg_source') - - # Resolve effective values at SQL level and exclude hidden channels - # so output ordering/display honors user overrides. - from apps.channels.managers import with_effective_values - channels = list( - with_effective_values(base_qs, select_related_fks=True) - .exclude(hidden_from_output=True) - .order_by("effective_channel_number") - # programme_index is a multi-MB JSON byte-offset index that EPG - # generation never reads; defer it so it isn't fetched and JSON-parsed - # once per channel (was ~13s of the request on large guides). - .defer("epg_data__epg_source__programme_index") - .prefetch_related( - Prefetch('streams', queryset=Stream.objects.order_by('channelstream__order')) - ) - ) - channel_count = len(channels) - - # For dummy EPG, use either the specified value or default to 3 days - dummy_days = num_days if num_days > 0 else 3 - - # Calculate cutoff dates for EPG data filtering - now = django_timezone.now() - cutoff_date = now + timedelta(days=num_days) if num_days > 0 else None - lookback_cutoff = now - timedelta(days=prev_days) - - # Build collision-free channel number mapping for XC clients (if user is authenticated) - # XC clients require integer channel numbers, so we need to ensure no conflicts - channel_num_map = {} - if user is not None: - # This is an XC client - build collision-free mapping - used_numbers = set() - - # First pass: assign integers for channels that already have integer numbers - for channel in channels: - effective_num = channel.effective_channel_number - if effective_num is not None and effective_num == int(effective_num): - num = int(effective_num) - channel_num_map[channel.id] = num - used_numbers.add(num) - - # Second pass: assign integers for channels with float numbers - for channel in channels: - effective_num = channel.effective_channel_number - if effective_num is not None and effective_num != int(effective_num): - candidate = int(effective_num) - while candidate in used_numbers: - candidate += 1 - channel_num_map[channel.id] = candidate - used_numbers.add(candidate) - - # Host/port/scheme are constant per request; precompute logo URL prefix once. - _base_url = build_absolute_uri_with_port(request, "") - _sample_logo_path = reverse("api:channels:logo-cache", args=[0]) - _logo_prefix_raw, _, _logo_suffix_raw = _sample_logo_path.partition("/0/") - _logo_url_prefix = _base_url + _logo_prefix_raw + "/" - _logo_url_suffix = "/" + _logo_suffix_raw - - dummy_program_list = [] - real_epg_map = {} - channel_xml_batch = [] - - for channel in channels: - effective_name = channel.effective_name - effective_epg_data = channel.effective_epg_data_obj - effective_epg_data_id = channel.effective_epg_data_id - effective_logo = channel.effective_logo_obj - effective_number = channel.effective_channel_number - - # user is set only for XC clients, which require integer channel numbers - if user is not None: - formatted_channel_number = channel_num_map[channel.id] - else: - formatted_channel_number = format_channel_number(effective_number) - - # Determine the channel ID based on the selected source - if tvg_id_source == 'tvg_id' and channel.effective_tvg_id: - channel_id = channel.effective_tvg_id - elif tvg_id_source == 'gracenote' and channel.effective_tvc_guide_stationid: - channel_id = channel.effective_tvc_guide_stationid - else: - channel_id = str(formatted_channel_number) if formatted_channel_number != "" else str(channel.id) - - tvg_logo = "" - - # Check if this is a custom dummy EPG with channel logo URL template - if effective_epg_data and effective_epg_data.epg_source and effective_epg_data.epg_source.source_type == 'dummy': - epg_source = effective_epg_data.epg_source - if epg_source.custom_properties: - custom_props = epg_source.custom_properties - channel_logo_url_template = custom_props.get('channel_logo_url', '') - - if channel_logo_url_template: - pattern_match_name, _ = _pattern_match_name_from_custom_props( - channel, effective_name, custom_props - ) - - # Try to extract groups from the channel/stream name and build the logo URL - title_pattern = custom_props.get('title_pattern', '') - if title_pattern: - try: - # Convert PCRE/JavaScript named groups to Python format - title_pattern = regex.sub(r'\(\?<(?![=!])([^>]+)>', r'(?P<\1>', title_pattern) - title_regex = regex.compile(title_pattern) - title_match = title_regex.search(pattern_match_name) - - if title_match: - groups = title_match.groupdict() - - # Add normalized versions of all groups for cleaner URLs - for key, value in list(groups.items()): - if value: - # Remove all non-alphanumeric characters and convert to lowercase - normalized = regex.sub(r'[^a-zA-Z0-9\s]', '', str(value)) - normalized = regex.sub(r'\s+', '', normalized).lower() - groups[f'{key}_normalize'] = normalized - - # Format the logo URL template with the matched groups (with URL encoding) - from urllib.parse import quote - for key, value in groups.items(): - if value: - encoded_value = quote(str(value), safe='') - channel_logo_url_template = channel_logo_url_template.replace(f'{{{key}}}', encoded_value) - else: - channel_logo_url_template = channel_logo_url_template.replace(f'{{{key}}}', '') - tvg_logo = channel_logo_url_template - logger.debug(f"Built channel logo URL from template: {tvg_logo}") - except Exception as e: - logger.warning(f"Failed to build channel logo URL for {effective_name}: {e}") - - # If no custom dummy logo, use regular logo logic - if not tvg_logo and effective_logo: - if use_cached_logos: - tvg_logo = f"{_logo_url_prefix}{effective_logo.id}{_logo_url_suffix}" - else: - # Use direct URL if available, otherwise fall back to cached version - direct_logo = effective_logo.url if effective_logo.url.startswith(('http://', 'https://')) else None - if direct_logo: - tvg_logo = direct_logo - else: - tvg_logo = f"{_logo_url_prefix}{effective_logo.id}{_logo_url_suffix}" - channel_xml_batch.append(f' ') - channel_xml_batch.append(f' {html.escape(effective_name)}') - channel_xml_batch.append(f' ') - channel_xml_batch.append(" ") - - if len(channel_xml_batch) >= _EPG_CHANNEL_XML_BATCH_SIZE * 4: - yield '\n'.join(channel_xml_batch) + '\n' - channel_xml_batch = [] - - pattern_match_name = effective_name - if effective_epg_data and effective_epg_data.epg_source: - epg_source = effective_epg_data.epg_source - if epg_source.custom_properties: - custom_props = epg_source.custom_properties - pattern_match_name, stream_lookup_failed = _pattern_match_name_from_custom_props( - channel, effective_name, custom_props - ) - if ( - custom_props.get('name_source') == 'stream' - and not stream_lookup_failed - and pattern_match_name != effective_name - ): - stream_index = custom_props.get('stream_index', 1) - 1 - logger.debug( - f"Using stream name for parsing: {pattern_match_name} " - f"(stream index: {stream_index})" - ) - elif stream_lookup_failed: - stream_index = custom_props.get('stream_index', 1) - 1 - logger.warning( - f"Stream index {stream_index} not found for channel " - f"{effective_name}, falling back to channel name" - ) - - if not effective_epg_data: - dummy_program_list.append((channel_id, pattern_match_name, None)) - elif effective_epg_data.epg_source and effective_epg_data.epg_source.source_type == 'dummy': - dummy_program_list.append((channel_id, pattern_match_name, effective_epg_data.epg_source)) - else: - real_epg_map.setdefault(effective_epg_data_id, []).append(channel_id) - - if channel_xml_batch: - yield '\n'.join(channel_xml_batch) + '\n' - - del channels - del channel_num_map - - batch_size = _EPG_PROGRAM_YIELD_BATCH_SIZE - - all_epg_ids = list(real_epg_map.keys()) - if all_epg_ids: - if num_days > 0: - programs_qs = ProgramData.objects.filter( - epg_id__in=all_epg_ids, - end_time__gte=lookback_cutoff, - start_time__lt=cutoff_date, - ) - else: - programs_qs = ProgramData.objects.filter( - epg_id__in=all_epg_ids, - end_time__gte=lookback_cutoff, - ) - - programs_base_qs = programs_qs.order_by('epg_id', 'id').values( - 'id', 'epg_id', 'start_time', 'end_time', 'title', 'sub_title', - 'description', 'custom_properties', - ) - - current_epg_id = None - channel_ids_for_epg = None - pending = [] - program_batch = [] - chunk_size = _EPG_PROGRAM_DB_CHUNK_SIZE - last_epg_id = 0 - last_id = 0 - _poster_url_base = build_absolute_uri_with_port(request, "/api/epg/programs/") - - def flush_pending(): - nonlocal program_batch, pending - if not pending: - return - pending.sort(key=lambda row: (row[0], row[1])) - escaped_primary = ( - html.escape(channel_ids_for_epg[0]) - if len(channel_ids_for_epg) > 1 else None - ) - for _, _, xml_text in pending: - program_batch.append(xml_text) - if escaped_primary: - for cid in channel_ids_for_epg[1:]: - program_batch.append(xml_text.replace( - f'channel="{escaped_primary}"', - f'channel="{html.escape(cid)}"', - 1, - )) - if len(program_batch) >= batch_size: - yield '\n'.join(program_batch) + '\n' - program_batch = [] - pending.clear() - - while True: - program_chunk = list( - programs_base_qs.filter(epg_id__gte=last_epg_id) - .exclude(epg_id=last_epg_id, id__lte=last_id)[:chunk_size] - ) - if not program_chunk: - break - - last_row = program_chunk[-1] - last_epg_id = last_row['epg_id'] - last_id = last_row['id'] - - for prog in program_chunk: - epg_id = prog['epg_id'] - - if epg_id != current_epg_id: - yield from flush_pending() - current_epg_id = epg_id - channel_ids_for_epg = real_epg_map[epg_id] - - primary_cid = channel_ids_for_epg[0] - # DB datetimes are UTC (USE_TZ=True, TIME_ZONE=UTC); format - # directly instead of strftime("%Y%m%d%H%M%S %z"), which is - # ~10x slower and dominates XML build over 750k rows. - st = prog['start_time'] - et = prog['end_time'] - start_str = f"{st.year:04d}{st.month:02d}{st.day:02d}{st.hour:02d}{st.minute:02d}{st.second:02d} +0000" - stop_str = f"{et.year:04d}{et.month:02d}{et.day:02d}{et.hour:02d}{et.minute:02d}{et.second:02d} +0000" - - program_xml = [f' '] - program_xml.append(f' {html.escape(prog["title"])}') - - if prog['sub_title']: - program_xml.append(f" {html.escape(prog['sub_title'])}") - - if prog['description']: - program_xml.append(f" {html.escape(prog['description'])}") - - custom_data = prog['custom_properties'] or {} - if custom_data: - - if "categories" in custom_data and custom_data["categories"]: - for category in custom_data["categories"]: - program_xml.append(f" {html.escape(category)}") - - if "keywords" in custom_data and custom_data["keywords"]: - for keyword in custom_data["keywords"]: - program_xml.append(f" {html.escape(keyword)}") - - # onscreen_episode takes priority over episode for the onscreen system - if "onscreen_episode" in custom_data: - program_xml.append(f' {html.escape(custom_data["onscreen_episode"])}') - elif "episode" in custom_data: - program_xml.append(f' E{custom_data["episode"]}') - - # Handle dd_progid format - if 'dd_progid' in custom_data: - program_xml.append(f' {html.escape(custom_data["dd_progid"])}') - - # Handle external database IDs - for system in ['thetvdb.com', 'themoviedb.org', 'imdb.com']: - if f'{system}_id' in custom_data: - program_xml.append(f' {html.escape(custom_data[f"{system}_id"])}') - - # Add season and episode numbers in xmltv_ns format if available - if "season" in custom_data and "episode" in custom_data: - season = ( - int(custom_data["season"]) - 1 - if str(custom_data["season"]).isdigit() - else 0 - ) - episode = ( - int(custom_data["episode"]) - 1 - if str(custom_data["episode"]).isdigit() - else 0 - ) - program_xml.append(f' {season}.{episode}.') - - if "language" in custom_data: - program_xml.append(f' {html.escape(custom_data["language"])}') - - if "original_language" in custom_data: - program_xml.append(f' {html.escape(custom_data["original_language"])}') - - if "length" in custom_data and isinstance(custom_data["length"], dict): - length_value = custom_data["length"].get("value", "") - length_units = custom_data["length"].get("units", "minutes") - program_xml.append(f' {html.escape(str(length_value))}') - - if "video" in custom_data and isinstance(custom_data["video"], dict): - program_xml.append(" ") - - if "audio" in custom_data and isinstance(custom_data["audio"], dict): - program_xml.append(" ") - - if "subtitles" in custom_data and isinstance(custom_data["subtitles"], list): - for subtitle in custom_data["subtitles"]: - if isinstance(subtitle, dict): - subtitle_type = subtitle.get("type", "") - type_attr = f' type="{html.escape(subtitle_type)}"' if subtitle_type else "" - program_xml.append(f" ") - if "language" in subtitle: - program_xml.append(f" {html.escape(subtitle['language'])}") - program_xml.append(" ") - - if "rating" in custom_data: - rating_system = custom_data.get("rating_system", "TV Parental Guidelines") - program_xml.append(f' ') - program_xml.append(f' {html.escape(custom_data["rating"])}') - program_xml.append(f" ") - - if "star_ratings" in custom_data and isinstance(custom_data["star_ratings"], list): - for star_rating in custom_data["star_ratings"]: - if isinstance(star_rating, dict) and "value" in star_rating: - system_attr = f' system="{html.escape(star_rating["system"])}"' if "system" in star_rating else "" - program_xml.append(f" ") - program_xml.append(f" {html.escape(star_rating['value'])}") - program_xml.append(" ") - - if "reviews" in custom_data and isinstance(custom_data["reviews"], list): - for review in custom_data["reviews"]: - if isinstance(review, dict) and "content" in review: - review_type = review.get("type", "text") - attrs = [f'type="{html.escape(review_type)}"'] - if "source" in review: - attrs.append(f'source="{html.escape(review["source"])}"') - if "reviewer" in review: - attrs.append(f'reviewer="{html.escape(review["reviewer"])}"') - attr_str = " ".join(attrs) - program_xml.append(f' {html.escape(review["content"])}') - - if "images" in custom_data and isinstance(custom_data["images"], list): - for image in custom_data["images"]: - if isinstance(image, dict) and "url" in image: - attrs = [] - for attr in ['type', 'size', 'orient', 'system']: - if attr in image: - attrs.append(f'{attr}="{html.escape(image[attr])}"') - attr_str = " " + " ".join(attrs) if attrs else "" - program_xml.append(f' {html.escape(image["url"])}') - - # Add enhanced credits handling - if "credits" in custom_data: - program_xml.append(" ") - credits = custom_data["credits"] - - for role in ['director', 'writer', 'adapter', 'producer', 'composer', 'editor', 'presenter', 'commentator', 'guest']: - if role in credits: - people = credits[role] - if isinstance(people, list): - for person in people: - program_xml.append(f" <{role}>{html.escape(person)}") - else: - program_xml.append(f" <{role}>{html.escape(people)}") - - # Handle actors separately to include role and guest attributes - if "actor" in credits: - actors = credits["actor"] - if isinstance(actors, list): - for actor in actors: - if isinstance(actor, dict): - name = actor.get("name", "") - role_attr = f' role="{html.escape(actor["role"])}"' if "role" in actor else "" - guest_attr = ' guest="yes"' if actor.get("guest") else "" - program_xml.append(f" {html.escape(name)}") - else: - program_xml.append(f" {html.escape(actor)}") - else: - program_xml.append(f" {html.escape(actors)}") - - program_xml.append(" ") - - if "date" in custom_data: - program_xml.append(f' {html.escape(custom_data["date"])}') - - if "country" in custom_data: - program_xml.append(f' {html.escape(custom_data["country"])}') - - if "icon" in custom_data: - program_xml.append(f' ') - elif "sd_icon" in custom_data: - program_xml.append(f' ') - - # Add special flags as proper tags with enhanced handling - if custom_data.get("previously_shown", False): - prev_shown_details = custom_data.get("previously_shown_details", {}) - attrs = [] - if "start" in prev_shown_details: - attrs.append(f'start="{html.escape(prev_shown_details["start"])}"') - if "channel" in prev_shown_details: - attrs.append(f'channel="{html.escape(prev_shown_details["channel"])}"') - attr_str = " " + " ".join(attrs) if attrs else "" - program_xml.append(f" ") - - if custom_data.get("premiere", False): - premiere_text = custom_data.get("premiere_text", "") - if premiere_text: - program_xml.append(f" {html.escape(premiere_text)}") - else: - program_xml.append(" ") - - if custom_data.get("last_chance", False): - last_chance_text = custom_data.get("last_chance_text", "") - if last_chance_text: - program_xml.append(f" {html.escape(last_chance_text)}") - else: - program_xml.append(" ") - - if custom_data.get("new", False): - program_xml.append(" ") - - if custom_data.get('live', False): - program_xml.append(' ') - - program_xml.append(" ") - - xml_text = '\n'.join(program_xml) - pending.append((prog['start_time'], prog['id'], xml_text)) - - del program_chunk - - yield from flush_pending() - - if program_batch: - yield '\n'.join(program_batch) + '\n' - - del real_epg_map - - for channel_id, pattern_match_name, epg_source in dummy_program_list: - program_length_hours = 4 - dummy_programs = generate_dummy_programs( - channel_id, pattern_match_name, - num_days=dummy_days, - program_length_hours=program_length_hours, - epg_source=epg_source, - export_lookback=lookback_cutoff, - export_cutoff=cutoff_date, - ) - if not dummy_programs: - continue - dummy_batch = [] - for program in dummy_programs: - start_str = program['start_time'].strftime("%Y%m%d%H%M%S %z") - stop_str = program['end_time'].strftime("%Y%m%d%H%M%S %z") - lines = [ - f' ', - f" {html.escape(program['title'])}", - ] - if program.get('sub_title'): - lines.append(f" {html.escape(program['sub_title'])}") - lines.append(f" {html.escape(program['description'])}") - custom_data = program.get('custom_properties', {}) - if 'categories' in custom_data: - for cat in custom_data['categories']: - lines.append(f" {html.escape(cat)}") - if 'date' in custom_data: - lines.append(f" {html.escape(custom_data['date'])}") - if custom_data.get('live', False): - lines.append(" ") - if custom_data.get('new', False): - lines.append(" ") - if 'icon' in custom_data: - lines.append(f' ') - lines.append(" ") - dummy_batch.append('\n'.join(lines)) - if len(dummy_batch) >= batch_size: - yield '\n'.join(dummy_batch) + '\n' - dummy_batch = [] - del dummy_programs - if dummy_batch: - yield '\n'.join(dummy_batch) + '\n' - - del dummy_program_list - - yield "\n" - - client_id, client_ip, user_agent = get_client_identifier(request) - event_cache_key = f"epg_download:{user.username if user else 'anonymous'}:{profile_name or 'all'}:{client_id}" - - def _log_epg_download(): - from django.core.cache import cache as event_cache - - if not event_cache.get(event_cache_key): - log_system_event( - event_type='epg_download', - profile=profile_name or 'all', - user=user.username if user else 'anonymous', - channels=channel_count, - client_ip=client_ip, - user_agent=user_agent, - ) - event_cache.set(event_cache_key, True, 2) - - try: - from core.utils import _is_gevent_monkey_patched - - if _is_gevent_monkey_patched(): - import gevent - - gevent.spawn(_log_epg_download) - else: - _log_epg_download() - except Exception: - _log_epg_download() - - def build_epg_stream(): - try: - yield from epg_generator() - finally: - _epg_export_teardown() - - return stream_epg_response(content_cache_key, build_epg_stream) - - def xc_get_user(request): username = request.GET.get("username") password = request.GET.get("password")