feat(epg): add Schedules Direct API integration

- Add EPGSource username field and SDScheduleMD5 model for MD5 delta sync
- Implement full Schedules Direct fetch with SHA1 auth, token caching,
  20-day schedule window, and MD5-based delta sync to minimize API usage
- Add lineup manager UI allowing users to search and manage SD lineups
- Add stations-only fetch on source creation for immediate EPG matching
- Enforce 2-hour minimum refresh interval to respect SD rate limits,
  with first-fetch bypass when no prior full refresh exists
- Add real-time progress updates via synchronous Redis WebSocket delivery,
  bypassing gevent.spawn for reliable delivery from Celery prefork workers
- Refactor EPGsTable status cell into standalone EPGStatusCell component
  with direct Zustand subscription for live progress bar updates
- Guard XMLTV parse trigger for channels linked to SD EPG sources
- Add 15 backend tests covering auth, credentials, signals, and serializers

Migrations: 0023 (username field), 0024 (api_key help_text), 0025 (SD fields + SDScheduleMD5)

apps/channels/signals.py

@@ -201,10 +201,19 @@ def refresh_epg_programs(sender, instance, created, **kwargs):
     if not created and kwargs.get('update_fields') and 'epg_data' in kwargs['update_fields']:
         logger.info(f"Channel {instance.id} ({instance.name}) EPG data updated, refreshing program data")
         if instance.epg_data:
+            # Skip XMLTV program parser for SD sources — program data is handled
+            # by fetch_schedules_direct() directly.
+            if instance.epg_data.epg_source and instance.epg_data.epg_source.source_type == 'schedules_direct':
+                logger.info(f"Skipping XMLTV parse for SD source {instance.epg_data.tvg_id}")
+                return
             logger.info(f"Triggering EPG program refresh for {instance.epg_data.tvg_id}")
             parse_programs_for_tvg_id.delay(instance.epg_data.id)
     # For new channels with EPG data, also refresh
     elif created and instance.epg_data:
+        # Skip XMLTV program parser for SD sources
+        if instance.epg_data.epg_source and instance.epg_data.epg_source.source_type == 'schedules_direct':
+            logger.info(f"Skipping XMLTV parse for SD source {instance.epg_data.tvg_id}")
+            return
         logger.info(f"New channel {instance.id} ({instance.name}) created with EPG data, refreshing program data")
         parse_programs_for_tvg_id.delay(instance.epg_data.id)
 

apps/epg/migrations/0023_epgsource_username.py

@@ -0,0 +1,23 @@
+# Generated migration — adds username field to EPGSource for Schedules Direct auth
+
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ('epg', '0022_alter_epgdata_name'),
+    ]
+
+    operations = [
+        migrations.AddField(
+            model_name='epgsource',
+            name='username',
+            field=models.CharField(
+                max_length=255,
+                blank=True,
+                null=True,
+                help_text='Username for Schedules Direct authentication',
+            ),
+        ),
+    ]

apps/epg/migrations/0024_alter_epgsource_api_key.py

@@ -0,0 +1,21 @@
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ('epg', '0023_epgsource_username'),
+    ]
+
+    operations = [
+        migrations.AlterField(
+            model_name='epgsource',
+            name='api_key',
+            field=models.CharField(
+                max_length=255,
+                blank=True,
+                null=True,
+                help_text='Password for Schedules Direct authentication',
+            ),
+        ),
+    ]

apps/epg/migrations/0025_sd_fields_and_schedule_md5.py

@@ -0,0 +1,65 @@
+from django.db import migrations, models
+import django.db.models.deletion
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ('epg', '0024_alter_epgsource_api_key'),
+    ]
+
+    operations = [
+        # Add sd_changes_remaining to EPGSource
+        migrations.AddField(
+            model_name='epgsource',
+            name='sd_changes_remaining',
+            field=models.IntegerField(
+                blank=True,
+                null=True,
+                help_text='Number of Schedules Direct lineup additions remaining today (resets at midnight UTC)',
+            ),
+        ),
+        # Add sd_changes_reset_at to EPGSource
+        migrations.AddField(
+            model_name='epgsource',
+            name='sd_changes_reset_at',
+            field=models.DateTimeField(
+                blank=True,
+                null=True,
+                help_text='UTC datetime when the Schedules Direct daily lineup change counter resets',
+            ),
+        ),
+        # Add sd_program_md5 to ProgramData
+        migrations.AddField(
+            model_name='programdata',
+            name='sd_program_md5',
+            field=models.CharField(
+                blank=True,
+                max_length=22,
+                null=True,
+                help_text='MD5 hash from Schedules Direct for delta detection',
+            ),
+        ),
+        # Create SDScheduleMD5 model
+        migrations.CreateModel(
+            name='SDScheduleMD5',
+            fields=[
+                ('id', models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
+                ('station_id', models.CharField(help_text='Schedules Direct stationID', max_length=20)),
+                ('date', models.DateField(help_text='Schedule date (UTC)')),
+                ('md5', models.CharField(help_text='MD5 hash of the schedule for this station/date from Schedules Direct', max_length=22)),
+                ('last_modified', models.DateTimeField(help_text='Last modified timestamp from Schedules Direct')),
+                ('epg_source', models.ForeignKey(
+                    on_delete=django.db.models.deletion.CASCADE,
+                    related_name='sd_schedule_md5s',
+                    to='epg.epgsource',
+                )),
+            ],
+            options={
+                'indexes': [
+                    models.Index(fields=['epg_source', 'station_id'], name='epg_sdsche_epg_sou_idx'),
+                ],
+                'unique_together': {('epg_source', 'station_id', 'date')},
+            },
+        ),
+    ]

apps/epg/models.py

@@ -30,7 +30,10 @@ class EPGSource(models.Model):
     name = models.CharField(max_length=255, unique=True)
     source_type = models.CharField(max_length=20, choices=SOURCE_TYPE_CHOICES)
     url = models.URLField(max_length=1000, blank=True, null=True)  # For XMLTV
-    api_key = models.CharField(max_length=255, blank=True, null=True)  # For Schedules Direct
+    username = models.CharField(max_length=255, blank=True, null=True,
+                               help_text='Username for Schedules Direct authentication')
+    api_key = models.CharField(max_length=255, blank=True, null=True,
+                               help_text='Password for Schedules Direct authentication')  # For Schedules Direct
     is_active = models.BooleanField(default=True)
     file_path = models.CharField(max_length=1024, blank=True, null=True)
     extracted_file_path = models.CharField(max_length=1024, blank=True, null=True,
@@ -49,6 +52,16 @@ class EPGSource(models.Model):
         default=0,
         help_text="Priority for EPG matching (higher numbers = higher priority). Used when multiple EPG sources have matching entries for a channel."
     )
+    sd_changes_remaining = models.IntegerField(
+        null=True,
+        blank=True,
+        help_text="Number of Schedules Direct lineup additions remaining today (resets at midnight UTC)"
+    )
+    sd_changes_reset_at = models.DateTimeField(
+        null=True,
+        blank=True,
+        help_text="UTC datetime when the Schedules Direct daily lineup change counter resets"
+    )
     status = models.CharField(
         max_length=20,
         choices=STATUS_CHOICES,
@@ -163,6 +176,47 @@ class ProgramData(models.Model):
     description = models.TextField(blank=True, null=True)
     tvg_id = models.CharField(max_length=255, null=True, blank=True)
     custom_properties = models.JSONField(default=dict, blank=True, null=True)
+    sd_program_md5 = models.CharField(
+        max_length=22,
+        null=True,
+        blank=True,
+        help_text="MD5 hash from Schedules Direct for delta detection"
+    )
 
     def __str__(self):
         return f"{self.title} ({self.start_time} - {self.end_time})"
+
+class SDScheduleMD5(models.Model):
+    """
+    Caches per-station per-date MD5 hashes from Schedules Direct.
+    Used to detect schedule changes and avoid unnecessary re-downloads,
+    minimizing API calls against SD's rate-limited endpoints.
+    """
+    epg_source = models.ForeignKey(
+        EPGSource,
+        on_delete=models.CASCADE,
+        related_name="sd_schedule_md5s",
+    )
+    station_id = models.CharField(
+        max_length=20,
+        help_text="Schedules Direct stationID"
+    )
+    date = models.DateField(
+        help_text="Schedule date (UTC)"
+    )
+    md5 = models.CharField(
+        max_length=22,
+        help_text="MD5 hash of the schedule for this station/date from Schedules Direct"
+    )
+    last_modified = models.DateTimeField(
+        help_text="Last modified timestamp from Schedules Direct"
+    )
+
+    class Meta:
+        unique_together = ('epg_source', 'station_id', 'date')
+        indexes = [
+            models.Index(fields=['epg_source', 'station_id']),
+        ]
+
+    def __str__(self):
+        return f"SDScheduleMD5: {self.station_id} / {self.date} ({self.epg_source.name})"

apps/epg/serializers.py

@@ -22,6 +22,7 @@ class EPGSourceSerializer(serializers.ModelSerializer):
             'name',
             'source_type',
             'url',
+            'username',
             'api_key',
             'is_active',
             'file_path',

apps/epg/signals.py

@@ -1,7 +1,7 @@
 from django.db.models.signals import post_save, post_delete, pre_save
 from django.dispatch import receiver
 from .models import EPGSource, EPGData
-from .tasks import refresh_epg_data, delete_epg_refresh_task_by_id
+from .tasks import refresh_epg_data, delete_epg_refresh_task_by_id, fetch_schedules_direct_stations
 from core.scheduling import create_or_update_periodic_task, delete_periodic_task
 from core.utils import is_protected_path, send_websocket_update
 import json
@@ -12,9 +12,16 @@ logger = logging.getLogger(__name__)
 
 @receiver(post_save, sender=EPGSource)
 def trigger_refresh_on_new_epg_source(sender, instance, created, **kwargs):
-    # Trigger refresh only if the source is newly created, active, and not a dummy EPG
+    # Trigger refresh only if the source is newly created, active, and not a dummy EPG.
+    # For Schedules Direct sources, run a stations-only fetch on creation so the user
+    # can run Auto-match EPG before committing to a full schedule/program fetch.
+    # A short countdown gives the frontend time to receive the API response and
+    # populate the store before WebSocket updates arrive.
     if created and instance.is_active and instance.source_type != 'dummy':
-        refresh_epg_data.delay(instance.id)
+        if instance.source_type == 'schedules_direct':
+            fetch_schedules_direct_stations.apply_async((instance.id,), countdown=3)
+        else:
+            refresh_epg_data.delay(instance.id)
 
 @receiver(post_save, sender=EPGSource)
 def create_dummy_epg_data(sender, instance, created, **kwargs):

apps/epg/tasks.py

@@ -29,6 +29,10 @@ from core.utils import acquire_task_lock, release_task_lock, TaskLockRenewer, se
 
 logger = logging.getLogger(__name__)
 
+SD_BASE_URL = 'https://json.schedulesdirect.org/20141201'
+SD_DAYS_TO_FETCH = 20
+SD_PROGRAM_BATCH_SIZE = 5000
+
 # DOCTYPE internal subset for XMLTV files.  Declares all 252 HTML 4 named
 # entities so lxml/libxml2 can resolve references like é correctly
 # instead of silently dropping them in recovery mode.
@@ -173,6 +177,64 @@ def send_epg_update(source_id, action, progress, **kwargs):
         gc.collect()
 
 
+def _sd_send_ws_sync(source_id, action, progress, **kwargs):
+    """
+    Sends a WebSocket progress update synchronously via Redis, bypassing gevent.spawn.
+
+    In Celery prefork workers that inherit gevent monkey-patching from uWSGI,
+    gevent.spawn schedules coroutines that never execute because there is no
+    running gevent hub in the worker process. This function writes directly to
+    Redis using the channels_redis 4.x wire format, guaranteeing delivery
+    regardless of the execution context.
+    """
+    try:
+        import msgpack
+        import redis as redis_lib
+        from django.conf import settings
+
+        data = {"progress": progress, "type": "epg_refresh", "source": source_id, "action": action}
+        data.update(kwargs)
+        message = {"type": "update", "data": data}
+
+        redis_url = getattr(settings, "CELERY_BROKER_URL", "redis://localhost:6379/0")
+        r = redis_lib.from_url(redis_url, decode_responses=False)
+
+        prefix = "asgi"
+        group_name = "updates"
+        group_key = f"{prefix}:group:{group_name}"
+        group_expiry = 86400
+        channel_expiry = 60
+        rand_len = 12
+        now = time.time()
+
+        r.zremrangebyscore(group_key, 0, now - group_expiry)
+        raw = r.zrange(group_key, 0, -1)
+        if not raw:
+            return
+
+        channels = [m.decode("utf-8") if isinstance(m, bytes) else m for m in raw]
+        nonlocal_map = {}
+        for ch in channels:
+            pos = ch.find("!")
+            nl = ch[:pos + 1] if pos >= 0 else ch
+            nonlocal_map.setdefault(nl, []).append(ch)
+
+        pipe = r.pipeline(transaction=False)
+        for nl, chs in nonlocal_map.items():
+            channel_key = prefix + nl
+            msg = dict(message)
+            msg["__asgi_channel__"] = chs
+            serialized = os.urandom(rand_len) + msgpack.packb(msg)
+            pipe.zadd(channel_key, {serialized: now})
+            pipe.expire(channel_key, channel_expiry)
+        pipe.execute()
+        r.close()
+    except Exception as e:
+        logger.warning(f"SD WebSocket sync send failed: {e}")
+        # Fall back to standard path
+        send_epg_update(source_id, action, progress, **kwargs)
+
+
 def delete_epg_refresh_task_by_id(epg_id):
     """
     Delete the periodic task associated with an EPG source ID.
@@ -1255,6 +1317,17 @@ def parse_channels_only(source):
 
 @shared_task(time_limit=3600, soft_time_limit=3500)
 def parse_programs_for_tvg_id(epg_id):
+    # Skip XMLTV file parsing for Schedules Direct sources — program data is
+    # fetched and persisted directly by fetch_schedules_direct().
+    try:
+        from apps.epg.models import EPGData
+        epg_obj = EPGData.objects.select_related('epg_source').filter(id=epg_id).first()
+        if epg_obj and epg_obj.epg_source and epg_obj.epg_source.source_type == 'schedules_direct':
+            logger.info(f"Skipping XMLTV parse for SD EPGData id={epg_id} (source: {epg_obj.epg_source.name})")
+            return "Skipped — Schedules Direct source"
+    except Exception as e:
+        logger.warning(f"Could not check EPG source type for id={epg_id}: {e}")
+
     if not acquire_task_lock('parse_epg_programs', epg_id):
         logger.info(f"Program parse for {epg_id} already in progress, skipping duplicate task")
         return "Task already running"
@@ -1885,75 +1958,845 @@ def parse_programs_for_source(epg_source, tvg_id=None):
             logger.info(f"[parse_programs_for_source] Final memory usage: {final_memory:.2f} MB difference: {final_memory - initial_memory:.2f} MB")
             # Explicitly clear the process object to prevent potential memory leaks
             process = None
-def fetch_schedules_direct(source):
-    logger.info(f"Fetching Schedules Direct data from source: {source.name}")
+@shared_task(bind=True)
+def fetch_schedules_direct_stations(self, source_id):
+    """
+    Lightweight Celery task that runs a stations-only Schedules Direct fetch.
+    Called on initial source creation so EPGData entries exist for auto-matching
+    before the user commits to a full schedule/program fetch.
+    """
     try:
-        # Get default user agent from settings
-        stream_settings = CoreSettings.get_stream_settings()
-        user_agent = "Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:138.0) Gecko/20100101 Firefox/138.0"  # Fallback default
-        default_user_agent_id = stream_settings.get('default_user_agent')
+        source = EPGSource.objects.get(id=source_id)
+    except EPGSource.DoesNotExist:
+        logger.error(f"EPGSource {source_id} not found for SD stations fetch")
+        return
+    fetch_schedules_direct(source, stations_only=True)
 
-        if default_user_agent_id:
-            try:
-                user_agent_obj = UserAgent.objects.filter(id=int(default_user_agent_id)).first()
-                if user_agent_obj and user_agent_obj.user_agent:
-                    user_agent = user_agent_obj.user_agent
-                    logger.debug(f"Using default user agent: {user_agent}")
-            except (ValueError, Exception) as e:
-                logger.warning(f"Error retrieving default user agent, using fallback: {e}")
 
-        api_url = ''
-        headers = {
-            'Content-Type': 'application/json',
-            'Authorization': f'Bearer {source.api_key}',
-            'User-Agent': user_agent
-        }
-        logger.debug(f"Requesting subscriptions from Schedules Direct using URL: {api_url}")
-        response = requests.get(api_url, headers=headers, timeout=30)
-        response.raise_for_status()
-        subscriptions = response.json()
-        logger.debug(f"Fetched subscriptions: {subscriptions}")
+def fetch_schedules_direct(source, stations_only=False):
+    """
+    Fetch EPG data from the Schedules Direct JSON API and persist it to the
+    EPGData / ProgramData models.
 
-        for sub in subscriptions:
-            tvg_id = sub.get('stationID')
-            logger.debug(f"Processing subscription for tvg_id: {tvg_id}")
-            schedules_url = f"/schedules/{tvg_id}"
-            logger.debug(f"Requesting schedules from URL: {schedules_url}")
-            sched_response = requests.get(schedules_url, headers=headers, timeout=30)
-            sched_response.raise_for_status()
-            schedules = sched_response.json()
-            logger.debug(f"Fetched schedules: {schedules}")
+    Authentication flow (as required by the SD API specification):
+      1. POST credentials to the token endpoint — password must be SHA1-hashed
+         as required by the Schedules Direct API specification.
+      2. Use the returned token for all subsequent requests via the 'token' header.
+      3. Tokens are valid for 24 hours; SD returns the current valid token if one
+         already exists for the account.
 
-            epg_data, created = EPGData.objects.get_or_create(
-                tvg_id=tvg_id,
-                defaults={'name': tvg_id}
-            )
-            if created:
-                logger.info(f"Created new EPGData for tvg_id '{tvg_id}'.")
-            else:
-                logger.debug(f"Found existing EPGData for tvg_id '{tvg_id}'.")
+    Data flow:
+      1. Fetch subscribed lineups for the account.
+      2. Fetch station metadata for each lineup.
+      3. Persist station metadata to EPGData.
+      4. If stations_only=True, stop here — used on initial source creation so
+         the user can run Auto-match EPG before the full program fetch.
+      5. Fetch schedule grids in 14-day date-batched requests per station.
+      6. Fetch program metadata in batched requests (up to 5000 programIDs per request).
+      7. Persist channels to EPGData and programs to ProgramData.
 
-            for sched in schedules.get('schedules', []):
-                title = sched.get('title', 'No Title')
-                desc = sched.get('description', '')
-                start_time = parse_schedules_direct_time(sched.get('startTime'))
-                end_time = parse_schedules_direct_time(sched.get('endTime'))
-                obj, created = ProgramData.objects.update_or_create(
-                    epg=epg_data,
-                    start_time=start_time,
-                    title=title,
-                    defaults={
-                        'end_time': end_time,
-                        'description': desc,
-                        'sub_title': ''
-                    }
+    Args:
+        source: EPGSource instance
+        stations_only: If True, only fetch and persist station metadata (no schedules/programs).
+                      Used on initial source creation to populate EPGData for auto-matching
+                      before channels are assigned.
+    """
+    import hashlib
+    from datetime import date
+
+
+    logger.info(f"Fetching Schedules Direct data for source: {source.name}")
+
+    # -------------------------------------------------------------------------
+    # Validate credentials
+    # -------------------------------------------------------------------------
+    username = (source.username or '').strip()
+    password = (source.api_key or '').strip()
+
+    if not username or not password:
+        msg = "Schedules Direct source requires both a username and password."
+        logger.error(msg)
+        source.status = EPGSource.STATUS_ERROR
+        source.last_message = msg
+        source.save(update_fields=['status', 'last_message'])
+        _sd_send_ws_sync(source.id, "refresh", 100, status="error", error=msg)
+        return
+
+    # -------------------------------------------------------------------------
+    # Enforce 2-hour minimum interval between full fetches (not stations-only).
+    # Schedules Direct enforces rate limits of ~200 requests per 2-hour window.
+    # This prevents automated abuse regardless of how the refresh was triggered.
+    #
+    # Exception: if no SDScheduleMD5 records exist yet, this is the first full
+    # refresh after initial source creation (stations-only runs first and updates
+    # updated_at, which would otherwise incorrectly trigger this guard). Always
+    # allow the first full refresh through so guide data is immediately available.
+    # -------------------------------------------------------------------------
+    if not stations_only and source.updated_at:
+        from apps.epg.models import SDScheduleMD5 as _SDScheduleMD5
+        has_prior_full_refresh = _SDScheduleMD5.objects.filter(epg_source=source).exists()
+        if has_prior_full_refresh:
+            elapsed = (timezone.now() - source.updated_at).total_seconds()
+            min_interval_seconds = 2 * 3600  # 2 hours
+            if elapsed < min_interval_seconds:
+                remaining_minutes = int((min_interval_seconds - elapsed) / 60)
+                msg = (
+                    f"Schedules Direct refresh skipped — minimum 2-hour interval not reached. "
+                    f"Last refreshed {int(elapsed / 60)} minutes ago. "
+                    f"Please wait {remaining_minutes} more minute(s)."
                 )
-                if created:
-                    logger.info(f"Created ProgramData '{title}' for tvg_id '{tvg_id}'.")
-                else:
-                    logger.info(f"Updated ProgramData '{title}' for tvg_id '{tvg_id}'.")
-    except Exception as e:
-        logger.error(f"Error fetching Schedules Direct data from {source.name}: {e}", exc_info=True)
+                logger.warning(f"SD source {source.id}: {msg}")
+                source.status = EPGSource.STATUS_IDLE
+                source.last_message = msg
+                source.save(update_fields=['status', 'last_message'])
+                _sd_send_ws_sync(source.id, "refresh", 100, status="idle", message=msg)
+                return
+        else:
+            logger.info(f"SD source {source.id}: No prior full refresh detected — skipping 2-hour guard for first full fetch.")
+
+    # -------------------------------------------------------------------------
+    # Resolve user agent
+    # -------------------------------------------------------------------------
+    stream_settings = CoreSettings.get_stream_settings()
+    user_agent = "Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:138.0) Gecko/20100101 Firefox/138.0"
+    default_user_agent_id = stream_settings.get('default_user_agent')
+    if default_user_agent_id:
+        try:
+            ua_obj = UserAgent.objects.filter(id=int(default_user_agent_id)).first()
+            if ua_obj and ua_obj.user_agent:
+                user_agent = ua_obj.user_agent
+        except (ValueError, Exception) as e:
+            logger.warning(f"Could not resolve default user agent, using fallback: {e}")
+
+    def _sd_headers(token=None):
+        h = {
+            'Content-Type': 'application/json',
+            'User-Agent': user_agent,
+        }
+        if token:
+            h['token'] = token
+        return h
+
+    # -------------------------------------------------------------------------
+    # Step 1 — Authenticate and obtain session token
+    # The SD API requires the password to be SHA1-hashed before transmission.
+    # This is a requirement of the Schedules Direct API specification, not an
+    # architectural choice.
+    # -------------------------------------------------------------------------
+    source.status = EPGSource.STATUS_FETCHING
+    source.last_message = "Authenticating with Schedules Direct..."
+    source.save(update_fields=['status', 'last_message'])
+    _sd_send_ws_sync(source.id, "parsing_programs", 2, message="Authenticating with Schedules Direct...")
+
+    try:
+        sha1_password = hashlib.sha1(password.encode('utf-8')).hexdigest()
+        token_response = requests.post(
+            f"{SD_BASE_URL}/token",
+            json={'username': username, 'password': sha1_password},
+            headers=_sd_headers(),
+            timeout=30,
+        )
+        token_response.raise_for_status()
+        token_data = token_response.json()
+
+        auth_code = token_data.get('code', 0)
+        if auth_code != 0:
+            if auth_code == 4007:
+                msg = "Schedules Direct: this application is not authorized. Please contact the Dispatcharr maintainers."
+            elif auth_code == 4004:
+                msg = "Schedules Direct: account locked due to too many failed login attempts. Try again in 15 minutes."
+            elif auth_code == 4009:
+                msg = "Schedules Direct: too many login attempts in 24 hours. Token is valid for 24 hours — check for misconfiguration."
+            elif auth_code == 4001:
+                msg = "Schedules Direct: account has expired. Please renew your subscription at schedulesdirect.org."
+            elif auth_code == 4008:
+                msg = "Schedules Direct: account is inactive. Please log in to schedulesdirect.org to reactivate."
+            else:
+                msg = f"Schedules Direct authentication failed (code {auth_code}): {token_data.get('message', 'Unknown error')}"
+            logger.error(msg)
+            source.status = EPGSource.STATUS_ERROR
+            source.last_message = msg
+            source.save(update_fields=['status', 'last_message'])
+            _sd_send_ws_sync(source.id, "refresh", 100, status="error", error=msg)
+            return
+
+        token = token_data.get('token')
+        if not token:
+            msg = "Schedules Direct returned no token."
+            logger.error(msg)
+            source.status = EPGSource.STATUS_ERROR
+            source.last_message = msg
+            source.save(update_fields=['status', 'last_message'])
+            _sd_send_ws_sync(source.id, "refresh", 100, status="error", error=msg)
+            return
+
+        logger.info("Schedules Direct authentication successful.")
+
+    except requests.exceptions.RequestException as e:
+        msg = f"Network error authenticating with Schedules Direct: {e}"
+        logger.error(msg, exc_info=True)
+        source.status = EPGSource.STATUS_ERROR
+        source.last_message = msg
+        source.save(update_fields=['status', 'last_message'])
+        _sd_send_ws_sync(source.id, "refresh", 100, status="error", error=msg)
+        return
+
+    # -------------------------------------------------------------------------
+    # Step 2 — Check account status (respect OFFLINE system status)
+    # -------------------------------------------------------------------------
+    try:
+        status_response = requests.get(
+            f"{SD_BASE_URL}/status",
+            headers=_sd_headers(token),
+            timeout=30,
+        )
+        status_response.raise_for_status()
+        status_data = status_response.json()
+        system_status = status_data.get('systemStatus', [{}])[0].get('status', 'Online')
+        if system_status == 'Offline':
+            # Per SD API spec: if system is offline, disconnect and do not
+            # retry for 1 hour. We set idle status rather than error since
+            # this is a temporary SD-side condition.
+            msg = "Schedules Direct system is currently offline. Per SD guidelines, retrying in 1 hour."
+            logger.warning(msg)
+            source.status = EPGSource.STATUS_IDLE
+            source.last_message = msg
+            source.save(update_fields=['status', 'last_message'])
+            _sd_send_ws_sync(source.id, "refresh", 100, status="idle", message=msg)
+            return
+        logger.debug(f"Schedules Direct system status: {system_status}")
+    except requests.exceptions.RequestException as e:
+        logger.warning(f"Could not fetch SD system status, proceeding anyway: {e}")
+
+    # -------------------------------------------------------------------------
+    # Step 3 — Fetch subscribed lineups and build station map
+    # -------------------------------------------------------------------------
+    _sd_send_ws_sync(source.id, "parsing_programs", 10, message="Fetching subscribed lineups...")
+    try:
+        lineups_response = requests.get(
+            f"{SD_BASE_URL}/lineups",
+            headers=_sd_headers(token),
+            timeout=30,
+        )
+        # SD returns 400 with code 4102 when no lineups are configured.
+        # This is a valid account state — the user needs to add lineups via
+        # the Manage Lineups UI. Treat as idle rather than error.
+        if lineups_response.status_code == 400:
+            sd_data = lineups_response.json()
+            if sd_data.get('code') == 4102:
+                msg = "No lineups configured. Use the Manage Lineups option in the EPG source settings to add a lineup."
+                logger.warning(f"SD source {source.id}: no lineups configured on account (4102).")
+                source.status = EPGSource.STATUS_IDLE
+                source.last_message = msg
+                source.save(update_fields=['status', 'last_message'])
+                _sd_send_ws_sync(source.id, "refresh", 100, status="idle", message=msg)
+                return
+        lineups_response.raise_for_status()
+        lineups_data = lineups_response.json()
+        lineups = [l for l in lineups_data.get('lineups', []) if not l.get('isDeleted', False)]
+        if not lineups:
+            msg = "No lineups configured. Use the Manage Lineups option in the EPG source settings to add a lineup."
+            logger.warning(f"SD source {source.id}: no active lineups found.")
+            source.status = EPGSource.STATUS_IDLE
+            source.last_message = msg
+            source.save(update_fields=['status', 'last_message'])
+            _sd_send_ws_sync(source.id, "refresh", 100, status="idle", message=msg)
+            return
+        logger.info(f"Found {len(lineups)} lineup(s) in SD account.")
+    except requests.exceptions.RequestException as e:
+        msg = f"Failed to fetch Schedules Direct lineups: {e}"
+        logger.error(msg, exc_info=True)
+        source.status = EPGSource.STATUS_ERROR
+        source.last_message = msg
+        source.save(update_fields=['status', 'last_message'])
+        _sd_send_ws_sync(source.id, "refresh", 100, status="error", error=msg)
+        return
+
+    # Build station metadata map: stationID -> {name, callsign, logo_url}
+    station_map = {}
+    _sd_send_ws_sync(source.id, "parsing_programs", 18, message=f"Fetching station metadata for {len(lineups)} lineup(s)...")
+    for lineup in lineups:
+        lineup_id = lineup.get('lineupID') or lineup.get('lineup')
+        if not lineup_id:
+            continue
+        try:
+            detail_response = requests.get(
+                f"{SD_BASE_URL}/lineups/{lineup_id}",
+                headers=_sd_headers(token),
+                timeout=30,
+            )
+            detail_response.raise_for_status()
+            detail_data = detail_response.json()
+            for station in detail_data.get('stations', []):
+                sid = station.get('stationID')
+                if not sid:
+                    continue
+                logo_url = None
+                logos = station.get('stationLogos') or station.get('logo') or []
+                if isinstance(logos, list) and logos:
+                    logo_url = logos[0].get('URL') or logos[0].get('url')
+                elif isinstance(logos, dict):
+                    logo_url = logos.get('URL') or logos.get('url')
+                station_map[sid] = {
+                    'name': station.get('name', sid),
+                    'callsign': station.get('callsign', ''),
+                    'logo_url': logo_url,
+                }
+            logger.debug(f"Fetched {len(detail_data.get('stations', []))} stations from lineup {lineup_id}")
+        except requests.exceptions.RequestException as e:
+            logger.warning(f"Failed to fetch lineup details for {lineup_id}: {e}")
+
+    if not station_map:
+        msg = "No stations found across all Schedules Direct lineups."
+        logger.warning(msg)
+        source.status = EPGSource.STATUS_ERROR
+        source.last_message = msg
+        source.save(update_fields=['status', 'last_message'])
+        _sd_send_ws_sync(source.id, "refresh", 100, status="error", error=msg)
+        return
+
+    logger.info(f"Built station map with {len(station_map)} stations.")
+
+    # -------------------------------------------------------------------------
+    # Step 4 — Persist station metadata to EPGData
+    # -------------------------------------------------------------------------
+    source.status = EPGSource.STATUS_PARSING
+    source.last_message = f"Syncing {len(station_map)} stations..."
+    source.save(update_fields=['status', 'last_message'])
+    _sd_send_ws_sync(source.id, "parsing_programs", 28, message=f"Syncing {len(station_map)} stations to database...")
+
+    existing_epg_map = {
+        epg.tvg_id: epg
+        for epg in EPGData.objects.filter(epg_source=source)
+    }
+
+    epgs_to_create = []
+    epgs_to_update = []
+    icon_max_length = EPGData._meta.get_field('icon_url').max_length
+    name_max_length = EPGData._meta.get_field('name').max_length
+
+    for sid, info in station_map.items():
+        display_name = (info['name'] or sid)[:name_max_length]
+        logo = info['logo_url']
+        if logo and len(logo) > icon_max_length:
+            logo = None
+
+        if sid in existing_epg_map:
+            epg_obj = existing_epg_map[sid]
+            needs_update = False
+            if epg_obj.name != display_name:
+                epg_obj.name = display_name
+                needs_update = True
+            if epg_obj.icon_url != logo:
+                epg_obj.icon_url = logo
+                needs_update = True
+            if needs_update:
+                epgs_to_update.append(epg_obj)
+        else:
+            epgs_to_create.append(EPGData(
+                tvg_id=sid,
+                name=display_name,
+                icon_url=logo,
+                epg_source=source,
+            ))
+
+    if epgs_to_create:
+        EPGData.objects.bulk_create(epgs_to_create, ignore_conflicts=True)
+        logger.info(f"Created {len(epgs_to_create)} new EPGData entries.")
+    if epgs_to_update:
+        EPGData.objects.bulk_update(epgs_to_update, ['name', 'icon_url'])
+        logger.info(f"Updated {len(epgs_to_update)} existing EPGData entries.")
+
+    gc.collect()
+
+    # Rebuild map with fresh DB ids for all stations
+    epg_id_map = {
+        epg.tvg_id: epg.id
+        for epg in EPGData.objects.filter(epg_source=source, tvg_id__in=list(station_map.keys()))
+    }
+
+    # Station sync complete — send progress continuing into programs phase
+    # We deliberately do NOT send parsing_channels at 100 with status=success here
+    # because that would cause the frontend to mark the source as complete and
+    # stop rendering progress updates for the subsequent program fetch phases.
+    _sd_send_ws_sync(source.id, "parsing_programs", 30,
+                    message=f"Stations synced ({len(station_map)} stations). Preparing schedule fetch...")
+
+    # -------------------------------------------------------------------------
+    # Stations-only mode — used on initial source creation.
+    # Stop here so the user can run Auto-match EPG before the full program fetch.
+    # -------------------------------------------------------------------------
+    if stations_only:
+        success_msg = (
+            f"{len(station_map)} stations loaded from Schedules Direct. "
+            f"Run Auto-match EPG to map your channels, then use the Refresh "
+            f"button to populate guide data."
+        )
+        source.status = EPGSource.STATUS_SUCCESS
+        source.last_message = success_msg
+        source.updated_at = timezone.now()
+        source.save(update_fields=['status', 'last_message', 'updated_at'])
+        _sd_send_ws_sync(source.id, "parsing_channels", 100, status="success",
+                        message=success_msg, channels_count=len(station_map))
+        logger.info(f"Stations-only fetch complete for source: {source.name} ({len(station_map)} stations)")
+        return
+
+    # -------------------------------------------------------------------------
+    # Step 5 — MD5-delta schedule fetch
+    # First fetch MD5 hashes for all stations/dates. Compare against our
+    # locally cached hashes to determine which schedules have changed.
+    # Only download schedules that have actually changed — this minimises
+    # API calls against SD's rate-limited endpoints.
+    # -------------------------------------------------------------------------
+    from apps.epg.models import SDScheduleMD5
+    from django.utils.dateparse import parse_datetime
+
+    _sd_send_ws_sync(source.id, "parsing_programs", 33, message=f"Checking schedule MD5s for {len(station_map)} stations over {SD_DAYS_TO_FETCH} days...")
+    station_ids = list(station_map.keys())
+    today = date.today()
+    date_list = [(today + timedelta(days=i)).strftime('%Y-%m-%d') for i in range(SD_DAYS_TO_FETCH)]
+
+    # Fetch MD5 hashes for all stations in batches of 5000
+    STATION_BATCH_SIZE = 5000
+    server_md5s = {}  # (station_id, date) -> {md5, last_modified}
+
+    logger.info(f"Fetching schedule MD5s for {len(station_ids)} stations over {SD_DAYS_TO_FETCH} days.")
+
+    station_batches = [station_ids[i:i + STATION_BATCH_SIZE] for i in range(0, len(station_ids), STATION_BATCH_SIZE)]
+    for batch in station_batches:
+        try:
+            md5_response = requests.post(
+                f"{SD_BASE_URL}/schedules/md5",
+                json=[{'stationID': sid, 'date': date_list} for sid in batch],
+                headers=_sd_headers(token),
+                timeout=120,
+            )
+            md5_response.raise_for_status()
+            md5_data = md5_response.json()
+            for sid, dates in md5_data.items():
+                for date_str, info in dates.items():
+                    if info.get('code', 0) == 0:
+                        server_md5s[(sid, date_str)] = {
+                            'md5': info.get('md5', ''),
+                            'last_modified': info.get('lastModified', ''),
+                        }
+        except requests.exceptions.RequestException as e:
+            logger.warning(f"Failed to fetch schedule MD5s: {e}")
+
+    # Load our cached MD5s from DB
+    cached_md5s = {
+        (r.station_id, r.date.strftime('%Y-%m-%d')): r.md5
+        for r in SDScheduleMD5.objects.filter(epg_source=source, station_id__in=station_ids)
+    }
+
+    # Determine which station/date combinations need downloading
+    changed_by_station = {}  # station_id -> [date_str, ...]
+    for (sid, date_str), server_info in server_md5s.items():
+        if date_str not in date_list:
+            continue
+        cached = cached_md5s.get((sid, date_str))
+        if cached != server_info['md5']:
+            changed_by_station.setdefault(sid, []).append(date_str)
+
+    total_changed = sum(len(v) for v in changed_by_station.items())
+    total_possible = len(station_ids) * len(date_list)
+    logger.info(f"Schedule MD5 check: {len(server_md5s)} hashes checked, {total_changed} station/date combinations changed (of {total_possible} possible).")
+    _sd_send_ws_sync(source.id, "parsing_programs", 38,
+                    message=f"MD5 check complete: {len(changed_by_station)} stations have schedule updates.")
+
+    # schedules_by_station: stationID -> list of {programID, airDateTime, duration, ...}
+    schedules_by_station = {sid: [] for sid in station_ids}
+    program_ids_needed = set()
+
+    # Existing cached program MD5s for delta detection in step 6
+    existing_program_md5s = {
+        p.tvg_id + '|' + p.title: p.sd_program_md5
+        for p in ProgramData.objects.filter(
+            epg__epg_source=source,
+            sd_program_md5__isnull=False,
+        ).only('tvg_id', 'title', 'sd_program_md5')
+    }
+
+    if not changed_by_station:
+        logger.info("No schedule changes detected — skipping schedule and program downloads.")
+        _sd_send_ws_sync(source.id, "parsing_programs", 100, status="success",
+                        message="No schedule changes detected since last refresh. Guide data is up to date.")
+        source.status = EPGSource.STATUS_SUCCESS
+        source.last_message = "No schedule changes detected. Guide data is up to date."
+        source.updated_at = timezone.now()
+        source.save(update_fields=['status', 'last_message', 'updated_at'])
+        return
+
+    # Download only changed schedules, batched by 7-day windows per station
+    SCHEDULE_BATCH_DAYS = 7
+    changed_station_ids = list(changed_by_station.keys())
+    date_batches = [date_list[i:i + SCHEDULE_BATCH_DAYS] for i in range(0, len(date_list), SCHEDULE_BATCH_DAYS)]
+    new_md5_records = []
+    updated_md5_records = []
+    existing_md5_map = {
+        (r.station_id, r.date.strftime('%Y-%m-%d')): r
+        for r in SDScheduleMD5.objects.filter(epg_source=source, station_id__in=changed_station_ids)
+    }
+
+    for batch_idx, date_batch in enumerate(date_batches):
+        # Notify frontend at the start of each batch so progress updates immediately
+        pre_progress = 38 + int((batch_idx / len(date_batches)) * 22)
+        logger.info(f"Fetching schedule batch {batch_idx + 1} of {len(date_batches)}...")
+        _sd_send_ws_sync(source.id, "parsing_programs", min(59, pre_progress),
+                        message=f"Fetching schedules: batch {batch_idx + 1} of {len(date_batches)}...")
+        # Yield to gevent hub so the WebSocket update is delivered before the blocking request
+        try:
+            import gevent; gevent.sleep(0)
+        except ImportError:
+            pass
+        # Only include stations that have changes in this date batch
+        request_body = [
+            {'stationID': sid, 'date': [d for d in date_batch if d in changed_by_station.get(sid, [])]}
+            for sid in changed_station_ids
+            if any(d in changed_by_station.get(sid, []) for d in date_batch)
+        ]
+        if not request_body:
+            continue
+        try:
+            sched_response = requests.post(
+                f"{SD_BASE_URL}/schedules",
+                json=request_body,
+                headers=_sd_headers(token),
+                timeout=120,
+            )
+            sched_response.raise_for_status()
+            sched_data = sched_response.json()
+
+            for station_sched in sched_data:
+                sid = station_sched.get('stationID')
+                if not sid:
+                    continue
+                programs = station_sched.get('programs', [])
+                schedules_by_station.setdefault(sid, []).extend(programs)
+                for prog in programs:
+                    pid = prog.get('programID')
+                    if pid:
+                        program_ids_needed.add(pid)
+
+                # Update MD5 cache for this station/date
+                meta = station_sched.get('metadata', {})
+                start_date = meta.get('startDate')
+                md5_val = meta.get('md5', '')
+                last_mod_str = meta.get('modified', '')
+                if start_date and md5_val:
+                    key = (sid, start_date)
+                    last_mod = parse_datetime(last_mod_str) if last_mod_str else timezone.now()
+                    if key in existing_md5_map:
+                        rec = existing_md5_map[key]
+                        rec.md5 = md5_val
+                        rec.last_modified = last_mod
+                        updated_md5_records.append(rec)
+                    else:
+                        import datetime as dt_module
+                        try:
+                            date_obj = dt_module.date.fromisoformat(start_date)
+                            new_md5_records.append(SDScheduleMD5(
+                                epg_source=source,
+                                station_id=sid,
+                                date=date_obj,
+                                md5=md5_val,
+                                last_modified=last_mod,
+                            ))
+                        except ValueError:
+                            pass
+
+            progress = 38 + int(((batch_idx + 1) / len(date_batches)) * 22)
+            _sd_send_ws_sync(source.id, "parsing_programs", min(60, progress),
+                            message=f"Fetching changed schedules: batch {batch_idx + 1}/{len(date_batches)} ({len(program_ids_needed):,} programs found)")
+
+        except requests.exceptions.RequestException as e:
+            logger.warning(f"Failed to fetch schedule batch {batch_idx + 1}: {e}")
+
+    # Persist updated MD5 cache
+    if new_md5_records:
+        SDScheduleMD5.objects.bulk_create(new_md5_records, ignore_conflicts=True)
+        logger.info(f"Cached {len(new_md5_records)} new schedule MD5s.")
+    if updated_md5_records:
+        SDScheduleMD5.objects.bulk_update(updated_md5_records, ['md5', 'last_modified'])
+        logger.info(f"Updated {len(updated_md5_records)} existing schedule MD5s.")
+
+    if not program_ids_needed:
+        msg = "No schedule data returned from Schedules Direct."
+        logger.warning(msg)
+        source.status = EPGSource.STATUS_ERROR
+        source.last_message = msg
+        source.save(update_fields=['status', 'last_message'])
+        _sd_send_ws_sync(source.id, "parsing_programs", 100, status="error", error=msg)
+        return
+
+    # -------------------------------------------------------------------------
+    # Step 6 — MD5-delta program metadata fetch
+    # The schedule response includes an MD5 hash per program airing.
+    # Compare against our cached program MD5s to only download programs
+    # whose metadata has changed since our last fetch.
+    # -------------------------------------------------------------------------
+
+    # Build map of programID -> md5 from schedule data
+    schedule_program_md5s = {}  # programID -> md5 from schedule
+    for sid, airings in schedules_by_station.items():
+        for airing in airings:
+            pid = airing.get('programID')
+            md5 = airing.get('md5')
+            if pid and md5:
+                schedule_program_md5s[pid] = md5
+
+    # Load cached program MD5s from DB keyed by programID via tvg_id
+    cached_prog_md5s = {
+        p.tvg_id: p.sd_program_md5
+        for p in ProgramData.objects.filter(
+            epg__epg_source=source,
+            sd_program_md5__isnull=False,
+        ).only('tvg_id', 'sd_program_md5').distinct('tvg_id')
+    }
+
+    # Only fetch programs where MD5 differs from our cache
+    programs_to_fetch = [
+        pid for pid in program_ids_needed
+        if schedule_program_md5s.get(pid) != cached_prog_md5s.get(pid)
+    ]
+
+    logger.info(
+        f"Program MD5 delta: {len(program_ids_needed)} programs in schedules, "
+        f"{len(programs_to_fetch)} need downloading ({len(program_ids_needed) - len(programs_to_fetch)} unchanged).")
+
+    program_metadata = {}
+    program_id_list = programs_to_fetch
+    total_batches = max(1, (len(program_id_list) + SD_PROGRAM_BATCH_SIZE - 1) // SD_PROGRAM_BATCH_SIZE)
+
+    if program_id_list:
+        logger.info(f"Fetching metadata for {len(program_id_list)} programs in {total_batches} batch(es).")
+        for batch_idx in range(total_batches):
+            # Notify frontend at the start of each batch so progress updates immediately
+            pre_progress = 60 + int((batch_idx / total_batches) * 20)
+            logger.info(f"Fetching program metadata batch {batch_idx + 1} of {total_batches} ({batch_idx * SD_PROGRAM_BATCH_SIZE:,} of {len(program_id_list):,} programs)...")
+            _sd_send_ws_sync(source.id, "parsing_programs", min(79, pre_progress),
+                            message=f"Fetching program data: batch {batch_idx + 1} of {total_batches} ({batch_idx * SD_PROGRAM_BATCH_SIZE:,} of {len(program_id_list):,} programs)")
+            # Yield to gevent hub so the WebSocket update is delivered before the blocking request
+            try:
+                import gevent; gevent.sleep(0)
+            except ImportError:
+                pass
+            batch = program_id_list[batch_idx * SD_PROGRAM_BATCH_SIZE:(batch_idx + 1) * SD_PROGRAM_BATCH_SIZE]
+            try:
+                prog_response = requests.post(
+                    f"{SD_BASE_URL}/programs",
+                    json=batch,
+                    headers=_sd_headers(token),
+                    timeout=120,
+                )
+                prog_response.raise_for_status()
+                prog_data = prog_response.json()
+                for prog in prog_data:
+                    pid = prog.get('programID')
+                    if pid:
+                        program_metadata[pid] = prog
+
+                progress = 60 + int(((batch_idx + 1) / total_batches) * 20)
+                _sd_send_ws_sync(source.id, "parsing_programs", min(80, progress),
+                                message=f"Fetching program details: batch {batch_idx + 1}/{total_batches} ({len(program_metadata):,} programs loaded)")
+                logger.debug(f"Fetched program metadata batch {batch_idx + 1}/{total_batches}")
+
+            except requests.exceptions.RequestException as e:
+                logger.warning(f"Failed to fetch program metadata batch {batch_idx + 1}: {e}")
+    else:
+        logger.info("All program metadata unchanged - skipping program download.")
+        _sd_send_ws_sync(source.id, "parsing_programs", 80, message="Program metadata unchanged - using cached data.")
+
+    gc.collect()
+
+    # -------------------------------------------------------------------------
+    # Step 7 — Build ProgramData records and persist atomically
+    # -------------------------------------------------------------------------
+    logger.info("Building program records...")
+    _sd_send_ws_sync(source.id, "parsing_programs", 80)
+
+    # Only process stations that are mapped to channels to match the existing
+    # XMLTV flow (parse_programs_for_source only processes mapped channels).
+    from apps.channels.models import Channel as ChannelModel
+    mapped_epg_ids = set(
+        ChannelModel.objects.filter(
+            epg_data__epg_source=source,
+            epg_data__isnull=False,
+        ).values_list('epg_data_id', flat=True)
+    )
+    mapped_tvg_ids = set(
+        EPGData.objects.filter(
+            id__in=mapped_epg_ids,
+            epg_source=source,
+        ).values_list('tvg_id', flat=True)
+    )
+
+    all_programs_to_create = []
+    total_programs = 0
+    skipped_unmapped = 0
+
+    for sid, airings in schedules_by_station.items():
+        if sid not in mapped_tvg_ids:
+            skipped_unmapped += len(airings)
+            continue
+
+        epg_db_id = epg_id_map.get(sid)
+        if not epg_db_id:
+            continue
+
+        for airing in airings:
+            pid = airing.get('programID')
+            air_time = airing.get('airDateTime')
+            duration_secs = airing.get('duration', 0)
+
+            if not pid or not air_time or not duration_secs:
+                continue
+
+            try:
+                start_dt = parse_schedules_direct_time(air_time)
+                end_dt = start_dt + timedelta(seconds=int(duration_secs))
+            except Exception as e:
+                logger.debug(f"Could not parse air time '{air_time}': {e}")
+                continue
+
+            meta = program_metadata.get(pid, {})
+            titles = meta.get('titles', [{}])
+            title = titles[0].get('title120', '') if titles else ''
+            if not title:
+                title = meta.get('episodeTitle150', '') or 'No Title'
+            title = title[:255]
+
+            descriptions = meta.get('descriptions', {})
+            desc = ''
+            for key in ('description1000', 'description255', 'description100'):
+                candidates = descriptions.get(key, [])
+                if candidates:
+                    desc = candidates[0].get('description', '')
+                    if desc:
+                        break
+
+            episode_title = meta.get('episodeTitle150', '')
+
+            # Build custom_properties following the same pattern as the XMLTV parser
+            custom_props = {}
+            metadata_block = meta.get('metadata', [{}])
+            if metadata_block:
+                m = metadata_block[0].get('Gracenote', {})
+                season = m.get('season')
+                episode = m.get('episode')
+                if season:
+                    custom_props['season'] = int(season)
+                if episode:
+                    custom_props['episode'] = int(episode)
+
+            content_rating = meta.get('contentRating', [])
+            if content_rating:
+                custom_props['rating'] = content_rating[0].get('code', '')
+                custom_props['rating_system'] = content_rating[0].get('body', '')
+
+            genres = meta.get('genres', [])
+            if genres:
+                custom_props['categories'] = genres
+
+            cast = meta.get('cast', [])
+            crew = meta.get('crew', [])
+            credits = {}
+            if cast:
+                credits['actor'] = [
+                    {'name': p.get('name', ''), 'role': p.get('role', '')}
+                    for p in cast if p.get('name')
+                ]
+            if crew:
+                for member in crew:
+                    role = member.get('role', '').lower()
+                    name = member.get('name', '')
+                    if not name:
+                        continue
+                    if 'director' in role:
+                        credits.setdefault('director', []).append(name)
+                    elif 'writer' in role or 'screenwriter' in role:
+                        credits.setdefault('writer', []).append(name)
+                    elif 'producer' in role:
+                        credits.setdefault('producer', []).append(name)
+            if credits:
+                custom_props['credits'] = credits
+
+            if airing.get('isLive'):
+                custom_props['live'] = True
+            if airing.get('isNew'):
+                custom_props['new'] = True
+            if airing.get('isPremiere'):
+                custom_props['premiere'] = True
+
+            year = meta.get('movie', {}).get('year') or meta.get('originalAirDate', '')[:4]
+            if year:
+                custom_props['date'] = str(year)
+
+            all_programs_to_create.append(ProgramData(
+                epg_id=epg_db_id,
+                start_time=start_dt,
+                end_time=end_dt,
+                title=title,
+                sub_title=episode_title or None,
+                description=desc or None,
+                tvg_id=sid,
+                custom_properties=custom_props or None,
+                sd_program_md5=schedule_program_md5s.get(pid),
+            ))
+            total_programs += 1
+
+    logger.info(f"Built {total_programs} program records "
+                f"({skipped_unmapped} skipped for unmapped stations).")
+
+    _sd_send_ws_sync(source.id, "parsing_programs", 88)
+
+    # Atomic delete + bulk insert — same pattern as parse_programs_for_source
+    BATCH_SIZE = 1000
+    try:
+        with transaction.atomic():
+            with connection.cursor() as cursor:
+                cursor.execute("SET LOCAL statement_timeout = '10min'")
+            deleted_count = ProgramData.objects.filter(epg_id__in=mapped_epg_ids).delete()[0]
+            logger.debug(f"Deleted {deleted_count} existing SD programs.")
+            for i in range(0, len(all_programs_to_create), BATCH_SIZE):
+                ProgramData.objects.bulk_create(all_programs_to_create[i:i + BATCH_SIZE])
+                progress = 88 + int(((i + BATCH_SIZE) / max(len(all_programs_to_create), 1)) * 10)
+                _sd_send_ws_sync(source.id, "parsing_programs", min(98, progress))
+
+        logger.info(f"Committed {total_programs} Schedules Direct programs to database.")
+    except Exception as db_error:
+        msg = f"Database error persisting Schedules Direct programs: {db_error}"
+        logger.error(msg, exc_info=True)
+        source.status = EPGSource.STATUS_ERROR
+        source.last_message = msg
+        source.save(update_fields=['status', 'last_message'])
+        _sd_send_ws_sync(source.id, "parsing_programs", 100, status="error", error=msg)
+        return
+    finally:
+        all_programs_to_create = None
+        gc.collect()
+
+    # -------------------------------------------------------------------------
+    # Done
+    # -------------------------------------------------------------------------
+    success_msg = (
+        f"Successfully fetched {total_programs:,} programs for "
+        f"{len(mapped_tvg_ids)} mapped stations from Schedules Direct "
+        f"({skipped_unmapped:,} programs skipped for unmapped stations)."
+    )
+    source.status = EPGSource.STATUS_SUCCESS
+    source.last_message = success_msg
+    source.updated_at = timezone.now()
+    source.save(update_fields=['status', 'last_message', 'updated_at'])
+    _sd_send_ws_sync(source.id, "parsing_programs", 100, status="success", message=success_msg)
+    log_system_event(
+        event_type='epg_refresh',
+        source_name=source.name,
+        programs=total_programs,
+        channels=len(mapped_tvg_ids),
+        skipped_programs=skipped_unmapped,
+    )
+    logger.info(f"Schedules Direct fetch complete for source: {source.name}")
 
 
 # -------------------------------

apps/epg/tests/test_schedules_direct.py

@@ -0,0 +1,263 @@
+"""
+Tests for the Schedules Direct EPG integration.
+
+Covers:
+- EPGSource model: username field presence and help text
+- EPGSource serializer: username field included in output
+- fetch_schedules_direct: credential validation
+- fetch_schedules_direct: SHA1 password hashing and token exchange
+- fetch_schedules_direct: graceful error handling on auth failure
+- parse_schedules_direct_time: correct UTC parsing
+- EPG signals: SD sources skip the XMLTV program parser
+"""
+
+import hashlib
+from datetime import datetime
+from unittest.mock import MagicMock, patch
+
+from django.test import TestCase
+from django.utils import timezone
+
+from apps.epg.models import EPGSource
+from apps.epg.serializers import EPGSourceSerializer
+
+
+# ---------------------------------------------------------------------------
+# Model tests
+# ---------------------------------------------------------------------------
+
+class EPGSourceUsernameFieldTests(TestCase):
+    """EPGSource.username must exist, be nullable, and carry help text."""
+
+    def test_username_field_exists(self):
+        source = EPGSource.objects.create(
+            name='SD Test',
+            source_type='schedules_direct',
+            username='testuser',
+            api_key='testpass',
+        )
+        source.refresh_from_db()
+        self.assertEqual(source.username, 'testuser')
+
+    def test_username_nullable(self):
+        source = EPGSource.objects.create(
+            name='SD Nullable',
+            source_type='schedules_direct',
+        )
+        source.refresh_from_db()
+        self.assertIsNone(source.username)
+
+    def test_username_help_text(self):
+        field = EPGSource._meta.get_field('username')
+        self.assertIn('Schedules Direct', field.help_text)
+
+
+# ---------------------------------------------------------------------------
+# Serializer tests
+# ---------------------------------------------------------------------------
+
+class EPGSourceSerializerSDTests(TestCase):
+    """EPGSourceSerializer must include the username field."""
+
+    def test_username_in_serializer_fields(self):
+        source = EPGSource.objects.create(
+            name='SD Serializer Test',
+            source_type='schedules_direct',
+            username='sduser',
+            api_key='sdpass',
+        )
+        data = EPGSourceSerializer(source).data
+        self.assertIn('username', data)
+        self.assertEqual(data['username'], 'sduser')
+
+    def test_api_key_in_serializer_fields(self):
+        source = EPGSource.objects.create(
+            name='SD API Key Test',
+            source_type='schedules_direct',
+            api_key='secret',
+        )
+        data = EPGSourceSerializer(source).data
+        self.assertIn('api_key', data)
+
+
+# ---------------------------------------------------------------------------
+# fetch_schedules_direct tests
+# ---------------------------------------------------------------------------
+
+class FetchSchedulesDirectCredentialTests(TestCase):
+    """fetch_schedules_direct must reject sources missing credentials."""
+
+    def _make_source(self, username=None, api_key=None):
+        return EPGSource.objects.create(
+            name='SD Cred Test',
+            source_type='schedules_direct',
+            username=username,
+            api_key=api_key,
+        )
+
+    def test_missing_username_sets_error_status(self):
+        from apps.epg.tasks import fetch_schedules_direct
+        source = self._make_source(username=None, api_key='pass')
+        fetch_schedules_direct(source)
+        source.refresh_from_db()
+        self.assertEqual(source.status, EPGSource.STATUS_ERROR)
+
+    def test_missing_password_sets_error_status(self):
+        from apps.epg.tasks import fetch_schedules_direct
+        source = self._make_source(username='user', api_key=None)
+        fetch_schedules_direct(source)
+        source.refresh_from_db()
+        self.assertEqual(source.status, EPGSource.STATUS_ERROR)
+
+    def test_empty_username_sets_error_status(self):
+        from apps.epg.tasks import fetch_schedules_direct
+        source = self._make_source(username='   ', api_key='pass')
+        fetch_schedules_direct(source)
+        source.refresh_from_db()
+        self.assertEqual(source.status, EPGSource.STATUS_ERROR)
+
+
+class FetchSchedulesDirectAuthTests(TestCase):
+    """fetch_schedules_direct must SHA1-hash the password before sending."""
+
+    @patch('apps.epg.tasks.requests.post')
+    @patch('apps.epg.tasks.requests.get')
+    def test_password_sha1_hashed_in_token_request(self, mock_get, mock_post):
+        """The token POST body must contain the SHA1 hash of the plaintext password."""
+        plaintext = 'mysecretpassword'
+        expected_hash = hashlib.sha1(plaintext.encode('utf-8')).hexdigest()
+
+        # Auth succeeds, status check returns empty data, lineups returns empty
+        mock_post.return_value = MagicMock(
+            status_code=200,
+            json=MagicMock(return_value={'code': 0, 'token': 'tok123'}),
+        )
+        mock_get.return_value = MagicMock(
+            status_code=200,
+            json=MagicMock(return_value={}),
+        )
+
+        from apps.epg.tasks import fetch_schedules_direct
+        source = EPGSource.objects.create(
+            name='SD Hash Test',
+            source_type='schedules_direct',
+            username='sduser',
+            api_key=plaintext,
+        )
+
+        with patch('apps.epg.tasks.send_epg_update'):
+            fetch_schedules_direct(source)
+
+        # Verify the POST was called and the body contained the hash
+        self.assertTrue(mock_post.called)
+        call_kwargs = mock_post.call_args
+        posted_json = call_kwargs[1].get('json') or call_kwargs[0][1]
+        self.assertEqual(posted_json.get('password'), expected_hash)
+        self.assertEqual(posted_json.get('username'), 'sduser')
+
+    @patch('apps.epg.tasks.requests.post')
+    def test_auth_failure_sets_error_status(self, mock_post):
+        """A non-zero SD response code must set STATUS_ERROR on the source."""
+        mock_post.return_value = MagicMock(
+            status_code=200,
+            json=MagicMock(return_value={
+                'code': 3000,
+                'message': 'Invalid credentials',
+            }),
+        )
+
+        from apps.epg.tasks import fetch_schedules_direct
+        source = EPGSource.objects.create(
+            name='SD Auth Fail',
+            source_type='schedules_direct',
+            username='baduser',
+            api_key='badpass',
+        )
+
+        with patch('apps.epg.tasks.send_epg_update'):
+            fetch_schedules_direct(source)
+
+        source.refresh_from_db()
+        self.assertEqual(source.status, EPGSource.STATUS_ERROR)
+
+    @patch('apps.epg.tasks.requests.post')
+    def test_network_error_sets_error_status(self, mock_post):
+        """A network-level exception must set STATUS_ERROR and not crash."""
+        import requests as req_lib
+        mock_post.side_effect = req_lib.exceptions.ConnectionError('timeout')
+
+        from apps.epg.tasks import fetch_schedules_direct
+        source = EPGSource.objects.create(
+            name='SD Network Error',
+            source_type='schedules_direct',
+            username='user',
+            api_key='pass',
+        )
+
+        with patch('apps.epg.tasks.send_epg_update'):
+            fetch_schedules_direct(source)  # Must not raise
+
+        source.refresh_from_db()
+        self.assertEqual(source.status, EPGSource.STATUS_ERROR)
+
+
+# ---------------------------------------------------------------------------
+# parse_schedules_direct_time tests
+# ---------------------------------------------------------------------------
+
+class ParseSchedulesDirectTimeTests(TestCase):
+    """parse_schedules_direct_time must parse SD ISO timestamps to UTC-aware datetimes."""
+
+    def test_parses_valid_timestamp(self):
+        from apps.epg.tasks import parse_schedules_direct_time
+        result = parse_schedules_direct_time('2026-05-16T20:00:00Z')
+        self.assertEqual(result.year, 2026)
+        self.assertEqual(result.month, 5)
+        self.assertEqual(result.day, 16)
+        self.assertEqual(result.hour, 20)
+        self.assertIsNotNone(result.tzinfo)
+
+    def test_result_is_utc_aware(self):
+        from apps.epg.tasks import parse_schedules_direct_time
+        result = parse_schedules_direct_time('2026-01-01T00:00:00Z')
+        # Should be timezone-aware
+        self.assertIsNotNone(result.tzinfo)
+
+    def test_raises_on_invalid_format(self):
+        from apps.epg.tasks import parse_schedules_direct_time
+        with self.assertRaises(Exception):
+            parse_schedules_direct_time('not-a-timestamp')
+
+
+# ---------------------------------------------------------------------------
+# Signal tests
+# ---------------------------------------------------------------------------
+
+class SDSourceSignalTests(TestCase):
+    """SD EPG sources must skip the XMLTV program parser signal."""
+
+    @patch('apps.channels.signals.parse_programs_for_tvg_id')
+    def test_sd_source_skips_xmltv_parse_on_channel_create(self, mock_parse):
+        """Creating a channel linked to an SD EPG source must not trigger
+        the XMLTV program parser — SD data is handled by fetch_schedules_direct."""
+        from apps.epg.models import EPGData
+        from apps.channels.models import Channel
+
+        sd_source = EPGSource.objects.create(
+            name='SD Signal Test',
+            source_type='schedules_direct',
+            username='u',
+            api_key='p',
+        )
+        epg_data = EPGData.objects.create(
+            tvg_id='sd-test-station',
+            name='SD Test Station',
+            epg_source=sd_source,
+        )
+
+        Channel.objects.create(
+            name='SD Channel',
+            epg_data=epg_data,
+        )
+
+        mock_parse.delay.assert_not_called()

frontend/src/WebSocket.jsx

@@ -51,6 +51,7 @@ export const WebsocketProvider = ({ children }) => {
   const epgs = useEPGsStore((s) => s.epgs);
   const updateEPG = useEPGsStore((s) => s.updateEPG);
   const updateEPGProgress = useEPGsStore((s) => s.updateEPGProgress);
+  const fetchEPGsForProgress = useEPGsStore((s) => s.fetchEPGs);
 
   const updatePlaylist = usePlaylistsStore((s) => s.updatePlaylist);
 
@@ -642,19 +643,18 @@ export const WebsocketProvider = ({ children }) => {
                   parsedEvent.data.source || parsedEvent.data.account;
                 const epg = epgs[sourceId];
 
-                // Only update progress if the EPG still exists in the store
-                // This prevents crashes when receiving updates for deleted EPGs
-                if (epg) {
-                  // Update the store with progress information
-                  updateEPGProgress(parsedEvent.data);
-                } else {
-                  // EPG was deleted, ignore this update
-                  console.debug(
-                    `Ignoring EPG refresh update for deleted EPG ${sourceId}`
-                  );
+                // If EPG not in store yet (e.g. newly created and refresh started
+                // before the store was updated), fetch the EPG list and update progress.
+                if (!epg) {
+                  fetchEPGsForProgress().then(() => {
+                    updateEPGProgress(parsedEvent.data);
+                  });
                   break;
                 }
 
+                // Update the store with progress information
+                updateEPGProgress(parsedEvent.data);
+
                 if (epg) {
                   // Check for any indication of an error (either via status or error field)
                   const hasError =

frontend/src/api.js

@@ -3817,4 +3817,49 @@ export default class API {
       errorNotification('Failed to fetch connect logs', e);
     }
   }
+
+  static async getSDLineups(sourceId) {
+    try {
+      const response = await request(`${host}/api/epg/sources/${sourceId}/sd-lineups/`);
+      return response;
+    } catch (e) {
+      errorNotification('Failed to retrieve Schedules Direct lineups', e);
+    }
+  }
+
+  static async addSDLineup(sourceId, lineup) {
+    try {
+      const response = await request(`${host}/api/epg/sources/${sourceId}/sd-lineups/`, {
+        method: 'POST',
+        body: { lineup },
+      });
+      return response;
+    } catch (e) {
+      errorNotification(`Failed to add lineup ${lineup}`, e);
+    }
+  }
+
+  static async deleteSDLineup(sourceId, lineup) {
+    try {
+      const response = await request(`${host}/api/epg/sources/${sourceId}/sd-lineups/`, {
+        method: 'DELETE',
+        body: { lineup },
+      });
+      return response;
+    } catch (e) {
+      errorNotification(`Failed to remove lineup ${lineup}`, e);
+    }
+  }
+
+  static async searchSDLineups(sourceId, country, postalcode) {
+    try {
+      const response = await request(`${host}/api/epg/sources/${sourceId}/sd-lineups/search/`, {
+        method: 'POST',
+        body: { country, postalcode },
+      });
+      return response;
+    } catch (e) {
+      errorNotification('Failed to search Schedules Direct lineups', e);
+    }
+  }
 }

frontend/src/components/forms/EPG.jsx

@@ -1,5 +1,5 @@
 // Modal.js
-import React, { useState, useEffect } from 'react';
+import React, { useState, useEffect, useCallback } from 'react';
 import {
   TextInput,
   Button,
@@ -12,15 +12,333 @@ import {
   Divider,
   Box,
   Text,
+  Alert,
+  Select,
+  Loader,
+  Badge,
+  ScrollArea,
+  Table,
 } from '@mantine/core';
+import { TriangleAlert, Trash2, Plus, Search } from 'lucide-react';
 import { isNotEmpty, useForm } from '@mantine/form';
 import ScheduleInput from './ScheduleInput';
 import { addEPG, updateEPG } from '../../utils/forms/DummyEpgUtils.js';
 import { showNotification } from '../../utils/notificationUtils.js';
+import API from '../../api.js';
+
+// Countries are fetched dynamically from the SD API on component mount.
+// Fallback list used if the API call fails.
+const SD_COUNTRIES_FALLBACK = [
+  { value: 'USA', label: 'United States' },
+  { value: 'CAN', label: 'Canada' },
+  { value: 'GBR', label: 'United Kingdom' },
+  { value: 'AUS', label: 'Australia' },
+  { value: 'AUT', label: 'Austria' },
+  { value: 'BEL', label: 'Belgium' },
+  { value: 'DNK', label: 'Denmark' },
+  { value: 'FIN', label: 'Finland' },
+  { value: 'FRA', label: 'France' },
+  { value: 'DEU', label: 'Germany' },
+  { value: 'IRL', label: 'Ireland' },
+  { value: 'ITA', label: 'Italy' },
+  { value: 'NLD', label: 'Netherlands' },
+  { value: 'NZL', label: 'New Zealand' },
+];
+
+const SDLineupManager = ({ sourceId }) => {
+  const [countries, setCountries] = useState(SD_COUNTRIES_FALLBACK);
+  const [activeLineups, setActiveLineups] = useState([]);
+  const [lineupNotice, setLineupNotice] = useState(null);
+  const [changesRemaining, setChangesRemaining] = useState(null);
+  const [changesResetAt, setChangesResetAt] = useState(null);
+  const [searchResults, setSearchResults] = useState([]);
+  const [country, setCountry] = useState('USA');
+  const [postalCode, setPostalCode] = useState('');
+  const [loadingLineups, setLoadingLineups] = useState(false);
+  const [searching, setSearching] = useState(false);
+  const [addingLineup, setAddingLineup] = useState(null);
+  const [removingLineup, setRemovingLineup] = useState(null);
+  const maxLineups = 4;
+  const SD_DOCS_URL = 'https://github.com/SchedulesDirect/JSON-Service/wiki/API-20141201#tasks-your-client-must-perform';
+
+  const fetchActiveLineups = useCallback(async () => {
+    setLoadingLineups(true);
+    try {
+      const data = await API.getSDLineups(sourceId);
+      if (data) {
+        setActiveLineups(data.lineups || []);
+        setLineupNotice(data.notice || null);
+        // Always update changesRemaining from server — includes null (unknown) and 0 (locked)
+        if (data.changes_remaining !== undefined) {
+          setChangesRemaining(data.changes_remaining);
+        }
+        if (data.changes_reset_at !== undefined) {
+          setChangesResetAt(data.changes_reset_at);
+        }
+      }
+    } finally {
+      setLoadingLineups(false);
+    }
+  }, [sourceId]);
+
+  useEffect(() => {
+    if (sourceId) {
+      fetchActiveLineups();
+      // Fetch country list from SD API per their recommendation to not hardcode
+      fetch('https://json.schedulesdirect.org/20141201/available/countries')
+        .then(r => r.json())
+        .then(data => {
+          const all = Object.values(data).flat();
+          const mapped = all
+            .filter(c => c.shortName && c.fullName)
+            .map(c => ({ value: c.shortName, label: c.fullName }))
+            .sort((a, b) => a.label.localeCompare(b.label));
+          if (mapped.length > 0) setCountries(mapped);
+        })
+        .catch(() => {}); // fallback list remains if fetch fails
+    }
+  }, [sourceId, fetchActiveLineups]);
+
+  const handleSearch = async () => {
+    if (!postalCode.trim()) return;
+    setSearching(true);
+    setSearchResults([]);
+    try {
+      const data = await API.searchSDLineups(sourceId, country, postalCode.trim());
+      if (data) {
+        setSearchResults(data.lineups || []);
+      }
+    } finally {
+      setSearching(false);
+    }
+  };
+
+  const handleAdd = async (lineup) => {
+    setAddingLineup(lineup.lineup);
+    try {
+      const result = await API.addSDLineup(sourceId, lineup.lineup);
+      if (!result) return;
+
+      // Update changesRemaining from response
+      if (result.changes_remaining !== undefined && result.changes_remaining !== null) {
+        setChangesRemaining(result.changes_remaining);
+      }
+
+      if (result.error === 'daily_limit_reached') {
+        setChangesRemaining(0);
+        await fetchActiveLineups(); // reload to get reset_at from server
+        return;
+      }
+
+      if (result.error === 'duplicate_lineup') {
+        showNotification({ title: 'Already added', message: result.message, color: 'yellow' });
+        return;
+      }
+
+      if (result.error === 'max_lineups_reached') {
+        showNotification({ title: 'Maximum lineups reached', message: result.message, color: 'orange' });
+        return;
+      }
+
+      if (result.error) {
+        showNotification({ title: 'Unable to add lineup', message: result.message, color: 'red' });
+        return;
+      }
+
+      if (result.code === 0) {
+        showNotification({ title: 'Lineup added', message: lineup.name, color: 'green' });
+        await fetchActiveLineups();
+      }
+    } finally {
+      setAddingLineup(null);
+    }
+  };
+
+  const handleRemove = async (lineup) => {
+    setRemovingLineup(lineup.lineup);
+    try {
+      const result = await API.deleteSDLineup(sourceId, lineup.lineup);
+      if (result && result.code === 0) {
+        showNotification({ title: 'Lineup removed', message: lineup.name, color: 'blue' });
+        if (result.changes_remaining !== undefined && result.changes_remaining !== null) {
+          setChangesRemaining(result.changes_remaining);
+        }
+        await fetchActiveLineups();
+        setSearchResults([]);
+      }
+    } finally {
+      setRemovingLineup(null);
+    }
+  };
+
+  const activeLineupIds = new Set(activeLineups.map((l) => l.lineup));
+  const atMax = activeLineups.length >= maxLineups;
+
+  return (
+    <Box mt="md">
+      <Divider my="sm" />
+      <Group justify="space-between" mb="xs">
+        <Text size="sm" fw={500}>
+          Manage Lineups
+        </Text>
+        <Badge color={atMax ? 'red' : 'blue'} variant="light">
+          {loadingLineups ? '...' : `${activeLineups.length} / ${maxLineups}`}
+        </Badge>
+      </Group>
+
+      {/* Active lineups */}
+      {loadingLineups ? (
+        <Group justify="center" py="sm">
+          <Loader size="sm" />
+        </Group>
+      ) : activeLineups.length === 0 ? (
+        <Text size="xs" c="dimmed" mb="sm">
+          {lineupNotice || 'No lineups configured. Search below to add one.'}
+        </Text>
+      ) : (
+        <Box mb="sm">
+          {activeLineups.map((lineup) => (
+            <Group
+              key={lineup.lineup}
+              justify="space-between"
+              px="sm"
+              py="xs"
+              mb={4}
+              style={{
+                background: 'var(--mantine-color-default)',
+                border: '1px solid var(--mantine-color-default-border)',
+                borderRadius: 'var(--mantine-radius-sm)',
+              }}
+            >
+              <Box>
+                <Text size="sm" fw={500}>{lineup.name}</Text>
+                <Text size="xs" c="dimmed">
+                  {lineup.transport} · {lineup.location} · {lineup.lineup}
+                </Text>
+              </Box>
+              <Button
+                size="xs"
+                color="red"
+                variant="subtle"
+                leftSection={<Trash2 size={12} />}
+                loading={removingLineup === lineup.lineup}
+                onClick={() => handleRemove(lineup)}
+              >
+                Remove
+              </Button>
+            </Group>
+          ))}
+        </Box>
+      )}
+
+      {/* Search section */}
+      <Text size="xs" fw={500} c="dimmed" mb="xs" tt="uppercase">
+        Add a lineup
+      </Text>
+
+      {atMax && (
+        <Alert color="orange" variant="light" mb="xs" icon={<TriangleAlert size={14} />}>
+          Maximum of {maxLineups} lineups reached. Remove one to add another.
+        </Alert>
+      )}
+
+      {changesRemaining === 0 && (
+        <Alert color="red" variant="light" mb="xs" icon={<TriangleAlert size={14} />}>
+          You have reached your daily Schedules Direct lineup addition limit. SD allows 6 adds per 24-hour period.{' '}
+          {changesResetAt && (
+            <span>Limit resets at <strong>{new Date(changesResetAt).toUTCString()}</strong>. </span>
+          )}
+          {!changesResetAt && <span>Limit resets 24 hours after the first add of the day. </span>}
+          <a href={SD_DOCS_URL} target="_blank" rel="noopener noreferrer">Learn more</a>
+        </Alert>
+      )}
+
+      {changesRemaining === 1 && (
+        <Alert color="orange" variant="light" mb="xs" icon={<TriangleAlert size={14} />}>
+          You have <strong>1 lineup addition remaining</strong> today. Use it carefully — Schedules Direct limits adds to 6 per 24-hour period.{' '}
+          <a href={SD_DOCS_URL} target="_blank" rel="noopener noreferrer">Learn more</a>
+        </Alert>
+      )}
+
+      {changesRemaining === 2 && (
+        <Alert color="yellow" variant="light" mb="xs" icon={<TriangleAlert size={14} />}>
+          You have <strong>2 lineup additions remaining</strong> today. Schedules Direct limits adds to 6 per 24-hour period.{' '}
+          <a href={SD_DOCS_URL} target="_blank" rel="noopener noreferrer">Learn more</a>
+        </Alert>
+      )}
+
+      <Group align="flex-end" mb="sm" gap="xs">
+        <Select
+          label="Country"
+          data={countries}
+          value={country}
+          onChange={(v) => setCountry(v)}
+          style={{ flex: 1 }}
+          size="sm"
+        />
+        <TextInput
+          label="Postal code"
+          placeholder="e.g. 07030"
+          value={postalCode}
+          onChange={(e) => setPostalCode(e.currentTarget.value)}
+          onKeyDown={(e) => e.key === 'Enter' && handleSearch()}
+          style={{ flex: 1 }}
+          size="sm"
+        />
+        <Button
+          size="sm"
+          leftSection={<Search size={14} />}
+          loading={searching}
+          onClick={handleSearch}
+          disabled={!postalCode.trim() || changesRemaining === 0}
+        >
+          Search
+        </Button>
+      </Group>
+
+      {searchResults.length > 0 && (
+        <ScrollArea h={180} style={{ border: '1px solid var(--mantine-color-default-border)', borderRadius: 'var(--mantine-radius-sm)' }}>
+          <Table striped highlightOnHover withRowBorders={false} fz="xs">
+            <Table.Tbody>
+              {searchResults.map((lineup) => {
+                const isActive = activeLineupIds.has(lineup.lineup);
+                return (
+                  <Table.Tr key={lineup.lineup}>
+                    <Table.Td>
+                      <Text size="xs" fw={500}>{lineup.name}</Text>
+                      <Text size="xs" c="dimmed">{lineup.transport} · {lineup.location}</Text>
+                    </Table.Td>
+                    <Table.Td style={{ textAlign: 'right', whiteSpace: 'nowrap' }}>
+                      {isActive ? (
+                        <Badge color="green" variant="light" size="sm">Active</Badge>
+                      ) : (
+                        <Button
+                          size="xs"
+                          variant="subtle"
+                          leftSection={<Plus size={12} />}
+                          loading={addingLineup === lineup.lineup}
+                          disabled={atMax || changesRemaining === 0}
+                          onClick={() => handleAdd(lineup)}
+                        >
+                          Add
+                        </Button>
+                      )}
+                    </Table.Td>
+                  </Table.Tr>
+                );
+              })}
+            </Table.Tbody>
+          </Table>
+        </ScrollArea>
+      )}
+    </Box>
+  );
+};
 
 const EPG = ({ epg = null, isOpen, onClose }) => {
   const [sourceType, setSourceType] = useState('xmltv');
   const [scheduleType, setScheduleType] = useState('interval');
+  const [savedEpgId, setSavedEpgId] = useState(null);
 
   const form = useForm({
     mode: 'uncontrolled',
@@ -28,6 +346,7 @@ const EPG = ({ epg = null, isOpen, onClose }) => {
       name: '',
       source_type: 'xmltv',
       url: '',
+      username: '',
       api_key: '',
       is_active: true,
       refresh_interval: 24,
@@ -44,11 +363,9 @@ const EPG = ({ epg = null, isOpen, onClose }) => {
   const onSubmit = async () => {
     const values = form.getValues();
 
-    // Determine which schedule type is active based on field values
     const hasCronExpression =
       values.cron_expression && values.cron_expression.trim() !== '';
 
-    // Clear the field that isn't active based on actual field values
     if (hasCronExpression) {
       values.refresh_interval = 0;
     } else {
@@ -56,7 +373,6 @@ const EPG = ({ epg = null, isOpen, onClose }) => {
     }
 
     if (epg?.id) {
-      // Validate that we have a valid EPG object before updating
       if (!epg || typeof epg !== 'object' || !epg.id) {
         showNotification({
           title: 'Error',
@@ -67,12 +383,16 @@ const EPG = ({ epg = null, isOpen, onClose }) => {
       }
 
       await updateEPG(values, epg);
+      onClose();
     } else {
-      await addEPG(values);
+      const result = await addEPG(values);
+      if (result?.id) {
+        setSavedEpgId(result.id);
+      } else {
+        form.reset();
+        onClose();
+      }
     }
-
-    form.reset();
-    onClose();
   };
 
   useEffect(() => {
@@ -81,6 +401,7 @@ const EPG = ({ epg = null, isOpen, onClose }) => {
         name: epg.name,
         source_type: epg.source_type,
         url: epg.url,
+        username: epg.username || '',
         api_key: epg.api_key,
         is_active: epg.is_active,
         refresh_interval: epg.refresh_interval,
@@ -89,7 +410,7 @@ const EPG = ({ epg = null, isOpen, onClose }) => {
       };
       form.setValues(values);
       setSourceType(epg.source_type);
-      // Determine schedule type from existing data - check both fields
+      setSavedEpgId(epg.id);
       setScheduleType(
         epg.cron_expression && epg.cron_expression.trim() !== ''
           ? 'cron'
@@ -99,23 +420,31 @@ const EPG = ({ epg = null, isOpen, onClose }) => {
       form.reset();
       setSourceType('xmltv');
       setScheduleType('interval');
+      setSavedEpgId(null);
     }
     // eslint-disable-next-line react-hooks/exhaustive-deps
   }, [epg]);
 
-  // Function to handle source type changes
   const handleSourceTypeChange = (value) => {
     form.setFieldValue('source_type', value);
     setSourceType(value);
   };
 
+  const handleClose = () => {
+    form.reset();
+    setSavedEpgId(null);
+    onClose();
+  };
+
   if (!isOpen) {
     return <></>;
   }
 
+  const isSD = sourceType === 'schedules_direct';
+
   return (
     <>
-      <Modal opened={isOpen} onClose={onClose} title="EPG Source" size={700}>
+      <Modal opened={isOpen} onClose={handleClose} title="EPG Source" size={700}>
         <form onSubmit={form.onSubmit(onSubmit)}>
           <Group justify="space-between" align="top">
             {/* Left Column */}
@@ -137,20 +466,28 @@ const EPG = ({ epg = null, isOpen, onClose }) => {
                 {...form.getInputProps('source_type')}
                 key={form.key('source_type')}
                 data={[
-                  {
-                    label: 'XMLTV',
-                    value: 'xmltv',
-                  },
-                  {
-                    label: 'Schedules Direct',
-                    value: 'schedules_direct',
-                  },
+                  { label: 'XMLTV', value: 'xmltv' },
+                  { label: 'Schedules Direct', value: 'schedules_direct' },
                 ]}
                 onChange={(event) =>
                   handleSourceTypeChange(event.currentTarget.value)
                 }
               />
 
+              {isSD && (
+                <Alert
+                  icon={<TriangleAlert size={16} />}
+                  color="yellow"
+                  variant="light"
+                  title="Schedules Direct API Limits"
+                >
+                  Schedules Direct enforces a limit of approximately 200 requests per 2 hours.
+                  Each refresh uses ~15 requests. Avoid frequent manual refreshes to prevent
+                  your account from being temporarily blocked. A 24-hour refresh interval
+                  is recommended.
+                </Alert>
+              )}
+
               <ScheduleInput
                 scheduleType={scheduleType}
                 onScheduleTypeChange={setScheduleType}
@@ -165,32 +502,80 @@ const EPG = ({ epg = null, isOpen, onClose }) => {
                 intervalLabel="Refresh Interval (hours)"
                 intervalDescription="How often to refresh EPG data (0 to disable)"
               />
+
+              {isSD && scheduleType === 'cron' && (
+                <Alert
+                  icon={<TriangleAlert size={16} />}
+                  color="orange"
+                  variant="light"
+                  title="Schedules Direct Rate Limit Warning"
+                >
+                  <strong>Exceeding SD's rate limits may result in a permanent account ban.</strong>{' '}
+                  Schedules Direct enforces a limit of ~200 API requests per 2-hour window.
+                  Your cron schedule must not trigger a refresh more than once every 2 hours.
+                  The recommended minimum schedule is{' '}
+                  <code>0 */2 * * *</code> (every 2 hours).{' '}
+                  <a
+                    href="https://github.com/SchedulesDirect/JSON-Service/wiki/API-20141201#tasks-your-client-must-perform"
+                    target="_blank"
+                    rel="noopener noreferrer"
+                  >
+                    Learn more
+                  </a>
+                </Alert>
+              )}
+
+              {isSD && scheduleType === 'interval' && form.getValues().refresh_interval > 0 && form.getValues().refresh_interval < 2 && (
+                <Alert
+                  icon={<TriangleAlert size={16} />}
+                  color="red"
+                  variant="light"
+                  title="Interval Too Short"
+                >
+                  <strong>Schedules Direct requires a minimum refresh interval of 2 hours.</strong>{' '}
+                  Setting a shorter interval may result in your account being banned.
+                  Please set the interval to 2 or higher, or use 0 to disable auto-refresh.
+                </Alert>
+              )}
             </Stack>
 
             <Divider size="sm" orientation="vertical" />
 
             {/* Right Column */}
             <Stack gap="md" style={{ flex: 1 }}>
-              <TextInput
-                id="url"
-                name="url"
-                label="URL"
-                description="Direct URL to the XMLTV file or API endpoint"
-                {...form.getInputProps('url')}
-                key={form.key('url')}
-              />
-
-              {sourceType === 'schedules_direct' && (
+              {!isSD && (
                 <TextInput
-                  id="api_key"
-                  name="api_key"
-                  label="API Key"
-                  description="API key for services that require authentication"
-                  {...form.getInputProps('api_key')}
-                  key={form.key('api_key')}
+                  id="url"
+                  name="url"
+                  label="URL"
+                  description="Direct URL to the XMLTV file or API endpoint"
+                  {...form.getInputProps('url')}
+                  key={form.key('url')}
                 />
               )}
 
+              {isSD && (
+                <>
+                  <TextInput
+                    id="username"
+                    name="username"
+                    label="Username"
+                    description="Your Schedules Direct account username"
+                    {...form.getInputProps('username')}
+                    key={form.key('username')}
+                  />
+                  <TextInput
+                    id="api_key"
+                    name="api_key"
+                    label="Password"
+                    description="Your Schedules Direct account password"
+                    type="password"
+                    {...form.getInputProps('api_key')}
+                    key={form.key('api_key')}
+                  />
+                </>
+              )}
+
               <NumberInput
                 min={0}
                 max={999}
@@ -200,7 +585,6 @@ const EPG = ({ epg = null, isOpen, onClose }) => {
                 key={form.key('priority')}
               />
 
-              {/* Put checkbox at the same level as Refresh Interval */}
               <Box style={{ marginTop: 0 }}>
                 <Text size="sm" fw={500} mb={3}>
                   Status
@@ -228,12 +612,23 @@ const EPG = ({ epg = null, isOpen, onClose }) => {
             </Stack>
           </Group>
 
+          {/* Lineup Manager — only shown for saved SD sources */}
+          {isSD && savedEpgId && (
+            <SDLineupManager sourceId={savedEpgId} />
+          )}
+
           {/* Full Width Section */}
           <Box mt="md">
             <Divider my="sm" />
 
+            {isSD && !savedEpgId && (
+              <Text size="xs" c="dimmed" mb="sm">
+                Save this source first to manage Schedules Direct lineups.
+              </Text>
+            )}
+
             <Group justify="end" mt="xl">
-              <Button variant="outline" onClick={onClose}>
+              <Button variant="outline" onClick={handleClose}>
                 Cancel
               </Button>
               <Button type="submit" variant="filled" disabled={form.submitting}>

frontend/src/components/tables/EPGsTable.jsx

@@ -99,6 +99,127 @@ const RowActions = ({ tableSize, row, editEPG, deleteEPG, refreshEPG }) => {
   );
 };
 
+
+const EPGStatusCell = ({ epg }) => {
+  // Direct Zustand subscription scoped to this source only.
+  // This component re-renders whenever its source's progress changes,
+  // independent of the parent table's render cycle.
+  const progress = useEPGsStore((s) => s.refreshProgress[epg.id]);
+  const theme = useMantineTheme();
+
+  const isDummyEPG = epg.source_type === 'dummy';
+  if (isDummyEPG) return null;
+
+  // Show progress bar if an active fetch is in progress
+  if (
+    progress &&
+    (progress.progress < 100 ||
+      progress.status === 'in_progress' ||
+      (progress.action === 'parsing_channels' && epg.status === 'parsing'))
+  ) {
+    let label = '';
+    switch (progress.action) {
+      case 'downloading':
+        label = 'Downloading';
+        break;
+      case 'extracting':
+        label = 'Extracting';
+        break;
+      case 'parsing_channels':
+        label = 'Parsing Channels';
+        break;
+      case 'parsing_programs':
+        label = 'Parsing Programs';
+        break;
+      default:
+        return null;
+    }
+
+    let additionalInfo = '';
+    if (progress.message) {
+      additionalInfo = progress.message;
+    } else if (progress.processed !== undefined && progress.channels !== undefined) {
+      additionalInfo = `${progress.processed.toLocaleString()} programs for ${progress.channels} channels`;
+    } else if (progress.processed !== undefined && progress.total !== undefined) {
+      additionalInfo = `${progress.processed.toLocaleString()} / ${progress.total.toLocaleString()}`;
+    }
+
+    return (
+      <Stack spacing={2}>
+        <Text size="xs">
+          {label}: {parseInt(progress.progress)}%
+        </Text>
+        <Progress
+          value={parseInt(progress.progress)}
+          size="xs"
+          style={{ margin: '2px 0' }}
+        />
+        {progress.speed && (
+          <Text size="xs" c="dimmed">
+            Speed: {parseInt(progress.speed)} KB/s
+          </Text>
+        )}
+        {additionalInfo && (
+          <Text size="xs" c="dimmed" lineClamp={1}>
+            {additionalInfo}
+          </Text>
+        )}
+      </Stack>
+    );
+  }
+
+  // Show error message
+  if (epg.status === 'error' && epg.last_message) {
+    return (
+      <Tooltip label={epg.last_message} multiline width={300}>
+        <Text
+          c="dimmed"
+          size="xs"
+          lineClamp={2}
+          style={{ color: theme.colors.red[6], lineHeight: 1.3 }}
+        >
+          {epg.last_message}
+        </Text>
+      </Tooltip>
+    );
+  }
+
+  // Show success message
+  if (epg.status === 'success') {
+    const successMessage = epg.last_message || 'EPG data refreshed successfully';
+    return (
+      <Tooltip label={successMessage} multiline width={300}>
+        <Text
+          c="dimmed"
+          size="xs"
+          lineClamp={2}
+          style={{ color: theme.colors.green[6], lineHeight: 1.3 }}
+        >
+          {successMessage}
+        </Text>
+      </Tooltip>
+    );
+  }
+
+  // Show idle message
+  if (epg.status === 'idle' && epg.last_message) {
+    return (
+      <Tooltip label={epg.last_message} multiline width={300}>
+        <Text
+          c="dimmed"
+          size="xs"
+          lineClamp={2}
+          style={{ lineHeight: 1.3 }}
+        >
+          {epg.last_message}
+        </Text>
+      </Tooltip>
+    );
+  }
+
+  return null;
+};
+
 const EPGsTable = () => {
   const [epg, setEPG] = useState(null);
   const [epgModalOpen, setEPGModalOpen] = useState(false);
@@ -111,7 +232,6 @@ const EPGsTable = () => {
   const [deleting, setDeleting] = useState(false);
 
   const epgs = useEPGsStore((s) => s.epgs);
-  const refreshProgress = useEPGsStore((s) => s.refreshProgress);
 
   const theme = useMantineTheme();
   const { fullDateTimeFormat } = useDateTimeFormat();
@@ -140,69 +260,6 @@ const EPGsTable = () => {
     }
   };
 
-  const buildProgressDisplay = (data) => {
-    const progress = refreshProgress[data.id] || null;
-
-    if (!progress) return null;
-
-    let label = '';
-    switch (progress.action) {
-      case 'downloading':
-        label = 'Downloading';
-        break;
-      case 'extracting':
-        label = 'Extracting';
-        break;
-      case 'parsing_channels':
-        label = 'Parsing Channels';
-        break;
-      case 'parsing_programs':
-        label = 'Parsing Programs';
-        break;
-      default:
-        return null;
-    }
-
-    // Build additional info string from progress data
-    let additionalInfo = '';
-    if (progress.message) {
-      additionalInfo = progress.message;
-    } else if (
-      progress.processed !== undefined &&
-      progress.channels !== undefined
-    ) {
-      additionalInfo = `${progress.processed.toLocaleString()} programs for ${progress.channels} channels`;
-    } else if (
-      progress.processed !== undefined &&
-      progress.total !== undefined
-    ) {
-      additionalInfo = `${progress.processed.toLocaleString()} / ${progress.total.toLocaleString()}`;
-    }
-
-    return (
-      <Stack spacing={2}>
-        <Text size="xs">
-          {label}: {parseInt(progress.progress)}%
-        </Text>
-        <Progress
-          value={parseInt(progress.progress)}
-          size="xs"
-          style={{ margin: '2px 0' }}
-        />
-        {progress.speed && (
-          <Text size="xs" c="dimmed">
-            Speed: {parseInt(progress.speed)} KB/s
-          </Text>
-        )}
-        {additionalInfo && (
-          <Text size="xs" c="dimmed" lineClamp={1}>
-            {additionalInfo}
-          </Text>
-        )}
-      </Stack>
-    );
-  };
-
   const columns = useMemo(
     //column definitions...
     () => [
@@ -214,21 +271,42 @@ const EPGsTable = () => {
       {
         header: 'Type',
         accessorKey: 'source_type',
-        size: 100,
+        size: 130,
+        cell: ({ cell }) => {
+          const typeMap = {
+            'xmltv': 'XMLTV',
+            'schedules_direct': 'Schedules Direct',
+            'dummy': 'Custom Dummy',
+          };
+          return typeMap[cell.getValue()] || cell.getValue();
+        },
       },
       {
-        header: 'URL / API Key / File Path',
+        header: 'Source / Credentials / File Path',
         accessorKey: 'url',
         enableSorting: false,
         minSize: 250,
         cell: ({ cell, row }) => {
-          const value =
-            cell.getValue() ||
-            row.original.api_key ||
-            row.original.file_path ||
-            '';
+          const sourceType = row.original.source_type;
+          let value = '';
+          let tooltip = '';
+
+          if (sourceType === 'schedules_direct') {
+            // Never expose credentials — show username only
+            const username = row.original.username || '';
+            value = username ? `User: ${username}` : '(credentials set)';
+            tooltip = value;
+          } else {
+            value =
+              cell.getValue() ||
+              row.original.api_key ||
+              row.original.file_path ||
+              '';
+            tooltip = value;
+          }
+
           return (
-            <Tooltip label={value} disabled={!value}>
+            <Tooltip label={tooltip} disabled={!tooltip}>
               <div
                 style={{
                   whiteSpace: 'nowrap',
@@ -267,76 +345,7 @@ const EPGsTable = () => {
         enableSorting: false,
         minSize: 250,
         grow: true,
-        cell: ({ row }) => {
-          const data = row.original;
-          const isDummyEPG = data.source_type === 'dummy';
-
-          // Dummy EPGs don't have status messages
-          if (isDummyEPG) {
-            return null;
-          }
-
-          // Check if there's an active progress for this EPG - show progress first if active
-          if (
-            refreshProgress[data.id] &&
-            refreshProgress[data.id].progress < 100
-          ) {
-            return buildProgressDisplay(data);
-          }
-
-          // Show error message when status is error
-          if (data.status === 'error' && data.last_message) {
-            return (
-              <Tooltip label={data.last_message} multiline width={300}>
-                <Text
-                  c="dimmed"
-                  size="xs"
-                  lineClamp={2}
-                  style={{ color: theme.colors.red[6], lineHeight: 1.3 }}
-                >
-                  {data.last_message}
-                </Text>
-              </Tooltip>
-            );
-          }
-
-          // Show success message for successful sources
-          if (data.status === 'success') {
-            const successMessage =
-              data.last_message || 'EPG data refreshed successfully';
-            return (
-              <Tooltip label={successMessage} multiline width={300}>
-                <Text
-                  c="dimmed"
-                  size="xs"
-                  lineClamp={2}
-                  style={{ color: theme.colors.green[6], lineHeight: 1.3 }}
-                >
-                  {successMessage}
-                </Text>
-              </Tooltip>
-            );
-          }
-
-          // Show last_message for idle sources (from previous refresh)
-          if (data.status === 'idle' && data.last_message) {
-            return (
-              <Tooltip label={data.last_message} multiline width={300}>
-                <Text
-                  c="dimmed"
-                  size="xs"
-                  lineClamp={2}
-                  style={{ lineHeight: 1.3 }}
-                >
-                  {data.last_message}
-                </Text>
-              </Tooltip>
-            );
-          }
-
-          // Otherwise return empty cell
-          return null;
-        },
+        cell: ({ row }) => <EPGStatusCell epg={row.original} />,
       },
       {
         header: 'Updated',
@@ -380,7 +389,7 @@ const EPGsTable = () => {
         size: tableSize == 'compact' ? 75 : 100,
       },
     ],
-    [refreshProgress, fullDateTimeFormat]
+    [fullDateTimeFormat]
   );
 
   const [isLoading, setIsLoading] = useState(true);