nirvana-7777/script.service.ultimate
1
0
Fork 0
mirror of https://github.com/nirvana-7777/script.service.ultimate.git synced 2026-09-16 06:02:35 +02:00
Code Issues Packages Projects Releases Wiki Activity

Reformat

Browse Source
This commit is contained in:
Nirvana
2026-01-20 15:26:41 +01:00
parent 08f5afb47a
commit e05205bbc8
14 changed files with 3352 additions and 2434 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Dockerfile
+16 -9
View File
@@ -47,24 +47,31 @@ RUN pip install --no-cache-dir --upgrade pip && \
RUN groupadd -g ${GROUP_ID} ${APP_USER} && \
useradd -u ${USER_ID} -g ${APP_USER} -m -s /bin/bash ${APP_USER}
# Create directories
RUN mkdir -p /config /logs /cache /drm-plugins && \
chown -R ${USER_ID}:${GROUP_ID} /config /logs /cache /drm-plugins
# Create directories for new structure
RUN mkdir -p /config /logs /cache /drm-plugins /app/routes && \
chown -R ${USER_ID}:${GROUP_ID} /config /logs /cache /drm-plugins /app/routes
# Copy application code
COPY --chown=${USER_ID}:${GROUP_ID} . .
# Copy application code with new structure
COPY --chown=${USER_ID}:${GROUP_ID} service.py .
# IMPORTANT: Copy the entire routes directory
COPY --chown=${USER_ID}:${GROUP_ID} routes/ /app/routes/
# Create the directory structure for DRM plugins
RUN mkdir -p /app/lib/streaming_providers/base/drm/plugins && \
chown -R ${USER_ID}:${GROUP_ID} /app/lib/streaming_providers/base/drm
# Copy entrypoint script
# Quick directory check (for debugging)
RUN echo "=== Directory structure ===" && \
ls -la /app && \
echo "---" && \
ls -la /app/routes/ 2>/dev/null || echo "routes directory not found" && \
echo "---" && \
ls -la /app/lib/ 2>/dev/null || echo "lib directory not found"
# Copy updated entrypoint script
COPY docker-entrypoint.sh /usr/local/bin/
RUN chmod +x /usr/local/bin/docker-entrypoint.sh
# Quick directory check
RUN ls -la /app && echo "---" && ls -la /app/lib/ 2>/dev/null || echo "lib directory not found"
# Switch to non-root user
USER ${USER_ID}
docker-compose.yml
+2 -1
View File
@@ -15,8 +15,9 @@ services:
environment:
- ULTIMATE_PORT=7777
- ULTIMATE_DEBUG=false
- ULTIMATE_EPG_URL=https://raw.githubusercontent.com/epgshare01/share01/master/epg.xml.gz
- ULTIMATE_EPG_URL=https://example.com/epg.xml.gz
- TZ=${TZ:-Europe/Berlin}
- MEDIA_PROXY_URL=http://media-proxy:8080 # if using media proxy
volumes:
- ./config:/config
- ./logs:/logs
docker-entrypoint.sh
+29
View File
@@ -19,5 +19,34 @@ if [ -z "$(ls -A /drm-plugins)" ] && [ -d "/app/lib/streaming_providers/base/drm
cp -r /app/lib/streaming_providers/base/drm/default-plugins/* /drm-plugins/
fi
# IMPORTANT: Ensure routes directory exists and has required files
# This is needed for the new split structure
if [ ! -d "/app/routes" ]; then
echo "ERROR: routes directory not found!"
echo "The service has been split into modules. Make sure routes/ directory exists in /app/"
exit 1
fi
# Verify all required route files exist
REQUIRED_ROUTES=("__init__.py" "providers.py" "streams.py" "m3u.py" "drm.py" "cache.py" "config.py" "epg.py")
MISSING_FILES=0
for route in "${REQUIRED_ROUTES[@]}"; do
if [ ! -f "/app/routes/$route" ]; then
echo "ERROR: Missing required route file: /app/routes/$route"
MISSING_FILES=$((MISSING_FILES + 1))
fi
done
if [ $MISSING_FILES -gt 0 ]; then
echo "ERROR: Missing $MISSING_FILES required route files. Service cannot start."
exit 1
fi
# Set Python path to include routes directory
export PYTHONPATH="${PYTHONPATH}:/app"
echo "Routes directory check passed. Starting service..."
# Execute the main command
exec "$@"
lib/streaming_providers/base/utils/mpd_rewriter.py
+62 -23
View File
@@ -2,63 +2,102 @@
import base64
import xml.etree.ElementTree as ET
from typing import Optional, Tuple
from urllib.parse import urljoin, urlparse
from urllib.parse import urljoin, urlparse, quote
from .logger import logger
class MPDRewriter:
"""
Utility for rewriting MPD (MPEG-DASH) manifest URLs to point to proxy endpoints
Utility for rewriting MPD (MPEG-DASH) manifest URLs to point to media proxy endpoints
Strategy:
- Remove all BaseURL elements
- Convert all relative URLs to absolute URLs
- Rewrite all absolute URLs to proxy endpoint
- Rewrite all absolute URLs to media proxy endpoint
- Keep template variables visible for client-side substitution
"""
# MPD namespace
MPD_NAMESPACE = {"mpd": "urn:mpeg:dash:schema:mpd:2011"}
def __init__(self, proxy_base_url: str, provider_name: str):
def __init__(self, media_proxy_url: str, provider_proxy_url: Optional[str] = None,
clearkey_keyids: Optional[dict] = None):
"""
Initialize MPD rewriter
Args:
proxy_base_url: Base URL of the proxy service (e.g., http://localhost:7777)
provider_name: Name of the provider for proxy routing
media_proxy_url: Base URL of the media proxy service (e.g., http://10.77.77.7:7775)
provider_proxy_url: Optional proxy URL for the provider (e.g., http://nordlynx_germany:8888)
clearkey_keyids: Optional dict of kid:key pairs for decrypted playback
"""
self.proxy_base_url = proxy_base_url.rstrip("/")
self.provider_name = provider_name
self.media_proxy_url = media_proxy_url.rstrip("/")
self.provider_proxy_url = provider_proxy_url
self.clearkey_keyids = clearkey_keyids or {}
@staticmethod
def encode_url(url: str) -> str:
"""Encode URL to base64 for use in proxy endpoint"""
return base64.urlsafe_b64encode(url.encode("utf-8")).decode("utf-8")
"""
Encode URL to base64 for use in media proxy endpoint.
Strips padding as required by media proxy.
"""
encoded = base64.urlsafe_b64encode(url.encode("utf-8")).decode("utf-8")
# Strip padding
return encoded.rstrip("=")
@staticmethod
def decode_url(encoded: str) -> str:
"""Decode base64 URL from proxy endpoint"""
"""
Decode base64 URL from media proxy endpoint.
Adds back padding if needed.
"""
# Add back padding if needed
padding = 4 - (len(encoded) % 4)
if padding != 4:
encoded += "=" * padding
return base64.urlsafe_b64decode(encoded.encode("utf-8")).decode("utf-8")
def build_proxy_url(self, original_url: str, template_pattern: Optional[str] = None) -> str:
"""
Build proxy URL for an original media URL
Build media proxy URL for an original media URL
Args:
original_url: Original URL to be proxied (base path for templates)
template_pattern: Optional template pattern to append (e.g., "segment-$Number$.m4s")
Returns:
Proxy URL
Media proxy URL
"""
encoded = self.encode_url(original_url)
proxy_url = f"{self.proxy_base_url}/api/proxy/{self.provider_name}/{encoded}"
# Choose endpoint based on whether we have clearkey data
if self.clearkey_keyids:
proxy_url = f"{self.media_proxy_url}/api/decrypt/{encoded}"
else:
proxy_url = f"{self.media_proxy_url}/api/proxy/{encoded}"
# Append template pattern if provided (keeps variables visible for client)
if template_pattern:
proxy_url += f"/{template_pattern}"
# URL encode the template pattern (same as current behavior)
encoded_pattern = quote(template_pattern, safe=".-_$")
proxy_url += f"/{encoded_pattern}"
# Build query parameters
query_params = []
# Add clearkey parameters if present
if self.clearkey_keyids:
for kid, key in self.clearkey_keyids.items():
query_params.append(f"kid={kid}")
query_params.append(f"key={key}")
# Add provider proxy parameter if configured
if self.provider_proxy_url:
query_params.append(f"proxy={self.provider_proxy_url}")
# Append query string if we have parameters
if query_params:
proxy_url += "?" + "&".join(query_params)
return proxy_url
@@ -98,13 +137,13 @@ class MPDRewriter:
def rewrite_mpd(self, mpd_content: str, manifest_url: str) -> str:
"""
Rewrite MPD content to use proxy URLs
Rewrite MPD content to use media proxy URLs
Strategy:
1. Parse MPD XML
2. Extract and remove all BaseURL elements
3. Resolve all relative URLs to absolute using BaseURLs and manifest URL
4. Rewrite all absolute URLs to proxy endpoints
4. Rewrite all absolute URLs to media proxy endpoints
5. Keep template variables visible for client substitution
Args:
@@ -124,7 +163,7 @@ class MPDRewriter:
# Get base URL for relative resolution
base_url = self._extract_base_url(root, manifest_url)
# Remove all BaseURL elements (Option 3 strategy)
# Remove all BaseURL elements
self._remove_base_urls(root)
# Rewrite all URLs in the MPD
@@ -137,7 +176,7 @@ class MPDRewriter:
if not rewritten.startswith("<?xml"):
rewritten = '<?xml version="1.0" encoding="UTF-8"?>\n' + rewritten
logger.debug(f"Successfully rewrote MPD for provider '{self.provider_name}'")
logger.debug(f"Successfully rewrote MPD for media proxy")
return rewritten
except ET.ParseError as e:
@@ -188,7 +227,7 @@ class MPDRewriter:
def _remove_base_urls(self, root: ET.Element) -> None:
"""
Remove all BaseURL elements from MPD (Option 3 strategy)
Remove all BaseURL elements from MPD
Args:
root: MPD root element
@@ -232,12 +271,12 @@ class MPDRewriter:
base_path, template_pattern = self.split_template_url(resolved)
element.attrib[attr] = self.build_proxy_url(base_path, template_pattern)
logger.debug(
f"Rewrote template URL: {original_url} -> proxy with template {template_pattern}"
f"Rewrote template URL: {original_url} -> media proxy with template {template_pattern}"
)
else:
# Regular URL without templates
element.attrib[attr] = self.build_proxy_url(resolved)
logger.debug(f"Rewrote URL: {original_url} -> proxy")
logger.debug(f"Rewrote URL: {original_url} -> media proxy")
# Handle SegmentURL elements (used in SegmentList)
if element.tag.endswith("SegmentURL"):
@@ -373,4 +412,4 @@ class MPDRewriter:
total_seconds = int(hours * 3600 + minutes * 60 + seconds)
logger.debug(f"Parsed ISO duration '{duration}' to {total_seconds}s")
return total_seconds
return total_seconds
requirements.txt
+7 -1
View File
@@ -9,4 +9,10 @@ python-dateutil>=2.8.2
urllib3>=2.0.7
chardet>=5.2.0
pycryptodome>=3.19.0
cryptography>=41.0.0
# Optional but recommended
aiohttp>=3.8.5
cryptography>=41.0.4
# For development/debugging
python-dotenv>=1.0.0
routes/__init__.py
+22
View File
@@ -0,0 +1,22 @@
#!/usr/bin/env python3
"""
Route handlers for Ultimate Backend Service
"""
from .cache import setup_cache_routes
from .config import setup_config_routes
from .drm import setup_drm_routes
from .epg import setup_epg_routes
from .m3u import setup_m3u_routes
from .providers import setup_provider_routes
from .streams import setup_stream_routes
__all__ = [
"setup_provider_routes",
"setup_stream_routes",
"setup_m3u_routes",
"setup_drm_routes",
"setup_cache_routes",
"setup_config_routes",
"setup_epg_routes",
]
routes/cache.py
+140
View File
@@ -0,0 +1,140 @@
#!/usr/bin/env python3
"""
Cache management route handlers
"""
from bottle import response
from streaming_providers.base.utils import logger
def setup_cache_routes(app, manager, service):
"""Setup cache management routes"""
@app.route("/api/cache/pssh", method="DELETE")
def clear_pssh_cache():
"""
Clear all PSSH cache entries.
This is useful for:
- Debugging
- Freeing memory
- Forcing re-extraction of all channels
"""
try:
cache_size = len(manager.drm_ops.pssh_cache.cache)
manager.drm_ops.pssh_cache.clear()
response.status = 200
return {"message": "PSSH cache cleared", "entries_cleared": cache_size}
except Exception as e:
logger.error(f"Error clearing cache: {e}")
response.status = 500
return {"error": "Failed to clear cache", "message": str(e)}
@app.route("/api/cache/pssh", method="GET")
def get_pssh_cache_stats():
"""
Get PSSH cache statistics.
Returns information about:
- Number of cached entries
- TTL configuration
- Memory usage estimate
"""
try:
cache = manager.drm_ops.pssh_cache
with cache.lock:
entries = []
total_size = 0
for key, (pssh_list, timestamp) in cache.cache.items():
import time
age = time.time() - timestamp
expires_in = cache.ttl - age
# Estimate size
size = sum(
len(p.pssh_box) + len(str(p.key_ids)) + len(p.system_id)
for p in pssh_list
)
total_size += size
entries.append(
{
"key": key,
"pssh_count": len(pssh_list),
"age_seconds": int(age),
"expires_in_seconds": int(expires_in),
"size_bytes": size,
}
)
response.status = 200
return {
"total_entries": len(entries),
"ttl_seconds": cache.ttl,
"total_size_bytes": total_size,
"total_size_mb": round(total_size / 1024 / 1024, 2),
"entries": sorted(
entries, key=lambda x: x["age_seconds"], reverse=True
),
}
except Exception as e:
logger.error(f"Error getting cache stats: {e}")
response.status = 500
return {"error": "Failed to get cache stats", "message": str(e)}
@app.route("/api/cache/mpd/clear")
def clear_mpd_cache():
"""Clear all cached MPD manifests"""
try:
service.mpd_cache.clear_all()
return {"success": True, "message": "MPD cache cleared"}
except Exception as e:
logger.error(f"Error clearing MPD cache: {e}")
response.status = 500
return {"error": str(e)}
@app.route("/api/cache/mpd/clear-expired")
def clear_expired_mpd_cache():
"""Clear expired MPD cache entries"""
try:
cleared = service.mpd_cache.clear_expired()
return {"success": True, "cleared": cleared}
except Exception as e:
logger.error(f"Error clearing expired MPD cache: {e}")
response.status = 500
return {"error": str(e)}
@app.route("/api/cache/mpd/<provider>/<channel_id>")
def get_mpd_cache_info(provider, channel_id):
"""Get cache information for a specific channel"""
try:
info = service.mpd_cache.get_cache_info(provider, channel_id)
if info:
return {"success": True, "cache_info": info}
else:
response.status = 404
return {"success": False, "message": "No cache found"}
except Exception as e:
logger.error(f"Error getting cache info: {e}")
response.status = 500
return {"error": str(e)}
@app.route("/api/cache/mpd/<provider>/<channel_id>/delete")
def delete_mpd_cache(provider, channel_id):
"""Delete cached MPD for a specific channel"""
try:
service.mpd_cache.delete(provider, channel_id)
return {
"success": True,
"message": f"Cache deleted for {provider}/{channel_id}",
}
except Exception as e:
logger.error(f"Error deleting cache: {e}")
response.status = 500
return {"error": str(e)}
routes/config.py
+327
View File
@@ -0,0 +1,327 @@
#!/usr/bin/env python3
"""
Configuration route handlers
"""
import json
import os
from bottle import request, response
from streaming_providers.base.utils import logger
def setup_config_routes(app, manager, service):
"""Setup configuration-related routes"""
@app.route("/api/config/export")
def export_config():
"""Export all configurations as JSON"""
try:
settings_manager = service._get_settings_manager()
# Use SettingsManager's export method
export_path = settings_manager.export_all_settings()
# Read the exported file
with open(export_path, "r", encoding="utf-8") as f:
config_data = json.load(f)
response.content_type = "application/json"
response.headers["Content-Disposition"] = (
f'attachment; filename="{os.path.basename(export_path)}"'
)
return json.dumps(config_data, indent=2)
except Exception as e:
logger.error(f"Error exporting config: {e}")
response.status = 500
return {"error": str(e)}
@app.route("/api/config/import", method="POST")
def import_config():
"""Import configurations from JSON"""
try:
import_data = request.json
except ValueError:
response.status = 400
return {"error": "Invalid JSON format"}
if not import_data:
response.status = 400
return {"error": "No data provided"}
# Validate it's a dict
if not isinstance(import_data, dict):
response.status = 400
return {"error": "Import data must be a JSON object"}
# Create temp file
try:
import json
import tempfile
import uuid
temp_dir = tempfile.gettempdir()
temp_file = os.path.join(temp_dir, f"import_{uuid.uuid4()}.json")
with open(temp_file, "w", encoding="utf-8") as f:
json.dump(import_data, f)
except (IOError, OSError, PermissionError) as file_err:
logger.error(f"Failed to create temp file: {file_err}")
response.status = 500
return {"error": "Failed to process import file"}
imported_count = 0
try:
# Use SettingsManager to import
settings_manager = service._get_settings_manager()
# Import credentials
credentials = import_data.get("providers", {})
for provider_name, provider_data in credentials.items():
# Validate provider data
if not isinstance(provider_data, dict):
logger.warning(
f"Skipping invalid provider data for {provider_name}"
)
continue
# Extract credential data if available
if "credentials" in provider_data:
cred_data = provider_data["credentials"]
if isinstance(cred_data, dict):
success, message = (
settings_manager.save_provider_credentials_from_api(
provider_name, cred_data
)
)
if success:
imported_count += 1
logger.info(f"Imported credentials for {provider_name}")
else:
logger.warning(
f"Failed to import credentials for {provider_name}: {message}"
)
# Import proxy data if available
if "proxy" in provider_data:
proxy_data = provider_data["proxy"]
if isinstance(proxy_data, dict):
success, message = (
settings_manager.save_provider_proxy_from_api(
provider_name, proxy_data
)
)
if success:
imported_count += 1
logger.info(f"Imported proxy for {provider_name}")
else:
logger.warning(
f"Failed to import proxy for {provider_name}: {message}"
)
except Exception as process_err:
logger.error(
f"Error during import processing: {process_err}", exc_info=True
)
response.status = 500
return {"error": f"Import failed: {str(process_err)}"}
finally:
# Always try to clean up temp file
try:
if os.path.exists(temp_file):
os.remove(temp_file)
except (FileNotFoundError, PermissionError, OSError) as cleanup_err:
logger.debug(f"Could not remove temp file {temp_file}: {cleanup_err}")
return {
"success": True,
"imported": imported_count,
"message": f"Imported {imported_count} configurations",
}
@app.route("/api/config/epg", method="GET")
def get_epg_config():
"""Get current EPG configuration"""
try:
from streaming_providers.base.utils.environment import (
get_environment_manager,
)
env_mgr = get_environment_manager()
config = {
"epg_url": env_mgr.get_config("epg_url", ""),
"epg_cache_ttl": env_mgr.get_config("epg_cache_ttl", 86400),
"source": (
"config.json" if env_mgr.get_config("epg_url") else "default"
),
}
# Also check environment variable for reference
import os
env_epg_url = os.environ.get("ULTIMATE_EPG_URL")
if env_epg_url:
config["environment_variable"] = env_epg_url
return {
"success": True,
"config": config,
"epg_manager_status": (
"initialized"
if hasattr(service, "epg_manager")
else "not_initialized"
),
}
except Exception as e:
logger.error(f"API Error in /api/config/epg: {str(e)}")
response.status = 500
return {"error": f"Internal server error: {str(e)}"}
@app.route("/api/config/epg", method="POST")
def set_epg_config():
"""Set EPG configuration"""
try:
# Parse JSON body
try:
epg_data = request.json
except Exception as json_err:
logger.error(f"Invalid JSON in request body: {json_err}")
response.status = 400
return {"error": "Invalid JSON in request body"}
if not epg_data:
response.status = 400
return {"error": "Request body must contain EPG configuration"}
# Validate it's a dictionary
if not isinstance(epg_data, dict):
response.status = 400
return {"error": "EPG data must be a JSON object"}
# Validate URL format
epg_url = epg_data.get("epg_url", "").strip()
if epg_url:
# Basic URL validation
if not (
epg_url.startswith("http://") or epg_url.startswith("https://")
):
response.status = 400
return {"error": "EPG URL must start with http:// or https://"}
# Validate it's an XML/GZ file
if not (
epg_url.endswith(".xml")
or epg_url.endswith(".xml.gz")
or epg_url.endswith(".gz")
):
logger.warning(f"EPG URL doesn't end with .xml or .gz: {epg_url}")
# Use environment manager
from streaming_providers.base.utils.environment import (
get_environment_manager,
)
env_mgr = get_environment_manager()
# Get current config
import json
import os
profile_path = env_mgr.get_config("profile_path", "")
config_file = os.path.join(profile_path, "config.json")
config_data = {}
if os.path.exists(config_file):
try:
with open(config_file, "r", encoding="utf-8") as f:
config_data = json.load(f)
except Exception as e:
logger.error(f"Error reading config.json: {e}")
config_data = {}
# Update with new values
config_data["epg_url"] = epg_url
# Optional: EPG cache TTL
if "epg_cache_ttl" in epg_data:
try:
ttl = int(epg_data["epg_cache_ttl"])
if ttl > 0:
config_data["epg_cache_ttl"] = ttl
except ValueError:
pass
# Save back to config.json
try:
with open(config_file, "w", encoding="utf-8") as f:
json.dump(config_data, f, indent=2, ensure_ascii=False)
# Update environment manager cache
env_mgr.set_config("epg_url", epg_url)
if "epg_cache_ttl" in config_data:
env_mgr.set_config("epg_cache_ttl", config_data["epg_cache_ttl"])
logger.info(f"Updated EPG configuration: URL={epg_url}")
return {
"success": True,
"message": "EPG configuration updated successfully",
"config": {
"epg_url": epg_url,
"epg_cache_ttl": config_data.get("epg_cache_ttl", 86400),
},
}
except Exception as e:
logger.error(f"Error writing config.json: {e}")
response.status = 500
return {"error": f"Failed to save configuration: {str(e)}"}
except Exception as e:
logger.error(f"API Error in POST /api/config/epg: {str(e)}")
response.status = 500
return {"error": f"Internal server error: {str(e)}"}
@app.route("/api/config/epg/clear-cache", method="POST")
def clear_epg_cache():
"""Clear EPG cache"""
try:
if hasattr(service, "epg_manager") and service.epg_manager:
success = service.epg_manager.clear_cache()
if success:
return {"success": True, "message": "EPG cache cleared"}
else:
response.status = 500
return {"error": "Failed to clear EPG cache"}
else:
response.status = 404
return {"error": "EPG manager not initialized"}
except Exception as e:
logger.error(f"API Error clearing EPG cache: {str(e)}")
response.status = 500
return {"error": f"Internal server error: {str(e)}"}
@app.route("/api/config/epg/cache-info", method="GET")
def get_epg_cache_info():
"""Get EPG cache information"""
try:
if hasattr(service, "epg_manager") and service.epg_manager:
cache_info = service.epg_manager.get_cache_info()
mapping_stats = service.epg_manager.get_mapping_stats()
return {
"success": True,
"cache_info": cache_info,
"mapping_stats": mapping_stats,
}
else:
response.status = 404
return {"error": "EPG manager not initialized"}
except Exception as e:
logger.error(f"API Error getting EPG cache info: {str(e)}")
response.status = 500
return {"error": f"Internal server error: {str(e)}"}
routes/drm.py
+250
View File
@@ -0,0 +1,250 @@
#!/usr/bin/env python3
"""
DRM and PSSH route handlers
"""
import traceback
from bottle import request, response
from streaming_providers.base.utils import logger
def setup_drm_routes(app, manager, service):
"""Setup DRM and PSSH-related routes"""
@app.route("/api/providers/<provider>/channels/<channel_id>/pssh")
def get_channel_pssh(provider, channel_id):
"""
Extract PSSH data for a channel.
Query parameters:
- country: Optional country code for geo-specific manifests
- force_refresh: If 'true', bypass cache and re-extract PSSH
Returns:
{
"provider": "provider_name",
"channel_id": "channel_id",
"manifest_url": "https://...",
"pssh_data": [
{
"system_id": "edef8ba9-79d6-4ace-a3c8-27dcd51d21ed",
"drm_system": "com.widevine.alpha",
"pssh_box": "AAAANHBzc2g...",
"key_ids": ["64656d6f..."],
"source": "mp4_segment"
}
],
"count": 1,
"cached": true
}
"""
try:
# Parse query parameters
country = request.params.get("country")
force_refresh = request.params.get("force_refresh", "").lower() == "true"
# Check cache status BEFORE clearing (to know if it was cached)
cache_key = f"{provider}:{channel_id}"
was_cached_before = manager.drm_ops.pssh_cache.get(cache_key) is not None
# Clear cache if force refresh requested
if force_refresh:
with manager.drm_ops.pssh_cache.lock:
if cache_key in manager.drm_ops.pssh_cache.cache:
del manager.drm_ops.pssh_cache.cache[cache_key]
logger.info(f"Cache cleared for force_refresh request: {cache_key}")
was_cached_before = False # Since we just cleared it
# Get DRM configs - this will check cache and extract if needed
try:
# This method handles caching internally
drm_configs = manager.drm_ops.get_channel_drm_configs(
provider_name=provider, channel_id=channel_id, country=country
)
except ValueError as e:
response.status = 404
return {
"error": "Provider not found",
"message": str(e),
"provider": provider,
}
except Exception as e:
response.status = 500
return {
"error": "Failed to get DRM configs",
"message": str(e),
"provider": provider,
"channel_id": channel_id,
}
# Get manifest URL for reference
try:
manifest_url = manager.get_channel_manifest(
provider_name=provider, channel_id=channel_id, country=country
)
except Exception as e:
manifest_url = None
logger.debug(f"Could not get manifest URL: {e}")
# Now get the PSSH data from cache (after get_channel_drm_configs has populated it)
pssh_data_list = manager.drm_ops.pssh_cache.get(cache_key)
was_cached_after = pssh_data_list is not None
if not pssh_data_list:
# If no PSSH data in cache, try to extract from manifest
if manifest_url:
try:
pssh_data_list = manager.drm_ops._extract_pssh_from_manifest(
manifest_url
)
# Cache the result
if pssh_data_list:
manager.drm_ops.pssh_cache.set(cache_key, pssh_data_list)
except Exception as extract_err:
logger.warning(
f"Failed to extract PSSH from manifest: {extract_err}"
)
pssh_data_list = []
else:
pssh_data_list = []
# Convert PSSH data to dictionary format
pssh_list = []
for pssh_data in pssh_data_list:
pssh_dict = {
"system_id": pssh_data.system_id,
"drm_system": (
pssh_data.drm_system.value if pssh_data.drm_system else None
),
"pssh_box": pssh_data.pssh_box if pssh_data.pssh_box else None,
"key_ids": pssh_data.key_ids if pssh_data.key_ids else [],
"source": pssh_data.source,
}
# Add human-readable system name
if pssh_data.drm_system:
pssh_dict["drm_system_name"] = {
"com.widevine.alpha": "Widevine",
"com.microsoft.playready": "PlayReady",
"com.apple.fps": "FairPlay",
"org.w3.clearkey": "ClearKey",
"com.huawei.wiseplay": "Wiseplay",
}.get(pssh_data.drm_system.value, pssh_data.drm_system.value)
# Remove None values for cleaner response
pssh_dict = {k: v for k, v in pssh_dict.items() if v is not None}
pssh_list.append(pssh_dict)
response.status = 200
return {
"provider": provider,
"channel_id": channel_id,
"manifest_url": manifest_url,
"pssh_data": pssh_list,
"count": len(pssh_list),
"cached": was_cached_after, # Whether data came from cache AFTER the operation
"was_cached_before": was_cached_before, # Whether data was in cache BEFORE the operation
"cache_ttl_seconds": manager.drm_ops.pssh_cache.ttl,
}
except Exception as e:
# Catch-all for unexpected errors
logger.error(f"Unexpected error in get_channel_pssh: {e}")
logger.error(traceback.format_exc())
response.status = 500
return {
"error": "Internal server error",
"message": str(e),
"provider": provider,
"channel_id": channel_id,
"traceback": (
traceback.format_exc() if app.config.get("debug") else None
),
}
@app.route(
"/api/providers/<provider>/channels/<channel_id>/pssh/refresh",
method="POST",
)
def refresh_channel_pssh(provider, channel_id):
"""
Force refresh PSSH data for a channel (clears cache and re-extracts).
This is useful when:
- Keys have been rotated
- Manifest structure has changed
- Previous extraction failed
"""
try:
# Parse query parameters
country = request.params.get("country")
# Generate cache key
cache_key = f"{provider}:{channel_id}"
# Check if entry exists in cache before clearing
was_cached = manager.drm_ops.pssh_cache.get(cache_key) is not None
# Clear the specific cache entry
with manager.drm_ops.pssh_cache.lock:
if cache_key in manager.drm_ops.pssh_cache.cache:
del manager.drm_ops.pssh_cache.cache[cache_key]
if was_cached:
logger.info(f"Cleared cache for {cache_key}")
else:
logger.info(f"No cache entry found for {cache_key}")
# Now extract fresh data by calling get_channel_drm_configs
# This will force a fresh extraction since cache was cleared
try:
drm_configs = manager.drm_ops.get_channel_drm_configs(
provider_name=provider, channel_id=channel_id, country=country
)
except ValueError as e:
response.status = 404
return {
"error": "Provider not found",
"message": str(e),
"provider": provider,
}
except Exception as e:
response.status = 500
return {
"error": "Failed to refresh DRM configs",
"message": str(e),
"provider": provider,
"channel_id": channel_id,
}
# Get the newly cached PSSH data
pssh_data_list = manager.drm_ops.pssh_cache.get(cache_key)
now_cached = pssh_data_list is not None
# Get manifest URL for reference
try:
manifest_url = manager.get_channel_manifest(
provider_name=provider, channel_id=channel_id, country=country
)
except Exception as e:
manifest_url = None
response.status = 200
return {
"message": "PSSH data refreshed successfully",
"provider": provider,
"channel_id": channel_id,
"manifest_url": manifest_url,
"pssh_count": len(pssh_data_list) if pssh_data_list else 0,
"was_cached": was_cached,
"now_cached": now_cached,
"extraction_successful": (
len(pssh_data_list) > 0 if pssh_data_list else False
),
}
except Exception as e:
logger.error(f"Error refreshing PSSH: {e}")
response.status = 500
return {"error": "Failed to refresh PSSH data", "message": str(e)}
routes/epg.py
+340
View File
@@ -0,0 +1,340 @@
#!/usr/bin/env python3
"""
EPG-related route handlers
"""
from bottle import request, response
from streaming_providers.base.utils import logger
def setup_epg_routes(app, manager, service):
"""Setup EPG-related routes"""
@app.route("/api/epg/status", method="GET")
def get_epg_status():
"""Get EPG configuration and cache status"""
try:
result = {
"configured": bool(service.epg_url)
and service.epg_url != "https://example.com/epg.xml.gz",
"epg_url": service.epg_url if service.epg_url else "Not configured",
"cache_valid": False,
"cache_path": None,
"channel_count": 0,
"environment_used": False,
}
# Check if we used the environment variable
import os
env_url = os.environ.get("ULTIMATE_EPG_URL")
if env_url and env_url == service.epg_url:
result["environment_used"] = True
if (
result["configured"]
and hasattr(service, "epg_manager")
and service.epg_manager
):
try:
cache = service.epg_manager.cache
xml_path = cache.get_cached_file_path()
if xml_path:
result["cache_valid"] = True
result["cache_path"] = xml_path
# Try to count channels
import gzip
import xml.etree.ElementTree as ET
def open_xml_file(file_path):
if file_path.endswith(".gz"):
return gzip.open(file_path, "rt", encoding="utf-8")
else:
return open(file_path, "r", encoding="utf-8")
channel_ids = set()
try:
with open_xml_file(xml_path) as xml_file:
context = ET.iterparse(xml_file, events=("start",))
for event, elem in context:
if elem.tag == "channel":
channel_id = elem.get("id")
if channel_id:
channel_ids.add(channel_id)
elem.clear()
result["channel_count"] = len(channel_ids)
except Exception as parse_err:
result["parse_error"] = str(parse_err)
except Exception as cache_err:
result["cache_error"] = str(cache_err)
else:
result["hint"] = "Please configure EPG URL in Advanced settings"
return result
except Exception as e:
logger.error(f"Error getting EPG status: {e}")
response.status = 500
return {"error": str(e)}
@app.route("/api/epg/xmltv-channels", method="GET")
def get_epg_xmltv_channels():
"""Get all unique channel IDs from EPG XML file with display names"""
try:
# Check if EPG manager is available
if not hasattr(service, "epg_manager") or not service.epg_manager:
response.status = 404
return {"error": "EPG module not available"}
# Check if we have a valid EPG URL configured
if (
not service.epg_url
or service.epg_url == "https://example.com/epg.xml.gz"
):
response.status = 400
return {
"error": "EPG URL not configured",
"hint": "Please configure a valid EPG URL in Advanced settings",
"current_url": service.epg_url,
}
# Get the cache manager from EPG manager
cache = service.epg_manager.cache
logger.info(f"EPG Channels: Using URL: {service.epg_url}")
# This will download if not cached, or return cached path
xml_path = cache.get_or_download(service.epg_url)
if not xml_path:
response.status = 404
return {
"error": f"EPG file not available from {service.epg_url}",
"details": "Failed to download or cache EPG file.",
"hint": "Check if the URL is accessible and contains valid XMLTV data.",
}
# Parse XML to get channel IDs and display names
import gzip
import os
import xml.etree.ElementTree as ET
if not os.path.exists(xml_path):
response.status = 404
return {"error": f"EPG file does not exist at path: {xml_path}"}
channel_ids = []
channel_map = {} # Map of id -> display name
def open_xml_file(file_path):
if file_path.endswith(".gz"):
return gzip.open(file_path, "rt", encoding="utf-8")
else:
return open(file_path, "r", encoding="utf-8")
logger.info(f"Parsing EPG file: {xml_path}")
file_size = os.path.getsize(xml_path)
logger.info(f"EPG file size: {file_size} bytes")
with open_xml_file(xml_path) as xml_file:
# Use iterparse for memory efficiency
context = ET.iterparse(xml_file, events=("start", "end"))
current_channel_id = None
current_display_names = []
for event, elem in context:
if event == "start" and elem.tag == "channel":
current_channel_id = elem.get("id")
current_display_names = []
elif event == "end" and elem.tag == "display-name":
if current_channel_id and elem.text:
current_display_names.append(elem.text.strip())
elif event == "end" and elem.tag == "channel":
if current_channel_id:
channel_ids.append(current_channel_id)
# Use the first display name as the primary name
if current_display_names:
channel_map[current_channel_id] = current_display_names[
0
]
else:
channel_map[current_channel_id] = current_channel_id
current_channel_id = None
# Clear element to save memory
if event == "end":
elem.clear()
logger.info(f"Found {len(channel_ids)} channels in EPG")
# Sort channels for consistent output
sorted_channels = sorted(channel_ids)
return {
"channels": sorted_channels,
"channel_map": channel_map, # NEW: Map of id -> display name
"count": len(sorted_channels),
"source_url": service.epg_url,
"cache_path": xml_path,
"cache_size_bytes": file_size,
}
except ET.ParseError as parse_err:
logger.error(f"XML parse error in EPG file: {parse_err}")
response.status = 500
return {
"error": f"Failed to parse EPG XML file: {str(parse_err)}",
"hint": "The EPG file may be malformed or not valid XMLTV format.",
}
except Exception as e:
logger.error(f"Error getting EPG channels: {e}", exc_info=True)
response.status = 500
return {"error": f"Failed to process EPG file: {str(e)}"}
@app.route("/api/providers/<provider>/epg-mapping", method="GET")
def get_epg_mapping(provider):
"""Get current EPG mapping for a provider"""
try:
from streaming_providers.base.utils.vfs import VFS
except ImportError:
# If VFS is not available, return empty mapping
return {"provider": provider, "mapping": {}, "exists": False}
try:
mapping_file = f"{provider}_epg_mapping.json"
vfs = VFS(addon_subdir="")
if vfs.exists(mapping_file):
mapping_data = vfs.read_json(mapping_file)
if mapping_data:
# The file structure is:
# {
# "_provider_name": "...",
# "channel_id": {"epg_id": "...", "name": "..."}
# }
# Extract mapping (skip internal fields starting with _)
internal_fields = [
"_provider_name",
"_created_at",
"_updated_at",
"_version",
]
actual_mapping = {
k: v
for k, v in mapping_data.items()
if k not in internal_fields
}
logger.info(
f"Loaded EPG mapping for {provider}: {len(actual_mapping)} channels"
)
logger.debug(f"Sample mappings: {list(actual_mapping.items())[:3]}")
return {
"provider": provider,
"mapping": actual_mapping,
"exists": True,
}
# Return empty mapping if file doesn't exist
logger.info(f"No EPG mapping file found for {provider}")
return {"provider": provider, "mapping": {}, "exists": False}
except Exception as e:
logger.error(
f"Error getting EPG mapping for {provider}: {e}", exc_info=True
)
response.status = 500
return {"error": f"Failed to load mapping: {str(e)}"}
@app.route("/api/providers/<provider>/epg-mapping", method="POST")
def save_epg_mapping(provider):
"""Save EPG mapping for a provider"""
try:
from streaming_providers.base.utils.vfs import VFS
except ImportError:
response.status = 500
return {"error": "VFS module not available"}
try:
# Get JSON data from request body using Bottle's request object
try:
mapping_data = request.json.get("mapping", {}) if request.json else {}
except Exception as json_err:
logger.error(f"Invalid JSON in request body: {json_err}")
response.status = 400
return {"error": "Invalid JSON in request body"}
# Get provider label if available
provider_label = provider
try:
provider_instance = manager.get_provider(provider)
if provider_instance:
provider_label = getattr(
provider_instance, "provider_label", provider
)
except:
pass
# Build the file structure:
# {
# "_provider_name": "Provider Label",
# "channel_id": {"epg_id": "...", "name": "..."}
# }
full_mapping = {"_provider_name": provider_label}
# Add each mapping entry
for channel_id, mapping_value in mapping_data.items():
if isinstance(mapping_value, dict):
# Already has structure {"epg_id": "...", "name": "..."}
full_mapping[channel_id] = mapping_value
elif isinstance(mapping_value, str):
# Simple string, convert to object
full_mapping[channel_id] = {
"epg_id": mapping_value,
"name": "", # Name not provided
}
# Save to file
vfs = VFS(addon_subdir="")
mapping_file = f"{provider}_epg_mapping.json"
success = vfs.write_json(mapping_file, full_mapping)
if success:
logger.info(
f"Saved EPG mapping for {provider}: {len(mapping_data)} channels"
)
# Clear mapping cache if it exists
try:
from streaming_providers.base.epg.epg_mapping import EPGMapping
mapping_manager = EPGMapping()
mapping_manager.reload_mapping(provider)
logger.info(f"Reloaded EPG mapping cache for {provider}")
except ImportError:
logger.debug("EPGMapping not available for cache reload")
pass
except Exception as reload_err:
logger.warning(f"Could not reload mapping cache: {reload_err}")
return {
"success": True,
"message": f"Mapping saved for {provider}",
"channels_mapped": len(mapping_data),
}
else:
response.status = 500
return {"error": "Failed to save mapping file"}
except Exception as e:
logger.error(f"Error saving EPG mapping for {provider}: {e}", exc_info=True)
response.status = 500
return {"error": f"Failed to save mapping: {str(e)}"}
routes/m3u.py
+350
View File
@@ -0,0 +1,350 @@
#!/usr/bin/env python3
"""
M3U playlist route handlers
"""
from bottle import request, response
from streaming_providers.base.utils import logger
def setup_m3u_routes(app, manager, service):
"""Setup M3U playlist-related routes"""
@app.route("/api/m3u")
def get_m3u_all():
"""
Generates M3U playlist for all configured providers.
Returns cached version if available, otherwise generates new one.
Example: http://localhost:7777/api/m3u
"""
try:
cache_file = "playlist.m3u"
# Try to read cached file
cached_content = service.vfs.read_text(cache_file)
if cached_content:
logger.info("Serving cached M3U playlist for all providers")
response.content_type = "audio/x-mpegurl; charset=utf-8"
response.headers["Content-Disposition"] = (
'attachment; filename="playlist.m3u8"'
)
return cached_content
# Cache doesn't exist or is corrupt, generate new M3U
logger.info(
"No valid cache found, generating M3U playlist for all providers"
)
return service._generate_m3u_all(save_to_cache=True)
except Exception as api_err:
logger.error(f"API Error in /api/m3u: {str(api_err)}")
response.status = 500
return {"error": f"Internal server error: {str(api_err)}"}
@app.route("/api/m3u/generate")
def generate_m3u_all():
"""
Forces regeneration of M3U playlist for all providers and saves to cache.
Example: http://localhost:7777/api/m3u/generate
"""
try:
logger.info("Force generating M3U playlist for all providers")
return service._generate_m3u_all(save_to_cache=True)
except Exception as api_err:
logger.error(f"API Error in /api/m3u/generate: {str(api_err)}")
response.status = 500
return {"error": f"Internal server error: {str(api_err)}"}
@app.route("/api/providers/<provider>/m3u")
def get_m3u_provider(provider):
"""
Generates M3U playlist for a specific provider.
Returns cached version if available, otherwise generates new one.
Example: http://localhost:7777/api/providers/rtlplus/m3u
"""
try:
cache_file = f"{provider}.m3u"
# Try to read cached file
cached_content = service.vfs.read_text(cache_file)
if cached_content:
logger.info(f"Serving cached M3U playlist for provider '{provider}'")
response.content_type = "audio/x-mpegurl; charset=utf-8"
response.headers["Content-Disposition"] = (
f'attachment; filename="{provider}_playlist.m3u8"'
)
return cached_content
# Cache doesn't exist or is corrupt, generate new M3U
logger.info(
f"No valid cache found, generating M3U playlist for provider '{provider}'"
)
return service._generate_m3u_provider(provider, save_to_cache=True)
except ValueError as val_err:
logger.error(f"API Error in /api/providers/{provider}/m3u: {str(val_err)}")
response.status = 404
return {"error": str(val_err)}
except Exception as api_err:
logger.error(f"API Error in /api/providers/{provider}/m3u: {str(api_err)}")
response.status = 500
return {"error": f"Internal server error: {str(api_err)}"}
@app.route("/api/providers/<provider>/m3u/generate")
def generate_m3u_provider(provider):
"""
Forces regeneration of M3U playlist for a specific provider and saves to cache.
Example: http://localhost:7777/api/providers/rtlplus/m3u/generate
"""
try:
logger.info(f"Force generating M3U playlist for provider '{provider}'")
return service._generate_m3u_provider(provider, save_to_cache=True)
except ValueError as val_err:
logger.error(
f"API Error in /api/providers/{provider}/m3u/generate: {str(val_err)}"
)
response.status = 404
return {"error": str(val_err)}
except Exception as api_err:
logger.error(
f"API Error in /api/providers/{provider}/m3u/generate: {str(api_err)}"
)
response.status = 500
return {"error": f"Internal server error: {str(api_err)}"}
@app.route("/api/m3u/decrypted")
def get_m3u_decrypted():
"""
Generates decrypted M3U playlist for all configured providers.
Only includes channels with ClearKey DRM or unencrypted channels.
Returns cached version if available, otherwise generates new one.
Example: http://localhost:7777/api/m3u/decrypted
"""
try:
cache_file = "playlist_decrypted.m3u"
# Try to read cached file
cached_content = service.vfs.read_text(cache_file)
if cached_content:
logger.info("Serving cached decrypted M3U playlist for all providers")
response.content_type = "audio/x-mpegurl; charset=utf-8"
response.headers["Content-Disposition"] = (
'attachment; filename="playlist_decrypted.m3u8"'
)
return cached_content
# Cache doesn't exist, generate new M3U
logger.info(
"No valid cache found, generating decrypted M3U playlist for all providers"
)
return service._generate_m3u_decrypted_all(save_to_cache=True)
except Exception as api_err:
logger.error(f"API Error in /api/m3u/decrypted: {str(api_err)}")
response.status = 500
return {"error": f"Internal server error: {str(api_err)}"}
@app.route("/api/m3u/decrypted/generate")
def generate_m3u_decrypted():
"""
Forces regeneration of decrypted M3U playlist for all providers and saves to cache.
Example: http://localhost:7777/api/m3u/decrypted/generate
"""
try:
logger.info("Force generating decrypted M3U playlist for all providers")
return service._generate_m3u_decrypted_all(save_to_cache=True)
except Exception as api_err:
logger.error(f"API Error in /api/m3u/decrypted/generate: {str(api_err)}")
response.status = 500
return {"error": f"Internal server error: {str(api_err)}"}
@app.route("/api/providers/<provider>/m3u/decrypted")
def get_m3u_decrypted_provider(provider):
"""
Generates decrypted M3U playlist for a specific provider.
Only includes channels with ClearKey DRM or unencrypted channels.
Returns cached version if available, otherwise generates new one.
Example: http://localhost:7777/api/providers/rtlplus/m3u/decrypted
"""
try:
cache_file = f"{provider}_decrypted.m3u"
# Try to read cached file
cached_content = service.vfs.read_text(cache_file)
if cached_content:
logger.info(
f"Serving cached decrypted M3U playlist for provider '{provider}'"
)
response.content_type = "audio/x-mpegurl; charset=utf-8"
response.headers["Content-Disposition"] = (
f'attachment; filename="{provider}_decrypted_playlist.m3u8"'
)
return cached_content
# Cache doesn't exist, generate new M3U
logger.info(
f"No valid cache found, generating decrypted M3U playlist for provider '{provider}'"
)
return service._generate_m3u_decrypted_provider(
provider, save_to_cache=True
)
except ValueError as val_err:
logger.error(
f"API Error in /api/providers/{provider}/m3u/decrypted: {str(val_err)}"
)
response.status = 404
return {"error": str(val_err)}
except Exception as api_err:
logger.error(
f"API Error in /api/providers/{provider}/m3u/decrypted: {str(api_err)}"
)
response.status = 500
return {"error": f"Internal server error: {str(api_err)}"}
@app.route("/api/providers/<provider>/m3u/decrypted/generate")
def generate_m3u_decrypted_provider(provider):
"""
Forces regeneration of decrypted M3U playlist for a specific provider and saves to cache.
Example: http://localhost:7777/api/providers/rtlplus/m3u/decrypted/generate
"""
try:
logger.info(
f"Force generating decrypted M3U playlist for provider '{provider}'"
)
return service._generate_m3u_decrypted_provider(
provider, save_to_cache=True
)
except ValueError as val_err:
logger.error(
f"API Error in /api/providers/{provider}/m3u/decrypted/generate: {str(val_err)}"
)
response.status = 404
return {"error": str(val_err)}
except Exception as api_err:
logger.error(
f"API Error in /api/providers/{provider}/m3u/decrypted/generate: {str(api_err)}"
)
response.status = 500
return {"error": f"Internal server error: {str(api_err)}"}
@app.route("/api/m3u/subscribed")
def get_m3u_subscribed():
"""
Generate M3U playlist with only subscribed channels.
Note: This uses the same caching mechanism as regular M3U,
but with '_subscribed' suffix in cache filename.
"""
try:
cache_file = "playlist_subscribed.m3u"
# Try cached version
cached_content = service.vfs.read_text(cache_file)
if cached_content:
logger.info("Serving cached subscribed M3U playlist")
response.content_type = "audio/x-mpegurl; charset=utf-8"
response.headers["Content-Disposition"] = (
'attachment; filename="playlist_subscribed.m3u8"'
)
return cached_content
# Generate new M3U with subscribed channels
base_url = f"{request.urlparts.scheme}://{request.urlparts.netloc}"
m3u_content = "#EXTM3U\n"
provider_list = manager.list_providers()
for provider_name in provider_list:
try:
channels = manager.get_subscribed_channels(provider_name)
for channel in channels:
# Generate M3U entry for each subscribed channel
channel_id = channel.channel_id
channel_name = channel.name
channel_logo = channel.logo_url or ""
# Get provider label
try:
provider_instance = manager.get_provider(provider_name)
provider_label = getattr(
provider_instance, "provider_label", provider_name
)
except:
provider_label = provider_name
# Build stream URL
stream_url = f"{base_url}/api/providers/{provider_name}/channels/{channel_id}/stream"
# Add M3U entry
m3u_content += f'#EXTINF:-1 tvg-id="{channel_id}" tvg-logo="{channel_logo}" group-title="{provider_label}",{channel_name}\n'
# Add DRM directives if available
try:
drm_configs = manager.get_channel_drm_configs(
provider_name, channel_id
)
if drm_configs:
drm_directives = service._generate_drm_directives(
drm_configs
)
m3u_content += drm_directives
except Exception as drm_err:
logger.debug(
f"Could not get DRM for {provider_name}/{channel_id}: {drm_err}"
)
m3u_content += f"{stream_url}\n"
except Exception as provider_err:
logger.warning(
f"Failed to process subscribed channels for '{provider_name}': {str(provider_err)}"
)
continue
# Cache the result
if service.vfs.write_text(cache_file, m3u_content):
logger.info(f"Subscribed M3U playlist cached to {cache_file}")
response.content_type = "audio/x-mpegurl; charset=utf-8"
response.headers["Content-Disposition"] = (
'attachment; filename="playlist_subscribed.m3u8"'
)
return m3u_content
except Exception as api_err:
logger.error(f"API Error in /api/m3u/subscribed: {str(api_err)}")
response.status = 500
return {"error": f"Internal server error: {str(api_err)}"}
@app.route("/api/m3u/subscribed/generate")
def generate_m3u_subscribed():
"""Force regenerate subscribed M3U playlist"""
try:
# Clear cache and regenerate
cache_file = "playlist_subscribed.m3u"
service.vfs.delete(cache_file) # Delete if exists
# Call the subscribed M3U endpoint which will regenerate
return get_m3u_subscribed()
except Exception as api_err:
logger.error(f"API Error in /api/m3u/subscribed/generate: {str(api_err)}")
response.status = 500
return {"error": f"Internal server error: {str(api_err)}"}
routes/providers.py
+812
View File
@@ -0,0 +1,812 @@
#!/usr/bin/env python3
"""
Provider-related route handlers
"""
from bottle import request, response
from streaming_providers.base.utils import logger
def setup_provider_routes(app, manager, service):
"""Setup provider-related routes"""
@app.route("/api/providers")
def list_providers():
try:
# Get metadata for ALL providers (enabled + disabled)
all_metadata = manager.get_all_providers_metadata()
# For backward compatibility, also get details for enabled providers
enabled_providers = []
for metadata in all_metadata:
if metadata["enabled"] and metadata["instance_ready"]:
provider_instance = manager.get_provider(metadata["name"])
if provider_instance:
# Get detailed auth info from instance
supported_auth_types = getattr(
provider_instance, "supported_auth_types", []
)
preferred_auth_type = getattr(
provider_instance, "preferred_auth_type", "unknown"
)
requires_stored_credentials = getattr(
provider_instance, "requires_stored_credentials", False
)
# Check specific auth type needs
needs_user_creds = "user_credentials" in supported_auth_types
needs_client_creds = (
"client_credentials" in supported_auth_types
)
is_network_based = "network_based" in supported_auth_types
is_anonymous = "anonymous" in supported_auth_types
uses_device_reg = "device_registration" in supported_auth_types
uses_embedded = "embedded_client" in supported_auth_types
provider_details = {
"name": metadata["name"],
"label": metadata["label"],
"logo": metadata["logo"],
"country": metadata["country"],
# Core authentication properties
"auth": {
"supported_auth_types": supported_auth_types,
"preferred_auth_type": preferred_auth_type,
"requires_stored_credentials": requires_stored_credentials,
# Specific auth type flags for easy UI decisions
"needs_user_credentials": needs_user_creds,
"needs_client_credentials": needs_client_creds,
"is_network_based": is_network_based,
"is_anonymous": is_anonymous,
"uses_device_registration": uses_device_reg,
"uses_embedded_client": uses_embedded,
# Derived summary for UI
"needs_user_input": needs_user_creds or uses_device_reg,
"needs_configuration": needs_user_creds
or needs_client_creds,
"is_automatic": is_network_based
or is_anonymous
or uses_embedded,
},
# Token properties
"primary_token_scope": getattr(
provider_instance, "primary_token_scope", None
),
"token_scopes": getattr(
provider_instance, "token_scopes", []
),
# Metadata fields
"enabled": metadata["enabled"],
"instance_ready": metadata["instance_ready"],
"requires_credentials": metadata["requires_credentials"],
}
enabled_providers.append(provider_details)
return {
"providers": enabled_providers,
"all_providers": all_metadata, # NEW: Include all providers metadata
"default_country": service.default_country,
}
except Exception as api_err:
logger.error(f"API Error in /api/providers: {str(api_err)}")
response.status = 500
return {"error": str(api_err)}
@app.route("/api/providers/<provider>/channels")
def get_channels(provider):
try:
channels = manager.get_channels(
provider_name=provider,
fetch_manifests=request.query.get("fetch_manifests", "false").lower()
== "true",
country=request.query.get("country"),
)
# Get provider instance to check catchup support
provider_instance = manager.get_provider(provider)
provider_catchup_hours = getattr(
provider_instance, "catchup_window", 0
) # CHANGE
# Build channel list with catchup info
channels_data = []
for c in channels:
channel_dict = c.to_dict()
# Add catchup hours - use channel-specific if available, else provider default
if hasattr(c, "catchup_hours"): # CHANGE
channel_dict["CatchupHours"] = c.catchup_hours # CHANGE
else:
channel_dict["CatchupHours"] = provider_catchup_hours # CHANGE
channels_data.append(channel_dict)
return {
"provider": provider,
"country": provider_instance.country if provider_instance else "DE",
"catchup_window_hours": provider_catchup_hours, # CHANGE
"channels": channels_data,
}
except Exception as api_err:
logger.error(f"API Error in /api/providers/{provider}: {str(api_err)}")
response.status = 500
return {"error": str(api_err)}
@app.route("/api/providers/<provider>/auth/status")
def get_provider_auth_status(provider):
"""Get authentication status from provider itself"""
try:
# Get provider instance
provider_instance = manager.get_provider(provider)
if not provider_instance:
response.status = 404
return {"error": f"Provider {provider} not found"}
# Get SettingsManager
settings_manager = service._get_settings_manager()
if not settings_manager:
response.status = 500
return {"error": "Settings manager not available"}
# Import and use new auth system
from streaming_providers.providers.auth_context import AuthContext
try:
auth_context = AuthContext(settings_manager)
auth_status = provider_instance.get_auth_status(auth_context)
return auth_status.to_dict()
except AttributeError as attr_err:
logger.error(
f"Provider {provider} missing required auth property: {attr_err}"
)
response.status = 501 # Not Implemented
return {
"error": f"Provider {provider} does not fully implement auth status",
"details": str(attr_err),
}
except Exception as e:
logger.error(f"Error getting auth status: {e}", exc_info=True)
response.status = 500
return {"error": str(e)}
except ImportError as import_error:
# This happens during development if modules not created yet
logger.warning(f"Auth modules not available: {import_error}")
return {
"provider": provider,
"auth_state": "not_implemented",
"is_ready": False,
"message": "New auth system in development",
}
except Exception as e:
logger.error(
f"Error getting auth status for {provider}: {e}", exc_info=True
)
response.status = 500
return {"error": f"Internal server error: {str(e)}"}
@app.route("/api/providers/<provider>/credentials", method="GET")
def get_provider_credentials(provider):
"""
GET: Retrieve current credentials (masked for security)
Example: GET /api/providers/joyn/credentials
Returns: {
"has_credentials": true,
"credential_type": "user_password",
"username_masked": "us***@example.com",
"username": "user@example.com" # Note: only included for pre-fill with user consent
}
"""
try:
settings_manager = service._get_settings_manager()
# Parse provider and country
provider_name, country = settings_manager.parse_provider_country(provider)
# Get credentials
credentials = settings_manager.get_provider_credentials(
provider_name, country
)
response_data = {
"provider": provider,
"has_credentials": credentials is not None,
"credential_type": None,
"username_masked": None,
"username": None, # We'll include this only if user explicitly allows
}
if credentials:
response_data["credential_type"] = credentials.credential_type
response_data["is_valid"] = credentials.validate()
# Get username if it exists (for user_password credentials)
if hasattr(credentials, "username") and credentials.username:
username = credentials.username
# Create masked version for display
if "@" in username: # Email address
parts = username.split("@")
if len(parts[0]) > 2:
masked = parts[0][:2] + "***@" + parts[1]
else:
masked = "***@" + parts[1]
else: # Username
if len(username) > 4:
masked = username[:2] + "***" + username[-2:]
else:
masked = "***"
response_data["username_masked"] = masked
# For pre-filling forms (security consideration - you can omit this)
# Only include if you trust your frontend and have HTTPS
response_data["username"] = username
# Log for debugging (remove in production)
logger.debug(
f"GET credentials for {provider}: type={credentials.credential_type}, has_username={hasattr(credentials, 'username')}"
)
return response_data
except Exception as e:
logger.error(f"GET credentials error for {provider}: {e}", exc_info=True)
response.status = 500
return {"error": f"Internal server error: {str(e)}"}
@app.route("/api/providers/<provider>/credentials", method="POST")
def save_provider_credentials(provider):
"""
Save credentials for a provider via API
Accepts JSON body with credentials:
- User/password: {"username": "...", "password": "..."}
- For updates: {"password": "..."} (keep existing username)
"""
try:
# Parse JSON body
try:
credentials_data = request.json
logger.debug(
f"Received credentials data for {provider}: {credentials_data}"
)
except Exception as json_err:
logger.error(f"Invalid JSON in request body: {json_err}")
response.status = 400
return {"error": "Invalid JSON in request body"}
if not credentials_data:
logger.error("No credentials data provided")
response.status = 400
return {"error": "Request body must contain credentials data"}
# Validate it's a dictionary
if not isinstance(credentials_data, dict):
logger.error(
f"Credentials data is not a dict: {type(credentials_data)}"
)
response.status = 400
return {"error": "Credentials data must be a JSON object"}
# Get settings manager
settings_manager = service._get_settings_manager()
# Parse provider and country
provider_name, country = settings_manager.parse_provider_country(provider)
# Check if we have existing credentials (for partial updates)
existing_credentials = settings_manager.get_provider_credentials(
provider_name, country
)
if existing_credentials and "username" not in credentials_data:
# Partial update - keep existing username, only update password
if hasattr(existing_credentials, "username"):
credentials_data["username"] = existing_credentials.username
else:
response.status = 400
return {
"error": "Cannot update - existing credentials do not have username"
}
# Save credentials
success, message = settings_manager.save_provider_credentials_from_api(
provider, credentials_data
)
logger.info(
f"Save result for {provider}: success={success}, message={message}"
)
if success:
# Reinitialize provider to pick up new credentials
reinit_success = manager.reinitialize_provider(provider)
if not reinit_success:
logger.warning(
f"Failed to reinitialize provider '{provider}' after credential change"
)
response.status = 200
response.content_type = "application/json; charset=utf-8"
return {
"success": True,
"provider": provider,
"message": message,
"action": "updated" if existing_credentials else "created",
"reinitialized": reinit_success,
}
else:
# Determine appropriate status code
if "not registered" in message.lower():
response.status = 404
elif (
"invalid" in message.lower()
or "validation failed" in message.lower()
):
response.status = 400
else:
response.status = 500
return {"error": message}
except Exception as api_err:
logger.error(
f"API Error in POST /api/providers/{provider}/credentials: {str(api_err)}",
exc_info=True,
)
response.status = 500
return {"error": f"Internal server error: {str(api_err)}"}
@app.route("/api/providers/<provider>/credentials", method="DELETE")
def delete_provider_credentials(provider):
"""
Delete credentials for a provider via API
Example: DELETE /api/providers/joyn_de/credentials
"""
try:
settings_manager = service._get_settings_manager()
success, message = settings_manager.delete_provider_credentials_from_api(
provider
)
if success:
# Reinitialize provider to clear any cached authentication
reinit_success = manager.reinitialize_provider(provider)
if not reinit_success:
logger.warning(
f"Failed to reinitialize provider '{provider}' after credential deletion"
)
response.status = 200
response.content_type = "application/json; charset=utf-8"
return {
"success": True,
"provider": provider,
"message": message,
"reinitialized": reinit_success,
}
else:
# Determine appropriate status code
if "not registered" in message.lower():
response.status = 404
else:
response.status = 500
return {"error": message}
except Exception as api_err:
logger.error(
f"API Error in DELETE /api/providers/{provider}/credentials: {str(api_err)}"
)
response.status = 500
return {"error": f"Internal server error: {str(api_err)}"}
@app.route("/api/providers/<provider>/proxy", method="GET")
def get_provider_proxy(provider):
"""
Get current proxy configuration for a provider
Example: GET /api/providers/joyn/proxy
"""
try:
settings_manager = service._get_settings_manager()
# Parse provider and country
provider_name, country = settings_manager.parse_provider_country(provider)
# Get proxy config
proxy_config = settings_manager.get_provider_proxy(provider_name, country)
if proxy_config:
return {
"success": True,
"provider": provider,
"proxy_config": (
proxy_config.to_dict()
if hasattr(proxy_config, "to_dict")
else proxy_config
),
}
else:
return {
"success": True,
"provider": provider,
"proxy_config": None,
"message": "No proxy configuration found",
}
except Exception as api_err:
logger.error(
f"API Error in GET /api/providers/{provider}/proxy: {str(api_err)}"
)
response.status = 500
return {"error": f"Internal server error: {str(api_err)}"}
@app.route("/api/providers/<provider>/proxy", method="POST")
def save_provider_proxy(provider):
"""
Save proxy configuration for a provider via API
Accepts JSON body with proxy configuration:
Required: {"host": "proxy.example.com", "port": 8080}
Optional: {
"proxy_type": "http", # http, https, socks4, socks5
"username": "proxyuser",
"password": "proxypass",
"timeout": 30,
"verify_ssl": true,
"scope": {
"api_calls": true,
"authentication": true,
"manifests": true,
"license": true,
"all": true
}
}
Example: POST /api/providers/joyn_de/proxy
Body: {"host": "proxy.example.com", "port": 8080}
"""
try:
# Parse JSON body
try:
proxy_data = request.json
except Exception as json_err:
logger.error(f"Invalid JSON in request body: {json_err}")
response.status = 400
return {"error": "Invalid JSON in request body"}
if not proxy_data:
response.status = 400
return {"error": "Request body must contain proxy configuration"}
# Validate it's a dictionary
if not isinstance(proxy_data, dict):
response.status = 400
return {"error": "Proxy data must be a JSON object"}
settings_manager = service._get_settings_manager()
success, message = settings_manager.save_provider_proxy_from_api(
provider, proxy_data
)
if success:
# Reinitialize provider to pick up new proxy configuration
reinit_success = manager.reinitialize_provider(provider)
if not reinit_success:
logger.warning(
f"Failed to reinitialize provider '{provider}' after proxy change"
)
response.status = 200
response.content_type = "application/json; charset=utf-8"
return {
"success": True,
"provider": provider,
"message": message,
"reinitialized": reinit_success,
}
else:
# Determine appropriate status code
if "not registered" in message.lower():
response.status = 404
elif (
"invalid" in message.lower()
or "validation failed" in message.lower()
):
response.status = 400
else:
response.status = 500
return {"error": message}
except Exception as api_err:
logger.error(
f"API Error in POST /api/providers/{provider}/proxy: {str(api_err)}"
)
response.status = 500
return {"error": f"Internal server error: {str(api_err)}"}
@app.route("/api/providers/<provider>/proxy", method="DELETE")
def delete_provider_proxy(provider):
"""
Delete proxy configuration for a provider via API
Example: DELETE /api/providers/joyn_de/proxy
"""
try:
settings_manager = service._get_settings_manager()
success, message = settings_manager.delete_provider_proxy_from_api(provider)
if success:
# Reinitialize provider to remove proxy configuration
reinit_success = manager.reinitialize_provider(provider)
if not reinit_success:
logger.warning(
f"Failed to reinitialize provider '{provider}' after proxy deletion"
)
response.status = 200
response.content_type = "application/json; charset=utf-8"
return {
"success": True,
"provider": provider,
"message": message,
"reinitialized": reinit_success,
}
else:
# Determine appropriate status code
if "not registered" in message.lower():
response.status = 404
else:
response.status = 500
return {"error": message}
except Exception as api_err:
logger.error(
f"API Error in DELETE /api/providers/{provider}/proxy: {str(api_err)}"
)
response.status = 500
return {"error": f"Internal server error: {str(api_err)}"}
@app.route("/api/providers/<provider>/reinitialize", method="POST")
def reinitialize_provider(provider):
"""
Manually reinitialize a provider (e.g., after external configuration changes)
Example: POST /api/providers/joyn_de/reinitialize
"""
try:
success = manager.reinitialize_provider(provider)
if success:
return {
"success": True,
"provider": provider,
"message": f"Provider {provider} reinitialized successfully",
}
else:
response.status = 500
return {
"success": False,
"provider": provider,
"message": f"Failed to reinitialize provider {provider}",
}
except Exception as e:
logger.error(f"Error reinitializing provider {provider}: {e}")
response.status = 500
return {"error": f"Internal server error: {str(e)}"}
@app.route("/api/providers/<provider>/subscription")
def get_provider_subscription(provider):
"""Get subscription status for a provider"""
try:
subscription = manager.get_subscription_status(provider)
if subscription:
return {"success": True, "subscription": subscription.to_dict()}
else:
return {
"success": True,
"subscription": None,
"message": "No subscription information available",
}
except ValueError as val_err:
response.status = 404
return {"error": str(val_err)}
except Exception as api_err:
logger.error(f"API Error in subscription endpoint: {str(api_err)}")
response.status = 500
return {"error": f"Internal server error: {str(api_err)}"}
@app.route("/api/providers/<provider>/channels/subscribed")
def get_subscribed_channels(provider):
"""Get channels the user is subscribed to for a provider"""
try:
channels = manager.get_subscribed_channels(provider)
# Get provider instance for additional info
provider_instance = manager.get_provider(provider)
provider_catchup_hours = (
getattr(provider_instance, "catchup_window", 0)
if provider_instance
else 0
)
channels_data = []
for c in channels:
channel_dict = c.to_dict()
# Add catchup hours
if hasattr(c, "catchup_hours"):
channel_dict["CatchupHours"] = c.catchup_hours
else:
channel_dict["CatchupHours"] = provider_catchup_hours
channels_data.append(channel_dict)
return {
"provider": provider,
"channels": channels_data,
"count": len(channels_data),
"is_filtered": True, # Indicates subscription filtering was applied
}
except ValueError as val_err:
response.status = 404
return {"error": str(val_err)}
except Exception as api_err:
logger.error(f"API Error in subscribed channels endpoint: {str(api_err)}")
response.status = 500
return {"error": f"Internal server error: {str(api_err)}"}
@app.route("/api/providers/<provider>/packages")
def get_provider_packages(provider):
"""Get available subscription packages for a provider"""
try:
packages = manager.get_available_packages(provider)
return {
"success": True,
"provider": provider,
"packages": [
{
"package_id": pkg.package_id,
"name": pkg.name,
"description": pkg.description,
"price_info": pkg.price_info,
"channel_count": pkg.channel_count,
"metadata": pkg.metadata,
}
for pkg in packages
],
"count": len(packages),
}
except ValueError as val_err:
response.status = 404
return {"error": str(val_err)}
except Exception as api_err:
logger.error(f"API Error in packages endpoint: {str(api_err)}")
response.status = 500
return {"error": f"Internal server error: {str(api_err)}"}
@app.route("/api/providers/channels/subscribed")
def get_all_subscribed_channels():
"""Get all subscribed channels across all providers"""
try:
all_subscribed = {}
provider_list = manager.list_providers()
for provider in provider_list:
try:
channels = manager.get_subscribed_channels(provider)
if channels:
all_subscribed[provider] = [c.to_dict() for c in channels]
except Exception as provider_err:
logger.warning(
f"Could not get subscribed channels for {provider}: {provider_err}"
)
continue
total_channels = sum(len(channels) for channels in all_subscribed.values())
return {
"success": True,
"providers": list(all_subscribed.keys()),
"channels_by_provider": all_subscribed,
"total_channels": total_channels,
"provider_count": len(all_subscribed),
}
except Exception as api_err:
logger.error(
f"API Error in all subscribed channels endpoint: {str(api_err)}"
)
response.status = 500
return {"error": f"Internal server error: {str(api_err)}"}
@app.route("/api/providers/<provider>/enabled", method="GET")
def get_provider_enabled(provider):
"""Get enabled status for specific provider"""
try:
# Validate provider exists
if not manager.get_provider(provider):
response.status = 404
return {"error": f"Provider {provider} not found"}
from streaming_providers.base.settings.provider_enable_manager import (
ProviderEnableManager,
)
enable_manager = ProviderEnableManager()
status = enable_manager.is_provider_enabled(provider)
source = enable_manager.get_enabled_source(provider)
# FIX: Convert enum to string value
source_value = source.value if hasattr(source, "value") else str(source)
return {
"success": True,
"provider": provider,
"enabled": status,
"source": source_value, # Now a string
"can_modify": source != "kodi",
}
except Exception as e:
logger.error(f"Error getting enabled status for {provider}: {e}")
response.status = 500
return {"error": str(e)}
@app.route("/api/providers/<provider>/enabled", method="POST")
def set_provider_enabled(provider):
"""Set enabled status for provider"""
try:
# Parse request
try:
data = request.json
if not data or "enabled" not in data:
response.status = 400
return {"error": 'Missing "enabled" field'}
enabled = bool(data["enabled"])
except ValueError:
response.status = 400
return {"error": "Invalid JSON"}
# Use the new manager method
success = manager.set_provider_enabled(provider, enabled)
if success:
# Get updated metadata
metadata = None
all_metadata = manager.get_all_providers_metadata()
for md in all_metadata:
if md["name"] == provider:
metadata = md
break
return {
"success": True,
"provider": provider,
"enabled": enabled,
"metadata": metadata,
"message": f'Provider {provider} {"enabled" if enabled else "disabled"}',
}
else:
response.status = 500
return {
"error": f'Failed to {"enable" if enabled else "disable"} provider {provider}'
}
except Exception as e:
logger.error(f"Error setting enabled status for {provider}: {e}")
response.status = 500
return {"error": str(e)}
routes/streams.py
+387
View File
@@ -0,0 +1,387 @@
#!/usr/bin/env python3
"""
Stream and manifest route handlers
"""
from datetime import datetime
from bottle import HTTPResponse, redirect, request, response
from streaming_providers.base.utils import logger
def setup_stream_routes(app, manager, service):
"""Setup stream and manifest-related routes"""
@app.route("/api/providers/<provider>/channels/<channel_id>/manifest")
def get_channel_manifest(provider, channel_id):
"""
Get channel manifest. Always returns JSON with manifest_url pointing to stream endpoint.
"""
try:
# Build the stream URL (which will handle both proxy and non-proxy)
base_url = f"{request.urlparts.scheme}://{request.urlparts.netloc}"
stream_url = (
f"{base_url}/api/providers/{provider}/channels/{channel_id}/stream"
)
# Add country parameter if provided
country = request.query.get("country")
if country:
stream_url += f"?country={country}"
return {
"provider": provider,
"channel_id": channel_id,
"manifest_url": stream_url, # Always point to /stream endpoint
}
except ValueError as val_err:
logger.error(
f"API Error in /api/providers/{provider}/channels/{channel_id}/manifest: {str(val_err)}"
)
response.status = 404
return {"error": str(val_err)}
except Exception as api_err:
logger.error(
f"API Error in /api/providers/{provider}/channels/{channel_id}/manifest: {str(api_err)}"
)
response.status = 500
return {"error": f"Internal server error: {str(api_err)}"}
@app.route("/api/providers/<provider>/channels/<channel_id>/stream")
def get_channel_stream(provider, channel_id):
"""
Returns HTTP 302 redirect to the actual manifest or rewritten manifest endpoint.
Supports both live and catchup streaming.
"""
try:
# Get optional catchup parameters
start_time = request.query.get("start_time")
end_time = request.query.get("end_time")
epg_id = request.query.get("epg_id")
country = request.query.get("country")
# Determine if this is a catchup request
is_catchup = bool(start_time and end_time)
if is_catchup:
logger.info(
f"Catchup stream request for {provider}/{channel_id}: "
f"start={start_time}, end={end_time}, epg_id={epg_id}"
)
# Convert Unix timestamps to integers
try:
start_time_int = int(start_time)
end_time_int = int(end_time)
except (ValueError, TypeError):
response.status = 400
return {"error": "Invalid start_time or end_time format"}
# Validate catchup is supported
provider_instance = manager.get_provider(provider)
catchup_hours = getattr(
provider_instance, "catchup_window", 0
) # Now in hours
if catchup_hours == 0:
response.status = 400
return {"error": f'Catchup not supported for provider "{provider}"'}
# Validate time is within catchup window (in HOURS)
import time
now = int(time.time())
max_age_seconds = catchup_hours * 3600 # Hours to seconds
if (now - start_time_int) > max_age_seconds:
response.status = 400
return {
"error": f"Content outside catchup window (max {catchup_hours} hours)"
}
# Check if provider needs proxy for catchup
if manager.needs_proxy(provider):
# Proxy mode: return rewritten MPD content directly
return service._get_proxied_catchup_manifest(
provider,
channel_id,
start_time_int,
end_time_int,
epg_id,
country,
)
else:
# Direct mode: get catchup manifest URL and redirect
manifest_url = manager.get_catchup_manifest(
provider_name=provider,
channel_id=channel_id,
start_time=start_time_int,
end_time=end_time_int,
epg_id=epg_id,
country=country,
)
if not manifest_url:
response.status = 404
return {
"error": f'Catchup manifest not available for channel "{channel_id}"'
}
logger.debug(f"Redirecting to catchup manifest: {manifest_url}")
redirect(manifest_url)
else:
# Live stream - existing logic
if manager.needs_proxy(provider):
return service._get_proxied_manifest(provider, channel_id)
else:
manifest_url = manager.get_channel_manifest(
provider_name=provider,
channel_id=channel_id,
country=country,
)
if not manifest_url:
response.status = 404
return {
"error": f'Manifest not available for channel "{channel_id}"'
}
logger.debug(f"Redirecting to manifest: {manifest_url}")
redirect(manifest_url)
except HTTPResponse:
raise
except ValueError as val_err:
logger.error(f"API Error in stream: {str(val_err)}")
response.status = 404
return {"error": str(val_err)}
except Exception as api_err:
logger.error(f"API Error in stream: {str(api_err)}")
response.status = 500
return {"error": f"Internal server error: {str(api_err)}"}
@app.route("/api/providers/<provider>/channels/<channel_id>/epg")
def get_channel_epg(provider, channel_id):
try:
# Parse optional parameters
kwargs = {"country": request.query.get("country")}
from datetime import timezone
# Handle start_time - can be Unix timestamp (from Kodi) or datetime
if request.query.get("start_time"):
start_time_str = request.query.get("start_time")
try:
# Try to parse as Unix timestamp (integer from Kodi PVR)
start_time_int = int(start_time_str)
kwargs["start_time"] = datetime.fromtimestamp(
start_time_int, tz=timezone.utc
)
except (ValueError, TypeError):
# Try to parse as ISO format string (for manual API calls)
try:
kwargs["start_time"] = datetime.fromisoformat(
start_time_str.replace("Z", "+00:00")
)
except ValueError:
logger.warning(f"Invalid start_time format: {start_time_str}")
# Continue without start_time filter
pass
# Handle end_time - can be Unix timestamp or datetime
if request.query.get("end_time"):
end_time_str = request.query.get("end_time")
try:
# Try to parse as Unix timestamp (integer from Kodi PVR)
end_time_int = int(end_time_str)
kwargs["end_time"] = datetime.fromtimestamp(
end_time_int, tz=timezone.utc
)
except (ValueError, TypeError):
# Try to parse as ISO format string (for manual API calls)
try:
kwargs["end_time"] = datetime.fromisoformat(
end_time_str.replace("Z", "+00:00")
)
except ValueError:
logger.warning(f"Invalid end_time format: {end_time_str}")
# Continue without end_time filter
pass
# Get EPG data from manager
epg_data = manager.get_channel_epg(
provider_name=provider, channel_id=channel_id, **kwargs
)
# Return as JSON
response.content_type = "application/json; charset=utf-8"
return {"provider": provider, "channel_id": channel_id, "epg": epg_data}
except ValueError as val_err:
# This handles the case where manager raises ValueError for unknown provider
logger.error(
f"API Error in /api/providers/{provider}/channels/{channel_id}/epg: {str(val_err)}"
)
response.status = 404
response.content_type = "application/json; charset=utf-8"
return {"error": str(val_err)}
except Exception as api_err:
logger.error(
f"API Error in /api/providers/{provider}/channels/{channel_id}/epg: {str(api_err)}"
)
response.status = 500
response.content_type = "application/json; charset=utf-8"
return {"error": f"Internal server error: {str(api_err)}"}
@app.route("/api/providers/<provider>/epg")
def get_provider_epg_xmltv(provider):
try:
# Set appropriate headers for XMLTV
response.content_type = "application/xml; charset=utf-8"
response.headers["Content-Disposition"] = (
f'attachment; filename="{provider}_epg.xml"'
)
# Get the XMLTV data from the provider
xmltv_data = manager.get_provider_epg_xmltv(
provider_name=provider, country=request.query.get("country")
)
if not xmltv_data:
response.status = 404
return {"error": f'EPG data not available for provider "{provider}"'}
return xmltv_data
except ValueError as val_err:
# Handle unknown provider
logger.error(f"API Error in /api/providers/{provider}/epg: {str(val_err)}")
response.status = 404
return {"error": str(val_err)}
except Exception as api_err:
logger.error(f"API Error in /api/providers/{provider}/epg: {str(api_err)}")
response.status = 500
return {"error": f"Internal server error: {str(api_err)}"}
@app.route("/api/providers/<provider>/channels/<channel_id>/drm")
def get_channel_drm(provider, channel_id):
try:
# Get optional catchup parameters
start_time = request.query.get("start_time")
end_time = request.query.get("end_time")
epg_id = request.query.get("epg_id")
country = request.query.get("country")
# Determine if this is a catchup request
is_catchup = bool(start_time and end_time)
if is_catchup:
logger.debug(
f"Catchup DRM request for {provider}/{channel_id}: "
f"epg_id={epg_id}"
)
# Convert timestamps
try:
start_time_int = int(start_time)
end_time_int = int(end_time)
except (ValueError, TypeError):
response.status = 400
return {"error": "Invalid start_time or end_time format"}
# Get catchup DRM configs
drm_configs = manager.get_catchup_drm_configs(
provider_name=provider,
channel_id=channel_id,
start_time=start_time_int,
end_time=end_time_int,
epg_id=epg_id,
country=country,
)
else:
# Live DRM - existing logic
drm_configs = manager.get_channel_drm_configs(
provider_name=provider, channel_id=channel_id, country=country
)
# Merge all DRM configs into a single dictionary
merged_drm_configs = {}
for config in drm_configs:
if hasattr(config, "to_dict"):
config_dict = config.to_dict()
else:
config_dict = config
merged_drm_configs.update(config_dict)
return {
"provider": provider,
"channel_id": channel_id,
"is_catchup": is_catchup,
"drm_configs": merged_drm_configs,
}
except ValueError as val_err:
logger.error(f"API Error in DRM endpoint: {str(val_err)}")
response.status = 404
return {"error": str(val_err)}
except Exception as api_err:
logger.error(f"API Error in DRM endpoint: {str(api_err)}")
response.status = 500
return {"error": f"Internal server error: {str(api_err)}"}
@app.route("/api/providers/<provider>/channels/<channel_id>/stream/decrypted")
def get_channel_stream_decrypted(provider, channel_id):
"""
Returns rewritten manifest for decrypted playback via media proxy.
This endpoint is used by the decrypted M3U playlists.
"""
try:
# Check if media proxy is configured
if not service.media_proxy_url:
response.status = 503
return {"error": "Media proxy not configured (MEDIA_PROXY_URL not set)"}
country = request.query.get("country")
# Get DRM configs to extract ClearKey data
drm_configs = manager.get_channel_drm_configs(
provider_name=provider, channel_id=channel_id, country=country
)
# Extract ClearKey data
clearkey_data = None
if isinstance(drm_configs, dict) and "org.w3.clearkey" in drm_configs:
clearkey_data = drm_configs["org.w3.clearkey"]
if not clearkey_data:
response.status = 400
return {"error": f'Channel "{channel_id}" does not have ClearKey DRM'}
# Extract KID:Key pairs
license_info = clearkey_data.get("license", {})
keyids = license_info.get("keyids", {})
if not keyids:
response.status = 400
return {"error": f'No ClearKey keyids found for channel "{channel_id}"'}
# Get manifest (check if needs proxy)
if manager.needs_proxy(provider):
# Get rewritten manifest with media proxy URLs + decrypt params
return service._get_decrypted_manifest(provider, channel_id, keyids)
else:
# No proxy needed - return error (decryption requires media proxy)
response.status = 400
return {
"error": "Decrypted playback requires provider proxy configuration"
}
except ValueError as val_err:
logger.error(f"API Error in decrypted stream: {str(val_err)}")
response.status = 404
return {"error": str(val_err)}
except Exception as api_err:
logger.error(f"API Error in decrypted stream: {str(api_err)}")
response.status = 500
return {"error": f"Internal server error: {str(api_err)}"}
service.py
+608 -2400
View File
File diff suppressed because it is too large Load Diff