diff --git a/apps/m3u/api_urls.py b/apps/m3u/api_urls.py index 41fc2fbc..80e54bb2 100644 --- a/apps/m3u/api_urls.py +++ b/apps/m3u/api_urls.py @@ -1,18 +1,38 @@ from django.urls import path, include from rest_framework.routers import DefaultRouter -from .api_views import M3UAccountViewSet, M3UFilterViewSet, ServerGroupViewSet, RefreshM3UAPIView, RefreshSingleM3UAPIView, UserAgentViewSet, M3UAccountProfileViewSet +from .api_views import ( + M3UAccountViewSet, + M3UFilterViewSet, + ServerGroupViewSet, + RefreshM3UAPIView, + RefreshSingleM3UAPIView, + UserAgentViewSet, + M3UAccountProfileViewSet, +) -app_name = 'm3u' +app_name = "m3u" router = DefaultRouter() -router.register(r'accounts', M3UAccountViewSet, basename='m3u-account') -router.register(r'accounts\/(?P\d+)\/profiles', M3UAccountProfileViewSet, basename='m3u-account-profiles') -router.register(r'filters', M3UFilterViewSet, basename='m3u-filter') -router.register(r'server-groups', ServerGroupViewSet, basename='server-group') +router.register(r"accounts", M3UAccountViewSet, basename="m3u-account") +router.register( + r"accounts\/(?P\d+)\/profiles", + M3UAccountProfileViewSet, + basename="m3u-account-profiles", +) +router.register( + r"accounts\/(?P\d+)\/filters", + M3UFilterViewSet, + basename="m3u-filters", +) +router.register(r"server-groups", ServerGroupViewSet, basename="server-group") urlpatterns = [ - path('refresh/', RefreshM3UAPIView.as_view(), name='m3u_refresh'), - path('refresh//', RefreshSingleM3UAPIView.as_view(), name='m3u_refresh_single'), + path("refresh/", RefreshM3UAPIView.as_view(), name="m3u_refresh"), + path( + "refresh//", + RefreshSingleM3UAPIView.as_view(), + name="m3u_refresh_single", + ), ] urlpatterns += router.urls diff --git a/apps/m3u/api_views.py b/apps/m3u/api_views.py index d3739f19..46676e93 100644 --- a/apps/m3u/api_views.py +++ b/apps/m3u/api_views.py @@ -183,8 +183,6 @@ class M3UAccountViewSet(viewsets.ModelViewSet): class M3UFilterViewSet(viewsets.ModelViewSet): - """Handles CRUD operations for M3U filters""" - queryset = M3UFilter.objects.all() serializer_class = M3UFilterSerializer @@ -194,6 +192,23 @@ class M3UFilterViewSet(viewsets.ModelViewSet): except KeyError: return [Authenticated()] + def get_queryset(self): + m3u_account_id = self.kwargs["account_id"] + return M3UFilter.objects.filter(m3u_account_id=m3u_account_id) + + def perform_create(self, serializer): + # Get the account ID from the URL + account_id = self.kwargs["account_id"] + + # # Get the M3UAccount instance for the account_id + # m3u_account = M3UAccount.objects.get(id=account_id) + + # Save the 'm3u_account' in the serializer context + serializer.context["m3u_account"] = account_id + + # Perform the actual save + serializer.save(m3u_account_id=account_id) + class ServerGroupViewSet(viewsets.ModelViewSet): """Handles CRUD operations for Server Groups""" diff --git a/apps/m3u/migrations/0013_alter_m3ufilter_filter_type.py b/apps/m3u/migrations/0013_alter_m3ufilter_filter_type.py new file mode 100644 index 00000000..0b0a8a1d --- /dev/null +++ b/apps/m3u/migrations/0013_alter_m3ufilter_filter_type.py @@ -0,0 +1,18 @@ +# Generated by Django 5.1.6 on 2025-07-22 21:16 + +from django.db import migrations, models + + +class Migration(migrations.Migration): + + dependencies = [ + ('m3u', '0012_alter_m3uaccount_refresh_interval'), + ] + + operations = [ + migrations.AlterField( + model_name='m3ufilter', + name='filter_type', + field=models.CharField(choices=[('group', 'Group'), ('name', 'Stream Name'), ('url', 'Stream URL')], default='group', help_text='Filter based on either group title or stream name.', max_length=50), + ), + ] diff --git a/apps/m3u/migrations/0014_alter_m3ufilter_options_m3ufilter_order.py b/apps/m3u/migrations/0014_alter_m3ufilter_options_m3ufilter_order.py new file mode 100644 index 00000000..3510bfc5 --- /dev/null +++ b/apps/m3u/migrations/0014_alter_m3ufilter_options_m3ufilter_order.py @@ -0,0 +1,22 @@ +# Generated by Django 5.1.6 on 2025-07-31 17:14 + +from django.db import migrations, models + + +class Migration(migrations.Migration): + + dependencies = [ + ('m3u', '0013_alter_m3ufilter_filter_type'), + ] + + operations = [ + migrations.AlterModelOptions( + name='m3ufilter', + options={'ordering': ['order']}, + ), + migrations.AddField( + model_name='m3ufilter', + name='order', + field=models.PositiveIntegerField(default=0), + ), + ] diff --git a/apps/m3u/models.py b/apps/m3u/models.py index 94ec88fc..b7993ef6 100644 --- a/apps/m3u/models.py +++ b/apps/m3u/models.py @@ -155,9 +155,11 @@ class M3UFilter(models.Model): """Defines filters for M3U accounts based on stream name or group title.""" FILTER_TYPE_CHOICES = ( - ("group", "Group Title"), + ("group", "Group"), ("name", "Stream Name"), + ("url", "Stream URL"), ) + m3u_account = models.ForeignKey( M3UAccount, on_delete=models.CASCADE, @@ -177,6 +179,7 @@ class M3UFilter(models.Model): default=True, help_text="If True, matching items are excluded; if False, only matches are included.", ) + order = models.PositiveIntegerField(default=0) def applies_to(self, stream_name, group_name): target = group_name if self.filter_type == "group" else stream_name @@ -226,9 +229,6 @@ class ServerGroup(models.Model): return self.name -from django.db import models - - class M3UAccountProfile(models.Model): """Represents a profile associated with an M3U Account.""" diff --git a/apps/m3u/serializers.py b/apps/m3u/serializers.py index a86227aa..3bf0e335 100644 --- a/apps/m3u/serializers.py +++ b/apps/m3u/serializers.py @@ -16,11 +16,9 @@ logger = logging.getLogger(__name__) class M3UFilterSerializer(serializers.ModelSerializer): """Serializer for M3U Filters""" - channel_groups = ChannelGroupM3UAccountSerializer(source="m3u_account", many=True) - class Meta: model = M3UFilter - fields = ["id", "filter_type", "regex_pattern", "exclude", "channel_groups"] + fields = ["id", "filter_type", "regex_pattern", "exclude", "order"] class M3UAccountProfileSerializer(serializers.ModelSerializer): @@ -64,7 +62,7 @@ class M3UAccountProfileSerializer(serializers.ModelSerializer): class M3UAccountSerializer(serializers.ModelSerializer): """Serializer for M3U Account""" - filters = M3UFilterSerializer(many=True, read_only=True) + filters = serializers.SerializerMethodField() # Include user_agent as a mandatory field using its primary key. user_agent = serializers.PrimaryKeyRelatedField( queryset=UserAgent.objects.all(), @@ -149,6 +147,10 @@ class M3UAccountSerializer(serializers.ModelSerializer): return instance + def get_filters(self, obj): + filters = obj.filters.order_by("order") + return M3UFilterSerializer(filters, many=True).data + class ServerGroupSerializer(serializers.ModelSerializer): """Serializer for Server Group""" diff --git a/apps/m3u/tasks.py b/apps/m3u/tasks.py index 40a395ce..588705a4 100644 --- a/apps/m3u/tasks.py +++ b/apps/m3u/tasks.py @@ -18,7 +18,12 @@ from channels.layers import get_channel_layer from django.utils import timezone import time import json -from core.utils import RedisClient, acquire_task_lock, release_task_lock, natural_sort_key +from core.utils import ( + RedisClient, + acquire_task_lock, + release_task_lock, + natural_sort_key, +) from core.models import CoreSettings, UserAgent from asgiref.sync import async_to_sync from core.xtream_codes import Client as XCClient @@ -29,6 +34,7 @@ logger = logging.getLogger(__name__) BATCH_SIZE = 1000 m3u_dir = os.path.join(settings.MEDIA_ROOT, "cached_m3u") + def fetch_m3u_lines(account, use_cache=False): os.makedirs(m3u_dir, exist_ok=True) file_path = os.path.join(m3u_dir, f"{account.id}.m3u") @@ -39,27 +45,35 @@ def fetch_m3u_lines(account, use_cache=False): try: # Try to get account-specific user agent first user_agent_obj = account.get_user_agent() - user_agent = user_agent_obj.user_agent if user_agent_obj else "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36" + user_agent = ( + user_agent_obj.user_agent + if user_agent_obj + else "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36" + ) - logger.debug(f"Using user agent: {user_agent} for M3U account: {account.name}") + logger.debug( + f"Using user agent: {user_agent} for M3U account: {account.name}" + ) headers = {"User-Agent": user_agent} logger.info(f"Fetching from URL {account.server_url}") # Set account status to FETCHING before starting download account.status = M3UAccount.Status.FETCHING account.last_message = "Starting download..." - account.save(update_fields=['status', 'last_message']) + account.save(update_fields=["status", "last_message"]) - response = requests.get(account.server_url, headers=headers, stream=True) + response = requests.get( + account.server_url, headers=headers, stream=True + ) response.raise_for_status() - total_size = int(response.headers.get('Content-Length', 0)) + total_size = int(response.headers.get("Content-Length", 0)) downloaded = 0 start_time = time.time() last_update_time = start_time progress = 0 - with open(file_path, 'wb') as file: + with open(file_path, "wb") as file: send_m3u_update(account.id, "downloading", 0) for chunk in response.iter_content(chunk_size=8192): if chunk: @@ -76,7 +90,11 @@ def fetch_m3u_lines(account, use_cache=False): progress = (downloaded / total_size) * 100 # Time remaining (in seconds) - time_remaining = (total_size - downloaded) / (speed * 1024) if speed > 0 else 0 + time_remaining = ( + (total_size - downloaded) / (speed * 1024) + if speed > 0 + else 0 + ) current_time = time.time() if current_time - last_update_time >= 0.5: @@ -85,26 +103,36 @@ def fetch_m3u_lines(account, use_cache=False): # Update the account's last_message with detailed progress info progress_msg = f"Downloading: {progress:.1f}% - {speed:.1f} KB/s - {time_remaining:.1f}s remaining" account.last_message = progress_msg - account.save(update_fields=['last_message']) + account.save(update_fields=["last_message"]) - send_m3u_update(account.id, "downloading", progress, - speed=speed, - elapsed_time=elapsed_time, - time_remaining=time_remaining, - message=progress_msg) + send_m3u_update( + account.id, + "downloading", + progress, + speed=speed, + elapsed_time=elapsed_time, + time_remaining=time_remaining, + message=progress_msg, + ) # Final update with 100% progress final_msg = f"Download complete. Size: {total_size/1024/1024:.2f} MB, Time: {time.time() - start_time:.1f}s" account.last_message = final_msg - account.save(update_fields=['last_message']) + account.save(update_fields=["last_message"]) send_m3u_update(account.id, "downloading", 100, message=final_msg) except Exception as e: logger.error(f"Error fetching M3U from URL {account.server_url}: {e}") # Update account status and send error notification account.status = M3UAccount.Status.ERROR account.last_message = f"Error downloading M3U file: {str(e)}" - account.save(update_fields=['status', 'last_message']) - send_m3u_update(account.id, "downloading", 100, status="error", error=f"Error downloading M3U file: {str(e)}") + account.save(update_fields=["status", "last_message"]) + send_m3u_update( + account.id, + "downloading", + 100, + status="error", + error=f"Error downloading M3U file: {str(e)}", + ) return [], False # Return empty list and False for success # Check if the file exists and is not empty @@ -113,45 +141,55 @@ def fetch_m3u_lines(account, use_cache=False): logger.error(error_msg) account.status = M3UAccount.Status.ERROR account.last_message = error_msg - account.save(update_fields=['status', 'last_message']) - send_m3u_update(account.id, "downloading", 100, status="error", error=error_msg) + account.save(update_fields=["status", "last_message"]) + send_m3u_update( + account.id, "downloading", 100, status="error", error=error_msg + ) return [], False # Return empty list and False for success try: - with open(file_path, 'r', encoding='utf-8') as f: + with open(file_path, "r", encoding="utf-8") as f: return f.readlines(), True except Exception as e: error_msg = f"Error reading M3U file: {str(e)}" logger.error(error_msg) account.status = M3UAccount.Status.ERROR account.last_message = error_msg - account.save(update_fields=['status', 'last_message']) - send_m3u_update(account.id, "downloading", 100, status="error", error=error_msg) + account.save(update_fields=["status", "last_message"]) + send_m3u_update( + account.id, "downloading", 100, status="error", error=error_msg + ) return [], False elif account.file_path: try: - if account.file_path.endswith('.gz'): - with gzip.open(account.file_path, 'rt', encoding='utf-8') as f: + if account.file_path.endswith(".gz"): + with gzip.open(account.file_path, "rt", encoding="utf-8") as f: return f.readlines(), True - elif account.file_path.endswith('.zip'): - with zipfile.ZipFile(account.file_path, 'r') as zip_file: + elif account.file_path.endswith(".zip"): + with zipfile.ZipFile(account.file_path, "r") as zip_file: for name in zip_file.namelist(): - if name.endswith('.m3u'): + if name.endswith(".m3u"): with zip_file.open(name) as f: - return [line.decode('utf-8') for line in f.readlines()], True + return [ + line.decode("utf-8") for line in f.readlines() + ], True - error_msg = f"No .m3u file found in ZIP archive: {account.file_path}" + error_msg = ( + f"No .m3u file found in ZIP archive: {account.file_path}" + ) logger.warning(error_msg) account.status = M3UAccount.Status.ERROR account.last_message = error_msg - account.save(update_fields=['status', 'last_message']) - send_m3u_update(account.id, "downloading", 100, status="error", error=error_msg) + account.save(update_fields=["status", "last_message"]) + send_m3u_update( + account.id, "downloading", 100, status="error", error=error_msg + ) return [], False else: - with open(account.file_path, 'r', encoding='utf-8') as f: + with open(account.file_path, "r", encoding="utf-8") as f: return f.readlines(), True except (IOError, OSError, zipfile.BadZipFile, gzip.BadGzipFile) as e: @@ -159,8 +197,10 @@ def fetch_m3u_lines(account, use_cache=False): logger.error(error_msg) account.status = M3UAccount.Status.ERROR account.last_message = error_msg - account.save(update_fields=['status', 'last_message']) - send_m3u_update(account.id, "downloading", 100, status="error", error=error_msg) + account.save(update_fields=["status", "last_message"]) + send_m3u_update( + account.id, "downloading", 100, status="error", error=error_msg + ) return [], False # Neither server_url nor uploaded_file is available @@ -168,10 +208,11 @@ def fetch_m3u_lines(account, use_cache=False): logger.error(error_msg) account.status = M3UAccount.Status.ERROR account.last_message = error_msg - account.save(update_fields=['status', 'last_message']) + account.save(update_fields=["status", "last_message"]) send_m3u_update(account.id, "downloading", 100, status="error", error=error_msg) return [], False + def get_case_insensitive_attr(attributes, key, default=""): """Get attribute value using case-insensitive key lookup.""" for attr_key, attr_value in attributes.items(): @@ -179,6 +220,7 @@ def get_case_insensitive_attr(attributes, key, default=""): return attr_value return default + def parse_extinf_line(line: str) -> dict: """ Parse an EXTINF line from an M3U file. @@ -192,7 +234,7 @@ def parse_extinf_line(line: str) -> dict: """ if not line.startswith("#EXTINF:"): return None - content = line[len("#EXTINF:"):].strip() + content = line[len("#EXTINF:") :].strip() # Split on the first comma that is not inside quotes. parts = re.split(r',(?=(?:[^"]*"[^"]*")*[^"]*$)', content, maxsplit=1) if len(parts) != 2: @@ -200,21 +242,9 @@ def parse_extinf_line(line: str) -> dict: attributes_part, display_name = parts[0], parts[1].strip() attrs = dict(re.findall(r'([^\s]+)=["\']([^"\']+)["\']', attributes_part)) # Use tvg-name attribute if available; otherwise, use the display name. - name = get_case_insensitive_attr(attrs, 'tvg-name', display_name) - return { - 'attributes': attrs, - 'display_name': display_name, - 'name': name - } + name = get_case_insensitive_attr(attrs, "tvg-name", display_name) + return {"attributes": attrs, "display_name": display_name, "name": name} -def _matches_filters(stream_name: str, group_name: str, filters): - """Check if a stream or group name matches a precompiled regex filter.""" - compiled_filters = [(re.compile(f.regex_pattern, re.IGNORECASE), f.exclude) for f in filters] - for pattern, exclude in compiled_filters: - target = group_name if f.filter_type == 'group' else stream_name - if pattern.search(target or ''): - return exclude - return False @shared_task def refresh_m3u_accounts(): @@ -229,6 +259,7 @@ def refresh_m3u_accounts(): logger.info(msg) return msg + def check_field_lengths(streams_to_create): for stream in streams_to_create: for field, value in stream.__dict__.items(): @@ -238,19 +269,44 @@ def check_field_lengths(streams_to_create): print("") print("") + @shared_task def process_groups(account, groups): - existing_groups = {group.name: group for group in ChannelGroup.objects.filter(name__in=groups.keys())} + existing_groups = { + group.name: group + for group in ChannelGroup.objects.filter(name__in=groups.keys()) + } logger.info(f"Currently {len(existing_groups)} existing groups") + compiled_filters = [ + (re.compile(f.regex_pattern), f) + for f in account.filters.order_by("order") + if f.filter_type == "group" + ] + group_objs = [] groups_to_create = [] for group_name, custom_props in groups.items(): logger.debug(f"Handling group for M3U account {account.id}: {group_name}") - if (group_name not in existing_groups): - groups_to_create.append(ChannelGroup( - name=group_name, - )) + + include = True + for pattern, filter in compiled_filters: + if pattern.search(group_name): + logger.debug( + f"Group {group_name} matches filter pattern {filter.regex_pattern}" + ) + include = not filter.exclude + break + + if not include: + continue + + if group_name not in existing_groups: + groups_to_create.append( + ChannelGroup( + name=group_name, + ) + ) else: group_objs.append(existing_groups[group_name]) @@ -264,17 +320,17 @@ def process_groups(account, groups): for group in group_objs: # Ensure we include the xc_id in the custom_properties custom_props = groups.get(group.name, {}) - relations.append(ChannelGroupM3UAccount( - channel_group=group, - m3u_account=account, - custom_properties=json.dumps(custom_props), - enabled=True, # Default to enabled - )) + relations.append( + ChannelGroupM3UAccount( + channel_group=group, + m3u_account=account, + custom_properties=json.dumps(custom_props), + enabled=True, # Default to enabled + ) + ) + + ChannelGroupM3UAccount.objects.bulk_create(relations, ignore_conflicts=True) - ChannelGroupM3UAccount.objects.bulk_create( - relations, - ignore_conflicts=True - ) @shared_task def process_xc_category(account_id, batch, groups, hash_keys): @@ -285,14 +341,21 @@ def process_xc_category(account_id, batch, groups, hash_keys): stream_hashes = {} try: - with XCClient(account.server_url, account.username, account.password, account.get_user_agent()) as xc_client: + with XCClient( + account.server_url, + account.username, + account.password, + account.get_user_agent(), + ) as xc_client: # Log the batch details to help with debugging logger.debug(f"Processing XC batch: {batch}") for group_name, props in batch.items(): # Check if we have a valid xc_id for this group - if 'xc_id' not in props: - logger.error(f"Missing xc_id for group {group_name} in batch {batch}") + if "xc_id" not in props: + logger.error( + f"Missing xc_id for group {group_name} in batch {batch}" + ) continue # Get actual group ID from the mapping @@ -302,14 +365,20 @@ def process_xc_category(account_id, batch, groups, hash_keys): continue try: - logger.debug(f"Fetching streams for XC category: {group_name} (ID: {props['xc_id']})") - streams = xc_client.get_live_category_streams(props['xc_id']) + logger.debug( + f"Fetching streams for XC category: {group_name} (ID: {props['xc_id']})" + ) + streams = xc_client.get_live_category_streams(props["xc_id"]) if not streams: - logger.warning(f"No streams found for XC category {group_name} (ID: {props['xc_id']})") + logger.warning( + f"No streams found for XC category {group_name} (ID: {props['xc_id']})" + ) continue - logger.debug(f"Found {len(streams)} streams for category {group_name}") + logger.debug( + f"Found {len(streams)} streams for category {group_name}" + ) for stream in streams: name = stream["name"] @@ -318,7 +387,9 @@ def process_xc_category(account_id, batch, groups, hash_keys): tvg_logo = stream.get("stream_icon", "") group_title = group_name - stream_hash = Stream.generate_hash_key(name, url, tvg_id, hash_keys) + stream_hash = Stream.generate_hash_key( + name, url, tvg_id, hash_keys + ) stream_props = { "name": name, "url": url, @@ -333,23 +404,38 @@ def process_xc_category(account_id, batch, groups, hash_keys): if stream_hash not in stream_hashes: stream_hashes[stream_hash] = stream_props except Exception as e: - logger.error(f"Error processing XC category {group_name} (ID: {props['xc_id']}): {str(e)}") + logger.error( + f"Error processing XC category {group_name} (ID: {props['xc_id']}): {str(e)}" + ) continue # Process all found streams - existing_streams = {s.stream_hash: s for s in Stream.objects.filter(stream_hash__in=stream_hashes.keys())} + existing_streams = { + s.stream_hash: s + for s in Stream.objects.filter(stream_hash__in=stream_hashes.keys()) + } for stream_hash, stream_props in stream_hashes.items(): if stream_hash in existing_streams: obj = existing_streams[stream_hash] - existing_attr = {field.name: getattr(obj, field.name) for field in Stream._meta.fields if field != 'channel_group_id'} - changed = any(existing_attr[key] != value for key, value in stream_props.items() if key != 'channel_group_id') + existing_attr = { + field.name: getattr(obj, field.name) + for field in Stream._meta.fields + if field != "channel_group_id" + } + changed = any( + existing_attr[key] != value + for key, value in stream_props.items() + if key != "channel_group_id" + ) if changed: for key, value in stream_props.items(): setattr(obj, key, value) obj.last_seen = timezone.now() - obj.updated_at = timezone.now() # Update timestamp only for changed streams + obj.updated_at = ( + timezone.now() + ) # Update timestamp only for changed streams streams_to_update.append(obj) del existing_streams[stream_hash] else: @@ -360,7 +446,9 @@ def process_xc_category(account_id, batch, groups, hash_keys): existing_streams[stream_hash] = obj else: stream_props["last_seen"] = timezone.now() - stream_props["updated_at"] = timezone.now() # Set initial updated_at for new streams + stream_props["updated_at"] = ( + timezone.now() + ) # Set initial updated_at for new streams streams_to_create.append(Stream(**stream_props)) try: @@ -370,14 +458,28 @@ def process_xc_category(account_id, batch, groups, hash_keys): if streams_to_update: # We need to split the bulk update to correctly handle updated_at # First, get the subset of streams that have content changes - changed_streams = [s for s in streams_to_update if hasattr(s, 'updated_at') and s.updated_at] - unchanged_streams = [s for s in streams_to_update if not hasattr(s, 'updated_at') or not s.updated_at] + changed_streams = [ + s + for s in streams_to_update + if hasattr(s, "updated_at") and s.updated_at + ] + unchanged_streams = [ + s + for s in streams_to_update + if not hasattr(s, "updated_at") or not s.updated_at + ] # Update changed streams with all fields including updated_at if changed_streams: Stream.objects.bulk_update( changed_streams, - {key for key in stream_props.keys() if key not in ["m3u_account", "stream_hash"] and key not in hash_keys} | {"last_seen", "updated_at"} + { + key + for key in stream_props.keys() + if key not in ["m3u_account", "stream_hash"] + and key not in hash_keys + } + | {"last_seen", "updated_at"}, ) # Update unchanged streams with only last_seen @@ -401,11 +503,18 @@ def process_xc_category(account_id, batch, groups, hash_keys): return retval + @shared_task def process_m3u_batch(account_id, batch, groups, hash_keys): """Processes a batch of M3U streams using bulk operations.""" account = M3UAccount.objects.get(id=account_id) + compiled_filters = [ + (re.compile(f.regex_pattern), f) + for f in account.filters.order_by("order") + if f.filter_type != "group" + ] + streams_to_create = [] streams_to_update = [] stream_hashes = {} @@ -415,12 +524,34 @@ def process_m3u_batch(account_id, batch, groups, hash_keys): for stream_info in batch: try: name, url = stream_info["name"], stream_info["url"] - tvg_id, tvg_logo = get_case_insensitive_attr(stream_info["attributes"], "tvg-id", ""), get_case_insensitive_attr(stream_info["attributes"], "tvg-logo", "") - group_title = get_case_insensitive_attr(stream_info["attributes"], "group-title", "Default Group") + + include = True + for pattern, filter in compiled_filters: + logger.debug(f"Checking filter patterh {pattern}") + target = url if filter.filter_type == "url" else name + if pattern.search(target or ""): + logger.debug( + f"Stream {name} - {url} matches filter pattern {filter.regex_pattern}" + ) + include = not filter.exclude + break + + if not include: + logger.debug(f"Stream excluded by filter, skipping.") + continue + + tvg_id, tvg_logo = get_case_insensitive_attr( + stream_info["attributes"], "tvg-id", "" + ), get_case_insensitive_attr(stream_info["attributes"], "tvg-logo", "") + group_title = get_case_insensitive_attr( + stream_info["attributes"], "group-title", "Default Group" + ) # Filter out disabled groups for this account if group_title not in groups: - logger.debug(f"Skipping stream in disabled group: {group_title}") + logger.debug( + f"Skipping stream in disabled or excluded group: {group_title}" + ) continue stream_hash = Stream.generate_hash_key(name, url, tvg_id, hash_keys) @@ -441,19 +572,32 @@ def process_m3u_batch(account_id, batch, groups, hash_keys): logger.error(f"Failed to process stream {name}: {e}") logger.error(json.dumps(stream_info)) - existing_streams = {s.stream_hash: s for s in Stream.objects.filter(stream_hash__in=stream_hashes.keys())} + existing_streams = { + s.stream_hash: s + for s in Stream.objects.filter(stream_hash__in=stream_hashes.keys()) + } for stream_hash, stream_props in stream_hashes.items(): if stream_hash in existing_streams: obj = existing_streams[stream_hash] - existing_attr = {field.name: getattr(obj, field.name) for field in Stream._meta.fields if field != 'channel_group_id'} - changed = any(existing_attr[key] != value for key, value in stream_props.items() if key != 'channel_group_id') + existing_attr = { + field.name: getattr(obj, field.name) + for field in Stream._meta.fields + if field != "channel_group_id" + } + changed = any( + existing_attr[key] != value + for key, value in stream_props.items() + if key != "channel_group_id" + ) if changed: for key, value in stream_props.items(): setattr(obj, key, value) obj.last_seen = timezone.now() - obj.updated_at = timezone.now() # Update timestamp only for changed streams + obj.updated_at = ( + timezone.now() + ) # Update timestamp only for changed streams streams_to_update.append(obj) del existing_streams[stream_hash] else: @@ -464,7 +608,9 @@ def process_m3u_batch(account_id, batch, groups, hash_keys): existing_streams[stream_hash] = obj else: stream_props["last_seen"] = timezone.now() - stream_props["updated_at"] = timezone.now() # Set initial updated_at for new streams + stream_props["updated_at"] = ( + timezone.now() + ) # Set initial updated_at for new streams streams_to_create.append(Stream(**stream_props)) try: @@ -474,14 +620,28 @@ def process_m3u_batch(account_id, batch, groups, hash_keys): if streams_to_update: # We need to split the bulk update to correctly handle updated_at # First, get the subset of streams that have content changes - changed_streams = [s for s in streams_to_update if hasattr(s, 'updated_at') and s.updated_at] - unchanged_streams = [s for s in streams_to_update if not hasattr(s, 'updated_at') or not s.updated_at] + changed_streams = [ + s + for s in streams_to_update + if hasattr(s, "updated_at") and s.updated_at + ] + unchanged_streams = [ + s + for s in streams_to_update + if not hasattr(s, "updated_at") or not s.updated_at + ] # Update changed streams with all fields including updated_at if changed_streams: Stream.objects.bulk_update( changed_streams, - {key for key in stream_props.keys() if key not in ["m3u_account", "stream_hash"] and key not in hash_keys} | {"last_seen", "updated_at"} + { + key + for key in stream_props.keys() + if key not in ["m3u_account", "stream_hash"] + and key not in hash_keys + } + | {"last_seen", "updated_at"}, ) # Update unchanged streams with only last_seen @@ -496,35 +656,37 @@ def process_m3u_batch(account_id, batch, groups, hash_keys): retval = f"M3U account: {account_id}, Batch processed: {len(streams_to_create)} created, {len(streams_to_update)} updated." # Aggressive garbage collection - #del streams_to_create, streams_to_update, stream_hashes, existing_streams - #from core.utils import cleanup_memory - #cleanup_memory(log_usage=True, force_collection=True) + # del streams_to_create, streams_to_update, stream_hashes, existing_streams + # from core.utils import cleanup_memory + # cleanup_memory(log_usage=True, force_collection=True) return retval + def cleanup_streams(account_id, scan_start_time=timezone.now): account = M3UAccount.objects.get(id=account_id, is_active=True) existing_groups = ChannelGroup.objects.filter( m3u_account__m3u_account=account, m3u_account__enabled=True, - ).values_list('id', flat=True) - logger.info(f"Found {len(existing_groups)} active groups for M3U account {account_id}") + ).values_list("id", flat=True) + logger.info( + f"Found {len(existing_groups)} active groups for M3U account {account_id}" + ) # Calculate cutoff date for stale streams stale_cutoff = scan_start_time - timezone.timedelta(days=account.stale_stream_days) - logger.info(f"Removing streams not seen since {stale_cutoff} for M3U account {account_id}") + logger.info( + f"Removing streams not seen since {stale_cutoff} for M3U account {account_id}" + ) # Delete streams that are not in active groups - streams_to_delete = Stream.objects.filter( - m3u_account=account - ).exclude( + streams_to_delete = Stream.objects.filter(m3u_account=account).exclude( channel_group__in=existing_groups ) # Also delete streams that haven't been seen for longer than stale_stream_days stale_streams = Stream.objects.filter( - m3u_account=account, - last_seen__lt=stale_cutoff + m3u_account=account, last_seen__lt=stale_cutoff ) deleted_count = streams_to_delete.count() @@ -534,20 +696,23 @@ def cleanup_streams(account_id, scan_start_time=timezone.now): stale_streams.delete() total_deleted = deleted_count + stale_count - logger.info(f"Cleanup for M3U account {account_id} complete: {deleted_count} streams removed due to group filter, {stale_count} removed as stale") + logger.info( + f"Cleanup for M3U account {account_id} complete: {deleted_count} streams removed due to group filter, {stale_count} removed as stale" + ) # Return the total count of deleted streams return total_deleted + @shared_task def refresh_m3u_groups(account_id, use_cache=False, full_refresh=False): - if not acquire_task_lock('refresh_m3u_account_groups', account_id): + if not acquire_task_lock("refresh_m3u_account_groups", account_id): return f"Task already running for account_id={account_id}.", None try: account = M3UAccount.objects.get(id=account_id, is_active=True) except M3UAccount.DoesNotExist: - release_task_lock('refresh_m3u_account_groups', account_id) + release_task_lock("refresh_m3u_account_groups", account_id) return f"M3UAccount with ID={account_id} not found or inactive.", None extinf_data = [] @@ -555,8 +720,12 @@ def refresh_m3u_groups(account_id, use_cache=False, full_refresh=False): if account.account_type == M3UAccount.Types.XC: # Log detailed information about the account - logger.info(f"Processing XC account {account_id} with URL: {account.server_url}") - logger.debug(f"Username: {account.username}, Has password: {'Yes' if account.password else 'No'}") + logger.info( + f"Processing XC account {account_id} with URL: {account.server_url}" + ) + logger.debug( + f"Username: {account.username}, Has password: {'Yes' if account.password else 'No'}" + ) # Validate required fields if not account.server_url: @@ -564,9 +733,11 @@ def refresh_m3u_groups(account_id, use_cache=False, full_refresh=False): logger.error(error_msg) account.status = M3UAccount.Status.ERROR account.last_message = error_msg - account.save(update_fields=['status', 'last_message']) - send_m3u_update(account_id, "processing_groups", 100, status="error", error=error_msg) - release_task_lock('refresh_m3u_account_groups', account_id) + account.save(update_fields=["status", "last_message"]) + send_m3u_update( + account_id, "processing_groups", 100, status="error", error=error_msg + ) + release_task_lock("refresh_m3u_account_groups", account_id) return error_msg, None if not account.username or not account.password: @@ -574,15 +745,19 @@ def refresh_m3u_groups(account_id, use_cache=False, full_refresh=False): logger.error(error_msg) account.status = M3UAccount.Status.ERROR account.last_message = error_msg - account.save(update_fields=['status', 'last_message']) - send_m3u_update(account_id, "processing_groups", 100, status="error", error=error_msg) - release_task_lock('refresh_m3u_account_groups', account_id) + account.save(update_fields=["status", "last_message"]) + send_m3u_update( + account_id, "processing_groups", 100, status="error", error=error_msg + ) + release_task_lock("refresh_m3u_account_groups", account_id) return error_msg, None try: # Ensure server URL is properly formatted - server_url = account.server_url.rstrip('/') - if not (server_url.startswith('http://') or server_url.startswith('https://')): + server_url = account.server_url.rstrip("/") + if not ( + server_url.startswith("http://") or server_url.startswith("https://") + ): server_url = f"http://{server_url}" # User agent handling - completely rewritten @@ -591,37 +766,63 @@ def refresh_m3u_groups(account_id, use_cache=False, full_refresh=False): logger.debug(f"Getting user agent for account {account.id}") # Use a hardcoded user agent string to avoid any issues with object structure - user_agent_string = "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36" + user_agent_string = ( + "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36" + ) try: # Try to get the user agent directly from the database if account.user_agent_id: ua_obj = UserAgent.objects.get(id=account.user_agent_id) - if ua_obj and hasattr(ua_obj, 'user_agent') and ua_obj.user_agent: + if ( + ua_obj + and hasattr(ua_obj, "user_agent") + and ua_obj.user_agent + ): user_agent_string = ua_obj.user_agent - logger.debug(f"Using user agent from account: {user_agent_string}") + logger.debug( + f"Using user agent from account: {user_agent_string}" + ) else: # Get default user agent from CoreSettings default_ua_id = CoreSettings.get_default_user_agent_id() - logger.debug(f"Default user agent ID from settings: {default_ua_id}") + logger.debug( + f"Default user agent ID from settings: {default_ua_id}" + ) if default_ua_id: ua_obj = UserAgent.objects.get(id=default_ua_id) - if ua_obj and hasattr(ua_obj, 'user_agent') and ua_obj.user_agent: + if ( + ua_obj + and hasattr(ua_obj, "user_agent") + and ua_obj.user_agent + ): user_agent_string = ua_obj.user_agent - logger.debug(f"Using default user agent: {user_agent_string}") + logger.debug( + f"Using default user agent: {user_agent_string}" + ) except Exception as e: - logger.warning(f"Error getting user agent, using fallback: {str(e)}") + logger.warning( + f"Error getting user agent, using fallback: {str(e)}" + ) logger.debug(f"Final user agent string: {user_agent_string}") except Exception as e: - user_agent_string = "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36" - logger.warning(f"Exception in user agent handling, using fallback: {str(e)}") + user_agent_string = ( + "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36" + ) + logger.warning( + f"Exception in user agent handling, using fallback: {str(e)}" + ) - logger.info(f"Creating XCClient with URL: {server_url}, Username: {account.username}, User-Agent: {user_agent_string}") + logger.info( + f"Creating XCClient with URL: {server_url}, Username: {account.username}, User-Agent: {user_agent_string}" + ) # Create XCClient with explicit error handling try: - with XCClient(server_url, account.username, account.password, user_agent_string) as xc_client: + with XCClient( + server_url, account.username, account.password, user_agent_string + ) as xc_client: logger.info(f"XCClient instance created successfully") # Authenticate with detailed error handling @@ -634,26 +835,42 @@ def refresh_m3u_groups(account_id, use_cache=False, full_refresh=False): logger.error(error_msg) account.status = M3UAccount.Status.ERROR account.last_message = error_msg - account.save(update_fields=['status', 'last_message']) - send_m3u_update(account_id, "processing_groups", 100, status="error", error=error_msg) - release_task_lock('refresh_m3u_account_groups', account_id) + account.save(update_fields=["status", "last_message"]) + send_m3u_update( + account_id, + "processing_groups", + 100, + status="error", + error=error_msg, + ) + release_task_lock("refresh_m3u_account_groups", account_id) return error_msg, None # Get categories with detailed error handling try: logger.info(f"Getting live categories from XC server") xc_categories = xc_client.get_live_categories() - logger.info(f"Found {len(xc_categories)} categories: {xc_categories}") + logger.info( + f"Found {len(xc_categories)} categories: {xc_categories}" + ) # Validate response if not isinstance(xc_categories, list): - error_msg = f"Unexpected response from XC server: {xc_categories}" + error_msg = ( + f"Unexpected response from XC server: {xc_categories}" + ) logger.error(error_msg) account.status = M3UAccount.Status.ERROR account.last_message = error_msg - account.save(update_fields=['status', 'last_message']) - send_m3u_update(account_id, "processing_groups", 100, status="error", error=error_msg) - release_task_lock('refresh_m3u_account_groups', account_id) + account.save(update_fields=["status", "last_message"]) + send_m3u_update( + account_id, + "processing_groups", + 100, + status="error", + error=error_msg, + ) + release_task_lock("refresh_m3u_account_groups", account_id) return error_msg, None if len(xc_categories) == 0: @@ -671,9 +888,15 @@ def refresh_m3u_groups(account_id, use_cache=False, full_refresh=False): logger.error(error_msg) account.status = M3UAccount.Status.ERROR account.last_message = error_msg - account.save(update_fields=['status', 'last_message']) - send_m3u_update(account_id, "processing_groups", 100, status="error", error=error_msg) - release_task_lock('refresh_m3u_account_groups', account_id) + account.save(update_fields=["status", "last_message"]) + send_m3u_update( + account_id, + "processing_groups", + 100, + status="error", + error=error_msg, + ) + release_task_lock("refresh_m3u_account_groups", account_id) return error_msg, None except Exception as e: @@ -681,25 +904,33 @@ def refresh_m3u_groups(account_id, use_cache=False, full_refresh=False): logger.error(error_msg) account.status = M3UAccount.Status.ERROR account.last_message = error_msg - account.save(update_fields=['status', 'last_message']) - send_m3u_update(account_id, "processing_groups", 100, status="error", error=error_msg) - release_task_lock('refresh_m3u_account_groups', account_id) + account.save(update_fields=["status", "last_message"]) + send_m3u_update( + account_id, + "processing_groups", + 100, + status="error", + error=error_msg, + ) + release_task_lock("refresh_m3u_account_groups", account_id) return error_msg, None except Exception as e: error_msg = f"Unexpected error occurred in XC Client: {str(e)}" logger.error(error_msg) account.status = M3UAccount.Status.ERROR account.last_message = error_msg - account.save(update_fields=['status', 'last_message']) - send_m3u_update(account_id, "processing_groups", 100, status="error", error=error_msg) - release_task_lock('refresh_m3u_account_groups', account_id) + account.save(update_fields=["status", "last_message"]) + send_m3u_update( + account_id, "processing_groups", 100, status="error", error=error_msg + ) + release_task_lock("refresh_m3u_account_groups", account_id) return error_msg, None else: # Here's the key change - use the success flag from fetch_m3u_lines lines, success = fetch_m3u_lines(account, use_cache) if not success: # If fetch failed, don't continue processing - release_task_lock('refresh_m3u_account_groups', account_id) + release_task_lock("refresh_m3u_account_groups", account_id) return f"Failed to fetch M3U data for account_id={account_id}.", None # Log basic file structure for debugging @@ -719,19 +950,25 @@ def refresh_m3u_groups(account_id, use_cache=False, full_refresh=False): extinf_count += 1 parsed = parse_extinf_line(line) if parsed: - group_title_attr = get_case_insensitive_attr(parsed["attributes"], "group-title", "") + group_title_attr = get_case_insensitive_attr( + parsed["attributes"], "group-title", "" + ) if group_title_attr: group_name = group_title_attr # Log new groups as they're discovered if group_name not in groups: - logger.debug(f"Found new group for M3U account {account_id}: '{group_name}'") + logger.debug( + f"Found new group for M3U account {account_id}: '{group_name}'" + ) groups[group_name] = {} extinf_data.append(parsed) else: # Log problematic EXTINF lines - logger.warning(f"Failed to parse EXTINF at line {line_index+1}: {line[:200]}") - problematic_lines.append((line_index+1, line[:200])) + logger.warning( + f"Failed to parse EXTINF at line {line_index+1}: {line[:200]}" + ) + problematic_lines.append((line_index + 1, line[:200])) elif extinf_data and line.startswith("http"): url_count += 1 @@ -741,49 +978,69 @@ def refresh_m3u_groups(account_id, use_cache=False, full_refresh=False): # Periodically log progress for large files if valid_stream_count % 1000 == 0: - logger.debug(f"Processed {valid_stream_count} valid streams so far for M3U account: {account_id}") + logger.debug( + f"Processed {valid_stream_count} valid streams so far for M3U account: {account_id}" + ) # Log summary statistics - logger.info(f"M3U parsing complete - Lines: {line_count}, EXTINF: {extinf_count}, URLs: {url_count}, Valid streams: {valid_stream_count}") + logger.info( + f"M3U parsing complete - Lines: {line_count}, EXTINF: {extinf_count}, URLs: {url_count}, Valid streams: {valid_stream_count}" + ) if problematic_lines: - logger.warning(f"Found {len(problematic_lines)} problematic lines during parsing") - for i, (line_num, content) in enumerate(problematic_lines[:10]): # Log max 10 examples + logger.warning( + f"Found {len(problematic_lines)} problematic lines during parsing" + ) + for i, (line_num, content) in enumerate( + problematic_lines[:10] + ): # Log max 10 examples logger.warning(f"Problematic line #{i+1} at line {line_num}: {content}") if len(problematic_lines) > 10: - logger.warning(f"... and {len(problematic_lines) - 10} more problematic lines") + logger.warning( + f"... and {len(problematic_lines) - 10} more problematic lines" + ) # Log group statistics - logger.info(f"Found {len(groups)} groups in M3U file: {', '.join(list(groups.keys())[:20])}" + - ("..." if len(groups) > 20 else "")) + logger.info( + f"Found {len(groups)} groups in M3U file: {', '.join(list(groups.keys())[:20])}" + + ("..." if len(groups) > 20 else "") + ) # Cache processed data cache_path = os.path.join(m3u_dir, f"{account_id}.json") - with open(cache_path, 'w', encoding='utf-8') as f: - json.dump({ - "extinf_data": extinf_data, - "groups": groups, - }, f) + with open(cache_path, "w", encoding="utf-8") as f: + json.dump( + { + "extinf_data": extinf_data, + "groups": groups, + }, + f, + ) logger.debug(f"Cached parsed M3U data to {cache_path}") send_m3u_update(account_id, "processing_groups", 0) process_groups(account, groups) - release_task_lock('refresh_m3u_account_groups', account_id) - - + release_task_lock("refresh_m3u_account_groups", account_id) if not full_refresh: # Use update() instead of save() to avoid triggering signals M3UAccount.objects.filter(id=account_id).update( status=M3UAccount.Status.PENDING_SETUP, - last_message="M3U groups loaded. Please select groups or refresh M3U to complete setup." + last_message="M3U groups loaded. Please select groups or refresh M3U to complete setup.", + ) + send_m3u_update( + account_id, + "processing_groups", + 100, + status="pending_setup", + message="M3U groups loaded. Please select groups or refresh M3U to complete setup.", ) - send_m3u_update(account_id, "processing_groups", 100, status="pending_setup", message="M3U groups loaded. Please select groups or refresh M3U to complete setup.") return extinf_data, groups + def delete_m3u_refresh_task_by_id(account_id): """ Delete the periodic task associated with an M3U account ID. @@ -797,6 +1054,7 @@ def delete_m3u_refresh_task_by_id(account_id): # Look for task by name try: from django_celery_beat.models import PeriodicTask, IntervalSchedule + task = PeriodicTask.objects.get(name=task_name) logger.debug(f"Found task by name: {task.id} for M3UAccount {account_id}") except PeriodicTask.DoesNotExist: @@ -807,12 +1065,16 @@ def delete_m3u_refresh_task_by_id(account_id): if task: # Store interval info before deleting the task interval_id = None - if hasattr(task, 'interval') and task.interval: + if hasattr(task, "interval") and task.interval: interval_id = task.interval.id # Count how many TOTAL tasks use this interval (including this one) - tasks_with_same_interval = PeriodicTask.objects.filter(interval_id=interval_id).count() - logger.debug(f"Interval {interval_id} is used by {tasks_with_same_interval} tasks total") + tasks_with_same_interval = PeriodicTask.objects.filter( + interval_id=interval_id + ).count() + logger.debug( + f"Interval {interval_id} is used by {tasks_with_same_interval} tasks total" + ) # Delete the task first task_id = task.id @@ -824,20 +1086,28 @@ def delete_m3u_refresh_task_by_id(account_id): if interval_id and tasks_with_same_interval == 1: try: interval = IntervalSchedule.objects.get(id=interval_id) - logger.debug(f"Deleting interval schedule {interval_id} (not shared with other tasks)") + logger.debug( + f"Deleting interval schedule {interval_id} (not shared with other tasks)" + ) interval.delete() logger.debug(f"Successfully deleted interval {interval_id}") except IntervalSchedule.DoesNotExist: logger.warning(f"Interval {interval_id} no longer exists") elif interval_id: - logger.debug(f"Not deleting interval {interval_id} as it's shared with {tasks_with_same_interval-1} other tasks") + logger.debug( + f"Not deleting interval {interval_id} as it's shared with {tasks_with_same_interval-1} other tasks" + ) return True return False except Exception as e: - logger.error(f"Error deleting periodic task for M3UAccount {account_id}: {str(e)}", exc_info=True) + logger.error( + f"Error deleting periodic task for M3UAccount {account_id}: {str(e)}", + exc_info=True, + ) return False + @shared_task def sync_auto_channels(account_id, scan_start_time=None): """ @@ -845,7 +1115,13 @@ def sync_auto_channels(account_id, scan_start_time=None): Preserves existing channel UUIDs to maintain M3U link integrity. Called after M3U refresh completes successfully. """ - from apps.channels.models import Channel, ChannelGroup, ChannelGroupM3UAccount, Stream, ChannelStream + from apps.channels.models import ( + Channel, + ChannelGroup, + ChannelGroupM3UAccount, + Stream, + ChannelStream, + ) from apps.epg.models import EPGData from django.utils import timezone @@ -862,10 +1138,8 @@ def sync_auto_channels(account_id, scan_start_time=None): # Get groups with auto sync enabled for this account auto_sync_groups = ChannelGroupM3UAccount.objects.filter( - m3u_account=account, - enabled=True, - auto_channel_sync=True - ).select_related('channel_group') + m3u_account=account, enabled=True, auto_channel_sync=True + ).select_related("channel_group") channels_created = 0 channels_updated = 0 @@ -890,7 +1164,9 @@ def sync_auto_channels(account_id, scan_start_time=None): force_dummy_epg = group_custom_props.get("force_dummy_epg", False) override_group_id = group_custom_props.get("group_override") name_regex_pattern = group_custom_props.get("name_regex_pattern") - name_replace_pattern = group_custom_props.get("name_replace_pattern") + name_replace_pattern = group_custom_props.get( + "name_replace_pattern" + ) name_match_regex = group_custom_props.get("name_match_regex") channel_profile_ids = group_custom_props.get("channel_profile_ids") channel_sort_order = group_custom_props.get("channel_sort_order") @@ -908,17 +1184,23 @@ def sync_auto_channels(account_id, scan_start_time=None): if override_group_id: try: target_group = ChannelGroup.objects.get(id=override_group_id) - logger.info(f"Using override group '{target_group.name}' instead of '{channel_group.name}' for auto-created channels") + logger.info( + f"Using override group '{target_group.name}' instead of '{channel_group.name}' for auto-created channels" + ) except ChannelGroup.DoesNotExist: - logger.warning(f"Override group with ID {override_group_id} not found, using original group '{channel_group.name}'") + logger.warning( + f"Override group with ID {override_group_id} not found, using original group '{channel_group.name}'" + ) - logger.info(f"Processing auto sync for group: {channel_group.name} (start: {start_number})") + logger.info( + f"Processing auto sync for group: {channel_group.name} (start: {start_number})" + ) # Get all current streams in this group for this M3U account, filter out stale streams current_streams = Stream.objects.filter( m3u_account=account, channel_group=channel_group, - last_seen__gte=scan_start_time + last_seen__gte=scan_start_time, ) # --- FILTER STREAMS BY NAME MATCH REGEX IF SPECIFIED --- @@ -928,33 +1210,38 @@ def sync_auto_channels(account_id, scan_start_time=None): name__iregex=name_match_regex ) except re.error as e: - logger.warning(f"Invalid name_match_regex '{name_match_regex}' for group '{channel_group.name}': {e}. Skipping name filter.") + logger.warning( + f"Invalid name_match_regex '{name_match_regex}' for group '{channel_group.name}': {e}. Skipping name filter." + ) # --- APPLY CHANNEL SORT ORDER --- streams_is_list = False # Track if we converted to list - if channel_sort_order and channel_sort_order != '': - if channel_sort_order == 'name': + if channel_sort_order and channel_sort_order != "": + if channel_sort_order == "name": # Use natural sorting for names to handle numbers correctly current_streams = list(current_streams) - current_streams.sort(key=lambda stream: natural_sort_key(stream.name)) + current_streams.sort( + key=lambda stream: natural_sort_key(stream.name) + ) streams_is_list = True - elif channel_sort_order == 'tvg_id': - current_streams = current_streams.order_by('tvg_id') - elif channel_sort_order == 'updated_at': - current_streams = current_streams.order_by('updated_at') + elif channel_sort_order == "tvg_id": + current_streams = current_streams.order_by("tvg_id") + elif channel_sort_order == "updated_at": + current_streams = current_streams.order_by("updated_at") else: - logger.warning(f"Unknown channel_sort_order '{channel_sort_order}' for group '{channel_group.name}'. Using provider order.") - current_streams = current_streams.order_by('id') + logger.warning( + f"Unknown channel_sort_order '{channel_sort_order}' for group '{channel_group.name}'. Using provider order." + ) + current_streams = current_streams.order_by("id") else: - current_streams = current_streams.order_by('id') + current_streams = current_streams.order_by("id") # If channel_sort_order is empty or None, use provider order (no additional sorting) # Get existing auto-created channels for this account (regardless of current group) # We'll find them by their stream associations instead of just group location existing_channels = Channel.objects.filter( - auto_created=True, - auto_created_by=account - ).select_related('logo', 'epg_data') + auto_created=True, auto_created_by=account + ).select_related("logo", "epg_data") # Create mapping of existing channels by their associated stream # This approach finds channels even if they've been moved to different groups @@ -964,8 +1251,8 @@ def sync_auto_channels(account_id, scan_start_time=None): channel_streams = ChannelStream.objects.filter( channel=channel, stream__m3u_account=account, - stream__channel_group=channel_group # Match streams from the original group - ).select_related('stream') + stream__channel_group=channel_group, # Match streams from the original group + ).select_related("stream") # Map each of our M3U account's streams to this channel for channel_stream in channel_streams: @@ -976,7 +1263,11 @@ def sync_auto_channels(account_id, scan_start_time=None): processed_stream_ids = set() # Check if we have streams - handle both QuerySet and list cases - has_streams = len(current_streams) > 0 if streams_is_list else current_streams.exists() + has_streams = ( + len(current_streams) > 0 + if streams_is_list + else current_streams.exists() + ) if not has_streams: logger.debug(f"No streams found in group {channel_group.name}") @@ -984,20 +1275,31 @@ def sync_auto_channels(account_id, scan_start_time=None): channels_to_delete = [ch for ch in existing_channel_map.values()] if channels_to_delete: deleted_count = len(channels_to_delete) - Channel.objects.filter(id__in=[ch.id for ch in channels_to_delete]).delete() + Channel.objects.filter( + id__in=[ch.id for ch in channels_to_delete] + ).delete() channels_deleted += deleted_count - logger.debug(f"Deleted {deleted_count} auto channels (no streams remaining)") + logger.debug( + f"Deleted {deleted_count} auto channels (no streams remaining)" + ) continue # Prepare profiles to assign to new channels from apps.channels.models import ChannelProfile, ChannelProfileMembership - if channel_profile_ids and isinstance(channel_profile_ids, list) and len(channel_profile_ids) > 0: + + if ( + channel_profile_ids + and isinstance(channel_profile_ids, list) + and len(channel_profile_ids) > 0 + ): # Convert all to int (in case they're strings) try: profile_ids = [int(pid) for pid in channel_profile_ids] except Exception: profile_ids = [] - profiles_to_assign = list(ChannelProfile.objects.filter(id__in=profile_ids)) + profiles_to_assign = list( + ChannelProfile.objects.filter(id__in=profile_ids) + ) else: profiles_to_assign = list(ChannelProfile.objects.all()) @@ -1010,10 +1312,11 @@ def sync_auto_channels(account_id, scan_start_time=None): temp_channel_number = start_number # Get all channel numbers that are already in use by other channels (not auto-created by this account) - used_numbers = set(Channel.objects.exclude( - auto_created=True, - auto_created_by=account - ).values_list('channel_number', flat=True)) + used_numbers = set( + Channel.objects.exclude( + auto_created=True, auto_created_by=account + ).values_list("channel_number", flat=True) + ) for stream in current_streams: if stream.id in existing_channel_map: @@ -1030,7 +1333,9 @@ def sync_auto_channels(account_id, scan_start_time=None): if channel.channel_number != target_number: channel.channel_number = target_number channels_to_renumber.append(channel) - logger.debug(f"Will renumber channel '{channel.name}' to {target_number}") + logger.debug( + f"Will renumber channel '{channel.name}' to {target_number}" + ) temp_channel_number += 1.0 if temp_channel_number % 1 != 0: # Has decimal @@ -1038,8 +1343,10 @@ def sync_auto_channels(account_id, scan_start_time=None): # Bulk update channel numbers if any need renumbering if channels_to_renumber: - Channel.objects.bulk_update(channels_to_renumber, ['channel_number']) - logger.info(f"Renumbered {len(channels_to_renumber)} channels to maintain sort order") + Channel.objects.bulk_update(channels_to_renumber, ["channel_number"]) + logger.info( + f"Renumbered {len(channels_to_renumber)} channels to maintain sort order" + ) # Reset channel number counter for processing new channels current_channel_number = start_number @@ -1048,7 +1355,11 @@ def sync_auto_channels(account_id, scan_start_time=None): processed_stream_ids.add(stream.id) try: # Parse custom properties for additional info - stream_custom_props = json.loads(stream.custom_properties) if stream.custom_properties else {} + stream_custom_props = ( + json.loads(stream.custom_properties) + if stream.custom_properties + else {} + ) tvc_guide_stationid = stream_custom_props.get("tvc-guide-stationid") # --- REGEX FIND/REPLACE LOGIC --- @@ -1056,11 +1367,19 @@ def sync_auto_channels(account_id, scan_start_time=None): new_name = original_name if name_regex_pattern is not None: # If replace is None, treat as empty string (remove match) - replace = name_replace_pattern if name_replace_pattern is not None else '' + replace = ( + name_replace_pattern + if name_replace_pattern is not None + else "" + ) try: - new_name = re.sub(name_regex_pattern, replace, original_name) + new_name = re.sub( + name_regex_pattern, replace, original_name + ) except re.error as e: - logger.warning(f"Regex error for group '{channel_group.name}': {e}. Using original name.") + logger.warning( + f"Regex error for group '{channel_group.name}': {e}. Using original name." + ) new_name = original_name # Check if we already have a channel for this stream @@ -1087,15 +1406,20 @@ def sync_auto_channels(account_id, scan_start_time=None): if existing_channel.channel_group != target_group: existing_channel.channel_group = target_group channel_updated = True - logger.info(f"Moved auto channel '{existing_channel.name}' from '{existing_channel.channel_group.name if existing_channel.channel_group else 'None'}' to '{target_group.name}'") + logger.info( + f"Moved auto channel '{existing_channel.name}' from '{existing_channel.channel_group.name if existing_channel.channel_group else 'None'}' to '{target_group.name}'" + ) # Handle logo updates current_logo = None if stream.logo_url: from apps.channels.models import Logo + current_logo, _ = Logo.objects.get_or_create( url=stream.logo_url, - defaults={"name": stream.name or stream.tvg_id or "Unknown"} + defaults={ + "name": stream.name or stream.tvg_id or "Unknown" + }, ) if existing_channel.logo != current_logo: @@ -1105,7 +1429,9 @@ def sync_auto_channels(account_id, scan_start_time=None): # Handle EPG data updates current_epg_data = None if stream.tvg_id and not force_dummy_epg: - current_epg_data = EPGData.objects.filter(tvg_id=stream.tvg_id).first() + current_epg_data = EPGData.objects.filter( + tvg_id=stream.tvg_id + ).first() if existing_channel.epg_data != current_epg_data: existing_channel.epg_data = current_epg_data @@ -1114,17 +1440,20 @@ def sync_auto_channels(account_id, scan_start_time=None): if channel_updated: existing_channel.save() channels_updated += 1 - logger.debug(f"Updated auto channel: {existing_channel.channel_number} - {existing_channel.name}") + logger.debug( + f"Updated auto channel: {existing_channel.channel_number} - {existing_channel.name}" + ) # Update channel profile memberships for existing channels current_memberships = set( ChannelProfileMembership.objects.filter( - channel=existing_channel, - enabled=True - ).values_list('channel_profile_id', flat=True) + channel=existing_channel, enabled=True + ).values_list("channel_profile_id", flat=True) ) - target_profile_ids = set(profile.id for profile in profiles_to_assign) + target_profile_ids = set( + profile.id for profile in profiles_to_assign + ) # Only update if memberships have changed if current_memberships != target_profile_ids: @@ -1135,16 +1464,20 @@ def sync_auto_channels(account_id, scan_start_time=None): # Enable/create memberships for target profiles for profile in profiles_to_assign: - membership, created = ChannelProfileMembership.objects.get_or_create( - channel_profile=profile, - channel=existing_channel, - defaults={'enabled': True} + membership, created = ( + ChannelProfileMembership.objects.get_or_create( + channel_profile=profile, + channel=existing_channel, + defaults={"enabled": True}, + ) ) if not created and not membership.enabled: membership.enabled = True membership.save() - logger.debug(f"Updated profile memberships for auto channel: {existing_channel.name}") + logger.debug( + f"Updated profile memberships for auto channel: {existing_channel.name}" + ) else: # Create new channel @@ -1164,19 +1497,19 @@ def sync_auto_channels(account_id, scan_start_time=None): channel_group=target_group, user_level=0, auto_created=True, - auto_created_by=account + auto_created_by=account, ) # Associate the stream with the channel ChannelStream.objects.create( - channel=channel, - stream=stream, - order=0 + channel=channel, stream=stream, order=0 ) # Assign to correct profiles memberships = [ - ChannelProfileMembership(channel_profile=profile, channel=channel, enabled=True) + ChannelProfileMembership( + channel_profile=profile, channel=channel, enabled=True + ) for profile in profiles_to_assign ] if memberships: @@ -1184,26 +1517,33 @@ def sync_auto_channels(account_id, scan_start_time=None): # Try to match EPG data if stream.tvg_id and not force_dummy_epg: - epg_data = EPGData.objects.filter(tvg_id=stream.tvg_id).first() + epg_data = EPGData.objects.filter( + tvg_id=stream.tvg_id + ).first() if epg_data: channel.epg_data = epg_data - channel.save(update_fields=['epg_data']) + channel.save(update_fields=["epg_data"]) elif stream.tvg_id and force_dummy_epg: channel.epg_data = None - channel.save(update_fields=['epg_data']) + channel.save(update_fields=["epg_data"]) # Handle logo if stream.logo_url: from apps.channels.models import Logo + logo, _ = Logo.objects.get_or_create( url=stream.logo_url, - defaults={"name": stream.name or stream.tvg_id or "Unknown"} + defaults={ + "name": stream.name or stream.tvg_id or "Unknown" + }, ) channel.logo = logo - channel.save(update_fields=['logo']) + channel.save(update_fields=["logo"]) channels_created += 1 - logger.debug(f"Created auto channel: {channel.channel_number} - {channel.name}") + logger.debug( + f"Created auto channel: {channel.channel_number} - {channel.name}" + ) # Increment channel number for next iteration current_channel_number += 1.0 @@ -1211,7 +1551,9 @@ def sync_auto_channels(account_id, scan_start_time=None): current_channel_number = int(current_channel_number) + 1.0 except Exception as e: - logger.error(f"Error processing auto channel for stream {stream.name}: {str(e)}") + logger.error( + f"Error processing auto channel for stream {stream.name}: {str(e)}" + ) continue # Delete channels for streams that no longer exist @@ -1222,21 +1564,28 @@ def sync_auto_channels(account_id, scan_start_time=None): if channels_to_delete: deleted_count = len(channels_to_delete) - Channel.objects.filter(id__in=[ch.id for ch in channels_to_delete]).delete() + Channel.objects.filter( + id__in=[ch.id for ch in channels_to_delete] + ).delete() channels_deleted += deleted_count - logger.debug(f"Deleted {deleted_count} auto channels for removed streams") + logger.debug( + f"Deleted {deleted_count} auto channels for removed streams" + ) - logger.info(f"Auto channel sync complete for account {account.name}: {channels_created} created, {channels_updated} updated, {channels_deleted} deleted") + logger.info( + f"Auto channel sync complete for account {account.name}: {channels_created} created, {channels_updated} updated, {channels_deleted} deleted" + ) return f"Auto sync: {channels_created} channels created, {channels_updated} updated, {channels_deleted} deleted" except Exception as e: logger.error(f"Error in auto channel sync for account {account_id}: {str(e)}") return f"Auto sync error: {str(e)}" + @shared_task def refresh_single_m3u_account(account_id): """Splits M3U processing into chunks and dispatches them as parallel tasks.""" - if not acquire_task_lock('refresh_single_m3u_account', account_id): + if not acquire_task_lock("refresh_single_m3u_account", account_id): return f"Task already running for account_id={account_id}." # Record start time @@ -1250,25 +1599,27 @@ def refresh_single_m3u_account(account_id): account = M3UAccount.objects.get(id=account_id, is_active=True) if not account.is_active: logger.debug(f"Account {account_id} is not active, skipping.") - release_task_lock('refresh_single_m3u_account', account_id) + release_task_lock("refresh_single_m3u_account", account_id) return # Set status to fetching account.status = M3UAccount.Status.FETCHING - account.save(update_fields=['status']) - - filters = list(account.filters.all()) + account.save(update_fields=["status"]) except M3UAccount.DoesNotExist: # The M3U account doesn't exist, so delete the periodic task if it exists - logger.warning(f"M3U account with ID {account_id} not found, but task was triggered. Cleaning up orphaned task.") + logger.warning( + f"M3U account with ID {account_id} not found, but task was triggered. Cleaning up orphaned task." + ) # Call the helper function to delete the task if delete_m3u_refresh_task_by_id(account_id): - logger.info(f"Successfully cleaned up orphaned task for M3U account {account_id}") + logger.info( + f"Successfully cleaned up orphaned task for M3U account {account_id}" + ) else: logger.debug(f"No orphaned task found for M3U account {account_id}") - release_task_lock('refresh_single_m3u_account', account_id) + release_task_lock("refresh_single_m3u_account", account_id) return f"M3UAccount with ID={account_id} not found or inactive, task cleaned up" # Fetch M3U lines and handle potential issues @@ -1278,14 +1629,16 @@ def refresh_single_m3u_account(account_id): cache_path = os.path.join(m3u_dir, f"{account_id}.json") if os.path.exists(cache_path): try: - with open(cache_path, 'r') as file: + with open(cache_path, "r") as file: data = json.load(file) - extinf_data = data['extinf_data'] - groups = data['groups'] + extinf_data = data["extinf_data"] + groups = data["groups"] except json.JSONDecodeError as e: # Handle corrupted JSON file - logger.error(f"Error parsing cached M3U data for account {account_id}: {str(e)}") + logger.error( + f"Error parsing cached M3U data for account {account_id}: {str(e)}" + ) # Backup the corrupted file for potential analysis backup_path = f"{cache_path}.corrupted" @@ -1293,7 +1646,9 @@ def refresh_single_m3u_account(account_id): os.rename(cache_path, backup_path) logger.info(f"Renamed corrupted cache file to {backup_path}") except OSError as rename_err: - logger.warning(f"Failed to rename corrupted cache file: {str(rename_err)}") + logger.warning( + f"Failed to rename corrupted cache file: {str(rename_err)}" + ) # Reset the data to empty structures extinf_data = [] @@ -1311,8 +1666,10 @@ def refresh_single_m3u_account(account_id): # Check for completely empty result or missing groups if not result or result[1] is None: - logger.error(f"Failed to refresh M3U groups for account {account_id}: {result}") - release_task_lock('refresh_single_m3u_account', account_id) + logger.error( + f"Failed to refresh M3U groups for account {account_id}: {result}" + ) + release_task_lock("refresh_single_m3u_account", account_id) return "Failed to update m3u account - download failed or other error" extinf_data, groups = result @@ -1329,15 +1686,23 @@ def refresh_single_m3u_account(account_id): logger.error(f"No streams found for non-XC account {account_id}") account.status = M3UAccount.Status.ERROR account.last_message = "No streams found in M3U source" - account.save(update_fields=['status', 'last_message']) - send_m3u_update(account_id, "parsing", 100, status="error", error="No streams found") + account.save(update_fields=["status", "last_message"]) + send_m3u_update( + account_id, "parsing", 100, status="error", error="No streams found" + ) except Exception as e: logger.error(f"Exception in refresh_m3u_groups: {str(e)}", exc_info=True) account.status = M3UAccount.Status.ERROR account.last_message = f"Error refreshing M3U groups: {str(e)}" - account.save(update_fields=['status', 'last_message']) - send_m3u_update(account_id, "parsing", 100, status="error", error=f"Error refreshing M3U groups: {str(e)}") - release_task_lock('refresh_single_m3u_account', account_id) + account.save(update_fields=["status", "last_message"]) + send_m3u_update( + account_id, + "parsing", + 100, + status="error", + error=f"Error refreshing M3U groups: {str(e)}", + ) + release_task_lock("refresh_single_m3u_account", account_id) return "Failed to update m3u account" # Only proceed with parsing if we actually have data and no errors were encountered @@ -1352,37 +1717,53 @@ def refresh_single_m3u_account(account_id): logger.error(f"No data to process for account {account_id}") account.status = M3UAccount.Status.ERROR account.last_message = "No data available for processing" - account.save(update_fields=['status', 'last_message']) - send_m3u_update(account_id, "parsing", 100, status="error", error="No data available for processing") - release_task_lock('refresh_single_m3u_account', account_id) + account.save(update_fields=["status", "last_message"]) + send_m3u_update( + account_id, + "parsing", + 100, + status="error", + error="No data available for processing", + ) + release_task_lock("refresh_single_m3u_account", account_id) return "Failed to update m3u account, no data available" hash_keys = CoreSettings.get_m3u_hash_key().split(",") - existing_groups = {group.name: group.id for group in ChannelGroup.objects.filter( - m3u_account__m3u_account=account, # Filter by the M3UAccount - m3u_account__enabled=True # Filter by the enabled flag in the join table - )} + existing_groups = { + group.name: group.id + for group in ChannelGroup.objects.filter( + m3u_account__m3u_account=account, # Filter by the M3UAccount + m3u_account__enabled=True, # Filter by the enabled flag in the join table + ) + } try: # Set status to parsing account.status = M3UAccount.Status.PARSING - account.save(update_fields=['status']) + account.save(update_fields=["status"]) if account.account_type == M3UAccount.Types.STADNARD: - logger.debug(f"Processing Standard account ({account_id}) with groups: {existing_groups}") + logger.debug( + f"Processing Standard account ({account_id}) with groups: {existing_groups}" + ) # Break into batches and process in parallel - batches = [extinf_data[i:i + BATCH_SIZE] for i in range(0, len(extinf_data), BATCH_SIZE)] - task_group = group(process_m3u_batch.s(account_id, batch, existing_groups, hash_keys) for batch in batches) + batches = [ + extinf_data[i : i + BATCH_SIZE] + for i in range(0, len(extinf_data), BATCH_SIZE) + ] + task_group = group( + process_m3u_batch.s(account_id, batch, existing_groups, hash_keys) + for batch in batches + ) else: # For XC accounts, get the groups with their custom properties containing xc_id logger.debug(f"Processing XC account with groups: {existing_groups}") # Get the ChannelGroupM3UAccount entries with their custom_properties channel_group_relationships = ChannelGroupM3UAccount.objects.filter( - m3u_account=account, - enabled=True - ).select_related('channel_group') + m3u_account=account, enabled=True + ).select_related("channel_group") filtered_groups = {} for rel in channel_group_relationships: @@ -1391,34 +1772,51 @@ def refresh_single_m3u_account(account_id): # Load the custom properties with the xc_id try: - custom_props = json.loads(rel.custom_properties) if rel.custom_properties else {} - if 'xc_id' in custom_props: + custom_props = ( + json.loads(rel.custom_properties) + if rel.custom_properties + else {} + ) + if "xc_id" in custom_props: filtered_groups[group_name] = { - 'xc_id': custom_props['xc_id'], - 'channel_group_id': group_id + "xc_id": custom_props["xc_id"], + "channel_group_id": group_id, } - logger.debug(f"Added group {group_name} with xc_id {custom_props['xc_id']}") + logger.debug( + f"Added group {group_name} with xc_id {custom_props['xc_id']}" + ) else: - logger.warning(f"No xc_id found in custom properties for group {group_name}") + logger.warning( + f"No xc_id found in custom properties for group {group_name}" + ) except (json.JSONDecodeError, KeyError) as e: - logger.error(f"Error parsing custom properties for group {group_name}: {str(e)}") + logger.error( + f"Error parsing custom properties for group {group_name}: {str(e)}" + ) - logger.info(f"Filtered {len(filtered_groups)} groups for processing: {filtered_groups}") + logger.info( + f"Filtered {len(filtered_groups)} groups for processing: {filtered_groups}" + ) # Batch the groups filtered_groups_list = list(filtered_groups.items()) batches = [ - dict(filtered_groups_list[i:i + 2]) + dict(filtered_groups_list[i : i + 2]) for i in range(0, len(filtered_groups_list), 2) ] logger.info(f"Created {len(batches)} batches for XC processing") - task_group = group(process_xc_category.s(account_id, batch, existing_groups, hash_keys) for batch in batches) + task_group = group( + process_xc_category.s(account_id, batch, existing_groups, hash_keys) + for batch in batches + ) total_batches = len(batches) completed_batches = 0 streams_processed = 0 # Track total streams processed - logger.debug(f"Dispatched {len(batches)} parallel tasks for account_id={account_id}.") + logger.debug( + f"Dispatched {len(batches)} parallel tasks for account_id={account_id}." + ) # result = task_group.apply_async() result = task_group.apply_async() @@ -1427,7 +1825,9 @@ def refresh_single_m3u_account(account_id): completed_task_ids = set() while completed_batches < total_batches: for async_result in result: - if async_result.ready() and async_result.id not in completed_task_ids: # If the task has completed and we haven't counted it + if ( + async_result.ready() and async_result.id not in completed_task_ids + ): # If the task has completed and we haven't counted it task_result = async_result.result # The result of the task logger.debug(f"Task completed with result: {task_result}") @@ -1447,7 +1847,9 @@ def refresh_single_m3u_account(account_id): pass completed_batches += 1 - completed_task_ids.add(async_result.id) # Mark this task as processed + completed_task_ids.add( + async_result.id + ) # Mark this task as processed # Calculate progress progress = int((completed_batches / total_batches) * 100) @@ -1471,7 +1873,7 @@ def refresh_single_m3u_account(account_id): progress, elapsed_time=current_elapsed, time_remaining=time_remaining, - streams_processed=streams_processed + streams_processed=streams_processed, ) # Optionally remove completed task from the group to prevent processing it again @@ -1480,9 +1882,13 @@ def refresh_single_m3u_account(account_id): logger.trace(f"Task is still running.") # Ensure all database transactions are committed before cleanup - logger.info(f"All {total_batches} tasks completed, ensuring DB transactions are committed before cleanup") + logger.info( + f"All {total_batches} tasks completed, ensuring DB transactions are committed before cleanup" + ) # Force a simple DB query to ensure connection sync - Stream.objects.filter(id=-1).exists() # This will never find anything but ensures DB sync + Stream.objects.filter( + id=-1 + ).exists() # This will never find anything but ensures DB sync # Now run cleanup streams_deleted = cleanup_streams(account_id, refresh_start_timestamp) @@ -1490,12 +1896,18 @@ def refresh_single_m3u_account(account_id): # Run auto channel sync after successful refresh auto_sync_message = "" try: - sync_result = sync_auto_channels(account_id, scan_start_time=str(refresh_start_timestamp)) - logger.info(f"Auto channel sync result for account {account_id}: {sync_result}") + sync_result = sync_auto_channels( + account_id, scan_start_time=str(refresh_start_timestamp) + ) + logger.info( + f"Auto channel sync result for account {account_id}: {sync_result}" + ) if sync_result and "created" in sync_result: auto_sync_message = f" {sync_result}." except Exception as e: - logger.error(f"Error running auto channel sync for account {account_id}: {str(e)}") + logger.error( + f"Error running auto channel sync for account {account_id}: {str(e)}" + ) # Calculate elapsed time elapsed_time = time.time() - start_time @@ -1508,7 +1920,7 @@ def refresh_single_m3u_account(account_id): f"Total processed: {streams_processed}.{auto_sync_message}" ) account.updated_at = timezone.now() - account.save(update_fields=['status', 'last_message', 'updated_at']) + account.save(update_fields=["status", "last_message", "updated_at"]) # Send final update with complete metrics and explicitly include success status send_m3u_update( @@ -1522,21 +1934,22 @@ def refresh_single_m3u_account(account_id): streams_created=streams_created, streams_updated=streams_updated, streams_deleted=streams_deleted, - message=account.last_message + message=account.last_message, ) except Exception as e: logger.error(f"Error processing M3U for account {account_id}: {str(e)}") account.status = M3UAccount.Status.ERROR account.last_message = f"Error processing M3U: {str(e)}" - account.save(update_fields=['status', 'last_message']) + account.save(update_fields=["status", "last_message"]) raise # Re-raise the exception for Celery to handle - release_task_lock('refresh_single_m3u_account', account_id) + release_task_lock("refresh_single_m3u_account", account_id) # Aggressive garbage collection del existing_groups, extinf_data, groups, batches from core.utils import cleanup_memory + cleanup_memory(log_usage=True, force_collection=True) # Clean up cache file since we've fully processed it @@ -1545,6 +1958,7 @@ def refresh_single_m3u_account(account_id): return f"Dispatched jobs complete." + def send_m3u_update(account_id, action, progress, **kwargs): # Start with the base data dictionary data = { @@ -1567,7 +1981,7 @@ def send_m3u_update(account_id, action, progress, **kwargs): # Add the additional key-value pairs from kwargs data.update(kwargs) - send_websocket_update('updates', 'update', data, collect_garbage=False) + send_websocket_update("updates", "update", data, collect_garbage=False) # Explicitly clear data reference to help garbage collection data = None diff --git a/frontend/src/api.js b/frontend/src/api.js index ddaccbc7..a6998bc2 100644 --- a/frontend/src/api.js +++ b/frontend/src/api.js @@ -256,7 +256,7 @@ export default class API { hasChannels: false, hasM3UAccounts: false, canEdit: true, - canDelete: true + canDelete: true, }; useChannelsStore.getState().addChannelGroup(processedGroup); // Refresh channel groups to update the UI @@ -736,10 +736,13 @@ export default class API { static async updateM3UGroupSettings(playlistId, groupSettings) { try { - const response = await request(`${host}/api/m3u/accounts/${playlistId}/group-settings/`, { - method: 'PATCH', - body: { group_settings: groupSettings }, - }); + const response = await request( + `${host}/api/m3u/accounts/${playlistId}/group-settings/`, + { + method: 'PATCH', + body: { group_settings: groupSettings }, + } + ); // Fetch the updated playlist and update the store const updatedPlaylist = await API.getPlaylist(playlistId); usePlaylistsStore.getState().updatePlaylist(updatedPlaylist); @@ -1110,6 +1113,48 @@ export default class API { } } + static async addM3UFilter(accountId, values) { + try { + const response = await request( + `${host}/api/m3u/accounts/${accountId}/filters/`, + { + method: 'POST', + body: values, + } + ); + + return response; + } catch (e) { + errorNotification(`Failed to add profile to account ${accountId}`, e); + } + } + + static async deleteM3UFilter(accountId, id) { + try { + await request(`${host}/api/m3u/accounts/${accountId}/filters/${id}/`, { + method: 'DELETE', + }); + } catch (e) { + errorNotification(`Failed to delete profile for account ${accountId}`, e); + } + } + + static async updateM3UFilter(accountId, filterId, values) { + const { id, ...payload } = values; + + try { + await request( + `${host}/api/m3u/accounts/${accountId}/filters/${filterId}/`, + { + method: 'PUT', + body: payload, + } + ); + } catch (e) { + errorNotification(`Failed to update profile for account ${accountId}`, e); + } + } + static async getSettings() { try { const response = await request(`${host}/api/core/settings/`); @@ -1230,7 +1275,9 @@ export default class API { static async getLogos(params = {}) { try { const queryParams = new URLSearchParams(params); - const response = await request(`${host}/api/channels/logos/?${queryParams.toString()}`); + const response = await request( + `${host}/api/channels/logos/?${queryParams.toString()}` + ); return response; } catch (e) { @@ -1369,7 +1416,7 @@ export default class API { }); // Remove multiple logos from store - ids.forEach(id => { + ids.forEach((id) => { useChannelsStore.getState().removeLogo(id); }); diff --git a/frontend/src/components/forms/M3U.jsx b/frontend/src/components/forms/M3U.jsx index 0e4d5643..3d55d31b 100644 --- a/frontend/src/components/forms/M3U.jsx +++ b/frontend/src/components/forms/M3U.jsx @@ -27,6 +27,7 @@ import usePlaylistsStore from '../../store/playlists'; import { notifications } from '@mantine/notifications'; import { isNotEmpty, useForm } from '@mantine/form'; import useEPGsStore from '../../store/epgs'; +import M3UFilters from './M3UFilters'; const M3U = ({ m3uAccount = null, @@ -45,6 +46,7 @@ const M3U = ({ const [file, setFile] = useState(null); const [profileModalOpen, setProfileModalOpen] = useState(false); const [groupFilterModalOpen, setGroupFilterModalOpen] = useState(false); + const [filterModalOpen, setFilterModalOpen] = useState(false); const [loadingText, setLoadingText] = useState(''); const [showCredentialFields, setShowCredentialFields] = useState(false); @@ -85,7 +87,11 @@ const M3U = ({ account_type: m3uAccount.account_type, username: m3uAccount.username ?? '', password: '', - stale_stream_days: m3uAccount.stale_stream_days !== undefined && m3uAccount.stale_stream_days !== null ? m3uAccount.stale_stream_days : 7, + stale_stream_days: + m3uAccount.stale_stream_days !== undefined && + m3uAccount.stale_stream_days !== null + ? m3uAccount.stale_stream_days + : 7, }); if (m3uAccount.account_type == 'XC') { @@ -145,7 +151,8 @@ const M3U = ({ if (values.account_type != 'XC') { notifications.show({ title: 'Fetching M3U Groups', - message: 'Configure group filters and auto sync settings once complete.', + message: + 'Configure group filters and auto sync settings once complete.', }); // Don't prompt for group filters, but keeping this here @@ -177,7 +184,10 @@ const M3U = ({ const closeGroupFilter = () => { setGroupFilterModalOpen(false); - close(); + }; + + const closeFilter = () => { + setFilterModalOpen(false); }; useEffect(() => { @@ -224,7 +234,12 @@ const M3U = ({ id="account_type" name="account_type" label="Account Type" - description={<>Standard for direct M3U URLs,
Xtream Codes for panel-based services} + description={ + <> + Standard for direct M3U URLs,
+ Xtream Codes for panel-based services + + } data={[ { value: 'STD', @@ -316,8 +331,13 @@ const M3U = ({ How often to automatically refresh M3U data
- (0 to disable automatic refreshes)} + description={ + <> + How often to automatically refresh M3U data +
+ (0 to disable automatic refreshes) + + } {...form.getInputProps('refresh_interval')} key={form.key('refresh_interval')} /> @@ -342,6 +362,13 @@ const M3U = ({ {playlist && ( <> +