mirror of
https://github.com/R0m1k3/xtremflow.git
synced 2026-10-19 13:08:05 +02:00
Multi-IPTV : chaque fonction utilise la playlist sélectionnée
Le serveur ignorait l'IPTV choisie : direct, VOD, guide, enregistrements, season passes et Auto NBA passaient tous par la playlist la plus récente (ou celle du premier utilisateur pour la capture FFmpeg), et les enregistrements s'affichaient sous toutes les playlists. - get_playlist respecte ?playlist=<id> (prioritaire) puis l'en-tête X-Playlist-Id, avec contrôle du propriétaire (admin : toutes). - Sessions FFmpeg HLS/VOD séparées par IPTV (même stream_id possible) ; le m3u8 propage ?playlist= aux segments, redirections et bascule .ts → HLS du lecteur web aussi. - tv_recordings, season_passes, auto_rules : colonne playlist_id (migrations v4-v6) ; lignes existantes rattachées à la playlist la plus récente de leur propriétaire ; URL de capture avec ?playlist= ; listes filtrées par IPTV ; une règle NBA par IPTV. - Proxy /api/xtream : cache d'allowlist par playlist, paramètre retiré avant l'appel au panneau ; CORS autorise X-Playlist-Id. - Client : AuthedHttp envoie X-Playlist-Id ; XtreamService ajoute ?playlist= aux flux, à l'EPG et aux appels xtream-api (cache distinct). Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
1 parent
29b4e11d13
commit
4f978755fd
15 files changed
+501
-70
No files matched your search
@@ -30,7 +30,11 @@ pub async fn auto_rules(state: &SharedState, req: &Req, rest: &str) -> Resp {
|
||||
};
|
||||
let parts = segments(rest);
|
||||
if let ("GET" | "HEAD", []) = (req.method.as_str(), parts.as_slice()) {
|
||||
let rules = state.db.get_auto_rules_for_user(&user.id);
|
||||
// Une règle NBA par IPTV : celle de la playlist choisie.
|
||||
let rules = match state.selected_playlist_id(req) {
|
||||
Some(pid) => state.db.get_auto_rules_for_playlist(&user.id, &pid),
|
||||
None => state.db.get_auto_rules_for_user(&user.id),
|
||||
};
|
||||
return json(200, &json!(rules));
|
||||
}
|
||||
if let ("POST", []) = (req.method.as_str(), parts.as_slice()) {
|
||||
@@ -144,12 +148,15 @@ async fn create(state: &SharedState, req: &Req, user: &User) -> Resp {
|
||||
Ok(b) => b,
|
||||
Err(r) => return r,
|
||||
};
|
||||
if state
|
||||
.db
|
||||
.get_auto_rules_for_user(&user.id)
|
||||
.iter()
|
||||
.any(|r| r.kind == "nba")
|
||||
{
|
||||
let playlist_id = state
|
||||
.get_playlist(req)
|
||||
.map(|p| p.id)
|
||||
.filter(|id| !id.is_empty());
|
||||
let existing = match &playlist_id {
|
||||
Some(pid) => state.db.get_auto_rules_for_playlist(&user.id, pid),
|
||||
None => state.db.get_auto_rules_for_user(&user.id),
|
||||
};
|
||||
if existing.iter().any(|r| r.kind == "nba") {
|
||||
return json(409, &json!({"error": "La règle NBA existe déjà"}));
|
||||
}
|
||||
let mut rule = AutoRule {
|
||||
@@ -164,6 +171,7 @@ async fn create(state: &SharedState, req: &Req, user: &User) -> Resp {
|
||||
pad_after_min: 45,
|
||||
team_filter: Vec::new(),
|
||||
keep_days: None,
|
||||
playlist_id: playlist_id.clone(),
|
||||
created_at: String::new(),
|
||||
updated_at: String::new(),
|
||||
};
|
||||
|
||||
@@ -578,6 +578,7 @@ mod tests {
|
||||
|
||||
fn playlist(base: &str) -> PlaylistConfig {
|
||||
PlaylistConfig {
|
||||
id: String::new(),
|
||||
dns: base.to_string(),
|
||||
username: "u".into(),
|
||||
password: "p".into(),
|
||||
|
||||
+27
-8
@@ -157,11 +157,21 @@ const ALLOWED_RESPONSE_HEADERS: [&str; 6] = [
|
||||
];
|
||||
const ALLOWED_REQUEST_HEADERS: [&str; 4] = ["user-agent", "accept", "range", "referer"];
|
||||
|
||||
/// Query sans le paramètre `playlist` (les autres sont recopiés tels quels).
|
||||
fn without_playlist_param(query: &str) -> String {
|
||||
query
|
||||
.split('&')
|
||||
.filter(|pair| !pair.is_empty() && pair.split('=').next() != Some("playlist"))
|
||||
.collect::<Vec<_>>()
|
||||
.join("&")
|
||||
}
|
||||
|
||||
pub struct ProxyHandler {
|
||||
asset_failures: AssetFailureCache,
|
||||
/// Playlist servant à l'allowlist, mise en cache 5 minutes pour éviter
|
||||
/// une requête SQL par segment.
|
||||
playlist_cache: Mutex<Option<(PlaylistConfig, Instant)>>,
|
||||
/// une requête SQL par segment. Clé : playlist demandée (vide = défaut),
|
||||
/// pour que deux IPTV aient chacune leur allowlist.
|
||||
playlist_cache: Mutex<std::collections::HashMap<String, (PlaylistConfig, Instant)>>,
|
||||
}
|
||||
|
||||
enum ProxyError {
|
||||
@@ -173,19 +183,26 @@ impl ProxyHandler {
|
||||
pub fn new() -> Self {
|
||||
Self {
|
||||
asset_failures: AssetFailureCache::new(3, Duration::from_secs(300)),
|
||||
playlist_cache: Mutex::new(None),
|
||||
playlist_cache: Mutex::new(std::collections::HashMap::new()),
|
||||
}
|
||||
}
|
||||
|
||||
fn cached_playlist(&self, state: &AppState, req: &Req) -> Option<PlaylistConfig> {
|
||||
if let Some((playlist, expiry)) = self.playlist_cache.lock().unwrap().as_ref()
|
||||
let key = req
|
||||
.query_param("playlist")
|
||||
.filter(|p| !p.trim().is_empty())
|
||||
.or_else(|| req.header("x-playlist-id").map(str::to_string))
|
||||
.unwrap_or_default();
|
||||
if let Some((playlist, expiry)) = self.playlist_cache.lock().unwrap().get(&key)
|
||||
&& Instant::now() < *expiry
|
||||
{
|
||||
return Some(playlist.clone());
|
||||
}
|
||||
let playlist = state.get_playlist(req)?;
|
||||
*self.playlist_cache.lock().unwrap() =
|
||||
Some((playlist.clone(), Instant::now() + Duration::from_secs(300)));
|
||||
self.playlist_cache.lock().unwrap().insert(
|
||||
key,
|
||||
(playlist.clone(), Instant::now() + Duration::from_secs(300)),
|
||||
);
|
||||
Some(playlist)
|
||||
}
|
||||
|
||||
@@ -218,9 +235,11 @@ impl ProxyHandler {
|
||||
);
|
||||
}
|
||||
let mut full_url = api_path;
|
||||
if !req.query.is_empty() {
|
||||
// `playlist` choisit l'IPTV côté XtremFlow : ne pas l'envoyer au panneau.
|
||||
let forwarded = without_playlist_param(&req.query);
|
||||
if !forwarded.is_empty() {
|
||||
full_url.push(if full_url.contains('?') { '&' } else { '?' });
|
||||
full_url.push_str(&req.query);
|
||||
full_url.push_str(&forwarded);
|
||||
}
|
||||
|
||||
let target = reqwest::Url::parse(&full_url).ok();
|
||||
|
||||
@@ -3,7 +3,7 @@
|
||||
|
||||
use crate::dart_time::DartDateTime;
|
||||
use crate::http::{self, Req, Resp, json, segments};
|
||||
use crate::models::{Recording, User};
|
||||
use crate::models::{Recording, User, stream_url_for_playlist};
|
||||
use crate::services::recording_scheduler::{RecordingScheduler, recordings_dir_path};
|
||||
use crate::state::SharedState;
|
||||
use crate::util::{media_probe, safe_path};
|
||||
@@ -58,10 +58,14 @@ async fn logs(state: &SharedState, req: &Req, id: &str) -> Resp {
|
||||
/// Liste les enregistrements de l'utilisateur (tous pour un admin),
|
||||
/// enrichis du suivi : taille, progression, relances, durée réelle.
|
||||
async fn list(state: &SharedState, req: &Req) -> Resp {
|
||||
let recordings = match &req.user {
|
||||
let mut recordings = match &req.user {
|
||||
Some(user) if !user.is_admin => state.db.get_user_recordings(&user.id),
|
||||
_ => state.db.get_all_recordings(),
|
||||
};
|
||||
// Une IPTV choisie : seulement ses enregistrements.
|
||||
if let Some(pid) = state.selected_playlist_id(req) {
|
||||
recordings.retain(|r| r.playlist_id.as_deref().is_none_or(|p| p == pid));
|
||||
}
|
||||
let now = Utc::now();
|
||||
let mut enriched = Vec::with_capacity(recordings.len());
|
||||
for r in &recordings {
|
||||
@@ -184,11 +188,26 @@ fn create(state: &SharedState, req: &Req) -> Resp {
|
||||
);
|
||||
}
|
||||
let title = to_text(data.get("title")).unwrap_or_else(|| "Sans Titre".into());
|
||||
// Capture sur l'IPTV choisie par le client (à défaut, la plus récente).
|
||||
let playlist_id = state
|
||||
.get_playlist(req)
|
||||
.map(|p| p.id)
|
||||
.filter(|id| !id.is_empty());
|
||||
let stream_url = match &playlist_id {
|
||||
Some(pid) => stream_url_for_playlist(&stream_url, pid),
|
||||
None => stream_url,
|
||||
};
|
||||
match state
|
||||
.db
|
||||
.create_recording(&user_id, &channel_id, &stream_url, &title, start, end)
|
||||
{
|
||||
Ok(recording) => json(200, &Value::Object(recording.to_map())),
|
||||
Ok(mut recording) => {
|
||||
if let Some(pid) = &playlist_id {
|
||||
state.db.set_recording_playlist(&recording.id, pid);
|
||||
recording.playlist_id = Some(pid.clone());
|
||||
}
|
||||
json(200, &Value::Object(recording.to_map()))
|
||||
}
|
||||
Err(e) => {
|
||||
println!("[RecordingsApi] Erreur à la création: {e}");
|
||||
json(500, &json!({"error": "Erreur lors de la programmation"}))
|
||||
@@ -261,11 +280,18 @@ pub fn season_passes(state: &SharedState, req: &Req, rest: &str) -> Resp {
|
||||
let Some(user) = &req.user else {
|
||||
return json(401, &json!({"error": "Authentification requise"}));
|
||||
};
|
||||
let passes = if user.is_admin {
|
||||
let mut passes = if user.is_admin {
|
||||
state.db.get_all_season_passes()
|
||||
} else {
|
||||
state.db.get_season_passes_for_user(&user.id)
|
||||
};
|
||||
if let Some(pid) = state.selected_playlist_id(req) {
|
||||
passes.retain(|p| {
|
||||
p.get("playlist_id")
|
||||
.and_then(Value::as_str)
|
||||
.is_none_or(|v| v == pid)
|
||||
});
|
||||
}
|
||||
json(
|
||||
200,
|
||||
&Value::Array(passes.into_iter().map(Value::Object).collect()),
|
||||
@@ -349,11 +375,28 @@ fn create_pass(state: &SharedState, req: &Req) -> Resp {
|
||||
&json!({"error": "Un Season Pass identique existe déjà"}),
|
||||
);
|
||||
}
|
||||
let playlist_id = state
|
||||
.get_playlist(req)
|
||||
.map(|p| p.id)
|
||||
.filter(|id| !id.is_empty());
|
||||
let stream_url = match &playlist_id {
|
||||
Some(pid) => stream_url_for_playlist(&stream_url, pid),
|
||||
None => stream_url,
|
||||
};
|
||||
match state
|
||||
.db
|
||||
.create_season_pass(&user.id, &show_title, &channel_id, &stream_url, match_mode)
|
||||
{
|
||||
Ok(pass) => {
|
||||
Ok(mut pass) => {
|
||||
if let Some(pid) = &playlist_id {
|
||||
let id = pass
|
||||
.get("id")
|
||||
.and_then(Value::as_str)
|
||||
.unwrap_or_default()
|
||||
.to_string();
|
||||
state.db.set_season_pass_playlist(&id, pid);
|
||||
pass.insert("playlist_id".into(), Value::from(pid.clone()));
|
||||
}
|
||||
println!("[SeasonPass] Créé: \"{show_title}\" sur chaîne {channel_id}");
|
||||
json(201, &Value::Object(pass))
|
||||
}
|
||||
|
||||
+98
-11
@@ -348,6 +348,53 @@ fn audio_args(with_filters: bool) -> Vec<String> {
|
||||
args
|
||||
}
|
||||
|
||||
/// Ajoute `?playlist=<id>` à chaque URI de segment d'un m3u8.
|
||||
///
|
||||
/// POURQUOI : le lecteur résout les segments relativement au m3u8 et perd
|
||||
/// sa query ; sans elle, le segment serait cherché dans la session d'une
|
||||
/// autre IPTV (même `stream_id`).
|
||||
fn with_segment_query(m3u8: &str, playlist_id: &str) -> String {
|
||||
let enc: String = url::form_urlencoded::byte_serialize(playlist_id.as_bytes()).collect();
|
||||
m3u8.lines()
|
||||
.map(|line| {
|
||||
let t = line.trim();
|
||||
if t.is_empty() || t.starts_with('#') {
|
||||
line.to_string()
|
||||
} else if t.contains('?') {
|
||||
format!("{t}&playlist={enc}")
|
||||
} else {
|
||||
format!("{t}?playlist={enc}")
|
||||
}
|
||||
})
|
||||
.collect::<Vec<_>>()
|
||||
.join("\n")
|
||||
+ "\n"
|
||||
}
|
||||
|
||||
/// Playlist explicitement demandée par la requête (à propager aux segments).
|
||||
fn requested_playlist(req: &Req) -> Option<String> {
|
||||
req.query_param("playlist").filter(|p| !p.trim().is_empty())
|
||||
}
|
||||
|
||||
/// Étiquette de l'IPTV servie, pour séparer les sessions de deux playlists.
|
||||
fn playlist_tag(state: &AppState, req: &Req) -> String {
|
||||
state
|
||||
.get_playlist(req)
|
||||
.map(|p| p.tag())
|
||||
.unwrap_or_else(|| "default".into())
|
||||
}
|
||||
|
||||
/// `?playlist=…` à recopier dans une redirection.
|
||||
fn playlist_suffix(req: &Req, first: bool) -> String {
|
||||
match requested_playlist(req) {
|
||||
Some(p) => {
|
||||
let enc: String = url::form_urlencoded::byte_serialize(p.as_bytes()).collect();
|
||||
format!("{}playlist={enc}", if first { '?' } else { '&' })
|
||||
}
|
||||
None => String::new(),
|
||||
}
|
||||
}
|
||||
|
||||
fn hls_headers(resp: Resp) -> Resp {
|
||||
http::with_headers(
|
||||
resp,
|
||||
@@ -514,7 +561,13 @@ impl Streaming {
|
||||
attempt
|
||||
}
|
||||
|
||||
async fn serve_session_playlist(&self, id: &str, attempt: Attempt, label: &str) -> Resp {
|
||||
async fn serve_session_playlist(
|
||||
&self,
|
||||
id: &str,
|
||||
attempt: Attempt,
|
||||
label: &str,
|
||||
playlist_query: Option<&str>,
|
||||
) -> Resp {
|
||||
let session = match attempt.session {
|
||||
Some(s) if attempt.ready => s,
|
||||
_ => {
|
||||
@@ -532,7 +585,13 @@ impl Streaming {
|
||||
};
|
||||
session.touch();
|
||||
match tokio::fs::read(session.playlist_path()).await {
|
||||
Ok(bytes) => hls_headers(http::response(200, bytes)),
|
||||
Ok(bytes) => {
|
||||
let body = match playlist_query {
|
||||
Some(q) => with_segment_query(&String::from_utf8_lossy(&bytes), q).into_bytes(),
|
||||
None => bytes,
|
||||
};
|
||||
hls_headers(http::response(200, body))
|
||||
}
|
||||
Err(_) => http::text(502, format!("{label} transcoder failed: playlist missing")),
|
||||
}
|
||||
}
|
||||
@@ -554,7 +613,10 @@ impl Streaming {
|
||||
// relatives des segments se résolvent sous le chemin de qualité.
|
||||
[id, "playlist.m3u8"] => {
|
||||
let quality = sanitize_quality(req.query_param("quality").as_deref());
|
||||
http::redirect(&format!("/api/live/{id}/{quality}/playlist.m3u8"))
|
||||
http::redirect(&format!(
|
||||
"/api/live/{id}/{quality}/playlist.m3u8{}",
|
||||
playlist_suffix(req, true)
|
||||
))
|
||||
}
|
||||
[id, "turbo.ts"] => self.turbo(state, req, id).await,
|
||||
[file] if file.len() > 3 && file.ends_with(".ts") => {
|
||||
@@ -564,7 +626,11 @@ impl Streaming {
|
||||
if !is_valid_stream_id(id) || segment.contains("..") {
|
||||
return http::text(400, "Invalid request");
|
||||
}
|
||||
let session_id = format!("live_{id}_{}", sanitize_quality(Some(quality)));
|
||||
let session_id = format!(
|
||||
"live_{}_{id}_{}",
|
||||
playlist_tag(state, req),
|
||||
sanitize_quality(Some(quality))
|
||||
);
|
||||
self.sessions.touch(&session_id);
|
||||
serve_segment(&self.sessions.base_dir.join(&session_id).join(segment), 60).await
|
||||
}
|
||||
@@ -590,7 +656,7 @@ impl Streaming {
|
||||
playlist.dns, playlist.username, playlist.password
|
||||
);
|
||||
let use_gpu = state.db.is_nvidia_gpu_enabled();
|
||||
let session_id = format!("live_{stream_id}_{quality}");
|
||||
let session_id = format!("live_{}_{stream_id}_{quality}", playlist.tag());
|
||||
if !self.sessions.contains(&session_id) {
|
||||
println!("[Live HLS] Starting {session_id}: {}", redact_url(&target));
|
||||
}
|
||||
@@ -657,8 +723,13 @@ impl Streaming {
|
||||
Some(&playlist),
|
||||
)
|
||||
.await;
|
||||
self.serve_session_playlist(&session_id, attempt, "Live")
|
||||
.await
|
||||
self.serve_session_playlist(
|
||||
&session_id,
|
||||
attempt,
|
||||
"Live",
|
||||
requested_playlist(req).as_deref(),
|
||||
)
|
||||
.await
|
||||
}
|
||||
|
||||
/// `/api/live/{id}/turbo.ts` — flux MPEG-TS continu pour mpegts.js : vidéo
|
||||
@@ -925,6 +996,9 @@ impl Streaming {
|
||||
query.append_pair(key, value);
|
||||
}
|
||||
}
|
||||
if let Some(p) = requested_playlist(req) {
|
||||
query.append_pair("playlist", &p);
|
||||
}
|
||||
let query = query.finish();
|
||||
let suffix = if query.is_empty() {
|
||||
String::new()
|
||||
@@ -937,7 +1011,11 @@ impl Streaming {
|
||||
if !is_valid_stream_id(id) || segment.contains("..") {
|
||||
return http::text(400, "Invalid request");
|
||||
}
|
||||
let session_id = format!("vod_{id}_{}", sanitize_quality(Some(quality)));
|
||||
let session_id = format!(
|
||||
"vod_{}_{id}_{}",
|
||||
playlist_tag(state, req),
|
||||
sanitize_quality(Some(quality))
|
||||
);
|
||||
self.sessions.touch(&session_id);
|
||||
serve_segment(
|
||||
&self.sessions.base_dir.join(&session_id).join(segment),
|
||||
@@ -983,7 +1061,7 @@ impl Streaming {
|
||||
vod_container_extension(params.get("ext").map(String::as_str))
|
||||
);
|
||||
let use_gpu = state.db.is_nvidia_gpu_enabled();
|
||||
let session_id = format!("vod_{stream_id}_{quality}");
|
||||
let session_id = format!("vod_{}_{stream_id}_{quality}", playlist.tag());
|
||||
if !self.sessions.contains(&session_id) {
|
||||
println!(
|
||||
"[VOD] Starting {session_id} ({content_type}): {}",
|
||||
@@ -1055,8 +1133,13 @@ impl Streaming {
|
||||
Some(&playlist),
|
||||
)
|
||||
.await;
|
||||
self.serve_session_playlist(&session_id, attempt, "VOD")
|
||||
.await
|
||||
self.serve_session_playlist(
|
||||
&session_id,
|
||||
attempt,
|
||||
"VOD",
|
||||
requested_playlist(req).as_deref(),
|
||||
)
|
||||
.await
|
||||
}
|
||||
|
||||
// ==========================================
|
||||
@@ -1314,6 +1397,10 @@ mod tests {
|
||||
assert_eq!(sanitize_quality(Some("bogus")), "high");
|
||||
assert_eq!(sanitize_quality(None), "high");
|
||||
assert_eq!(live_video_args("source", true), vec!["-c:v", "copy"]);
|
||||
assert_eq!(
|
||||
with_segment_query("#EXTM3U\n#EXTINF:4,\nseg_000.ts\nseg_001.ts?x=1\n", "ab-1"),
|
||||
"#EXTM3U\n#EXTINF:4,\nseg_000.ts?playlist=ab-1\nseg_001.ts?x=1&playlist=ab-1\n"
|
||||
);
|
||||
assert!(live_video_args("high", true).contains(&"h264_nvenc".to_string()));
|
||||
assert!(vod_video_args("low", false).contains(&"scale=-2:720".to_string()));
|
||||
}
|
||||
|
||||
Reference in new issue
Block a user