From 6180b4ffef6aaf522e39cd6683ca8b53632471d7 Mon Sep 17 00:00:00 2001 From: SergeantPanda Date: Mon, 28 Jul 2025 21:40:29 -0500 Subject: [PATCH 01/27] Save stream stats to database. --- ...am_stats_stream_stream_stats_updated_at.py | 23 ++++++++ apps/channels/models.py | 13 +++++ apps/channels/serializers.py | 2 + .../ts_proxy/services/channel_service.py | 58 +++++++++++++++++-- apps/proxy/ts_proxy/stream_manager.py | 14 ++++- 5 files changed, 102 insertions(+), 8 deletions(-) create mode 100644 apps/channels/migrations/0023_stream_stream_stats_stream_stream_stats_updated_at.py diff --git a/apps/channels/migrations/0023_stream_stream_stats_stream_stream_stats_updated_at.py b/apps/channels/migrations/0023_stream_stream_stats_stream_stream_stats_updated_at.py new file mode 100644 index 00000000..1b0fdbe8 --- /dev/null +++ b/apps/channels/migrations/0023_stream_stream_stats_stream_stream_stats_updated_at.py @@ -0,0 +1,23 @@ +# Generated by Django 5.1.6 on 2025-07-29 02:39 + +from django.db import migrations, models + + +class Migration(migrations.Migration): + + dependencies = [ + ('dispatcharr_channels', '0022_channel_auto_created_channel_auto_created_by_and_more'), + ] + + operations = [ + migrations.AddField( + model_name='stream', + name='stream_stats', + field=models.JSONField(blank=True, help_text='JSON object containing stream statistics like video codec, resolution, etc.', null=True), + ), + migrations.AddField( + model_name='stream', + name='stream_stats_updated_at', + field=models.DateTimeField(blank=True, db_index=True, help_text='When stream statistics were last updated', null=True), + ), + ] diff --git a/apps/channels/models.py b/apps/channels/models.py index f53a9875..d6c3faef 100644 --- a/apps/channels/models.py +++ b/apps/channels/models.py @@ -95,6 +95,19 @@ class Stream(models.Model): ) last_seen = models.DateTimeField(db_index=True, default=datetime.now) custom_properties = models.TextField(null=True, blank=True) + + # Stream statistics fields + stream_stats = models.JSONField( + null=True, + blank=True, + help_text="JSON object containing stream statistics like video codec, resolution, etc." + ) + stream_stats_updated_at = models.DateTimeField( + null=True, + blank=True, + help_text="When stream statistics were last updated", + db_index=True + ) class Meta: # If you use m3u_account, you might do unique_together = ('name','url','m3u_account') diff --git a/apps/channels/serializers.py b/apps/channels/serializers.py index 9273b265..32fd4a74 100644 --- a/apps/channels/serializers.py +++ b/apps/channels/serializers.py @@ -104,6 +104,8 @@ class StreamSerializer(serializers.ModelSerializer): "is_custom", "channel_group", "stream_hash", + "stream_stats", + "stream_stats_updated_at", ] def get_fields(self): diff --git a/apps/proxy/ts_proxy/services/channel_service.py b/apps/proxy/ts_proxy/services/channel_service.py index 026aa883..932479ea 100644 --- a/apps/proxy/ts_proxy/services/channel_service.py +++ b/apps/proxy/ts_proxy/services/channel_service.py @@ -417,8 +417,8 @@ class ChannelService: return False, None, None, {"error": f"Exception: {str(e)}"} @staticmethod - def parse_and_store_stream_info(channel_id, stream_info_line, stream_type="video"): - """Parse FFmpeg stream info line and store in Redis metadata""" + def parse_and_store_stream_info(channel_id, stream_info_line, stream_type="video", stream_id=None): + """Parse FFmpeg stream info line and store in Redis metadata and database""" try: if stream_type == "input": # Example lines: @@ -432,6 +432,9 @@ class ChannelService: # Store in Redis if we have valid data if input_format: ChannelService._update_stream_info_in_redis(channel_id, None, None, None, None, None, None, None, None, None, None, None, input_format) + # Save to database if stream_id is provided + if stream_id: + ChannelService._update_stream_stats_in_db(stream_id, stream_type=input_format) logger.debug(f"Input format info - Format: {input_format} for channel {channel_id}") @@ -480,6 +483,16 @@ class ChannelService: # Store in Redis if we have valid data if any(x is not None for x in [video_codec, resolution, source_fps, pixel_format, video_bitrate]): ChannelService._update_stream_info_in_redis(channel_id, video_codec, resolution, width, height, source_fps, pixel_format, video_bitrate, None, None, None, None, None) + # Save to database if stream_id is provided + if stream_id: + ChannelService._update_stream_stats_in_db( + stream_id, + video_codec=video_codec, + resolution=resolution, + source_fps=source_fps, + pixel_format=pixel_format, + video_bitrate=video_bitrate + ) logger.info(f"Video stream info - Codec: {video_codec}, Resolution: {resolution}, " f"Source FPS: {source_fps}, Pixel Format: {pixel_format}, " @@ -511,9 +524,15 @@ class ChannelService: # Store in Redis if we have valid data if any(x is not None for x in [audio_codec, sample_rate, channels, audio_bitrate]): ChannelService._update_stream_info_in_redis(channel_id, None, None, None, None, None, None, None, audio_codec, sample_rate, channels, audio_bitrate, None) - - logger.info(f"Audio stream info - Codec: {audio_codec}, Sample Rate: {sample_rate} Hz, " - f"Channels: {channels}, Audio Bitrate: {audio_bitrate} kb/s") + # Save to database if stream_id is provided + if stream_id: + ChannelService._update_stream_stats_in_db( + stream_id, + audio_codec=audio_codec, + sample_rate=sample_rate, + audio_channels=channels, + audio_bitrate=audio_bitrate + ) except Exception as e: logger.debug(f"Error parsing FFmpeg {stream_type} stream info: {e}") @@ -575,6 +594,35 @@ class ChannelService: logger.error(f"Error updating stream info in Redis: {e}") return False + @staticmethod + def _update_stream_stats_in_db(stream_id, **stats): + """Update stream stats in database""" + try: + from apps.channels.models import Stream + from django.utils import timezone + + stream = Stream.objects.get(id=stream_id) + + # Get existing stats or create new dict + current_stats = stream.stream_stats or {} + + # Update with new stats + for key, value in stats.items(): + if value is not None: + current_stats[key] = value + + # Save updated stats and timestamp + stream.stream_stats = current_stats + stream.stream_stats_updated_at = timezone.now() + stream.save(update_fields=['stream_stats', 'stream_stats_updated_at']) + + logger.debug(f"Updated stream stats in database for stream {stream_id}: {stats}") + return True + + except Exception as e: + logger.error(f"Error updating stream stats in database for stream {stream_id}: {e}") + return False + # Helper methods for Redis operations @staticmethod diff --git a/apps/proxy/ts_proxy/stream_manager.py b/apps/proxy/ts_proxy/stream_manager.py index f7c538c2..e80d4527 100644 --- a/apps/proxy/ts_proxy/stream_manager.py +++ b/apps/proxy/ts_proxy/stream_manager.py @@ -587,9 +587,9 @@ class StreamManager: from .services.channel_service import ChannelService if "video:" in content_lower: - ChannelService.parse_and_store_stream_info(self.channel_id, content, "video") + ChannelService.parse_and_store_stream_info(self.channel_id, content, "video", self.current_stream_id) elif "audio:" in content_lower: - ChannelService.parse_and_store_stream_info(self.channel_id, content, "audio") + ChannelService.parse_and_store_stream_info(self.channel_id, content, "audio", self.current_stream_id) # Determine log level based on content if any(keyword in content_lower for keyword in ['error', 'failed', 'cannot', 'invalid', 'corrupt']): @@ -605,7 +605,7 @@ class StreamManager: if content.startswith('Input #0'): # If it's input 0, parse stream info from .services.channel_service import ChannelService - ChannelService.parse_and_store_stream_info(self.channel_id, content, "input") + ChannelService.parse_and_store_stream_info(self.channel_id, content, "input", self.current_stream_id) else: # Everything else at debug level logger.debug(f"FFmpeg stderr for channel {self.channel_id}: {content}") @@ -649,6 +649,14 @@ class StreamManager: if any(x is not None for x in [ffmpeg_speed, ffmpeg_fps, actual_fps, ffmpeg_output_bitrate]): self._update_ffmpeg_stats_in_redis(ffmpeg_speed, ffmpeg_fps, actual_fps, ffmpeg_output_bitrate) + # Also save ffmpeg_output_bitrate to database if we have stream_id + if ffmpeg_output_bitrate is not None and self.current_stream_id: + from .services.channel_service import ChannelService + ChannelService._update_stream_stats_in_db( + self.current_stream_id, + ffmpeg_output_bitrate=ffmpeg_output_bitrate + ) + # Fix the f-string formatting actual_fps_str = f"{actual_fps:.1f}" if actual_fps is not None else "N/A" ffmpeg_output_bitrate_str = f"{ffmpeg_output_bitrate:.1f}" if ffmpeg_output_bitrate is not None else "N/A" From b222dbe5a32eb33ee557a8452b0e8feaf1b6456f Mon Sep 17 00:00:00 2001 From: SergeantPanda Date: Tue, 29 Jul 2025 11:21:44 -0500 Subject: [PATCH 02/27] Display stream qualites in channel table. --- .../components/tables/ChannelTableStreams.jsx | 134 ++++++++++++++++-- .../src/components/tables/StreamsTable.jsx | 8 +- 2 files changed, 132 insertions(+), 10 deletions(-) diff --git a/frontend/src/components/tables/ChannelTableStreams.jsx b/frontend/src/components/tables/ChannelTableStreams.jsx index 097a2ba8..3e034fb3 100644 --- a/frontend/src/components/tables/ChannelTableStreams.jsx +++ b/frontend/src/components/tables/ChannelTableStreams.jsx @@ -8,7 +8,11 @@ import { Text, useMantineTheme, Center, + Badge, + Group, + Tooltip, } from '@mantine/core'; +import { notifications } from '@mantine/notifications'; import { useReactTable, getCoreRowModel, @@ -141,6 +145,19 @@ const ChannelStreams = ({ channel, isExpanded }) => { await API.requeryChannels(); }; + // Create M3U account map for quick lookup + const m3uAccountsMap = useMemo(() => { + const map = {}; + if (playlists && Array.isArray(playlists)) { + playlists.forEach((account) => { + if (account.id) { + map[account.id] = account.name; + } + }); + } + return map; + }, [playlists]); + const table = useReactTable({ columns: useMemo( () => [ @@ -152,14 +169,114 @@ const ChannelStreams = ({ channel, isExpanded }) => { }, { id: 'name', - header: 'Name', + header: 'Stream Info', accessorKey: 'name', - }, - { - id: 'm3u', - header: 'M3U', - accessorFn: (row) => - playlists.find((playlist) => playlist.id === row.m3u_account)?.name, + cell: ({ row }) => { + const stream = row.original; + const playlistName = playlists[stream.m3u_account]?.name || 'Unknown'; + const accountName = m3uAccountsMap[stream.m3u_account] || playlistName; + + return ( + + {stream.name} + + + {accountName} + + {stream.quality && ( + + {stream.quality} + + )} + {stream.url && ( + + { + e.stopPropagation(); + navigator.clipboard.writeText(stream.url); + notifications.show({ + title: 'URL Copied', + message: 'Stream URL copied to clipboard', + color: 'green', + }); + }} + > + URL + + + )} + + {stream.stream_stats && ( + + {/* Video Information */} + {(stream.stream_stats.video_codec || stream.stream_stats.resolution || stream.stream_stats.video_bitrate || stream.stream_stats.source_fps) && ( + <> + Video: + {stream.stream_stats.resolution && ( + + {stream.stream_stats.resolution} + + )} + {stream.stream_stats.video_bitrate && ( + + {stream.stream_stats.video_bitrate} KbPS + + )} + {stream.stream_stats.source_fps && ( + + {stream.stream_stats.source_fps} FPS + + )} + {stream.stream_stats.video_codec && ( + + {stream.stream_stats.video_codec.toUpperCase()} + + )} + + )} + + {/* Audio Information */} + {(stream.stream_stats.audio_codec || stream.stream_stats.audio_channels || stream.stream_stats.audio_bitrate) && ( + <> + Audio: + {stream.stream_stats.audio_channels && ( + + {stream.stream_stats.audio_channels} + + )} + {stream.stream_stats.audio_codec && ( + + {stream.stream_stats.audio_codec.toUpperCase()} + + )} + {stream.stream_stats.audio_bitrate && ( + + {stream.stream_stats.audio_bitrate} KbPS + + )} + + )} + + {/* Output Bitrate */} + {(stream.stream_stats.ffmpeg_output_bitrate) && ( + <> + Output Bitrate: + {stream.stream_stats.ffmpeg_output_bitrate && ( + + {stream.stream_stats.ffmpeg_output_bitrate} KbPS + + )} + + )} + + )} + + ); + }, }, { id: 'actions', @@ -178,7 +295,7 @@ const ChannelStreams = ({ channel, isExpanded }) => { ), }, ], - [data, playlists] + [data, playlists, m3uAccountsMap] ), data, state: { @@ -292,4 +409,5 @@ const ChannelStreams = ({ channel, isExpanded }) => { ); }; + export default ChannelStreams; diff --git a/frontend/src/components/tables/StreamsTable.jsx b/frontend/src/components/tables/StreamsTable.jsx index c149e2a7..616a52a3 100644 --- a/frontend/src/components/tables/StreamsTable.jsx +++ b/frontend/src/components/tables/StreamsTable.jsx @@ -102,10 +102,14 @@ const StreamRowActions = ({ 'ID:', row.original.id, 'Hash:', - row.original.stream_hash + row.original.stream_hash, + 'Quality:', + row.original.quality, + 'Stats:', + row.original.stream_stats ); handleWatchStream(row.original.stream_hash); - }, [row.original.id]); // Add proper dependency to ensure correct stream + }, [row.original, handleWatchStream]); // Add proper dependencies to ensure correct stream const iconSize = tableSize == 'default' ? 'sm' : tableSize == 'compact' ? 'xs' : 'md'; From 5b076fd00bdf27acc19839e9c5e0d5ba74352d04 Mon Sep 17 00:00:00 2001 From: SergeantPanda Date: Tue, 29 Jul 2025 11:35:22 -0500 Subject: [PATCH 03/27] Allow for clicking url badge and copy to clipboard. --- frontend/src/components/Sidebar.jsx | 24 ++++----- .../components/tables/ChannelTableStreams.jsx | 11 ++-- .../src/components/tables/ChannelsTable.jsx | 53 ++++++++----------- frontend/src/utils.js | 13 +++-- 4 files changed, 48 insertions(+), 53 deletions(-) diff --git a/frontend/src/components/Sidebar.jsx b/frontend/src/components/Sidebar.jsx index 08114d99..03ad831e 100644 --- a/frontend/src/components/Sidebar.jsx +++ b/frontend/src/components/Sidebar.jsx @@ -1,5 +1,6 @@ import React, { useRef, useEffect, useState } from 'react'; import { Link, useLocation } from 'react-router-dom'; +import { copyToClipboard } from '../utils'; import { ListOrdered, Play, @@ -27,6 +28,7 @@ import { ActionIcon, Menu, } from '@mantine/core'; +import { notifications } from '@mantine/notifications'; import logo from '../images/logo.png'; import useChannelsStore from '../store/channels'; import './sidebar.css'; @@ -168,19 +170,15 @@ const Sidebar = ({ collapsed, toggleDrawer, drawerWidth, miniDrawerWidth }) => { }, []); const copyPublicIP = async () => { - try { - await navigator.clipboard.writeText(environment.public_ip); - } catch (err) { - const inputElement = publicIPRef.current; // Get the actual input - console.log(inputElement); - - if (inputElement) { - inputElement.focus(); - inputElement.select(); - - // For older browsers - document.execCommand('copy'); - } + const success = await copyToClipboard(environment.public_ip); + if (success) { + notifications.show({ + title: 'Success', + message: 'Public IP copied to clipboard', + color: 'green', + }); + } else { + console.error('Failed to copy public IP to clipboard'); } }; diff --git a/frontend/src/components/tables/ChannelTableStreams.jsx b/frontend/src/components/tables/ChannelTableStreams.jsx index 3e034fb3..391f4d30 100644 --- a/frontend/src/components/tables/ChannelTableStreams.jsx +++ b/frontend/src/components/tables/ChannelTableStreams.jsx @@ -1,5 +1,6 @@ import React, { useMemo, useState, useEffect } from 'react'; import API from '../../api'; +import { copyToClipboard } from '../../utils'; import { GripHorizontal, SquareMinus } from 'lucide-react'; import { Box, @@ -195,13 +196,13 @@ const ChannelStreams = ({ channel, isExpanded }) => { variant="light" color="indigo" style={{ cursor: 'pointer' }} - onClick={(e) => { + onClick={async (e) => { e.stopPropagation(); - navigator.clipboard.writeText(stream.url); + const success = await copyToClipboard(stream.url); notifications.show({ - title: 'URL Copied', - message: 'Stream URL copied to clipboard', - color: 'green', + title: success ? 'URL Copied' : 'Copy Failed', + message: success ? 'Stream URL copied to clipboard' : 'Failed to copy URL to clipboard', + color: success ? 'green' : 'red', }); }} > diff --git a/frontend/src/components/tables/ChannelsTable.jsx b/frontend/src/components/tables/ChannelsTable.jsx index 3d34ca55..3380fbd2 100644 --- a/frontend/src/components/tables/ChannelsTable.jsx +++ b/frontend/src/components/tables/ChannelsTable.jsx @@ -524,24 +524,12 @@ const ChannelsTable = ({ }) => { }; const handleCopy = async (textToCopy, ref) => { - try { - await navigator.clipboard.writeText(textToCopy); - notifications.show({ - title: 'Copied!', - // style: { width: '200px', left: '200px' }, - }); - } catch (err) { - const inputElement = ref.current; // Get the actual input - - if (inputElement) { - inputElement.focus(); - inputElement.select(); - - // For older browsers - document.execCommand('copy'); - notifications.show({ title: 'Copied!' }); - } - } + const success = await copyToClipboard(textToCopy); + notifications.show({ + title: success ? 'Copied!' : 'Copy Failed', + message: success ? undefined : 'Failed to copy to clipboard', + color: success ? 'green' : 'red', + }); }; // Build URLs with parameters const buildM3UUrl = () => { @@ -564,25 +552,30 @@ const ChannelsTable = ({ }) => { return params.toString() ? `${baseUrl}?${params.toString()}` : baseUrl; }; // Example copy URLs - const copyM3UUrl = () => { - copyToClipboard(buildM3UUrl()); + const copyM3UUrl = async () => { + const success = await copyToClipboard(buildM3UUrl()); notifications.show({ - title: 'M3U URL Copied!', - message: 'The M3U URL has been copied to your clipboard.', + title: success ? 'M3U URL Copied!' : 'Copy Failed', + message: success ? 'The M3U URL has been copied to your clipboard.' : 'Failed to copy M3U URL to clipboard', + color: success ? 'green' : 'red', }); }; - const copyEPGUrl = () => { - copyToClipboard(buildEPGUrl()); + + const copyEPGUrl = async () => { + const success = await copyToClipboard(buildEPGUrl()); notifications.show({ - title: 'EPG URL Copied!', - message: 'The EPG URL has been copied to your clipboard.', + title: success ? 'EPG URL Copied!' : 'Copy Failed', + message: success ? 'The EPG URL has been copied to your clipboard.' : 'Failed to copy EPG URL to clipboard', + color: success ? 'green' : 'red', }); }; - const copyHDHRUrl = () => { - copyToClipboard(hdhrUrl); + + const copyHDHRUrl = async () => { + const success = await copyToClipboard(hdhrUrl); notifications.show({ - title: 'HDHR URL Copied!', - message: 'The HDHR URL has been copied to your clipboard.', + title: success ? 'HDHR URL Copied!' : 'Copy Failed', + message: success ? 'The HDHR URL has been copied to your clipboard.' : 'Failed to copy HDHR URL to clipboard', + color: success ? 'green' : 'red', }); }; diff --git a/frontend/src/utils.js b/frontend/src/utils.js index a488ce8d..c995fd89 100644 --- a/frontend/src/utils.js +++ b/frontend/src/utils.js @@ -65,24 +65,27 @@ export const getDescendantProp = (obj, path) => path.split('.').reduce((acc, part) => acc && acc[part], obj); export const copyToClipboard = async (value) => { - let copied = false; if (navigator.clipboard) { // Modern method, using navigator.clipboard try { await navigator.clipboard.writeText(value); - copied = true; + return true; } catch (err) { console.error('Failed to copy: ', err); } } - if (!copied) { - // Fallback method for environments without clipboard support + // Fallback method for environments without clipboard support + try { const textarea = document.createElement('textarea'); textarea.value = value; document.body.appendChild(textarea); textarea.select(); - document.execCommand('copy'); + const successful = document.execCommand('copy'); document.body.removeChild(textarea); + return successful; + } catch (err) { + console.error('Failed to copy with fallback method: ', err); + return false; } }; From 72ecb88c76a9b1be8f4ef88b7ef0b38c40f28cf7 Mon Sep 17 00:00:00 2001 From: SergeantPanda Date: Tue, 29 Jul 2025 14:59:03 -0500 Subject: [PATCH 04/27] Fix bug where creating a channel from a stream that isn't displayed in the table has invalid stream name. --- frontend/src/components/tables/StreamsTable.jsx | 14 +++++++++----- 1 file changed, 9 insertions(+), 5 deletions(-) diff --git a/frontend/src/components/tables/StreamsTable.jsx b/frontend/src/components/tables/StreamsTable.jsx index 616a52a3..4731c019 100644 --- a/frontend/src/components/tables/StreamsTable.jsx +++ b/frontend/src/components/tables/StreamsTable.jsx @@ -103,10 +103,6 @@ const StreamRowActions = ({ row.original.id, 'Hash:', row.original.stream_hash, - 'Quality:', - row.original.quality, - 'Stats:', - row.original.stream_stats ); handleWatchStream(row.original.stream_hash); }, [row.original, handleWatchStream]); // Add proper dependencies to ensure correct stream @@ -414,8 +410,16 @@ const StreamsTable = ({ }) => { try { const selectedChannelProfileId = useChannelsStore.getState().selectedProfileId; + // Try to fetch the actual stream data for selected streams + let streamsData = []; + try { + streamsData = await API.getStreamsByIds(selectedStreamIds); + } catch (error) { + console.warn('Could not fetch stream details, using IDs only:', error); + } + const streamData = selectedStreamIds.map(streamId => { - const stream = data.find(s => s.id === streamId); + const stream = streamsData.find(s => s.id === streamId); return { stream_id: streamId, name: stream?.name || `Stream ${streamId}`, From cb49172e988200210bee52aacc2902792138b0c5 Mon Sep 17 00:00:00 2001 From: SergeantPanda Date: Tue, 29 Jul 2025 15:02:13 -0500 Subject: [PATCH 05/27] Fix bug where creating a channel from a stream that isn't displayed in the table has invalid stream name. --- frontend/src/api.js | 14 ++++++++++++++ 1 file changed, 14 insertions(+) diff --git a/frontend/src/api.js b/frontend/src/api.js index effc0bdd..ddaccbc7 100644 --- a/frontend/src/api.js +++ b/frontend/src/api.js @@ -1675,4 +1675,18 @@ export default class API { errorNotification('Failed to trigger stream rehash', e); } } + + static async getStreamsByIds(ids) { + try { + const params = new URLSearchParams(); + params.append('ids', ids.join(',')); + const response = await request( + `${host}/api/channels/streams/?${params.toString()}` + ); + + return response.results || response; + } catch (e) { + errorNotification('Failed to retrieve streams by IDs', e); + } + } } From 44f8d4576805bdf1ec9e4ffc789e13883d5cc1bf Mon Sep 17 00:00:00 2001 From: SergeantPanda Date: Tue, 29 Jul 2025 15:02:35 -0500 Subject: [PATCH 06/27] Add the ability to see advanced stats for channel streams. --- .../components/tables/ChannelTableStreams.jsx | 168 +++++++++++++++++- 1 file changed, 163 insertions(+), 5 deletions(-) diff --git a/frontend/src/components/tables/ChannelTableStreams.jsx b/frontend/src/components/tables/ChannelTableStreams.jsx index 391f4d30..373427bb 100644 --- a/frontend/src/components/tables/ChannelTableStreams.jsx +++ b/frontend/src/components/tables/ChannelTableStreams.jsx @@ -1,7 +1,7 @@ import React, { useMemo, useState, useEffect } from 'react'; import API from '../../api'; import { copyToClipboard } from '../../utils'; -import { GripHorizontal, SquareMinus } from 'lucide-react'; +import { GripHorizontal, SquareMinus, ChevronDown, ChevronRight } from 'lucide-react'; import { Box, ActionIcon, @@ -12,6 +12,8 @@ import { Badge, Group, Tooltip, + Collapse, + Button, } from '@mantine/core'; import { notifications } from '@mantine/notifications'; import { @@ -159,6 +161,122 @@ const ChannelStreams = ({ channel, isExpanded }) => { return map; }, [playlists]); + // Add state for tracking which streams have advanced stats expanded + const [expandedAdvancedStats, setExpandedAdvancedStats] = useState(new Set()); + + // Helper function to categorize stream stats + const categorizeStreamStats = (stats) => { + if (!stats) return { basic: {}, video: {}, audio: {}, technical: {}, other: {} }; + + const categories = { + basic: {}, + video: {}, + audio: {}, + technical: {}, + other: {} + }; + + // Define which stats go in which category + const categoryMapping = { + basic: ['resolution', 'video_codec', 'source_fps', 'audio_codec', 'audio_channels'], + video: ['video_bitrate', 'pixel_format', 'width', 'height', 'aspect_ratio', 'frame_rate'], + audio: ['audio_bitrate', 'sample_rate', 'audio_format', 'audio_channels_layout'], + technical: ['stream_type', 'container_format', 'duration', 'file_size', 'ffmpeg_output_bitrate'], + other: [] // Will catch anything not categorized above + }; + + // Categorize each stat + Object.entries(stats).forEach(([key, value]) => { + let categorized = false; + + for (const [category, keys] of Object.entries(categoryMapping)) { + if (keys.includes(key)) { + categories[category][key] = value; + categorized = true; + break; + } + } + + // If not categorized, put it in 'other' + if (!categorized) { + categories.other[key] = value; + } + }); + + return categories; + }; + + // Function to format stat values for display + const formatStatValue = (key, value) => { + if (value === null || value === undefined) return 'N/A'; + + // Handle specific formatting cases + switch (key) { + case 'video_bitrate': + case 'audio_bitrate': + case 'ffmpeg_output_bitrate': + return `${value} kbps`; + case 'source_fps': + case 'frame_rate': + return `${value} fps`; + case 'sample_rate': + return `${value} Hz`; + case 'file_size': + // Convert bytes to appropriate unit + if (typeof value === 'number') { + if (value < 1024) return `${value} B`; + if (value < 1024 * 1024) return `${(value / 1024).toFixed(2)} KB`; + if (value < 1024 * 1024 * 1024) return `${(value / (1024 * 1024)).toFixed(2)} MB`; + return `${(value / (1024 * 1024 * 1024)).toFixed(2)} GB`; + } + return value; + case 'duration': + // Format duration if it's in seconds + if (typeof value === 'number') { + const hours = Math.floor(value / 3600); + const minutes = Math.floor((value % 3600) / 60); + const seconds = Math.floor(value % 60); + return `${hours.toString().padStart(2, '0')}:${minutes.toString().padStart(2, '0')}:${seconds.toString().padStart(2, '0')}`; + } + return value; + default: + return value.toString(); + } + }; + + // Function to render a stats category + const renderStatsCategory = (categoryName, stats) => { + if (!stats || Object.keys(stats).length === 0) return null; + + return ( + + + {categoryName} + + + {Object.entries(stats).map(([key, value]) => ( + + + {key.replace(/_/g, ' ').replace(/\b\w/g, l => l.toUpperCase())}: {formatStatValue(key, value)} + + + ))} + + + ); + }; + + // Function to toggle advanced stats for a stream + const toggleAdvancedStats = (streamId) => { + const newExpanded = new Set(expandedAdvancedStats); + if (newExpanded.has(streamId)) { + newExpanded.delete(streamId); + } else { + newExpanded.add(streamId); + } + setExpandedAdvancedStats(newExpanded); + }; + const table = useReactTable({ columns: useMemo( () => [ @@ -177,6 +295,12 @@ const ChannelStreams = ({ channel, isExpanded }) => { const playlistName = playlists[stream.m3u_account]?.name || 'Unknown'; const accountName = m3uAccountsMap[stream.m3u_account] || playlistName; + // Categorize stream stats + const categorizedStats = categorizeStreamStats(stream.stream_stats); + const hasAdvancedStats = Object.values(categorizedStats).some(category => + Object.keys(category).length > 0 + ); + return ( {stream.name} @@ -211,6 +335,8 @@ const ChannelStreams = ({ channel, isExpanded }) => { )} + + {/* Basic Stream Stats (always shown) */} {stream.stream_stats && ( {/* Video Information */} @@ -224,7 +350,7 @@ const ChannelStreams = ({ channel, isExpanded }) => { )} {stream.stream_stats.video_bitrate && ( - {stream.stream_stats.video_bitrate} KbPS + {stream.stream_stats.video_bitrate} kbps )} {stream.stream_stats.source_fps && ( @@ -256,7 +382,7 @@ const ChannelStreams = ({ channel, isExpanded }) => { )} {stream.stream_stats.audio_bitrate && ( - {stream.stream_stats.audio_bitrate} KbPS + {stream.stream_stats.audio_bitrate} kbps )} @@ -268,13 +394,45 @@ const ChannelStreams = ({ channel, isExpanded }) => { Output Bitrate: {stream.stream_stats.ffmpeg_output_bitrate && ( - {stream.stream_stats.ffmpeg_output_bitrate} KbPS + {stream.stream_stats.ffmpeg_output_bitrate} kbps )} )} )} + + {/* Advanced Stats Toggle Button */} + {hasAdvancedStats && ( + + + + )} + + {/* Advanced Stats (expandable) */} + + + {renderStatsCategory('Video', categorizedStats.video)} + {renderStatsCategory('Audio', categorizedStats.audio)} + {renderStatsCategory('Technical', categorizedStats.technical)} + {renderStatsCategory('Other', categorizedStats.other)} + + {/* Show when stats were last updated */} + {stream.stream_stats_updated_at && ( + + Last updated: {new Date(stream.stream_stats_updated_at).toLocaleString()} + + )} + + ); }, @@ -296,7 +454,7 @@ const ChannelStreams = ({ channel, isExpanded }) => { ), }, ], - [data, playlists, m3uAccountsMap] + [data, playlists, m3uAccountsMap, expandedAdvancedStats] ), data, state: { From 7551869a2eaa8101926af557839947fa68288b6b Mon Sep 17 00:00:00 2001 From: SergeantPanda Date: Tue, 29 Jul 2025 15:12:14 -0500 Subject: [PATCH 07/27] Remove audio bitrate from basic stats. --- frontend/src/components/tables/ChannelTableStreams.jsx | 7 +------ 1 file changed, 1 insertion(+), 6 deletions(-) diff --git a/frontend/src/components/tables/ChannelTableStreams.jsx b/frontend/src/components/tables/ChannelTableStreams.jsx index 373427bb..2da4002d 100644 --- a/frontend/src/components/tables/ChannelTableStreams.jsx +++ b/frontend/src/components/tables/ChannelTableStreams.jsx @@ -367,7 +367,7 @@ const ChannelStreams = ({ channel, isExpanded }) => { )} {/* Audio Information */} - {(stream.stream_stats.audio_codec || stream.stream_stats.audio_channels || stream.stream_stats.audio_bitrate) && ( + {(stream.stream_stats.audio_codec || stream.stream_stats.audio_channels) && ( <> Audio: {stream.stream_stats.audio_channels && ( @@ -380,11 +380,6 @@ const ChannelStreams = ({ channel, isExpanded }) => { {stream.stream_stats.audio_codec.toUpperCase()} )} - {stream.stream_stats.audio_bitrate && ( - - {stream.stream_stats.audio_bitrate} kbps - - )} )} From e26ecad013c0fba5f76c5df0fbecc721462871de Mon Sep 17 00:00:00 2001 From: SergeantPanda Date: Tue, 29 Jul 2025 15:17:18 -0500 Subject: [PATCH 08/27] Move m3u and url badges to same line as stream name. --- frontend/src/components/tables/ChannelTableStreams.jsx | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/frontend/src/components/tables/ChannelTableStreams.jsx b/frontend/src/components/tables/ChannelTableStreams.jsx index 2da4002d..d226c52a 100644 --- a/frontend/src/components/tables/ChannelTableStreams.jsx +++ b/frontend/src/components/tables/ChannelTableStreams.jsx @@ -303,8 +303,8 @@ const ChannelStreams = ({ channel, isExpanded }) => { return ( - {stream.name} - + + {stream.name} {accountName} From 613c0d8bb559dd25c37229e68dc15f38ae77a914 Mon Sep 17 00:00:00 2001 From: SergeantPanda Date: Tue, 29 Jul 2025 15:43:44 -0500 Subject: [PATCH 09/27] Add input_bitrate to technical for future use. --- frontend/src/components/tables/ChannelTableStreams.jsx | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/frontend/src/components/tables/ChannelTableStreams.jsx b/frontend/src/components/tables/ChannelTableStreams.jsx index d226c52a..11fa9600 100644 --- a/frontend/src/components/tables/ChannelTableStreams.jsx +++ b/frontend/src/components/tables/ChannelTableStreams.jsx @@ -181,7 +181,7 @@ const ChannelStreams = ({ channel, isExpanded }) => { basic: ['resolution', 'video_codec', 'source_fps', 'audio_codec', 'audio_channels'], video: ['video_bitrate', 'pixel_format', 'width', 'height', 'aspect_ratio', 'frame_rate'], audio: ['audio_bitrate', 'sample_rate', 'audio_format', 'audio_channels_layout'], - technical: ['stream_type', 'container_format', 'duration', 'file_size', 'ffmpeg_output_bitrate'], + technical: ['stream_type', 'container_format', 'duration', 'file_size', 'ffmpeg_output_bitrate', 'input_bitrate'], other: [] // Will catch anything not categorized above }; From 4ae66e0bc9572001c4a1bad5cd771861833a019f Mon Sep 17 00:00:00 2001 From: SergeantPanda Date: Thu, 31 Jul 2025 09:52:02 -0500 Subject: [PATCH 10/27] Add membership creation in UpdateChannelMembershipAPIView if not found. Fixes #275 --- apps/channels/api_views.py | 14 +++++++++++--- 1 file changed, 11 insertions(+), 3 deletions(-) diff --git a/apps/channels/api_views.py b/apps/channels/api_views.py index 636d4875..0221a266 100644 --- a/apps/channels/api_views.py +++ b/apps/channels/api_views.py @@ -1508,9 +1508,17 @@ class UpdateChannelMembershipAPIView(APIView): """Enable or disable a channel for a specific group""" channel_profile = get_object_or_404(ChannelProfile, id=profile_id) channel = get_object_or_404(Channel, id=channel_id) - membership = get_object_or_404( - ChannelProfileMembership, channel_profile=channel_profile, channel=channel - ) + try: + membership = ChannelProfileMembership.objects.get( + channel_profile=channel_profile, channel=channel + ) + except ChannelProfileMembership.DoesNotExist: + # Create the membership if it does not exist (for custom channels) + membership = ChannelProfileMembership.objects.create( + channel_profile=channel_profile, + channel=channel, + enabled=False # Default to False, will be updated below + ) serializer = ChannelProfileMembershipSerializer( membership, data=request.data, partial=True From e029cd8b3dbbff550b9a1926370e156de13571af Mon Sep 17 00:00:00 2001 From: SergeantPanda Date: Thu, 31 Jul 2025 10:22:43 -0500 Subject: [PATCH 11/27] Fix XML escaping for channel ID in generate_dummy_epg function --- apps/output/views.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/apps/output/views.py b/apps/output/views.py index 8d58a1b3..3fcd512b 100644 --- a/apps/output/views.py +++ b/apps/output/views.py @@ -266,7 +266,7 @@ def generate_dummy_epg( # Create program entry with escaped channel name xml_lines.append( - f' ' + f' ' ) xml_lines.append(f" {html.escape(program['title'])}") xml_lines.append(f" {html.escape(program['description'])}") From 5a887cc55ab6cabe58bdda7a15bccc66b58a84d8 Mon Sep 17 00:00:00 2001 From: SergeantPanda Date: Thu, 31 Jul 2025 13:54:20 -0500 Subject: [PATCH 12/27] Bump Postgres to version 17. --- docker/DispatcharrBase | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/docker/DispatcharrBase b/docker/DispatcharrBase index 4360ced3..957c8573 100644 --- a/docker/DispatcharrBase +++ b/docker/DispatcharrBase @@ -32,11 +32,11 @@ RUN curl -fsSL https://packages.redis.io/gpg | gpg --dearmor -o /usr/share/keyri apt-get update && apt-get install -y redis-server && \ apt-get clean && rm -rf /var/lib/apt/lists/* -# --- Set up PostgreSQL 14.x --- +# --- Set up PostgreSQL 17.x --- RUN curl -fsSL https://www.postgresql.org/media/keys/ACCC4CF8.asc | gpg --dearmor -o /usr/share/keyrings/postgresql-keyring.gpg && \ echo "deb [signed-by=/usr/share/keyrings/postgresql-keyring.gpg] http://apt.postgresql.org/pub/repos/apt $(lsb_release -cs)-pgdg main" | \ tee /etc/apt/sources.list.d/pgdg.list && \ - apt-get update && apt-get install -y postgresql-14 postgresql-contrib-14 && \ + apt-get update && apt-get install -y postgresql-17 postgresql-contrib-17 && \ apt-get clean && rm -rf /var/lib/apt/lists/* # Create render group for hardware acceleration support with GID 109 From 108a99264333a2135cbd84b0cba101903b01c264 Mon Sep 17 00:00:00 2001 From: SergeantPanda Date: Thu, 31 Jul 2025 14:53:55 -0500 Subject: [PATCH 13/27] Detect mismatched Postgres version and automatically run pg_upgrade --- docker/entrypoint.sh | 9 ++--- docker/init/02-postgres.sh | 67 ++++++++++++++++++++++++++++++++++---- 2 files changed, 66 insertions(+), 10 deletions(-) diff --git a/docker/entrypoint.sh b/docker/entrypoint.sh index 412cf808..8d204a5b 100755 --- a/docker/entrypoint.sh +++ b/docker/entrypoint.sh @@ -33,7 +33,8 @@ export POSTGRES_USER=${POSTGRES_USER:-dispatch} export POSTGRES_PASSWORD=${POSTGRES_PASSWORD:-secret} export POSTGRES_HOST=${POSTGRES_HOST:-localhost} export POSTGRES_PORT=${POSTGRES_PORT:-5432} - +export PG_VERSION=$(ls /usr/lib/postgresql/ | sort -V | tail -n 1) +export PG_BINDIR="/usr/lib/postgresql/${PG_VERSION}/bin" export REDIS_HOST=${REDIS_HOST:-localhost} export REDIS_DB=${REDIS_DB:-0} export DISPATCHARR_PORT=${DISPATCHARR_PORT:-9191} @@ -107,13 +108,13 @@ echo "Starting init process..." # Start PostgreSQL echo "Starting Postgres..." -su - postgres -c "/usr/lib/postgresql/14/bin/pg_ctl -D ${POSTGRES_DIR} start -w -t 300 -o '-c port=${POSTGRES_PORT}'" +su - postgres -c "$PG_BINDIR/pg_ctl -D ${POSTGRES_DIR} start -w -t 300 -o '-c port=${POSTGRES_PORT}'" # Wait for PostgreSQL to be ready -until su - postgres -c "/usr/lib/postgresql/14/bin/pg_isready -h ${POSTGRES_HOST} -p ${POSTGRES_PORT}" >/dev/null 2>&1; do +until su - postgres -c "$PG_BINDIR/pg_isready -h ${POSTGRES_HOST} -p ${POSTGRES_PORT}" >/dev/null 2>&1; do echo_with_timestamp "Waiting for PostgreSQL to be ready..." sleep 1 done -postgres_pid=$(su - postgres -c "/usr/lib/postgresql/14/bin/pg_ctl -D ${POSTGRES_DIR} status" | sed -n 's/.*PID: \([0-9]\+\).*/\1/p') +postgres_pid=$(su - postgres -c "$PG_BINDIR/pg_ctl -D ${POSTGRES_DIR} status" | sed -n 's/.*PID: \([0-9]\+\).*/\1/p') echo "✅ Postgres started with PID $postgres_pid" pids+=("$postgres_pid") diff --git a/docker/init/02-postgres.sh b/docker/init/02-postgres.sh index 69a81dd4..aebce1a4 100644 --- a/docker/init/02-postgres.sh +++ b/docker/init/02-postgres.sh @@ -27,6 +27,61 @@ if [ -e "/data/postgresql.conf" ]; then echo "Migration completed successfully." fi +PG_VERSION_FILE="${POSTGRES_DIR}/PG_VERSION" + +# Detect current version from data directory, if present +if [ -f "$PG_VERSION_FILE" ]; then + CURRENT_VERSION=$(cat "$PG_VERSION_FILE") +else + CURRENT_VERSION="" +fi + +# Set binary paths for upgrade if needed +OLD_PG_VERSION="$CURRENT_VERSION" +OLD_BINDIR="/usr/lib/postgresql/${OLD_PG_VERSION}/bin" +NEW_BINDIR="/usr/lib/postgresql/${PG_VERSION}/bin" + +# Only run upgrade if current version is set and not the target +PG_INSTALLED_BY_SCRIPT=0 + +if [ -n "$CURRENT_VERSION" ] && [ "$CURRENT_VERSION" != "$PG_VERSION" ]; then + echo "Detected PostgreSQL data directory version $CURRENT_VERSION, upgrading to $PG_VERSION..." + if [ ! -d "$OLD_BINDIR" ]; then + echo "PostgreSQL binaries for version $CURRENT_VERSION not found. Installing..." + apt update && apt install -y postgresql-$CURRENT_VERSION postgresql-contrib-$CURRENT_VERSION + if [ $? -ne 0 ]; then + echo "Failed to install PostgreSQL version $CURRENT_VERSION. Exiting." + exit 1 + fi + PG_INSTALLED_BY_SCRIPT=1 + fi + + # Prepare new data directory + NEW_POSTGRES_DIR="${POSTGRES_DIR}_$PG_VERSION" + mkdir -p "$NEW_POSTGRES_DIR" + chown -R postgres:postgres "$NEW_POSTGRES_DIR" + chmod 700 "$NEW_POSTGRES_DIR" + + # Initialize new data directory + su - postgres -c "$NEW_BINDIR/initdb -D $NEW_POSTGRES_DIR" + + # Run pg_upgrade + su - postgres -c "$NEW_BINDIR/pg_upgrade -b $OLD_BINDIR -B $NEW_BINDIR -d $POSTGRES_DIR -D $NEW_POSTGRES_DIR" + + # Move old data directory for backup, move new into place + mv "$POSTGRES_DIR" "${POSTGRES_DIR}_backup_${CURRENT_VERSION}_$(date +%s)" + mv "$NEW_POSTGRES_DIR" "$POSTGRES_DIR" + + echo "Upgrade complete. Old data directory backed up." + + # Uninstall PostgreSQL if we installed it just for upgrade + if [ "$PG_INSTALLED_BY_SCRIPT" -eq 1 ]; then + echo "Uninstalling temporary PostgreSQL $CURRENT_VERSION packages..." + apt remove -y postgresql-$CURRENT_VERSION postgresql-contrib-$CURRENT_VERSION + apt autoremove -y + fi +fi + # Initialize PostgreSQL database if [ -z "$(ls -A $POSTGRES_DIR)" ]; then echo "Initializing PostgreSQL database..." @@ -35,21 +90,21 @@ if [ -z "$(ls -A $POSTGRES_DIR)" ]; then chmod 700 $POSTGRES_DIR # Initialize PostgreSQL - su - postgres -c "/usr/lib/postgresql/14/bin/initdb -D ${POSTGRES_DIR}" + su - postgres -c "$PG_BINDIR/initdb -D ${POSTGRES_DIR}" # Configure PostgreSQL echo "host all all 0.0.0.0/0 md5" >> "${POSTGRES_DIR}/pg_hba.conf" echo "listen_addresses='*'" >> "${POSTGRES_DIR}/postgresql.conf" # Start PostgreSQL echo "Starting Postgres..." - su - postgres -c "/usr/lib/postgresql/14/bin/pg_ctl -D ${POSTGRES_DIR} start -w -t 300 -o '-c port=${POSTGRES_PORT}'" + su - postgres -c "$PG_BINDIR/pg_ctl -D ${POSTGRES_DIR} start -w -t 300 -o '-c port=${POSTGRES_PORT}'" # Wait for PostgreSQL to be ready - until su - postgres -c "/usr/lib/postgresql/14/bin/pg_isready -h ${POSTGRES_HOST} -p ${POSTGRES_PORT}" >/dev/null 2>&1; do + until su - postgres -c "$PG_BINDIR/pg_isready -h ${POSTGRES_HOST} -p ${POSTGRES_PORT}" >/dev/null 2>&1; do echo "Waiting for PostgreSQL to be ready..." sleep 1 done - postgres_pid=$(su - postgres -c "/usr/lib/postgresql/14/bin/pg_ctl -D ${POSTGRES_DIR} status" | sed -n 's/.*PID: \([0-9]\+\).*/\1/p') + postgres_pid=$(su - postgres -c "$PG_BINDIR/pg_ctl -D ${POSTGRES_DIR} status" | sed -n 's/.*PID: \([0-9]\+\).*/\1/p') # Setup database if needed if ! su - postgres -c "psql -p ${POSTGRES_PORT} -tAc \"SELECT 1 FROM pg_database WHERE datname = '$POSTGRES_DB';\"" | grep -q 1; then @@ -69,8 +124,8 @@ END \$\$; EOF echo "Setting PostgreSQL user privileges..." - su postgres -c "/usr/lib/postgresql/14/bin/psql -p ${POSTGRES_PORT} -c \"ALTER DATABASE ${POSTGRES_DB} OWNER TO $POSTGRES_USER;\"" - su postgres -c "/usr/lib/postgresql/14/bin/psql -p ${POSTGRES_PORT} -c \"GRANT ALL PRIVILEGES ON DATABASE ${POSTGRES_DB} TO $POSTGRES_USER;\"" + su postgres -c "$PG_BINDIR/psql -p ${POSTGRES_PORT} -c \"ALTER DATABASE ${POSTGRES_DB} OWNER TO $POSTGRES_USER;\"" + su postgres -c "$PG_BINDIR/psql -p ${POSTGRES_PORT} -c \"GRANT ALL PRIVILEGES ON DATABASE ${POSTGRES_DB} TO $POSTGRES_USER;\"" # Finished setting up PosgresSQL database echo "PostgreSQL database setup complete." fi From 406ac37fb97c9803e8cc3778425d11c75db53e79 Mon Sep 17 00:00:00 2001 From: SergeantPanda Date: Thu, 31 Jul 2025 15:01:28 -0500 Subject: [PATCH 14/27] Delete temp folder if it exists during upgrade. --- docker/init/02-postgres.sh | 21 +++++++++++++-------- 1 file changed, 13 insertions(+), 8 deletions(-) diff --git a/docker/init/02-postgres.sh b/docker/init/02-postgres.sh index aebce1a4..4deb921d 100644 --- a/docker/init/02-postgres.sh +++ b/docker/init/02-postgres.sh @@ -36,16 +36,13 @@ else CURRENT_VERSION="" fi -# Set binary paths for upgrade if needed -OLD_PG_VERSION="$CURRENT_VERSION" -OLD_BINDIR="/usr/lib/postgresql/${OLD_PG_VERSION}/bin" -NEW_BINDIR="/usr/lib/postgresql/${PG_VERSION}/bin" - # Only run upgrade if current version is set and not the target -PG_INSTALLED_BY_SCRIPT=0 - if [ -n "$CURRENT_VERSION" ] && [ "$CURRENT_VERSION" != "$PG_VERSION" ]; then echo "Detected PostgreSQL data directory version $CURRENT_VERSION, upgrading to $PG_VERSION..." + # Set binary paths for upgrade if needed + OLD_BINDIR="/usr/lib/postgresql/${CURRENT_VERSION}/bin" + NEW_BINDIR="/usr/lib/postgresql/${PG_VERSION}/bin" + PG_INSTALLED_BY_SCRIPT=0 if [ ! -d "$OLD_BINDIR" ]; then echo "PostgreSQL binaries for version $CURRENT_VERSION not found. Installing..." apt update && apt install -y postgresql-$CURRENT_VERSION postgresql-contrib-$CURRENT_VERSION @@ -58,13 +55,21 @@ if [ -n "$CURRENT_VERSION" ] && [ "$CURRENT_VERSION" != "$PG_VERSION" ]; then # Prepare new data directory NEW_POSTGRES_DIR="${POSTGRES_DIR}_$PG_VERSION" + + # Remove new data directory if it already exists (from a failed/partial upgrade) + if [ -d "$NEW_POSTGRES_DIR" ]; then + echo "Warning: $NEW_POSTGRES_DIR already exists. Removing it to avoid upgrade issues." + rm -rf "$NEW_POSTGRES_DIR" + fi + mkdir -p "$NEW_POSTGRES_DIR" chown -R postgres:postgres "$NEW_POSTGRES_DIR" chmod 700 "$NEW_POSTGRES_DIR" # Initialize new data directory + echo "Initializing new PostgreSQL data directory at $NEW_POSTGRES_DIR..." su - postgres -c "$NEW_BINDIR/initdb -D $NEW_POSTGRES_DIR" - + echo "Running pg_upgrade from $OLD_BINDIR to $NEW_BINDIR..." # Run pg_upgrade su - postgres -c "$NEW_BINDIR/pg_upgrade -b $OLD_BINDIR -B $NEW_BINDIR -d $POSTGRES_DIR -D $NEW_POSTGRES_DIR" From 59b75c18fce916fc97780ef60708751caa18c13b Mon Sep 17 00:00:00 2001 From: SergeantPanda Date: Thu, 31 Jul 2025 15:54:24 -0500 Subject: [PATCH 15/27] Add ability to preview streams under a channel. --- .../components/tables/ChannelTableStreams.jsx | 66 +++++++++++++------ 1 file changed, 45 insertions(+), 21 deletions(-) diff --git a/frontend/src/components/tables/ChannelTableStreams.jsx b/frontend/src/components/tables/ChannelTableStreams.jsx index 11fa9600..991b7074 100644 --- a/frontend/src/components/tables/ChannelTableStreams.jsx +++ b/frontend/src/components/tables/ChannelTableStreams.jsx @@ -1,7 +1,7 @@ import React, { useMemo, useState, useEffect } from 'react'; import API from '../../api'; import { copyToClipboard } from '../../utils'; -import { GripHorizontal, SquareMinus, ChevronDown, ChevronRight } from 'lucide-react'; +import { GripHorizontal, SquareMinus, ChevronDown, ChevronRight, Eye } from 'lucide-react'; import { Box, ActionIcon, @@ -14,6 +14,7 @@ import { Tooltip, Collapse, Button, + } from '@mantine/core'; import { notifications } from '@mantine/notifications'; import { @@ -24,6 +25,8 @@ import { import './table.css'; import useChannelsTableStore from '../../store/channelsTable'; import usePlaylistsStore from '../../store/playlists'; +import useVideoStore from '../../store/useVideoStore'; +import useSettingsStore from '../../store/settings'; import { DndContext, KeyboardSensor, @@ -130,6 +133,15 @@ const ChannelStreams = ({ channel, isExpanded }) => { ); const playlists = usePlaylistsStore((s) => s.playlists); const authUser = useAuthStore((s) => s.user); + const showVideo = useVideoStore((s) => s.showVideo); + const env_mode = useSettingsStore((s) => s.environment.env_mode); + function handleWatchStream(streamHash) { + let vidUrl = `/proxy/ts/stream/${streamHash}`; + if (env_mode === 'dev') { + vidUrl = `${window.location.protocol}//${window.location.hostname}:5656${vidUrl}`; + } + showVideo(vidUrl); + } const [data, setData] = useState(channelStreams || []); @@ -314,25 +326,38 @@ const ChannelStreams = ({ channel, isExpanded }) => { )} {stream.url && ( - - { - e.stopPropagation(); - const success = await copyToClipboard(stream.url); - notifications.show({ - title: success ? 'URL Copied' : 'Copy Failed', - message: success ? 'Stream URL copied to clipboard' : 'Failed to copy URL to clipboard', - color: success ? 'green' : 'red', - }); - }} - > - URL - - + <> + + { + e.stopPropagation(); + const success = await copyToClipboard(stream.url); + notifications.show({ + title: success ? 'URL Copied' : 'Copy Failed', + message: success ? 'Stream URL copied to clipboard' : 'Failed to copy URL to clipboard', + color: success ? 'green' : 'red', + }); + }} + > + URL + + + + handleWatchStream(stream.stream_hash || stream.id)} + style={{ marginLeft: 2 }} + > + + + + )} @@ -563,5 +588,4 @@ const ChannelStreams = ({ channel, isExpanded }) => { ); }; - export default ChannelStreams; From 953db7947644ebf3789d71bd8fdc9f0e83decabb Mon Sep 17 00:00:00 2001 From: SergeantPanda Date: Thu, 31 Jul 2025 21:26:59 -0500 Subject: [PATCH 16/27] Display stream logo and name in channel card when previewing streams. --- frontend/src/pages/Stats.jsx | 26 ++++++++++++++++++++++---- 1 file changed, 22 insertions(+), 4 deletions(-) diff --git a/frontend/src/pages/Stats.jsx b/frontend/src/pages/Stats.jsx index c3709c17..19520a21 100644 --- a/frontend/src/pages/Stats.jsx +++ b/frontend/src/pages/Stats.jsx @@ -94,6 +94,7 @@ const ChannelCard = ({ const [activeStreamId, setActiveStreamId] = useState(null); const [currentM3UProfile, setCurrentM3UProfile] = useState(null); // Add state for current M3U profile const [data, setData] = useState([]); + const [previewedStream, setPreviewedStream] = useState(null); // Get M3U account data from the playlists store const m3uAccounts = usePlaylistsStore((s) => s.playlists); @@ -425,12 +426,29 @@ const ChannelCard = ({ // Get logo URL from the logos object if available const logoUrl = - channel.logo_id && logos && logos[channel.logo_id] + (channel.logo_id && logos && logos[channel.logo_id] ? logos[channel.logo_id].cache_url - : null; + : null) || + (previewedStream && previewedStream.logo_url) || + null; - // Ensure these values exist to prevent errors - const channelName = channel.name || 'Unnamed Channel'; + useEffect(() => { + let isMounted = true; + // Only fetch if we have a stream_id and NO channel.name + if (!channel.name && channel.stream_id) { + API.getStreamsByIds([channel.stream_id]).then((streams) => { + if (isMounted && streams && streams.length > 0) { + setPreviewedStream(streams[0]); + } + }); + } + return () => { isMounted = false; }; + }, [channel.name, channel.stream_id]); + + const channelName = + channel.name || + previewedStream?.name || + 'Unnamed Channel'; const uptime = channel.uptime || 0; const bitrates = channel.bitrates || []; const totalBytes = channel.total_bytes || 0; From 7b5a617bf829f91f26f7dafa1dbc0e1dfea7fa6a Mon Sep 17 00:00:00 2001 From: SergeantPanda Date: Fri, 1 Aug 2025 11:28:51 -0500 Subject: [PATCH 17/27] Use custom validator for urls fields to allow for non fqdn hostnames. Fixes #63 --- apps/channels/serializers.py | 11 +++++++++-- apps/epg/serializers.py | 7 +++++++ apps/m3u/serializers.py | 7 +++++++ core/utils.py | 32 ++++++++++++++++++++++++++++++++ 4 files changed, 55 insertions(+), 2 deletions(-) diff --git a/apps/channels/serializers.py b/apps/channels/serializers.py index 32fd4a74..7c5ddd54 100644 --- a/apps/channels/serializers.py +++ b/apps/channels/serializers.py @@ -16,6 +16,7 @@ from apps.epg.models import EPGData from django.urls import reverse from rest_framework import serializers from django.utils import timezone +from core.utils import validate_flexible_url class LogoSerializer(serializers.ModelSerializer): @@ -32,10 +33,10 @@ class LogoSerializer(serializers.ModelSerializer): """Validate that the URL is unique for creation or update""" if self.instance and self.instance.url == value: return value - + if Logo.objects.filter(url=value).exists(): raise serializers.ValidationError("A logo with this URL already exists.") - + return value def create(self, validated_data): @@ -79,6 +80,12 @@ class LogoSerializer(serializers.ModelSerializer): # Stream # class StreamSerializer(serializers.ModelSerializer): + url = serializers.CharField( + required=False, + allow_blank=True, + allow_null=True, + validators=[validate_flexible_url] + ) stream_profile_id = serializers.PrimaryKeyRelatedField( queryset=StreamProfile.objects.all(), source="stream_profile", diff --git a/apps/epg/serializers.py b/apps/epg/serializers.py index 09390237..2f97cebf 100644 --- a/apps/epg/serializers.py +++ b/apps/epg/serializers.py @@ -1,3 +1,4 @@ +from core.utils import validate_flexible_url from rest_framework import serializers from .models import EPGSource, EPGData, ProgramData from apps.channels.models import Channel @@ -5,6 +6,12 @@ from apps.channels.models import Channel class EPGSourceSerializer(serializers.ModelSerializer): epg_data_ids = serializers.SerializerMethodField() read_only_fields = ['created_at', 'updated_at'] + url = serializers.CharField( + required=False, + allow_blank=True, + allow_null=True, + validators=[validate_flexible_url] + ) class Meta: model = EPGSource diff --git a/apps/m3u/serializers.py b/apps/m3u/serializers.py index 7394f00b..a86227aa 100644 --- a/apps/m3u/serializers.py +++ b/apps/m3u/serializers.py @@ -1,3 +1,4 @@ +from core.utils import validate_flexible_url from rest_framework import serializers from rest_framework.response import Response from .models import M3UAccount, M3UFilter, ServerGroup, M3UAccountProfile @@ -76,6 +77,12 @@ class M3UAccountSerializer(serializers.ModelSerializer): channel_groups = ChannelGroupM3UAccountSerializer( source="channel_group", many=True, required=False ) + server_url = serializers.CharField( + required=False, + allow_blank=True, + allow_null=True, + validators=[validate_flexible_url], + ) class Meta: model = M3UAccount diff --git a/core/utils.py b/core/utils.py index 932af979..36ac5fef 100644 --- a/core/utils.py +++ b/core/utils.py @@ -9,6 +9,8 @@ from redis.exceptions import ConnectionError, TimeoutError from django.core.cache import cache from asgiref.sync import async_to_sync from channels.layers import get_channel_layer +from django.core.validators import URLValidator +from django.core.exceptions import ValidationError import gc logger = logging.getLogger(__name__) @@ -354,3 +356,33 @@ def is_protected_path(file_path): return True return False + +def validate_flexible_url(value): + """ + Custom URL validator that accepts URLs with hostnames that aren't FQDNs. + This allows URLs like "http://hostname/" which + Django's standard URLValidator rejects. + """ + if not value: + return # Allow empty values since the field is nullable + + # Create a standard Django URL validator + url_validator = URLValidator() + + try: + # First try the standard validation + url_validator(value) + except ValidationError as e: + # If standard validation fails, check if it's a non-FQDN hostname + import re + + # More flexible pattern for non-FQDN hostnames with paths + # Matches: http://hostname, http://hostname/, http://hostname:port/path/to/file.xml + non_fqdn_pattern = r'^https?://[a-zA-Z0-9]([a-zA-Z0-9\-]{0,61}[a-zA-Z0-9])?(\:[0-9]+)?(/[^\s]*)?$' + non_fqdn_match = re.match(non_fqdn_pattern, value) + + if non_fqdn_match: + return # Accept non-FQDN hostnames + + # If it doesn't match our flexible patterns, raise the original error + raise ValidationError("Enter a valid URL.") From ead76fe6611476d5cdd163b51e6d0910c4c7d3ce Mon Sep 17 00:00:00 2001 From: dekzter Date: Fri, 1 Aug 2025 15:02:43 -0400 Subject: [PATCH 18/27] first run at m3u filtering --- apps/m3u/api_urls.py | 36 +- apps/m3u/api_views.py | 19 +- .../0013_alter_m3ufilter_filter_type.py | 18 + ...alter_m3ufilter_options_m3ufilter_order.py | 22 + apps/m3u/models.py | 8 +- apps/m3u/serializers.py | 10 +- apps/m3u/tasks.py | 1016 ++++++++++++----- frontend/src/api.js | 61 +- frontend/src/components/forms/M3U.jsx | 44 +- frontend/src/components/forms/M3UFilter.jsx | 126 ++ frontend/src/components/forms/M3UFilters.jsx | 327 ++++++ frontend/src/constants.js | 27 +- frontend/src/store/playlists.jsx | 24 +- 13 files changed, 1400 insertions(+), 338 deletions(-) create mode 100644 apps/m3u/migrations/0013_alter_m3ufilter_filter_type.py create mode 100644 apps/m3u/migrations/0014_alter_m3ufilter_options_m3ufilter_order.py create mode 100644 frontend/src/components/forms/M3UFilter.jsx create mode 100644 frontend/src/components/forms/M3UFilters.jsx 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 && ( <> + - - - - )} +