Mirrors/Dispatcharr
1
0
Fork 1
mirror of https://github.com/Dispatcharr/Dispatcharr.git synced 2026-07-20 16:51:10 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 7

Merge pull request #924 from Dispatcharr/hash-refactor

Browse source 
Hashing refactor
This commit is contained in:
SergeantPanda 2026-01-31 18:26:33 -06:00 • committed by GitHub
commit e20858b7b5
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
9 changed files with 503 additions and 115 deletions
Download patch file Download diff file Expand all files Collapse all files

3
CHANGELOG.md
View file

@ -9,6 +9,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
### Added
- Frontend now automatically refreshes streams and channels after a stream rehash completes, ensuring the UI is always up-to-date following backend merge operations.
- Frontend Unit Tests: Added comprehensive unit tests for React hooks and Zustand stores, including:
- `useLocalStorage` hook tests with localStorage mocking and error handling
- `useSmartLogos` hook tests for logo loading and management
@ -24,6 +25,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
### Changed
- Stream Identity Stability: Added `stream_id` (provider stream identifier) and `stream_chno` (provider channel number) fields to Stream model. For XC accounts, the stream hash now uses the stable `stream_id` instead of the URL when hashing, ensuring XC streams maintain their identity and channel associations even when account credentials or server URLs change. Supports both XC `num` and M3U `tvg-chno`/`channel-number` attributes.
- Swagger/OpenAPI Migration: Migrated from `drf-yasg` (OpenAPI 2.0) to `drf-spectacular` (OpenAPI 3.0) for API documentation. This provides:
- Native Bearer token authentication support in Swagger UI - users can now enter just the JWT token and the "Bearer " prefix is automatically added
- Modern OpenAPI 3.0 specification compliance
@ -33,6 +35,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
### Fixed
- Stream rehash/merge logic now guarantees unique stream_hash and always preserves the stream with the best channel ordering and relationships. This prevents duplicate key errors and ensures the correct stream is retained when merging. (Fixes #892)
- Admin URL Conflict with XC Streams: Updated nginx configuration to only redirect exact `/admin` and `/admin/` paths to login in production, preventing interference with stream URLs that use "admin" as a username (e.g., `/admin/password/stream_id` now properly routes to stream handling instead of being redirected).
- EPG Channel ID XML Escaping: Fixed XML parsing errors in EPG output when channel IDs contain special characters (&, <, >, \") by properly escaping them in XML attributes. (Fixes #765) - Thanks [@CodeBormen](https://github.com/CodeBormen)
- Fixed NumPy baseline detection in Docker entrypoint. Now properly detects when NumPy crashes on import due to CPU baseline incompatibility and installs legacy NumPy version. Previously, if NumPy failed to import, the script would skip legacy installation assuming it was already compatible.

18
apps/channels/api_views.py
View file

@ -907,18 +907,13 @@ class ChannelViewSet(viewsets.ModelViewSet):
if name is None:
name = stream.name
# Check if client provided a channel_number; if not, auto-assign one.
stream_custom_props = stream.custom_properties or {}
# Check if client provided a channel_number; if not, use stream_chno or auto-assign
channel_number = request.data.get("channel_number")
if channel_number is None:
# Channel number not provided by client, check stream properties or auto-assign
if "tvg-chno" in stream_custom_props:
channel_number = float(stream_custom_props["tvg-chno"])
elif "channel-number" in stream_custom_props:
channel_number = float(stream_custom_props["channel-number"])
elif "num" in stream_custom_props:
channel_number = float(stream_custom_props["num"])
# Channel number not provided by client, check stream's channel number or auto-assign
if stream.stream_chno is not None:
channel_number = stream.stream_chno
elif channel_number == 0:
# Special case: 0 means ignore provider numbers and auto-assign
channel_number = None
@ -939,9 +934,8 @@ class ChannelViewSet(viewsets.ModelViewSet):
if Channel.objects.filter(channel_number=channel_number).exists():
channel_number = Channel.get_next_available_channel_number(channel_number)
# Get the tvc_guide_stationid from custom properties if it exists
tvc_guide_stationid = None
if "tvc-guide-stationid" in stream_custom_props:
tvc_guide_stationid = stream_custom_props["tvc-guide-stationid"]
stream_custom_props = stream.custom_properties or {}
tvc_guide_stationid = stream_custom_props.get("tvc-guide-stationid")
channel_data = {
"channel_number": channel_number,

205
apps/channels/migrations/0033_stream_id_stream_chno.py Normal file
View file

@ -0,0 +1,205 @@
# Generated by Django - Add stream_id and channel_number fields with data migration
from django.db import migrations, models
import hashlib
import json
import logging
logger = logging.getLogger(__name__)
def populate_fields_and_rehash(apps, schema_editor):
"""
Populate stream_id and stream_chno from custom_properties for XC account streams,
populate stream_chno from tvg-chno for standard M3U accounts,
then rehash XC streams using stable hash keys.
"""
Stream = apps.get_model('dispatcharr_channels', 'Stream')
M3UAccount = apps.get_model('m3u', 'M3UAccount')
CoreSettings = apps.get_model('core', 'CoreSettings')
# Get hash keys from settings
try:
stream_settings = CoreSettings.objects.get(key='stream_settings')
hash_key_str = stream_settings.value.get('m3u_hash_key', '') if stream_settings.value else ''
keys = [k.strip() for k in hash_key_str.split(',') if k.strip()] if hash_key_str else []
except CoreSettings.DoesNotExist:
keys = []
logger.info(f"Using hash keys: {keys}")
# Get XC account IDs
xc_account_ids = set(
M3UAccount.objects.filter(account_type='XC').values_list('id', flat=True)
)
logger.info(f"Found {len(xc_account_ids)} XC accounts")
# Track hash collisions for XC streams
hash_map = {} # new_hash -> stream_id
duplicates_to_delete = []
# Process all streams in batches
batch_size = 1000
processed = 0
updated = 0
total_count = Stream.objects.count()
logger.info(f"Processing {total_count} total streams")
streams_to_update = []
for stream in Stream.objects.select_related('channel_group', 'm3u_account').iterator(chunk_size=batch_size):
processed += 1
needs_update = False
custom_props = stream.custom_properties or {}
is_xc = stream.m3u_account_id in xc_account_ids if stream.m3u_account_id else False
# Extract stream_id (XC accounts only)
if is_xc and isinstance(custom_props, dict):
provider_stream_id = custom_props.get('stream_id')
if provider_stream_id:
try:
stream.stream_id = int(provider_stream_id)
needs_update = True
except (ValueError, TypeError):
pass
# Extract stream_chno
channel_num = None
if isinstance(custom_props, dict):
if is_xc:
# XC accounts use 'num'
channel_num = custom_props.get('num')
else:
# Standard M3U accounts use 'tvg-chno' or 'channel-number' (case insensitive check)
for key in ['tvg-chno', 'TVG-CHNO', 'tvg-Chno', 'Tvg-Chno', 'channel-number', 'Channel-Number', 'CHANNEL-NUMBER']:
if key in custom_props:
channel_num = custom_props.get(key)
break
if channel_num is not None:
try:
stream.stream_chno = float(channel_num)
needs_update = True
except (ValueError, TypeError):
pass
# Rehash XC streams only when 'url' is in hash keys (otherwise hash wouldn't change)
if is_xc and stream.stream_id and keys and 'url' in keys:
# For XC accounts, use stream_id instead of url when 'url' is in the hash keys
# This ensures credential/URL changes don't break stream identity
effective_url = stream.stream_id
# Get group name
group_name = stream.channel_group.name if stream.channel_group else None
# Build hash parts
stream_parts = {
"name": stream.name,
"url": effective_url,
"tvg_id": stream.tvg_id,
"m3u_id": stream.m3u_account_id,
"group": group_name
}
hash_parts = {key: stream_parts[key] for key in keys if key in stream_parts}
# When using stream_id instead of URL, we MUST include m3u_id to prevent
# collisions across different XC accounts (stream_id is only unique per account)
if 'm3u_id' not in hash_parts:
hash_parts['m3u_id'] = stream.m3u_account_id
# Generate hash
serialized_obj = json.dumps(hash_parts, sort_keys=True)
new_hash = hashlib.sha256(serialized_obj.encode()).hexdigest()
# Check for collisions
if new_hash in hash_map:
# Duplicate - mark for deletion (keep the first one)
duplicates_to_delete.append(stream.id)
continue
hash_map[new_hash] = stream.id
stream.stream_hash = new_hash
needs_update = True
if needs_update:
streams_to_update.append(stream)
updated += 1
# Bulk update in batches
if len(streams_to_update) >= batch_size:
Stream.objects.bulk_update(
streams_to_update,
['stream_id', 'stream_chno', 'stream_hash'],
batch_size=500
)
logger.info(f"Updated batch: {processed}/{total_count} streams processed")
streams_to_update = []
# Final batch
if streams_to_update:
Stream.objects.bulk_update(
streams_to_update,
['stream_id', 'stream_chno', 'stream_hash'],
batch_size=500
)
# Delete duplicates if any
if duplicates_to_delete:
logger.warning(f"Deleting {len(duplicates_to_delete)} duplicate streams due to hash collisions")
Stream.objects.filter(id__in=duplicates_to_delete).delete()
logger.info(f"Migration complete: {updated} streams updated, {len(duplicates_to_delete)} duplicates removed")
def reverse_migration(apps, schema_editor):
"""
Reverse migration - clear fields but don't attempt to reverse hash changes.
"""
Stream = apps.get_model('dispatcharr_channels', 'Stream')
Stream.objects.all().update(stream_id=None, stream_chno=None)
logger.info("Cleared stream_id and stream_chno fields. Note: stream hashes were not reverted.")
class Migration(migrations.Migration):
dependencies = [
('dispatcharr_channels', '0032_channel_is_adult_stream_is_adult'),
('m3u', '0018_add_profile_custom_properties'),
('core', '0020_change_coresettings_value_to_jsonfield'),
]
operations = [
# Schema changes - add fields WITHOUT indexes first
migrations.AddField(
model_name='stream',
name='stream_id',
field=models.IntegerField(
blank=True,
help_text='Provider stream ID (e.g., XC stream_id) for stable identity across credential changes',
null=True,
),
),
migrations.AddField(
model_name='stream',
name='stream_chno',
field=models.FloatField(
blank=True,
help_text='Provider channel number (XC num or M3U tvg-chno) for ordering - supports decimals like 2.1',
null=True,
),
),
# Data migration (may delete duplicates, which would conflict with pending index creation)
migrations.RunPython(populate_fields_and_rehash, reverse_migration),
# Add indexes AFTER data migration completes
migrations.AddIndex(
model_name='stream',
index=models.Index(fields=['stream_id'], name='dispatcharr_stream_id_idx'),
),
migrations.AddIndex(
model_name='stream',
index=models.Index(fields=['stream_chno'], name='dispatcharr_stream_chno_idx'),
),
]

30
apps/channels/models.py
View file

@ -106,6 +106,19 @@ class Stream(models.Model):
)
custom_properties = models.JSONField(default=dict, blank=True, null=True)
stream_id = models.IntegerField(
null=True,
blank=True,
db_index=True,
help_text="Provider stream ID (e.g., XC stream_id) for stable identity across credential changes"
)
stream_chno = models.FloatField(
null=True,
blank=True,
db_index=True,
help_text="Provider channel number (XC num or M3U tvg-chno) for ordering - supports decimals like 2.1"
)
# Stream statistics fields
stream_stats = models.JSONField(
null=True,
@ -129,14 +142,27 @@ class Stream(models.Model):
return self.name or self.url or f"Stream ID {self.id}"
@classmethod
def generate_hash_key(cls, name, url, tvg_id, keys=None, m3u_id=None, group=None):
def generate_hash_key(cls, name, url, tvg_id, keys=None, m3u_id=None, group=None,
account_type=None, stream_id=None):
if keys is None:
keys = CoreSettings.get_m3u_hash_key().split(",")
stream_parts = {"name": name, "url": url, "tvg_id": tvg_id, "m3u_id": m3u_id, "group": group}
# For XC accounts, use stream_id instead of url when 'url' is in the hash keys
# This ensures credential/URL changes don't break stream identity
effective_url = url
use_stream_id = account_type == 'XC' and stream_id and 'url' in keys
if use_stream_id:
effective_url = stream_id
stream_parts = {"name": name, "url": effective_url, "tvg_id": tvg_id, "m3u_id": m3u_id, "group": group}
hash_parts = {key: stream_parts[key] for key in keys if key in stream_parts}
# When using stream_id instead of URL, we MUST include m3u_id to prevent
# collisions across different XC accounts (stream_id is only unique per account)
if use_stream_id and 'm3u_id' not in hash_parts:
hash_parts['m3u_id'] = m3u_id
# Serialize and hash the dictionary
serialized_obj = json.dumps(
hash_parts, sort_keys=True

4
apps/channels/serializers.py
View file

@ -104,7 +104,7 @@ class StreamSerializer(serializers.ModelSerializer):
allow_null=True,
required=False,
)
read_only_fields = ["is_custom", "m3u_account", "stream_hash"]
read_only_fields = ["is_custom", "m3u_account", "stream_hash", "stream_id", "stream_chno"]
class Meta:
model = Stream
@ -127,6 +127,8 @@ class StreamSerializer(serializers.ModelSerializer):
"stream_hash",
"stream_stats",
"stream_stats_updated_at",
"stream_id",
"stream_chno",
]
def get_fields(self):

10
apps/channels/tasks.py
View file

@ -2629,13 +2629,9 @@ def bulk_create_channels_from_streams(self, stream_ids, channel_profile_ids=None
channel_number = None
if starting_channel_number is None:
# Mode 1: Use provider numbers when available
if "tvg-chno" in stream_custom_props:
channel_number = float(stream_custom_props["tvg-chno"])
elif "channel-number" in stream_custom_props:
channel_number = float(stream_custom_props["channel-number"])
elif "num" in stream_custom_props:
channel_number = float(stream_custom_props["num"])
# Mode 1: Use provider numbers when available (from stream_chno field)
if stream.stream_chno is not None:
channel_number = stream.stream_chno
# For modes 2 and 3 (starting_channel_number == 0 or specific number),
# ignore provider numbers and use sequential assignment

79
apps/m3u/tasks.py
View file

@ -741,6 +741,7 @@ def collect_xc_streams(account_id, enabled_groups):
"group-title": group_info["name"],
# Preserve all XC stream properties as custom attributes
"stream_id": str(stream.get("stream_id", "")),
"num": stream.get("num"),
"category_id": category_id,
"stream_type": stream.get("stream_type", ""),
"added": stream.get("added", ""),
@ -749,7 +750,7 @@ def collect_xc_streams(account_id, enabled_groups):
# Include any other properties that might be present
**{k: str(v) for k, v in stream.items() if k not in [
"name", "stream_id", "epg_channel_id", "stream_icon",
"category_id", "stream_type", "added", "is_adult", "custom_sid"
"category_id", "stream_type", "added", "is_adult", "custom_sid", "num"
] and v is not None}
}
}
@ -817,13 +818,28 @@ def process_xc_category_direct(account_id, batch, groups, hash_keys):
for stream in streams:
name = stream["name"]
raw_stream_id = stream.get("stream_id", "")
provider_stream_id = None
if raw_stream_id:
try:
provider_stream_id = int(raw_stream_id)
except (ValueError, TypeError):
pass
url = xc_client.get_stream_url(stream["stream_id"])
tvg_id = stream.get("epg_channel_id", "")
tvg_logo = stream.get("stream_icon", "")
group_title = group_name
stream_chno = stream.get("num")
# Convert stream_chno to float if valid, otherwise None
if stream_chno is not None:
try:
stream_chno = float(stream_chno)
except (ValueError, TypeError):
stream_chno = None
stream_hash = Stream.generate_hash_key(
name, url, tvg_id, hash_keys, m3u_id=account_id, group=group_title
name, url, tvg_id, hash_keys, m3u_id=account_id, group=group_title,
account_type='XC', stream_id=provider_stream_id
)
stream_props = {
"name": name,
@ -836,6 +852,8 @@ def process_xc_category_direct(account_id, batch, groups, hash_keys):
"custom_properties": stream,
"is_adult": int(stream.get("is_adult", 0)) == 1,
"is_stale": False,
"stream_id": provider_stream_id,
"stream_chno": stream_chno,
}
if stream_hash not in stream_hashes:
@ -850,7 +868,7 @@ def process_xc_category_direct(account_id, batch, groups, hash_keys):
existing_streams = {
s.stream_hash: s
for s in Stream.objects.filter(stream_hash__in=stream_hashes.keys()).select_related('m3u_account').only(
'id', 'stream_hash', 'name', 'url', 'logo_url', 'tvg_id', 'custom_properties', 'last_seen', 'updated_at', 'm3u_account'
'id', 'stream_hash', 'name', 'url', 'logo_url', 'tvg_id', 'custom_properties', 'last_seen', 'updated_at', 'm3u_account', 'stream_id', 'stream_chno'
)
}
@ -864,7 +882,9 @@ def process_xc_category_direct(account_id, batch, groups, hash_keys):
obj.logo_url != stream_props["logo_url"] or
obj.tvg_id != stream_props["tvg_id"] or
obj.custom_properties != stream_props["custom_properties"] or
obj.is_adult != stream_props["is_adult"]
obj.is_adult != stream_props["is_adult"] or
obj.stream_id != stream_props["stream_id"] or
obj.stream_chno != stream_props["stream_chno"]
)
if changed:
@ -900,7 +920,7 @@ def process_xc_category_direct(account_id, batch, groups, hash_keys):
# Simplified bulk update for better performance
Stream.objects.bulk_update(
streams_to_update,
['name', 'url', 'logo_url', 'tvg_id', 'custom_properties', 'is_adult', 'last_seen', 'updated_at', 'is_stale'],
['name', 'url', 'logo_url', 'tvg_id', 'custom_properties', 'is_adult', 'last_seen', 'updated_at', 'is_stale', 'stream_id', 'stream_chno'],
batch_size=150 # Smaller batch size for XC processing
)
@ -1003,7 +1023,42 @@ def process_m3u_batch_direct(account_id, batch, groups, hash_keys):
)
continue
stream_hash = Stream.generate_hash_key(name, url, tvg_id, hash_keys, m3u_id=account_id, group=group_title)
# Determine provider-specific fields first
provider_stream_id = None
channel_num = None
account_type_for_hash = None
if account.account_type == M3UAccount.Types.XC:
account_type_for_hash = 'XC'
raw_stream_id = stream_info["attributes"].get("stream_id", "")
if raw_stream_id:
try:
provider_stream_id = int(raw_stream_id)
except (ValueError, TypeError):
pass
raw_num = stream_info["attributes"].get("num")
if raw_num is not None:
try:
channel_num = float(raw_num)
except (ValueError, TypeError):
pass
else:
# For standard M3U accounts, check for tvg-chno or channel-number
tvg_chno = get_case_insensitive_attr(stream_info["attributes"], "tvg-chno", None)
if tvg_chno is None:
tvg_chno = get_case_insensitive_attr(stream_info["attributes"], "channel-number", None)
if tvg_chno is not None:
try:
channel_num = float(tvg_chno)
except (ValueError, TypeError):
pass
# Generate hash once with all parameters
stream_hash = Stream.generate_hash_key(
name, url, tvg_id, hash_keys, m3u_id=account_id, group=group_title,
account_type=account_type_for_hash, stream_id=provider_stream_id
)
stream_props = {
"name": name,
"url": url,
@ -1015,6 +1070,8 @@ def process_m3u_batch_direct(account_id, batch, groups, hash_keys):
"custom_properties": stream_info["attributes"],
"is_adult": int(stream_info["attributes"].get("is_adult", 0)) == 1,
"is_stale": False,
"stream_id": provider_stream_id,
"stream_chno": channel_num,
}
if stream_hash not in stream_hashes:
@ -1026,7 +1083,7 @@ def process_m3u_batch_direct(account_id, batch, groups, hash_keys):
existing_streams = {
s.stream_hash: s
for s in Stream.objects.filter(stream_hash__in=stream_hashes.keys()).select_related('m3u_account').only(
'id', 'stream_hash', 'name', 'url', 'logo_url', 'tvg_id', 'custom_properties', 'last_seen', 'updated_at', 'm3u_account'
'id', 'stream_hash', 'name', 'url', 'logo_url', 'tvg_id', 'custom_properties', 'last_seen', 'updated_at', 'm3u_account', 'stream_id', 'stream_chno'
)
}
@ -1040,7 +1097,9 @@ def process_m3u_batch_direct(account_id, batch, groups, hash_keys):
obj.logo_url != stream_props["logo_url"] or
obj.tvg_id != stream_props["tvg_id"] or
obj.custom_properties != stream_props["custom_properties"] or
obj.is_adult != stream_props["is_adult"]
obj.is_adult != stream_props["is_adult"] or
obj.stream_id != stream_props["stream_id"] or
obj.stream_chno != stream_props["stream_chno"]
)
# Always update last_seen
@ -1054,6 +1113,8 @@ def process_m3u_batch_direct(account_id, batch, groups, hash_keys):
obj.tvg_id = stream_props["tvg_id"]
obj.custom_properties = stream_props["custom_properties"]
obj.is_adult = stream_props["is_adult"]
obj.stream_id = stream_props["stream_id"]
obj.stream_chno = stream_props["stream_chno"]
obj.updated_at = timezone.now()
# Always mark as not stale since we saw it in this refresh
@ -1076,7 +1137,7 @@ def process_m3u_batch_direct(account_id, batch, groups, hash_keys):
# Update all streams in a single bulk operation
Stream.objects.bulk_update(
streams_to_update,
['name', 'url', 'logo_url', 'tvg_id', 'custom_properties', 'is_adult', 'last_seen', 'updated_at', 'is_stale'],
['name', 'url', 'logo_url', 'tvg_id', 'custom_properties', 'is_adult', 'last_seen', 'updated_at', 'is_stale', 'stream_id', 'stream_chno'],
batch_size=200
)
except Exception as e:

258
core/tasks.py
View file

@ -480,15 +480,18 @@ def rehash_streams(keys):
try:
batch_size = 1000
queryset = Stream.objects.all()
# Track statistics
total_processed = 0
duplicates_merged = 0
# hash_keys maps new_hash -> stream_id for streams we've already processed
hash_keys = {}
# Track IDs of streams that have been deleted to avoid stale references
deleted_stream_ids = set()
total_records = queryset.count()
logger.info(f"Starting rehash of {total_records} streams with keys: {keys}")
# Get initial count for progress reporting
initial_total_records = Stream.objects.count()
logger.info(f"Starting rehash of {initial_total_records} streams with keys: {keys}")
# Send initial WebSocket update
send_websocket_update(
@ -499,103 +502,115 @@ def rehash_streams(keys):
"type": "stream_rehash",
"action": "starting",
"progress": 0,
"total_records": total_records,
"message": f"Starting rehash of {total_records} streams"
"total_records": initial_total_records,
"message": f"Starting rehash of {initial_total_records} streams"
}
)
for start in range(0, total_records, batch_size):
# Use ID-based pagination to handle deletions correctly
# This ensures we don't skip records when items are deleted
last_processed_id = 0
batch_number = 0
while True:
batch_number += 1
batch_processed = 0
batch_duplicates = 0
with transaction.atomic():
batch = queryset[start:start + batch_size]
# Fetch batch by ID ordering, using select_for_update to lock records
# This prevents race conditions and ensures we process each record exactly once
batch = list(
Stream.objects.filter(id__gt=last_processed_id)
.select_for_update(skip_locked=True, of=('self',))
.select_related('channel_group', 'm3u_account')
.order_by('id')[:batch_size]
)
if not batch:
# No more records to process
break
for obj in batch:
# Generate new hash
# Update the last processed ID for next batch
last_processed_id = obj.id
# Generate new hash - handle XC accounts differently
group_name = obj.channel_group.name if obj.channel_group else None
new_hash = Stream.generate_hash_key(obj.name, obj.url, obj.tvg_id, keys, m3u_id=obj.m3u_account_id, group=group_name)
account_type = obj.m3u_account.account_type if obj.m3u_account else None
stream_id_val = obj.stream_id if hasattr(obj, 'stream_id') else None
# Check if this hash already exists in our tracking dict or in database
new_hash = Stream.generate_hash_key(
obj.name, obj.url, obj.tvg_id, keys,
m3u_id=obj.m3u_account_id, group=group_name,
account_type=account_type, stream_id=stream_id_val
)
# Check if this hash already exists in our tracking dict
if new_hash in hash_keys:
# Found duplicate in current batch - merge the streams
existing_stream_id = hash_keys[new_hash]
existing_stream = Stream.objects.get(id=existing_stream_id)
# Move any channel relationships from duplicate to existing stream
# Handle potential unique constraint violations
for channel_stream in ChannelStream.objects.filter(stream_id=obj.id):
# Check if this channel already has a relationship with the target stream
existing_relationship = ChannelStream.objects.filter(
channel_id=channel_stream.channel_id,
stream_id=existing_stream_id
).first()
# Verify the target stream still exists and hasn't been deleted
if existing_stream_id in deleted_stream_ids:
# The target was deleted, so this stream becomes the new canonical one
obj.stream_hash = new_hash
obj.save(update_fields=['stream_hash'])
hash_keys[new_hash] = obj.id
batch_processed += 1
continue
if existing_relationship:
# Relationship already exists, just delete the duplicate
channel_stream.delete()
else:
# Safe to update the relationship
channel_stream.stream_id = existing_stream_id
channel_stream.save()
try:
existing_stream = Stream.objects.get(id=existing_stream_id)
except Stream.DoesNotExist:
# Target stream was deleted externally, make this the canonical one
deleted_stream_ids.add(existing_stream_id)
obj.stream_hash = new_hash
obj.save(update_fields=['stream_hash'])
hash_keys[new_hash] = obj.id
batch_processed += 1
continue
# Update the existing stream with the most recent data
if obj.updated_at > existing_stream.updated_at:
existing_stream.name = obj.name
existing_stream.url = obj.url
existing_stream.logo_url = obj.logo_url
existing_stream.tvg_id = obj.tvg_id
existing_stream.m3u_account = obj.m3u_account
existing_stream.channel_group = obj.channel_group
existing_stream.custom_properties = obj.custom_properties
existing_stream.last_seen = obj.last_seen
existing_stream.updated_at = obj.updated_at
existing_stream.save()
# Determine which stream to keep based on channel ordering
stream_to_keep, stream_to_delete = _determine_stream_to_keep(existing_stream, obj)
# Delete the duplicate
obj.delete()
# Move channel relationships from the stream being deleted to the one being kept
_merge_stream_relationships(stream_to_delete, stream_to_keep)
# Delete the duplicate FIRST to free up the unique hash constraint
deleted_stream_ids.add(stream_to_delete.id)
stream_to_delete.delete()
batch_duplicates += 1
# Now safely set the hash on the kept stream (after deletion freed it up)
if stream_to_keep.stream_hash != new_hash:
stream_to_keep.stream_hash = new_hash
stream_to_keep.save(update_fields=['stream_hash'])
# Update hash_keys to point to the kept stream
hash_keys[new_hash] = stream_to_keep.id
else:
# Check if hash already exists in database (from previous batches or existing data)
# Check if hash already exists in database (from streams not yet processed)
existing_stream = Stream.objects.filter(stream_hash=new_hash).exclude(id=obj.id).first()
if existing_stream:
# Found duplicate in database - merge the streams
# Move any channel relationships from duplicate to existing stream
# Handle potential unique constraint violations
for channel_stream in ChannelStream.objects.filter(stream_id=obj.id):
# Check if this channel already has a relationship with the target stream
existing_relationship = ChannelStream.objects.filter(
channel_id=channel_stream.channel_id,
stream_id=existing_stream.id
).first()
# Found duplicate in database - determine which to keep based on channel ordering
stream_to_keep, stream_to_delete = _determine_stream_to_keep(existing_stream, obj)
if existing_relationship:
# Relationship already exists, just delete the duplicate
channel_stream.delete()
else:
# Safe to update the relationship
channel_stream.stream_id = existing_stream.id
channel_stream.save()
# Move channel relationships from the stream being deleted to the one being kept
_merge_stream_relationships(stream_to_delete, stream_to_keep)
# Update the existing stream with the most recent data
if obj.updated_at > existing_stream.updated_at:
existing_stream.name = obj.name
existing_stream.url = obj.url
existing_stream.logo_url = obj.logo_url
existing_stream.tvg_id = obj.tvg_id
existing_stream.m3u_account = obj.m3u_account
existing_stream.channel_group = obj.channel_group
existing_stream.custom_properties = obj.custom_properties
existing_stream.last_seen = obj.last_seen
existing_stream.updated_at = obj.updated_at
existing_stream.save()
# Delete the duplicate
obj.delete()
# Delete the duplicate FIRST to free up the unique hash constraint
deleted_stream_ids.add(stream_to_delete.id)
stream_to_delete.delete()
batch_duplicates += 1
hash_keys[new_hash] = existing_stream.id
# Now safely set the hash on the kept stream (after deletion freed it up)
if stream_to_keep.stream_hash != new_hash:
stream_to_keep.stream_hash = new_hash
stream_to_keep.save(update_fields=['stream_hash'])
hash_keys[new_hash] = stream_to_keep.id
else:
# Update hash for this stream
# No duplicate - update hash for this stream
obj.stream_hash = new_hash
obj.save(update_fields=['stream_hash'])
hash_keys[new_hash] = obj.id
@ -605,10 +620,9 @@ def rehash_streams(keys):
total_processed += batch_processed
duplicates_merged += batch_duplicates
# Calculate progress percentage
progress_percent = int((total_processed / total_records) * 100)
current_batch = start // batch_size + 1
total_batches = (total_records // batch_size) + 1
# Calculate progress percentage based on initial count
# Cap at 99% until we're actually done to avoid showing 100% prematurely
progress_percent = min(99, int((total_processed / max(initial_total_records, 1)) * 100))
# Send progress update via WebSocket
send_websocket_update(
@ -619,15 +633,14 @@ def rehash_streams(keys):
"type": "stream_rehash",
"action": "processing",
"progress": progress_percent,
"batch": current_batch,
"total_batches": total_batches,
"batch": batch_number,
"processed": total_processed,
"duplicates_merged": duplicates_merged,
"message": f"Processed batch {current_batch}/{total_batches}: {batch_processed} streams, {batch_duplicates} duplicates merged"
"message": f"Processed batch {batch_number}: {batch_processed} streams, {batch_duplicates} duplicates merged"
}
)
logger.info(f"Rehashed batch {current_batch}/{total_batches}: "
logger.info(f"Rehashed batch {batch_number}: "
f"{batch_processed} processed, {batch_duplicates} duplicates merged")
logger.info(f"Rehashing complete: {total_processed} streams processed, "
@ -654,7 +667,7 @@ def rehash_streams(keys):
return f"Successfully rehashed {total_processed} streams"
except Exception as e:
logger.error(f"Error during stream rehash: {e}")
logger.error(f"Error during stream rehash: {e}", exc_info=True)
raise
finally:
# Always release all acquired M3U locks
@ -663,6 +676,83 @@ def rehash_streams(keys):
logger.info(f"Released M3U task locks for {len(acquired_locks)} accounts")
def _merge_stream_relationships(source_stream, target_stream):
"""
Move channel relationships from source_stream to target_stream.
Handles unique constraint violations by preserving existing relationships.
Preserves the best ordering when merging relationships.
"""
for channel_stream in ChannelStream.objects.filter(stream_id=source_stream.id):
# Check if this channel already has a relationship with the target stream
existing_relationship = ChannelStream.objects.filter(
channel_id=channel_stream.channel_id,
stream_id=target_stream.id
).first()
if existing_relationship:
# Relationship already exists - keep the one with better ordering (lower order value)
if channel_stream.order < existing_relationship.order:
existing_relationship.order = channel_stream.order
existing_relationship.save(update_fields=['order'])
# Delete the duplicate relationship
channel_stream.delete()
else:
# Safe to update the relationship
channel_stream.stream_id = target_stream.id
channel_stream.save()
def _get_best_channel_order(stream):
"""
Get the best (lowest) channel order for a stream.
Returns None if stream has no channel relationships.
Lower order value = better/higher position in the channel list.
"""
best_order = ChannelStream.objects.filter(stream_id=stream.id).order_by('order').values_list('order', flat=True).first()
return best_order
def _determine_stream_to_keep(stream_a, stream_b):
"""
Determine which stream should be kept when merging duplicates.
Priority:
1. Stream with better (lower) channel order wins
2. If both have same order or neither has channel relationships,
keep the one with more recent updated_at
3. If still tied, keep the one with the lower ID (more stable)
Returns: (stream_to_keep, stream_to_delete)
"""
order_a = _get_best_channel_order(stream_a)
order_b = _get_best_channel_order(stream_b)
# If one has channel relationships and the other doesn't, keep the one with relationships
if order_a is not None and order_b is None:
return (stream_a, stream_b)
if order_b is not None and order_a is None:
return (stream_b, stream_a)
# If both have channel relationships, keep the one with better (lower) order
if order_a is not None and order_b is not None:
if order_a < order_b:
return (stream_a, stream_b)
elif order_b < order_a:
return (stream_b, stream_a)
# Same order, fall through to other criteria
# Neither has relationships, or same order - use updated_at
if stream_a.updated_at > stream_b.updated_at:
return (stream_a, stream_b)
elif stream_b.updated_at > stream_a.updated_at:
return (stream_b, stream_a)
# Same updated_at - keep lower ID for stability
if stream_a.id < stream_b.id:
return (stream_a, stream_b)
return (stream_b, stream_a)
@shared_task
def cleanup_vod_persistent_connections():
"""Clean up stale VOD persistent connections"""

11
frontend/src/WebSocket.jsx
View file

@ -700,6 +700,17 @@ export const WebsocketProvider = ({ children }) => {
withCloseButton: true, // Allow manual close
loading: false, // Remove loading indicator
});
// Requery streams and channels after rehash completes
try {
await API.requeryChannels();
await API.requeryStreams();
await useChannelsStore.getState().fetchChannels();
} catch (error) {
console.error(
'Error refreshing channels/streams after rehash:',
error
);
}
} else if (parsedEvent.data.action === 'blocked') {
// Handle blocked rehash attempt
notifications.show({