diff --git a/apps/channels/migrations/0008_stream_stream_hash.py b/apps/channels/migrations/0008_stream_stream_hash.py index d8109a2b..5279dd19 100644 --- a/apps/channels/migrations/0008_stream_stream_hash.py +++ b/apps/channels/migrations/0008_stream_stream_hash.py @@ -13,6 +13,11 @@ class Migration(migrations.Migration): migrations.AddField( model_name='stream', name='stream_hash', - field=models.CharField(help_text='Unique hash for this stream from the M3U account', max_length=255, null=True, unique=True), + field=models.CharField(db_index=True, help_text='Unique hash for this stream from the M3U account', max_length=255, null=True, unique=True), + ), + migrations.AlterField( + model_name='stream', + name='logo_url', + field=models.TextField(blank=True, null=True), ), ] diff --git a/apps/channels/models.py b/apps/channels/models.py index 8706e635..df291e09 100644 --- a/apps/channels/models.py +++ b/apps/channels/models.py @@ -25,6 +25,16 @@ class ChannelGroup(models.Model): def __str__(self): return self.name + @classmethod + def bulk_create_and_fetch(cls, objects): + # Perform the bulk create operation + cls.objects.bulk_create(objects) + + # Use a unique field to fetch the created objects (assuming 'name' is unique) + created_objects = cls.objects.filter(name__in=[obj.name for obj in objects]) + + return created_objects + class Stream(models.Model): """ Represents a single stream (e.g. from an M3U source or custom URL). @@ -38,7 +48,7 @@ class Stream(models.Model): blank=True, related_name="streams", ) - logo_url = models.URLField(max_length=2000, blank=True, null=True) + logo_url = models.TextField(blank=True, null=True) tvg_id = models.CharField(max_length=255, blank=True, null=True) local_file = models.FileField(upload_to='uploads/', blank=True, null=True) current_viewers = models.PositiveIntegerField(default=0) @@ -65,7 +75,8 @@ class Stream(models.Model): max_length=255, null=True, unique=True, - help_text="Unique hash for this stream from the M3U account" + help_text="Unique hash for this stream from the M3U account", + db_index=True, ) last_seen = models.DateTimeField(db_index=True, default=datetime.now) @@ -79,21 +90,15 @@ class Stream(models.Model): return self.name or self.url or f"Stream ID {self.id}" @classmethod - def generate_hash_key(cls, stream): - # Check if the passed object is an instance or a dictionary - if isinstance(stream, dict): - # Handle dictionary case (e.g., when the input is a dict of stream data) - hash_parts = {key: stream[key] for key in CoreSettings.get_m3u_hash_key().split(",") if key in stream} - if 'm3u_account_id' in stream: - hash_parts['m3u_account_id'] = stream['m3u_account_id'] - elif isinstance(stream, Stream): - # Handle the case where the input is a Stream instance - key_parts = CoreSettings.get_m3u_hash_key().split(",") - hash_parts = {key: getattr(stream, key) for key in key_parts if hasattr(stream, key)} - if stream.m3u_account: - hash_parts['m3u_account_id'] = stream.m3u_account.id - else: - raise ValueError("stream must be either a dictionary or a Stream instance") + def generate_hash_key(cls, name, url, tvg_id, keys=None): + if keys is None: + keys = CoreSettings.get_m3u_hash_key().split(",") + + stream_parts = { + "name": name, "url": url, "tvg_id": tvg_id + } + + hash_parts = {key: stream_parts[key] for key in keys if key in stream_parts} # Serialize and hash the dictionary serialized_obj = json.dumps(hash_parts, sort_keys=True) # sort_keys ensures consistent ordering diff --git a/apps/m3u/tasks.py b/apps/m3u/tasks.py index 354754ad..01787cf1 100644 --- a/apps/m3u/tasks.py +++ b/apps/m3u/tasks.py @@ -4,7 +4,8 @@ import re import requests import os from celery.app.control import Inspect -from celery import shared_task, current_app +from celery.result import AsyncResult +from celery import shared_task, current_app, group from django.conf import settings from django.core.cache import cache from .models import M3UAccount @@ -12,10 +13,44 @@ from apps.channels.models import Stream, ChannelGroup, ChannelGroupM3UAccount from asgiref.sync import async_to_sync from channels.layers import get_channel_layer from django.utils import timezone +import time +from channels.layers import get_channel_layer +import json +from core.utils import redis_client +from core.models import CoreSettings +from asgiref.sync import async_to_sync logger = logging.getLogger(__name__) -LOCK_EXPIRE = 120 # Lock expires after 120 seconds +LOCK_EXPIRE = 300 +BATCH_SIZE = 10000 +SKIP_EXTS = {} + +def fetch_m3u_lines(account): + """Fetch M3U file lines efficiently.""" + if account.server_url: + headers = {"User-Agent": account.user_agent.user_agent} + logger.info(f"Fetching from URL {account.server_url}") + try: + # Perform the HTTP request with stream and handle any potential issues + with requests.get(account.server_url, timeout=60, headers=headers, stream=True) as response: + response.raise_for_status() # This will raise an HTTPError if the status is not 200 + # Return an iterator for the lines + return response.iter_lines(decode_unicode=True) + except requests.exceptions.RequestException as e: + logger.error(f"Error fetching M3U from URL {account.server_url}: {e}") + return [] # Return an empty list in case of error + elif account.uploaded_file: + try: + # Open the file and return the lines as a list or iterator + with open(account.uploaded_file.path, 'r', encoding='utf-8') as f: + return f.readlines() # Ensure you return lines from the file, not the file object + except IOError as e: + logger.error(f"Error opening file {account.uploaded_file.path}: {e}") + return [] # Return an empty list in case of error + + # Return an empty list if neither server_url nor uploaded_file is available + return [] def parse_extinf_line(line: str) -> dict: """ @@ -36,7 +71,7 @@ def parse_extinf_line(line: str) -> dict: if len(parts) != 2: return None attributes_part, display_name = parts[0], parts[1].strip() - attrs = dict(re.findall(r'(\w+)=["\']([^"\']+)["\']', attributes_part)) + attrs = dict(re.findall(r'([^\s]+)=["\']([^"\']+)["\']', attributes_part)) # Use tvg-name attribute if available; otherwise, use the display name. name = attrs.get('tvg-name', display_name) return { @@ -45,20 +80,18 @@ def parse_extinf_line(line: str) -> dict: 'name': name } -def _get_group_title(extinf_line: str) -> str: - """Extract group title from EXTINF line.""" - match = re.search(r'group-title="([^"]*)"', extinf_line) - return match.group(1) if match else "Default Group" +import re +import logging -def _matches_filters(stream_name: str, group_name: str, filters) -> bool: - logger.info("Testing filter") - for f in filters: - pattern = f.regex_pattern +logger = logging.getLogger(__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 - logger.info(f"Testing {pattern} on: {target}") - if re.search(pattern, target or '', re.IGNORECASE): - logger.debug(f"Filter matched: {pattern} on {target}. Exclude={f.exclude}") - return f.exclude + if pattern.search(target or ''): + return exclude return False def acquire_lock(task_name, account_id): @@ -87,138 +120,254 @@ 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(): + if isinstance(value, str) and len(value) > 255: + print(f"{field} --- {value}") + + print("") + print("") + +@shared_task +def process_groups(account, group_names): + existing_groups = {group.name: group for group in ChannelGroup.objects.filter(name__in=group_names)} + logger.info(f"Currently {len(existing_groups)} existing groups") + + groups = [] + groups_to_create = [] + for group_name in group_names: + if group_name in existing_groups: + groups.append(existing_groups[group_name]) + else: + groups_to_create.append(ChannelGroup( + name=group_name, + )) + + if groups_to_create: + logger.info(f"Creating {len(groups_to_create)} groups") + created = ChannelGroup.bulk_create_and_fetch(groups_to_create) + logger.info(f"Created {len(created)} groups") + groups.extend(created) + + relations = [] + for group in groups: + relations.append(ChannelGroupM3UAccount( + channel_group=group, + m3u_account=account, + )) + + ChannelGroupM3UAccount.objects.bulk_create( + relations, + ignore_conflicts=True + ) + +@shared_task +def process_m3u_batch(account_id, batch, group_names, hash_keys): + """Processes a batch of M3U streams using bulk operations.""" + account = M3UAccount.objects.get(id=account_id) + existing_groups = {group.name: group for group in ChannelGroup.objects.filter(name__in=group_names)} + + streams_to_create = [] + streams_to_update = [] + stream_hashes = {} + + # compiled_filters = [(f.filter_type, re.compile(f.regex_pattern, re.IGNORECASE)) for f in filters] + + logger.info(f"Processing batch of {len(batch)}") + for stream_info in batch: + name, url = stream_info["name"], stream_info["url"] + tvg_id, tvg_logo = stream_info["attributes"].get("tvg-id", ""), stream_info["attributes"].get("tvg-logo", "") + group_title = stream_info["attributes"].get("group-title", "Default Group") + + # if any(url.lower().endswith(ext) for ext in SKIP_EXTS) or len(url) > 2000: + # continue + + # if _matches_filters(name, group_title, account.filters.all()): + # continue + + # if any(compiled_pattern.search(current_info['name']) for ftype, compiled_pattern in compiled_filters if ftype == 'name'): + # excluded_count += 1 + # current_info = None + # continue + + try: + stream_hash = Stream.generate_hash_key(name, url, tvg_id, hash_keys) + if redis_client.exists(f"m3u_refresh:{stream_hash}"): + # duplicate already processed by another batch + continue + + redis_client.set(f"m3u_refresh:{stream_hash}", "true") + stream_props = { + "name": name, + "url": url, + "logo_url": tvg_logo, + "tvg_id": tvg_id, + "m3u_account": account, + "channel_group": existing_groups[group_title], + "stream_hash": stream_hash, + } + + if stream_hash not in stream_hashes: + stream_hashes[stream_hash] = stream_props + except Exception as e: + 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())} + logger.info(f"Hashed {len(stream_hashes.keys())} unique streams") + + for stream_hash, stream_props in stream_hashes.items(): + if stream_hash in existing_streams: + obj = existing_streams[stream_hash] + changed = False + for key, value in stream_props.items(): + if getattr(obj, key) == value: + continue + changed = True + setattr(obj, key, value) + + obj.last_seen = timezone.now() + if changed: + streams_to_update.append(obj) + del existig_streams[stream_hash] + else: + existing_streams[stream_hash] = obj + else: + streams_to_create.append(Stream(**stream_props)) + + try: + if streams_to_create: + Stream.objects.bulk_create(streams_to_create, ignore_conflicts=True) + if streams_to_update: + Stream.objects.bulk_update(streams_to_update, stream_props.keys()) + if len(existing_streams.keys()) > 0: + Stream.objects.bulk_update(existing_streams.values(), ["last_seen"]) + except Exception as e: + logger.error(f"Bulk create failed: {str(e)}") + check_field_lengths(streams_to_create) + # check_field_lengths(streams_to_update) + + + return f"Batch processed: {len(streams_to_create)} created, {len(streams_to_update)} updated." + @shared_task def refresh_single_m3u_account(account_id): - logger.info(f"Task {refresh_single_m3u_account.request.id}: Starting refresh for account_id={account_id}") - + """Splits M3U processing into chunks and dispatches them as parallel tasks.""" if not acquire_lock('refresh_single_m3u_account', account_id): return f"Task already running for account_id={account_id}." + # Record start time + start_time = time.time() + send_progress_update(0, account_id) + try: account = M3UAccount.objects.get(id=account_id, is_active=True) filters = list(account.filters.all()) - logger.info(f"Found active M3UAccount (id={account.id}, name={account.name}).") except M3UAccount.DoesNotExist: - msg = f"M3UAccount with ID={account_id} not found or inactive." - logger.warning(msg) release_lock('refresh_single_m3u_account', account_id) - return msg - except Exception as e: - logger.error(f"Error fetching M3UAccount {account_id}: {e}") - release_lock('refresh_single_m3u_account', account_id) - return str(e) + return f"M3UAccount with ID={account_id} not found or inactive." - try: - lines = [] - if account.server_url: - if not account.user_agent: - err_msg = f"User-Agent not provided for account id {account_id}." - logger.error(err_msg) - release_lock('refresh_single_m3u_account', account_id) - return err_msg + # Fetch M3U lines and handle potential issues + # lines = fetch_m3u_lines(account) # Extracted fetch logic into separate function - headers = {"User-Agent": account.user_agent.user_agent} - response = requests.get(account.server_url, timeout=60, headers=headers) - response.raise_for_status() - lines = response.text.splitlines() - elif account.uploaded_file: - file_path = account.uploaded_file.path - with open(file_path, 'r', encoding='utf-8') as f: - lines = f.read().splitlines() - else: - err_msg = f"No server_url or uploaded_file provided for account_id={account_id}." + lines = [] + if account.server_url: + if not account.user_agent: + err_msg = f"User-Agent not provided for account id {account_id}." logger.error(err_msg) release_lock('refresh_single_m3u_account', account_id) return err_msg - except Exception as e: - err_msg = f"Failed fetching M3U: {e}" - logger.error(err_msg) - release_lock('refresh_single_m3u_account', account_id) - return err_msg - logger.info(f"M3U has {len(lines)} lines. Now parsing for Streams.") - skip_exts = ('.mkv', '.mp4', '.m4v', '.wav', '.avi', '.flv', '.m4p', '.mpg', - '.mpeg', '.m2v', '.mp2', '.mpe', '.mpv') + headers = {"User-Agent": account.user_agent.user_agent} + response = requests.get(account.server_url, timeout=60, headers=headers) + response.raise_for_status() + lines = response.text.splitlines() + elif account.uploaded_file: + file_path = account.uploaded_file.path + with open(file_path, 'r', encoding='utf-8') as f: + lines = f.read().splitlines() - created_count, updated_count, excluded_count = 0, 0, 0 - current_info = None + extinf_data = [] + stream_hashes = [] + groups = set("Default Group") for line in lines: line = line.strip() - if line.startswith('#EXTINF'): - extinf = parse_extinf_line(line) - if not extinf: - continue - name = extinf['name'] - tvg_id = extinf['attributes'].get('tvg-id', '') - tvg_logo = extinf['attributes'].get('tvg-logo', '') - # Prefer group-title from attributes if available. - group_title = extinf['attributes'].get('group-title', _get_group_title(line)) - logger.debug(f"Parsed EXTINF: name={name}, logo_url={tvg_logo}, tvg_id={tvg_id}, group_title={group_title}") - current_info = { - "name": name, - "logo_url": tvg_logo, - "group_title": group_title, - "tvg_id": tvg_id, - } - elif current_info and line.startswith('http'): - lower_line = line.lower() - if any(lower_line.endswith(ext) for ext in skip_exts): - logger.debug(f"Skipping file with unsupported extension: {line}") - current_info = None - continue + if line.startswith("#EXTINF"): + parsed = parse_extinf_line(line) + if parsed: + groups.add(parsed["attributes"].get("group-title", "Default Group")) + extinf_data.append(parsed) + elif extinf_data and line.startswith("http"): + # Associate URL with the last EXTINF line + extinf_data[-1]["url"] = line - if len(line) > 2000: - logger.warning(f"Stream URL too long, skipping: {line}") - excluded_count += 1 - current_info = None - continue + if not extinf_data: + release_lock('refresh_single_m3u_account', account_id) + return "No valid EXTINF data found." - if _matches_filters(current_info['name'], current_info['group_title'], filters): - logger.info(f"Stream excluded by filter: {current_info['name']} in group {current_info['group_title']}") - excluded_count += 1 - current_info = None - continue + groups = list(groups) + # Retrieve all unique groups so we can create / associate them before + # processing the streams themselves + process_groups(account, groups) - defaults = { - "logo_url": current_info["logo_url"], - "tvg_id": current_info["tvg_id"] - } - try: - channel_group, created = ChannelGroup.objects.get_or_create(name=current_info["group_title"]) - ChannelGroupM3UAccount.objects.get_or_create( - channel_group=channel_group, - m3u_account=account, - ) + hash_keys = CoreSettings.get_m3u_hash_key().split(",") - stream_props = defaults | { - "name": current_info["name"], - "url": line, - "m3u_account": account, - "channel_group": channel_group, - "last_seen": timezone.now(), - } + # 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, groups, hash_keys) for batch in batches) - stream_hash = Stream.generate_hash_key(stream_props) - obj, created = Stream.update_or_create_by_hash(stream_hash, **stream_props) + total_batches = len(batches) + completed_batches = 0 + logger.debug(f"Dispatched {len(batches)} parallel tasks for account_id={account_id}.") - if created: - created_count += 1 - else: - updated_count += 1 - except Exception as e: - logger.error(f"Failed to update/create stream {current_info['name']}: {e}") - finally: - current_info = None + # result = task_group.apply_async() + result = task_group.apply_async() + + while completed_batches < total_batches: + for async_result in result: + if async_result.ready(): # If the task has completed + task_result = async_result.result # The result of the task + logger.debug(f"Task completed with result: {task_result}") + completed_batches += 1 + + # Calculate progress + progress = int((completed_batches / total_batches) * 100) + + # Send progress update via Channels + send_progress_update(progress, account_id) + + # Optionally remove completed task from the group to prevent processing it again + result.remove(async_result) + else: + logger.debug(f"Task is still running.") + + end_time = time.time() + + # Calculate elapsed time + elapsed_time = end_time - start_time + + print(f"Function took {elapsed_time} seconds to execute.") - logger.info(f"Completed parsing. Created {created_count} new Streams, updated {updated_count} existing Streams, excluded {excluded_count} Streams.") release_lock('refresh_single_m3u_account', account_id) + + cursor = 0 + while True: + cursor, keys = redis_client.scan(cursor, match=f"m3u_refresh:*", count=BATCH_SIZE) + if keys: + redis_client.delete(*keys) # Delete the matching keys + if cursor == 0: + break + + return f"Dispatched jobs complete." + +def send_progress_update(progress, account_id): channel_layer = get_channel_layer() async_to_sync(channel_layer.group_send)( - "updates", + 'updates', { - "type": "update", - "data": {"success": True, "type": "m3u_refresh", "message": "M3U refresh completed successfully"} - }, + 'type': 'update', + "data": {"progress": progress, "type": "m3u_refresh", "account": account_id} + } ) - return f"Account {account_id} => Created {created_count}, updated {updated_count}, excluded {excluded_count} Streams." diff --git a/frontend/src/App.jsx b/frontend/src/App.jsx index 4d2476b4..bfe43fc1 100644 --- a/frontend/src/App.jsx +++ b/frontend/src/App.jsx @@ -26,6 +26,7 @@ import './index.css'; import mantineTheme from './mantineTheme'; import API from './api'; import { Notifications } from '@mantine/notifications'; +import M3URefreshNotification from './components/M3URefreshNotification'; const drawerWidth = 240; const miniDrawerWidth = 60; @@ -143,6 +144,7 @@ const App = () => { + diff --git a/frontend/src/WebSocket.jsx b/frontend/src/WebSocket.jsx index d361409b..92670fca 100644 --- a/frontend/src/WebSocket.jsx +++ b/frontend/src/WebSocket.jsx @@ -8,6 +8,7 @@ import React, { import useStreamsStore from './store/streams'; import { notifications } from '@mantine/notifications'; import useChannelsStore from './store/channels'; +import usePlaylistsStore from './store/playlists'; export const WebsocketContext = createContext(false, null, () => {}); @@ -17,6 +18,7 @@ export const WebsocketProvider = ({ children }) => { const { fetchStreams } = useStreamsStore(); const { setChannelStats } = useChannelsStore(); + const { setRefreshProgress } = usePlaylistsStore(); const ws = useRef(null); @@ -52,12 +54,16 @@ export const WebsocketProvider = ({ children }) => { event = JSON.parse(event.data); switch (event.data.type) { case 'm3u_refresh': + console.log('inside m3u_refresh event'); if (event.data.success) { fetchStreams(); notifications.show({ message: event.data.message, color: 'green.5', }); + } else if (event.data.progress) { + console.log('calling set progress'); + setRefreshProgress(event.data.account, event.data.progress); } break; diff --git a/frontend/src/components/M3URefreshNotification.jsx b/frontend/src/components/M3URefreshNotification.jsx new file mode 100644 index 00000000..34d50daf --- /dev/null +++ b/frontend/src/components/M3URefreshNotification.jsx @@ -0,0 +1,76 @@ +// frontend/src/components/FloatingVideo.js +import React, { useState } from 'react'; +import usePlaylistsStore from '../store/playlists'; +import { notifications } from '@mantine/notifications'; +import { IconCheck } from '@tabler/icons-react'; + +export default function M3URefreshNotification() { + const { playlists, refreshProgress, removeRefreshProgress } = + usePlaylistsStore(); + const [progress, setProgress] = useState({}); + + const clearAccountNotification = (id) => { + removeRefreshProgress(id); + setProgress({ + ...progress, + [id]: null, + }); + }; + + for (const id in refreshProgress) { + const playlist = playlists.find((pl) => pl.id == id); + if (!progress[id]) { + if (refreshProgress[id] == 100) { + const notificationId = notifications.show({ + loading: false, + title: `M3U Refresh: ${playlist.name}`, + message: `Refresh complete!`, + icon: , + }); + setProgress({ + ...progress, + [id]: notificationId, + }); + setTimeout(() => clearAccountNotification(id), 2000); + + return; + } + const notificationId = notifications.show({ + loading: true, + title: `M3U Refresh: ${playlist.name}`, + message: `Updating M3U: ${refreshProgress[id]}%`, + autoClose: false, + withCloseButton: false, + }); + + setProgress({ + ...progress, + [id]: notificationId, + }); + } else { + if (refreshProgress[id] == 0) { + notifications.update({ + id: progress[id], + message: `Starting...`, + }); + } else if (refreshProgress[id] == 100) { + notifications.update({ + id: progress[id], + message: `Refresh complete!`, + loading: false, + autoClose: 2000, + icon: , + }); + + setTimeout(() => clearAccountNotification(id), 2000); + } else { + notifications.update({ + id: progress[id], + message: `Updating M3U: ${refreshProgress[id]}%`, + }); + } + } + } + + return <>; +} diff --git a/frontend/src/store/auth.jsx b/frontend/src/store/auth.jsx index b1ff0c33..f5302713 100644 --- a/frontend/src/store/auth.jsx +++ b/frontend/src/store/auth.jsx @@ -119,7 +119,7 @@ const useAuthStore = create((set, get) => ({ const refreshToken = localStorage.getItem('refreshToken') || null; if (refreshToken) { - const loggedIn = await get().refreshToken(); + const loggedIn = await get().getRefreshToken(); if (loggedIn) { return true; } diff --git a/frontend/src/store/playlists.jsx b/frontend/src/store/playlists.jsx index 274c943a..1a496b3d 100644 --- a/frontend/src/store/playlists.jsx +++ b/frontend/src/store/playlists.jsx @@ -4,6 +4,7 @@ import api from '../api'; const usePlaylistsStore = create((set) => ({ playlists: [], profiles: {}, + refreshProgress: {}, isLoading: false, error: null, @@ -60,6 +61,24 @@ const usePlaylistsStore = create((set) => ({ ), // @TODO: remove playlist profiles here })), + + setRefreshProgress: (id, progress) => + set((state) => ({ + refreshProgress: { + ...state.refreshProgress, + [id]: progress, + }, + })), + + removeRefreshProgress: (id) => + set((state) => { + const updatedProgress = { ...state.refreshProgress }; + delete updatedProgress[id]; + + return { + refreshProgress: updatedProgress, + }; + }), })); export default usePlaylistsStore;