Mirrors/Dispatcharr
1
0
Fork 1
mirror of https://github.com/Dispatcharr/Dispatcharr.git synced 2026-07-18 00:55:50 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 4

outstanding commits, may revisit later

Browse source
This commit is contained in:
dekzter 2026-01-27 15:23:07 -05:00
parent 2ca2a51e2e
commit dba70b3c58
14 changed files with 242 additions and 123 deletions
Download patch file Download diff file Expand all files Collapse all files

18
apps/accounts/migrations/0004_user_stream_priority.py Normal file
View file

@ -0,0 +1,18 @@
# Generated by Django 5.2.4 on 2025-10-21 19:32
from django.db import migrations, models
class Migration(migrations.Migration):
dependencies = [
('accounts', '0003_alter_user_custom_properties'),
]
operations = [
migrations.AddField(
model_name='user',
name='stream_priority',
field=models.IntegerField(default=0, help_text='Priority level for streaming tasks. Lower values indicate higher priority.'),
),
]

153
apps/channels/models.py
View file

@ -166,7 +166,7 @@ class Stream(models.Model):
return stream_profile
def get_stream(self):
def get_stream(self, requester=None):
"""
Finds an available stream for the requested channel and returns the selected stream and profile.
"""
@ -349,7 +349,145 @@ class Channel(models.Model):
return stream_profile
def get_stream(self):
def _pick_channel_to_preempt(
self,
profile_id,
requester_level,
redis_client,
exclude_channel_ids=None,
cooldown_seconds=30,
):
"""
Pick the lowest-impact channel to terminate on the given profile.
Returns: Optional[int] channel_id to preempt
"""
exclude_channel_ids = set(exclude_channel_ids or [])
candidates = []
# 1) Try to get active channel IDs for this profile from an index set if available
ch_set_key = f"ts_proxy:profile:{profile_id}:channels"
try:
ch_ids = { (int(x) if not isinstance(x, int) else x) for x in (redis_client.smembers(ch_set_key) or set()) }
except Exception:
ch_ids = set()
logger.debug("Candidate channels for preemption:")
logger.debug(ch_ids)
# 2) Fallback: scan metadata keys and filter by m3u_profile == profile_id
if not ch_ids:
cursor = 0
pattern = "ts_proxy:channel:*:metadata"
while True:
cursor, keys = redis_client.scan(cursor=cursor, match=pattern, count=500)
if keys:
# Prefer HGET m3u_profile if metadata is a hash
pipe = redis_client.pipeline()
for k in keys:
pipe.hget(k, "m3u_profile")
prof_vals = pipe.execute()
for k, prof_val in zip(keys, prof_vals):
try:
pid = int(prof_val) if prof_val is not None else None
except Exception:
pid = None
if pid == profile_id:
parts = k.split(":") # ts_proxy:channel:{id}:metadata
if len(parts) >= 4:
try:
ch_ids.add(int(parts[2]))
except Exception:
pass
if cursor == 0:
break
logger.debug("Candidate channels for preemption:")
logger.debug(ch_ids)
if not ch_ids:
return None
# 3) Score candidates
for ch_id in ch_ids:
if ch_id in exclude_channel_ids:
continue
# Skip if recently preempted
last_preempt_key = f"ts_proxy:channel:{ch_id}:last_preempt"
try:
last_preempt = float(redis_client.get(last_preempt_key) or 0.0)
except Exception:
last_preempt = 0.0
if last_preempt and (time.time() - last_preempt) < cooldown_seconds:
continue
# Clients and their levels
clients_key = f"ts_proxy:channel:{ch_id}:clients"
member_ids = list(redis_client.smembers(clients_key) or [])
viewer_count = len(member_ids)
max_viewer_level = 0
if viewer_count:
pipe = redis_client.pipeline()
for cid in member_ids:
pipe.hget(f"ts_proxy:channel:{ch_id}:clients:{cid}", "user_level")
levels_raw = pipe.execute()
levels = []
for lv in levels_raw:
try:
levels.append(int(lv or 0))
except Exception:
levels.append(0)
max_viewer_level = max(levels or [0])
# Only preempt if requester strictly outranks this channel's viewers
if requester_level <= max_viewer_level:
continue
# Metadata (protected/recording/started_at_ts)
meta_key = f"ts_proxy:channel:{ch_id}:metadata"
try:
protected, recording, started_at_ts = redis_client.hmget(
meta_key, "protected", "recording", "started_at_ts"
)
except Exception:
protected = recording = started_at_ts = None
protected = str(protected or "0") in ("1", "true", "True")
recording = str(recording or "0") in ("1", "true", "True")
if protected or recording:
continue
try:
started_at_ts = float(started_at_ts) if started_at_ts is not None else None
except Exception:
started_at_ts = None
if started_at_ts is None:
started_at_ts = time.time() # treat unknown as newest
# Score: lower is safer to terminate
has_viewers = 1 if viewer_count > 0 else 0
score = (has_viewers, max_viewer_level, viewer_count, started_at_ts)
candidates.append((score, ch_id))
logger.debug("Candidate channels after scoring:")
logger.debug(candidates)
if not candidates:
return None
candidates.sort(key=lambda x: x[0])
victim_id = candidates[0][1]
# Mark preempt timestamp to avoid thrashing
try:
redis_client.set(f"ts_proxy:channel:{victim_id}:last_preempt", str(time.time()), ex=3600)
except Exception:
pass
return victim_id
def get_stream(self, requester=None):
"""
Finds an available stream for the requested channel and returns the selected stream and profile.
@ -438,6 +576,17 @@ class Channel(models.Model):
None,
) # Return newly assigned stream and matched profile
else:
# At capacity: try to preempt a lower-impact channel on this profile
victim_channel_id = self._pick_channel_to_preempt(
profile_id=profile.id,
requester_level=requester.user_level if requester else 100,
redis_client=redis_client,
exclude_channel_ids=None,
)
if victim_channel_id:
logger.info(f"Preempting channel {victim_channel_id} for new stream on profile {profile.id}")
# return self.id, profile.id, victim_channel_id
# This profile is at max connections
has_streams_but_maxed_out = True
logger.debug(

5
apps/channels/tasks.py
View file

@ -1900,11 +1900,8 @@ def run_recording(recording_id, channel_id, start_time_str, end_time_str):
metadata_key = RedisKeys.channel_metadata(str(channel.uuid))
md = r.hgetall(metadata_key)
if md:
def _gv(bkey):
return md.get(bkey)
def _d(bkey, cast=str):
v = _gv(bkey)
v = md.get(key)
try:
if v is None:
return None

16
apps/proxy/ts_proxy/client_manager.py
View file

@ -1,10 +1,8 @@
"""Client connection management for TS streams"""
import threading
import logging
import time
import json
import gevent
from typing import Set, Optional
from apps.proxy.config import TSConfig as Config
from redis.exceptions import ConnectionError, TimeoutError
@ -123,7 +121,7 @@ class ClientManager:
# Check for stale activity using last_active field
last_active = self.redis_client.hget(client_key, "last_active")
if last_active:
last_active_time = float(last_active.decode('utf-8'))
last_active_time = float(last_active)
ghost_timeout = self.heartbeat_interval * getattr(Config, 'GHOST_CLIENT_MULTIPLIER', 5.0)
if current_time - last_active_time > ghost_timeout:
@ -224,7 +222,7 @@ class ClientManager:
except Exception as e:
logger.error(f"Error notifying owner of client activity: {e}")
def add_client(self, client_id, client_ip, user_agent=None):
def add_client(self, client_id, client_ip, user_agent=None, user=None):
"""Add a client with duplicate prevention"""
if client_id in self._registered_clients:
logger.debug(f"Client {client_id} already registered, skipping")
@ -242,7 +240,8 @@ class ClientManager:
"ip_address": client_ip,
"connected_at": current_time,
"last_active": current_time,
"worker_id": self.worker_id or "unknown"
"worker_id": self.worker_id or "unknown",
"user_level": user.user_level if user is not None else 100, # default to a high value since no user means the non-user specific M3U/HDHR
}
try:
@ -303,8 +302,6 @@ class ClientManager:
def remove_client(self, client_id):
"""Remove a client from this channel and Redis"""
client_ip = None
with self.lock:
if client_id in self.clients:
self.clients.remove(client_id)
@ -317,11 +314,6 @@ class ClientManager:
if self.redis_client:
# Get client IP before removing the data
client_key = f"ts_proxy:channel:{self.channel_id}:clients:{client_id}"
client_data = self.redis_client.hgetall(client_key)
if client_data and b'ip_address' in client_data:
client_ip = client_data[b'ip_address'].decode('utf-8')
elif client_data and 'ip_address' in client_data:
client_ip = client_data['ip_address']
# Remove from channel's client set
self.redis_client.srem(self.client_set_key, client_id)

6
apps/proxy/ts_proxy/server.py
View file

@ -1121,12 +1121,6 @@ class ProxyServer:
else:
# Grace period expired with clients - mark channel as active
logger.info(f"Grace period expired with {total_clients} clients - marking channel {channel_id} as active")
<<<<<<< HEAD
old_state = "unknown"
if metadata and 'state' in metadata:
old_state = metadata['state']
=======
>>>>>>> origin/dev
if self.update_channel_state(channel_id, ChannelState.ACTIVE, {
"grace_period_ended_at": str(time.time()),
"clients_at_activation": str(total_clients)

4
apps/proxy/ts_proxy/utils.py
View file

@ -83,7 +83,7 @@ def create_ts_packet(packet_type='null', message=None):
# Add message to payload if provided
if message:
msg_bytes = message.encode('utf-8')
msg_bytes = message
packet[4:4+min(len(msg_bytes), 180)] = msg_bytes[:180]
return bytes(packet)
@ -113,4 +113,4 @@ def get_logger(component_name=None):
# Default if detection fails
logger_name = "ts_proxy"
return logging.getLogger(logger_name)
return logging.getLogger(logger_name)

26
apps/proxy/ts_proxy/views.py
View file

@ -45,7 +45,7 @@ logger = get_logger()
@api_view(["GET"])
def stream_ts(request, channel_id):
def stream_ts(request, channel_id, user=None):
if not network_access_allowed(request, "STREAMS"):
return JsonResponse({"error": "Forbidden"}, status=403)
@ -80,9 +80,9 @@ def stream_ts(request, channel_id):
metadata_key = RedisKeys.channel_metadata(channel_id)
if proxy_server.redis_client.exists(metadata_key):
metadata = proxy_server.redis_client.hgetall(metadata_key)
state_field = ChannelMetadataField.STATE.encode("utf-8")
state_field = ChannelMetadataField.STATE
if state_field in metadata:
channel_state = metadata[state_field].decode("utf-8")
channel_state = metadata[state_field]
# Active/running states - channel is operational, don't reinitialize
if channel_state in [
@ -120,7 +120,7 @@ def stream_ts(request, channel_id):
else:
owner_field = ChannelMetadataField.OWNER.encode("utf-8")
if owner_field in metadata:
owner = metadata[owner_field].decode("utf-8")
owner = metadata[owner_field]
owner_heartbeat_key = f"ts_proxy:worker:{owner}:heartbeat"
if proxy_server.redis_client.exists(owner_heartbeat_key):
# Owner is still active with unknown state - don't reinitialize
@ -373,7 +373,7 @@ def stream_ts(request, channel_id):
metadata_key, ChannelMetadataField.STATE
)
if state_bytes:
current_state = state_bytes.decode("utf-8")
current_state = state_bytes
logger.debug(
f"[{client_id}] Current state of channel {channel_id}: {current_state}"
)
@ -449,12 +449,12 @@ def stream_ts(request, channel_id):
)
if url_bytes:
url = url_bytes.decode("utf-8")
url = url_bytes
if ua_bytes:
stream_user_agent = ua_bytes.decode("utf-8")
stream_user_agent = ua_bytes
# Extract transcode setting from Redis
if profile_bytes:
profile_str = profile_bytes.decode("utf-8")
profile_str = profile_bytes
use_transcode = (
profile_str == PROXY_PROFILE_NAME or profile_str == "None"
)
@ -490,7 +490,7 @@ def stream_ts(request, channel_id):
# Register client
buffer = proxy_server.stream_buffers[channel_id]
client_manager = proxy_server.client_managers[channel_id]
client_manager.add_client(client_id, client_ip, client_user_agent)
client_manager.add_client(client_id, client_ip, client_user_agent, user)
logger.info(f"[{client_id}] Client registered with channel {channel_id}")
# Create a stream generator for this client
@ -553,7 +553,7 @@ def stream_xc(request, username, password, channel_id):
channel = get_object_or_404(Channel, id=channel_id)
# @TODO: we've got the file 'type' via extension, support this when we support multiple outputs
return stream_ts(request._request, str(channel.uuid))
return stream_ts(request._request, str(channel.uuid), user)
@csrf_exempt
@ -681,7 +681,7 @@ def channel_status(request, channel_id=None):
)
for key in keys:
channel_id_match = re.search(
r"ts_proxy:channel:(.*):metadata", key.decode("utf-8")
r"ts_proxy:channel:(.*):metadata", key
)
if channel_id_match:
ch_id = channel_id_match.group(1)
@ -802,7 +802,7 @@ def next_stream(request, channel_id):
metadata_key, ChannelMetadataField.STREAM_ID
)
if stream_id_bytes:
current_stream_id = int(stream_id_bytes.decode("utf-8"))
current_stream_id = int(stream_id_bytes)
logger.info(
f"Found current stream ID {current_stream_id} in Redis for channel {channel_id}"
)
@ -812,7 +812,7 @@ def next_stream(request, channel_id):
metadata_key, ChannelMetadataField.M3U_PROFILE
)
if profile_id_bytes:
profile_id = int(profile_id_bytes.decode("utf-8"))
profile_id = int(profile_id_bytes)
logger.info(
f"Found M3U profile ID {profile_id} in Redis for channel {channel_id}"
)

2
docker/init/02-postgres.sh
View file

@ -162,5 +162,3 @@ ensure_utf8_encoding() {
echo "Database $POSTGRES_DB converted to UTF8 and permissions set."
fi
}

2
docker/init/03-init-dispatcharr.sh
View file

@ -64,4 +64,4 @@ if [ "$(id -u)" = "0" ]; then
fi
chmod +x /data
fi
fi

50
docker/init/99-init-dev.sh
View file

@ -1,25 +1,33 @@
#!/bin/bash
echo "🚀 Development Mode - Setting up Frontend..."
if [ ! -e "/tmp/init" ]; then
echo "🚀 Development Mode - Setting up Frontend..."
# Install Node.js
if ! command -v node 2>&1 >/dev/null
then
echo "=== setting up nodejs ==="
curl -sL https://deb.nodesource.com/setup_23.x -o /tmp/nodesource_setup.sh
bash /tmp/nodesource_setup.sh
apt-get update
apt-get install -y --no-install-recommends \
nodejs
fi
# Install frontend dependencies
cd /app/frontend && npm install
# Install pip dependencies
cd /app && pip install -r requirements.txt
# Install debugpy for remote debugging
if [ "$DISPATCHARR_DEBUG" = "true" ]; then
echo "=== setting up debugpy ==="
pip install debugpy
# Install Node.js
if ! command -v node 2>&1 >/dev/null
then
echo "=== setting up nodejs ==="
curl -sL https://deb.nodesource.com/setup_23.x -o /tmp/nodesource_setup.sh
bash /tmp/nodesource_setup.sh
apt-get update
apt-get install -y --no-install-recommends \
nodejs
fi
# Install frontend dependencies
cd /app/frontend && npm install
# Install pip dependencies
cd /app && pip install -r requirements.txt
# Install debugpy for remote debugging
if [ "$DISPATCHARR_DEBUG" = "true" ]; then
echo "=== setting up debugpy ==="
pip install debugpy
fi
if [[ "$DISPATCHARR_ENV" = "dev" ]]; then
touch /tmp/init
fi
else
echo "Development mode initialization already done. Skipping dev setup."
fi

4
docker/uwsgi.dev.ini
View file

@ -8,7 +8,7 @@
exec-pre = python /app/scripts/wait_for_redis.py
; Start Redis first
attach-daemon = redis-server
attach-daemon = redis-server --protected-mode no
; Then start other services with configurable nice level (default: 5 for low priority)
; Users can override via CELERY_NICE_LEVEL environment variable in docker-compose
attach-daemon = nice -n $(CELERY_NICE_LEVEL) celery -A dispatcharr worker --autoscale=6,1
@ -59,4 +59,4 @@ logformat-strftime = true
log-date = %%Y-%%m-%%d %%H:%%M:%%S,000
# Use formatted time with environment variable for log level
log-format = %(ftime) $(DISPATCHARR_LOG_LEVEL) uwsgi.requests Worker ID: %(wid) %(method) %(status) %(uri) %(msecs)ms
log-buffering = 1024 # Add buffer size limit for logging
log-buffering = 1024 # Add buffer size limit for logging

4
frontend/prettier.config.js
View file

@ -3,8 +3,8 @@ export default {
semi: true, // Add semicolons at the end of statements
singleQuote: true, // Use single quotes instead of double
tabWidth: 2, // Set the indentation width
trailingComma: "es5", // Add trailing commas where valid in ES5
trailingComma: 'es5', // Add trailing commas where valid in ES5
printWidth: 80, // Wrap lines at 80 characters
bracketSpacing: true, // Add spaces inside object braces
arrowParens: "always", // Always include parentheses around arrow function parameters
arrowParens: 'always', // Always include parentheses around arrow function parameters
};

40
frontend/src/components/forms/User.jsx
View file

@ -12,8 +12,9 @@ import {
Stack,
MultiSelect,
ActionIcon,
NumberInput,
} from '@mantine/core';
import { RotateCcwKey, X } from 'lucide-react';
import { RotateCcwKey } from 'lucide-react';
import { useForm } from '@mantine/form';
import useChannelsStore from '../../store/channels';
import { USER_LEVELS, USER_LEVEL_LABELS } from '../../constants';
@ -209,26 +210,23 @@ const User = ({ user = null, isOpen, onClose }) => {
key={form.key('last_name')}
/>
<Group align="flex-end">
<TextInput
label="XC Password"
description="Clear to disable XC API"
{...form.getInputProps('xc_password')}
key={form.key('xc_password')}
style={{ flex: 1 }}
rightSectionWidth={30}
rightSection={
<ActionIcon
variant="transparent"
size="sm"
color="white"
onClick={generateXCPassword}
>
<RotateCcwKey />
</ActionIcon>
}
/>
</Group>
<TextInput
label="XC Password"
description="Clear to disable XC API"
{...form.getInputProps('xc_password')}
key={form.key('xc_password')}
rightSectionWidth={30}
rightSection={
<ActionIcon
variant="transparent"
size="sm"
color="white"
onClick={generateXCPassword}
>
<RotateCcwKey />
</ActionIcon>
}
/>
{showPermissions && (
<MultiSelect

35
frontend/src/pages/Users.jsx
View file

@ -2,49 +2,14 @@ import React, { useState } from 'react';
import UsersTable from '../components/tables/UsersTable';
import { Box } from '@mantine/core';
import useAuthStore from '../store/auth';
import { USER_LEVELS } from '../constants';
const UsersPage = () => {
const authUser = useAuthStore((s) => s.user);
const [selectedUser, setSelectedUser] = useState(null);
const [userModalOpen, setUserModalOpen] = useState(false);
const [confirmDeleteOpen, setConfirmDeleteOpen] = useState(false);
const [deleteTarget, setDeleteTarget] = useState(null);
const [userToDelete, setUserToDelete] = useState(null);
if (!authUser.id) {
return <></>;
}
const closeUserModal = () => {
setSelectedUser(null);
setUserModalOpen(false);
};
const editUser = (user) => {
setSelectedUser(user);
setUserModalOpen(true);
};
const deleteUser = (id) => {
// Get user details for the confirmation dialog
const user = users.find((u) => u.id === id);
setUserToDelete(user);
setDeleteTarget(id);
// Skip warning if it's been suppressed
if (isWarningSuppressed('delete-user')) {
return executeDeleteUser(id);
}
setConfirmDeleteOpen(true);
};
const executeDeleteUser = async (id) => {
await API.deleteUser(id);
setConfirmDeleteOpen(false);
};
return (
<Box style={{ padding: 10 }}>
<UsersTable />