830 lines
28 KiB
Python
830 lines
28 KiB
Python
"""AllDebrid API integration for converting free links to direct downloads.
|
|
|
|
AllDebrid is a debrid service that unlocks free file hosters and provides direct download links.
|
|
API docs: https://docs.alldebrid.com/#general-informations
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import sys
|
|
|
|
from helper.logger import log, debug
|
|
import time
|
|
import logging
|
|
from pathlib import Path
|
|
from typing import Any, Dict, Optional, Set, List, Sequence
|
|
from urllib.parse import urlencode, urlparse
|
|
from .http_client import HTTPClient
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class AllDebridError(Exception):
|
|
"""Raised when AllDebrid API request fails."""
|
|
pass
|
|
|
|
|
|
# Cache for supported hosters (domain -> host info)
|
|
_SUPPORTED_HOSTERS_CACHE: Optional[Dict[str, Dict[str, Any]]] = None
|
|
_CACHE_TIMESTAMP: float = 0
|
|
_CACHE_DURATION: float = 3600 # 1 hour
|
|
|
|
|
|
class AllDebridClient:
|
|
"""Client for AllDebrid API."""
|
|
|
|
# Try both v4 and v3 APIs
|
|
BASE_URLS = [
|
|
"https://api.alldebrid.com/v4",
|
|
"https://api.alldebrid.com/v3",
|
|
]
|
|
|
|
def __init__(self, api_key: str):
|
|
"""Initialize AllDebrid client with API key.
|
|
|
|
Args:
|
|
api_key: AllDebrid API key from config
|
|
"""
|
|
self.api_key = api_key.strip()
|
|
if not self.api_key:
|
|
raise AllDebridError("AllDebrid API key is empty")
|
|
self.base_url = self.BASE_URLS[0] # Start with v4
|
|
|
|
def _request(self, endpoint: str, params: Optional[Dict[str, str]] = None) -> Dict[str, Any]:
|
|
"""Make a request to AllDebrid API.
|
|
|
|
Args:
|
|
endpoint: API endpoint (e.g., "user/profile", "link/unlock")
|
|
params: Query parameters
|
|
|
|
Returns:
|
|
Parsed JSON response
|
|
|
|
Raises:
|
|
AllDebridError: If request fails or API returns error
|
|
"""
|
|
if params is None:
|
|
params = {}
|
|
|
|
# Add API key to params
|
|
params['apikey'] = self.api_key
|
|
|
|
url = f"{self.base_url}/{endpoint}"
|
|
query_string = urlencode(params)
|
|
full_url = f"{url}?{query_string}"
|
|
|
|
logger.debug(f"[AllDebrid] {endpoint} request to {full_url[:80]}...")
|
|
|
|
try:
|
|
# Pass timeout to HTTPClient init, not to get()
|
|
with HTTPClient(timeout=30.0, headers={'User-Agent': 'downlow/1.0'}) as client:
|
|
try:
|
|
response = client.get(full_url)
|
|
response.raise_for_status()
|
|
except Exception as req_err:
|
|
# Log detailed error info
|
|
logger.error(f"[AllDebrid] Request error to {full_url[:80]}: {req_err}", exc_info=True)
|
|
if hasattr(req_err, 'response') and req_err.response is not None: # type: ignore
|
|
try:
|
|
error_body = req_err.response.content.decode('utf-8') # type: ignore
|
|
logger.error(f"[AllDebrid] Response body: {error_body[:200]}")
|
|
except:
|
|
pass
|
|
raise
|
|
|
|
data = json.loads(response.content.decode('utf-8'))
|
|
logger.debug(f"[AllDebrid] Response status: {response.status_code}")
|
|
|
|
# Check for API errors
|
|
if data.get('status') == 'error':
|
|
error_msg = data.get('error', {}).get('message', 'Unknown error')
|
|
logger.error(f"[AllDebrid] API error: {error_msg}")
|
|
raise AllDebridError(f"AllDebrid API error: {error_msg}")
|
|
|
|
return data
|
|
except AllDebridError:
|
|
raise
|
|
except Exception as exc:
|
|
error_msg = f"AllDebrid request failed: {exc}"
|
|
logger.error(f"[AllDebrid] {error_msg}", exc_info=True)
|
|
raise AllDebridError(error_msg)
|
|
|
|
def unlock_link(self, link: str) -> Optional[str]:
|
|
"""Unlock a restricted link and get direct download URL.
|
|
|
|
Args:
|
|
link: Restricted link to unlock
|
|
|
|
Returns:
|
|
Direct download URL, or None if already unrestricted
|
|
|
|
Raises:
|
|
AllDebridError: If unlock fails
|
|
"""
|
|
if not link.startswith(('http://', 'https://')):
|
|
raise AllDebridError(f"Invalid URL: {link}")
|
|
|
|
try:
|
|
response = self._request('link/unlock', {'link': link})
|
|
|
|
# Check if unlock was successful
|
|
if response.get('status') == 'success':
|
|
data = response.get('data', {})
|
|
|
|
# AllDebrid returns the download info in 'link' field
|
|
if 'link' in data:
|
|
return data['link']
|
|
|
|
# Alternative: check for 'file' field
|
|
if 'file' in data:
|
|
return data['file']
|
|
|
|
# If no direct link, return the input link
|
|
return link
|
|
|
|
return None
|
|
except AllDebridError:
|
|
raise
|
|
except Exception as exc:
|
|
raise AllDebridError(f"Failed to unlock link: {exc}")
|
|
|
|
def check_host(self, hostname: str) -> Dict[str, Any]:
|
|
"""Check if a host is supported by AllDebrid.
|
|
|
|
Args:
|
|
hostname: Hostname to check (e.g., "uploadhaven.com")
|
|
|
|
Returns:
|
|
Host information dict with support status
|
|
|
|
Raises:
|
|
AllDebridError: If request fails
|
|
"""
|
|
try:
|
|
response = self._request('host', {'name': hostname})
|
|
|
|
if response.get('status') == 'success':
|
|
return response.get('data', {})
|
|
|
|
return {}
|
|
except AllDebridError:
|
|
raise
|
|
except Exception as exc:
|
|
raise AllDebridError(f"Failed to check host: {exc}")
|
|
|
|
def get_user_info(self) -> Dict[str, Any]:
|
|
"""Get current user account information.
|
|
|
|
Returns:
|
|
User information dict
|
|
|
|
Raises:
|
|
AllDebridError: If request fails
|
|
"""
|
|
try:
|
|
response = self._request('user/profile')
|
|
|
|
if response.get('status') == 'success':
|
|
return response.get('data', {})
|
|
|
|
return {}
|
|
except AllDebridError:
|
|
raise
|
|
except Exception as exc:
|
|
raise AllDebridError(f"Failed to get user info: {exc}")
|
|
|
|
def get_supported_hosters(self) -> Dict[str, Dict[str, Any]]:
|
|
"""Get list of all supported hosters from AllDebrid API.
|
|
|
|
Returns:
|
|
Dict mapping domain to host info (status, name, etc)
|
|
|
|
Raises:
|
|
AllDebridError: If request fails
|
|
"""
|
|
try:
|
|
response = self._request('hosts/domains')
|
|
|
|
if response.get('status') == 'success':
|
|
data = response.get('data', {})
|
|
# The API returns hosts keyed by domain
|
|
return data if isinstance(data, dict) else {}
|
|
|
|
return {}
|
|
except AllDebridError:
|
|
raise
|
|
except Exception as exc:
|
|
raise AllDebridError(f"Failed to get supported hosters: {exc}")
|
|
|
|
def magnet_add(self, magnet_uri: str) -> Dict[str, Any]:
|
|
"""Submit a magnet link or torrent hash to AllDebrid for processing.
|
|
|
|
AllDebrid will download the torrent content and store it in the account.
|
|
Processing time varies based on torrent size and availability.
|
|
|
|
Args:
|
|
magnet_uri: Magnet URI (magnet:?xt=urn:btih:...) or torrent hash
|
|
|
|
Returns:
|
|
Dict with magnet info:
|
|
- id: Magnet ID (int) - needed for status checks
|
|
- name: Torrent name
|
|
- hash: Torrent hash
|
|
- size: Total file size (bytes)
|
|
- ready: Boolean - True if already available
|
|
|
|
Raises:
|
|
AllDebridError: If submit fails (requires premium, invalid magnet, etc)
|
|
"""
|
|
if not magnet_uri:
|
|
raise AllDebridError("Magnet URI is empty")
|
|
|
|
try:
|
|
# API endpoint: POST /v4/magnet/upload
|
|
# Format: /magnet/upload?apikey=key&magnets[]=magnet:?xt=...
|
|
response = self._request('magnet/upload', {'magnets[]': magnet_uri})
|
|
|
|
if response.get('status') == 'success':
|
|
data = response.get('data', {})
|
|
magnets = data.get('magnets', [])
|
|
|
|
if magnets and len(magnets) > 0:
|
|
magnet_info = magnets[0]
|
|
|
|
# Check for errors in the magnet response
|
|
if 'error' in magnet_info:
|
|
error = magnet_info['error']
|
|
error_msg = error.get('message', 'Unknown error')
|
|
raise AllDebridError(f"Magnet error: {error_msg}")
|
|
|
|
return magnet_info
|
|
|
|
raise AllDebridError("No magnet data in response")
|
|
|
|
raise AllDebridError(f"API error: {response.get('error', 'Unknown')}")
|
|
except AllDebridError:
|
|
raise
|
|
except Exception as exc:
|
|
raise AllDebridError(f"Failed to submit magnet: {exc}")
|
|
|
|
def magnet_status(self, magnet_id: int, include_files: bool = False) -> Dict[str, Any]:
|
|
"""Get status of a magnet currently being processed or stored.
|
|
|
|
Status codes:
|
|
0-3: Processing (in queue, downloading, compressing, uploading)
|
|
4: Ready (files available for download)
|
|
5-15: Error (upload failed, not downloaded in 20min, too big, etc)
|
|
|
|
Args:
|
|
magnet_id: Magnet ID from magnet_add()
|
|
include_files: If True, includes file list in response
|
|
|
|
Returns:
|
|
Dict with status info:
|
|
- id: Magnet ID
|
|
- filename: Torrent name
|
|
- size: Total size (bytes)
|
|
- status: Human-readable status
|
|
- statusCode: Numeric code (0-15)
|
|
- downloaded: Bytes downloaded so far
|
|
- uploaded: Bytes uploaded so far
|
|
- seeders: Number of seeders
|
|
- downloadSpeed: Current speed (bytes/sec)
|
|
- uploadSpeed: Current speed (bytes/sec)
|
|
- files: (optional) Array of file objects when include_files=True
|
|
Each file: {n: name, s: size, l: download_link}
|
|
|
|
Raises:
|
|
AllDebridError: If status check fails
|
|
"""
|
|
if not isinstance(magnet_id, int) or magnet_id <= 0:
|
|
raise AllDebridError(f"Invalid magnet ID: {magnet_id}")
|
|
|
|
try:
|
|
# Use v4.1 endpoint for better response format
|
|
# Temporarily override base_url for this request
|
|
old_base = self.base_url
|
|
self.base_url = "https://api.alldebrid.com/v4.1"
|
|
|
|
try:
|
|
response = self._request('magnet/status', {'id': str(magnet_id)})
|
|
finally:
|
|
self.base_url = old_base
|
|
|
|
if response.get('status') == 'success':
|
|
data = response.get('data', {})
|
|
magnets = data.get('magnets', {})
|
|
|
|
# Handle both list and dict responses
|
|
if isinstance(magnets, list) and len(magnets) > 0:
|
|
return magnets[0]
|
|
elif isinstance(magnets, dict) and magnets:
|
|
return magnets
|
|
|
|
raise AllDebridError(f"No magnet found with ID {magnet_id}")
|
|
|
|
raise AllDebridError(f"API error: {response.get('error', 'Unknown')}")
|
|
except AllDebridError:
|
|
raise
|
|
except Exception as exc:
|
|
raise AllDebridError(f"Failed to get magnet status: {exc}")
|
|
|
|
def magnet_status_live(self, magnet_id: int, session: int = None, counter: int = 0) -> Dict[str, Any]:
|
|
"""Get live status of a magnet using delta sync mode.
|
|
|
|
The live mode endpoint provides real-time progress by only sending
|
|
deltas (changed fields) instead of full status on each call. This
|
|
reduces bandwidth and server load compared to regular polling.
|
|
|
|
Note: The "live" designation refers to the delta-sync mode where you
|
|
maintain state locally and apply diffs from the API, not a streaming
|
|
endpoint. Regular magnet_status() polling is simpler for single magnets.
|
|
|
|
Docs: https://docs.alldebrid.com/#get-status-live-mode
|
|
|
|
Args:
|
|
magnet_id: Magnet ID from magnet_add()
|
|
session: Session ID (use same ID across multiple calls). If None, will query current status
|
|
counter: Counter value from previous response (starts at 0)
|
|
|
|
Returns:
|
|
Dict with magnet status. May contain only changed fields if counter > 0.
|
|
For single-magnet tracking, use magnet_status() instead.
|
|
|
|
Raises:
|
|
AllDebridError: If request fails
|
|
"""
|
|
if not isinstance(magnet_id, int) or magnet_id <= 0:
|
|
raise AllDebridError(f"Invalid magnet ID: {magnet_id}")
|
|
|
|
try:
|
|
# For single magnet queries, just use regular endpoint with ID
|
|
# The "live mode" with session/counter is for multi-magnet dashboards
|
|
# where bandwidth savings from diffs matter
|
|
response = self._request('magnet/status', {'id': magnet_id})
|
|
|
|
if response.get('status') == 'success':
|
|
data = response.get('data', {})
|
|
magnets = data.get('magnets', [])
|
|
|
|
# Handle list response
|
|
if isinstance(magnets, list) and len(magnets) > 0:
|
|
return magnets[0]
|
|
|
|
raise AllDebridError(f"No magnet found with ID {magnet_id}")
|
|
|
|
raise AllDebridError(f"API error: {response.get('error', 'Unknown')}")
|
|
except AllDebridError:
|
|
raise
|
|
except Exception as exc:
|
|
raise AllDebridError(f"Failed to get magnet live status: {exc}")
|
|
|
|
def magnet_links(self, magnet_ids: list) -> Dict[str, Any]:
|
|
"""Get files and download links for one or more magnets.
|
|
|
|
Use this after magnet_status shows statusCode == 4 (Ready).
|
|
Returns the file tree structure with direct download links.
|
|
|
|
Args:
|
|
magnet_ids: List of magnet IDs to get files for
|
|
|
|
Returns:
|
|
Dict mapping magnet_id (as string) -> magnet_info:
|
|
- id: Magnet ID
|
|
- files: Array of file/folder objects
|
|
File: {n: name, s: size, l: direct_download_link}
|
|
Folder: {n: name, e: [sub_items]}
|
|
|
|
Raises:
|
|
AllDebridError: If request fails
|
|
"""
|
|
if not magnet_ids:
|
|
raise AllDebridError("No magnet IDs provided")
|
|
|
|
try:
|
|
# Build parameter: id[]=123&id[]=456 style
|
|
params = {}
|
|
for i, magnet_id in enumerate(magnet_ids):
|
|
params[f'id[{i}]'] = str(magnet_id)
|
|
|
|
response = self._request('magnet/files', params)
|
|
|
|
if response.get('status') == 'success':
|
|
data = response.get('data', {})
|
|
magnets = data.get('magnets', [])
|
|
|
|
# Convert list to dict keyed by ID (as string) for easier access
|
|
result = {}
|
|
for magnet_info in magnets:
|
|
magnet_id = magnet_info.get('id')
|
|
if magnet_id:
|
|
result[str(magnet_id)] = magnet_info
|
|
|
|
return result
|
|
|
|
raise AllDebridError(f"API error: {response.get('error', 'Unknown')}")
|
|
except AllDebridError:
|
|
raise
|
|
except Exception as exc:
|
|
raise AllDebridError(f"Failed to get magnet files: {exc}")
|
|
|
|
def instant_available(self, magnet_hash: str) -> Optional[List[Dict[str, Any]]]:
|
|
"""Check if magnet is available for instant streaming without downloading.
|
|
|
|
AllDebrid's "instant" feature checks if a magnet can be streamed directly
|
|
without downloading all the data. Returns available video/audio files.
|
|
|
|
Args:
|
|
magnet_hash: Torrent hash (with or without magnet: prefix)
|
|
|
|
Returns:
|
|
List of available files for streaming, or None if not available
|
|
Each file: {n: name, s: size, e: extension, t: type}
|
|
Returns empty list if torrent not found or not available
|
|
|
|
Raises:
|
|
AllDebridError: If API request fails
|
|
"""
|
|
try:
|
|
# Parse magnet hash if needed
|
|
if magnet_hash.startswith('magnet:'):
|
|
# Extract hash from magnet URI
|
|
import re
|
|
match = re.search(r'xt=urn:btih:([a-fA-F0-9]+)', magnet_hash)
|
|
if not match:
|
|
return None
|
|
hash_value = match.group(1)
|
|
else:
|
|
hash_value = magnet_hash.strip()
|
|
|
|
if not hash_value or len(hash_value) < 32:
|
|
return None
|
|
|
|
response = self._request('magnet/instant', {'magnet': hash_value})
|
|
|
|
if response.get('status') == 'success':
|
|
data = response.get('data', {})
|
|
# Returns 'files' array if available, or empty
|
|
return data.get('files', [])
|
|
|
|
# Not available is not an error, just return empty list
|
|
return []
|
|
|
|
except AllDebridError:
|
|
raise
|
|
except Exception as exc:
|
|
logger.debug(f"[AllDebrid] instant_available check failed: {exc}")
|
|
return None
|
|
|
|
def magnet_delete(self, magnet_id: int) -> bool:
|
|
"""Delete a magnet from the AllDebrid account.
|
|
|
|
Args:
|
|
magnet_id: Magnet ID to delete
|
|
|
|
Returns:
|
|
True if deletion was successful
|
|
|
|
Raises:
|
|
AllDebridError: If deletion fails
|
|
"""
|
|
if not isinstance(magnet_id, int) or magnet_id <= 0:
|
|
raise AllDebridError(f"Invalid magnet ID: {magnet_id}")
|
|
|
|
try:
|
|
response = self._request('magnet/delete', {'id': str(magnet_id)})
|
|
|
|
if response.get('status') == 'success':
|
|
return True
|
|
|
|
raise AllDebridError(f"API error: {response.get('error', 'Unknown')}")
|
|
except AllDebridError:
|
|
raise
|
|
except Exception as exc:
|
|
raise AllDebridError(f"Failed to delete magnet: {exc}")
|
|
|
|
|
|
def _get_cached_supported_hosters(api_key: str) -> Set[str]:
|
|
"""Get cached list of supported hoster domains.
|
|
|
|
Uses AllDebrid API to fetch the list once per hour,
|
|
caching the result to avoid repeated API calls.
|
|
|
|
Args:
|
|
api_key: AllDebrid API key
|
|
|
|
Returns:
|
|
Set of supported domain names (lowercased)
|
|
"""
|
|
global _SUPPORTED_HOSTERS_CACHE, _CACHE_TIMESTAMP
|
|
|
|
now = time.time()
|
|
|
|
# Return cached result if still valid
|
|
if _SUPPORTED_HOSTERS_CACHE is not None and (now - _CACHE_TIMESTAMP) < _CACHE_DURATION:
|
|
return set(_SUPPORTED_HOSTERS_CACHE.keys())
|
|
|
|
# Fetch fresh list from API
|
|
try:
|
|
client = AllDebridClient(api_key)
|
|
hosters_dict = client.get_supported_hosters()
|
|
|
|
if hosters_dict:
|
|
# API returns: hosts (list), streams (list), redirectors (list)
|
|
# Combine all into a single set
|
|
all_domains: Set[str] = set()
|
|
|
|
# Add hosts
|
|
if 'hosts' in hosters_dict and isinstance(hosters_dict['hosts'], list):
|
|
all_domains.update(hosters_dict['hosts'])
|
|
|
|
# Add streams
|
|
if 'streams' in hosters_dict and isinstance(hosters_dict['streams'], list):
|
|
all_domains.update(hosters_dict['streams'])
|
|
|
|
# Add redirectors
|
|
if 'redirectors' in hosters_dict and isinstance(hosters_dict['redirectors'], list):
|
|
all_domains.update(hosters_dict['redirectors'])
|
|
|
|
# Cache as dict for consistency
|
|
_SUPPORTED_HOSTERS_CACHE = {domain: {} for domain in all_domains}
|
|
_CACHE_TIMESTAMP = now
|
|
|
|
if all_domains:
|
|
debug(f"✓ Cached {len(all_domains)} supported hosters")
|
|
|
|
return all_domains
|
|
except Exception as exc:
|
|
log(f"⚠ Failed to fetch supported hosters: {exc}", file=sys.stderr)
|
|
# Return any cached hosters even if expired
|
|
if _SUPPORTED_HOSTERS_CACHE:
|
|
return set(_SUPPORTED_HOSTERS_CACHE.keys())
|
|
|
|
# Fallback: empty set if no cache available
|
|
return set()
|
|
|
|
|
|
def is_link_restrictable_hoster(url: str, api_key: str) -> bool:
|
|
"""Check if a URL is from a hoster that AllDebrid can unlock.
|
|
|
|
Intelligently queries the AllDebrid API to detect if the URL is
|
|
from a supported restricted hoster.
|
|
|
|
Args:
|
|
url: URL to check
|
|
api_key: AllDebrid API key
|
|
|
|
Returns:
|
|
True if URL is from a supported restrictable hoster
|
|
"""
|
|
if not url or not api_key:
|
|
return False
|
|
|
|
try:
|
|
# Extract domain from URL
|
|
parsed = urlparse(url)
|
|
domain = parsed.netloc.lower()
|
|
|
|
# Remove www. prefix for comparison
|
|
if domain.startswith('www.'):
|
|
domain = domain[4:]
|
|
|
|
# Get supported hosters (cached)
|
|
supported = _get_cached_supported_hosters(api_key)
|
|
|
|
if not supported:
|
|
# API check failed, fall back to manual detection
|
|
# Check for common restricted hosters
|
|
common_hosters = {
|
|
'uploadhaven.com', 'uploaded.to', 'uploaded.net',
|
|
'datafile.com', 'rapidfile.io', 'nitroflare.com',
|
|
'1fichier.com', 'mega.nz', 'mediafire.com'
|
|
}
|
|
return any(host in url.lower() for host in common_hosters)
|
|
|
|
# Check if domain is in supported list
|
|
# Need to check exact match and with/without www
|
|
return domain in supported or f"www.{domain}" in supported
|
|
except Exception as exc:
|
|
log(f"⚠ Hoster detection failed: {exc}", file=sys.stderr)
|
|
return False
|
|
|
|
|
|
def convert_link_with_debrid(link: str, api_key: str) -> Optional[str]:
|
|
"""Convert a restricted link to a direct download URL using AllDebrid.
|
|
|
|
Args:
|
|
link: Restricted link
|
|
api_key: AllDebrid API key
|
|
|
|
Returns:
|
|
Direct download URL, or original link if already unrestricted
|
|
"""
|
|
if not api_key:
|
|
return None
|
|
|
|
try:
|
|
client = AllDebridClient(api_key)
|
|
direct_link = client.unlock_link(link)
|
|
|
|
if direct_link and direct_link != link:
|
|
debug(f"✓ Converted link: {link[:60]}... → {direct_link[:60]}...")
|
|
return direct_link
|
|
|
|
return None
|
|
except AllDebridError as exc:
|
|
log(f"⚠ Failed to convert link: {exc}", file=sys.stderr)
|
|
return None
|
|
except Exception as exc:
|
|
log(f"⚠ Unexpected error: {exc}", file=sys.stderr)
|
|
return None
|
|
|
|
|
|
def is_magnet_link(uri: str) -> bool:
|
|
"""Check if a URI is a magnet link.
|
|
|
|
Magnet links start with 'magnet:?xt=urn:btih:' or just 'magnet:'
|
|
|
|
Args:
|
|
uri: URI to check
|
|
|
|
Returns:
|
|
True if URI is a magnet link
|
|
"""
|
|
if not uri:
|
|
return False
|
|
return uri.lower().startswith('magnet:')
|
|
|
|
|
|
def is_torrent_hash(text: str) -> bool:
|
|
"""Check if text looks like a torrent hash (40 or 64 hex characters).
|
|
|
|
Common formats:
|
|
- Info hash v1: 40 hex chars (SHA-1)
|
|
- Info hash v2: 64 hex chars (SHA-256)
|
|
|
|
Args:
|
|
text: Text to check
|
|
|
|
Returns:
|
|
True if text matches torrent hash format
|
|
"""
|
|
if not text or not isinstance(text, str):
|
|
return False
|
|
|
|
text = text.strip()
|
|
|
|
# Check if it's 40 hex chars (SHA-1) or 64 hex chars (SHA-256)
|
|
if len(text) not in (40, 64):
|
|
return False
|
|
|
|
try:
|
|
# Try to parse as hex
|
|
int(text, 16)
|
|
return True
|
|
except ValueError:
|
|
return False
|
|
|
|
|
|
def is_torrent_file(path: str) -> bool:
|
|
"""Check if a file path is a .torrent file.
|
|
|
|
Args:
|
|
path: File path to check
|
|
|
|
Returns:
|
|
True if file has .torrent extension
|
|
"""
|
|
if not path:
|
|
return False
|
|
return path.lower().endswith('.torrent')
|
|
|
|
|
|
def parse_magnet_or_hash(uri: str) -> Optional[str]:
|
|
"""Parse a magnet URI or hash into a format for AllDebrid API.
|
|
|
|
AllDebrid's magnet/upload endpoint accepts:
|
|
- Full magnet URIs: magnet:?xt=urn:btih:...
|
|
- Info hashes: 40 or 64 hex characters
|
|
|
|
Args:
|
|
uri: Magnet URI or hash
|
|
|
|
Returns:
|
|
Normalized input for AllDebrid API, or None if invalid
|
|
"""
|
|
if not uri:
|
|
return None
|
|
|
|
uri = uri.strip()
|
|
|
|
# Already a magnet link - just return it
|
|
if is_magnet_link(uri):
|
|
return uri
|
|
|
|
# Check if it's a valid hash
|
|
if is_torrent_hash(uri):
|
|
return uri
|
|
|
|
# Not a recognized format
|
|
return None
|
|
|
|
|
|
# ============================================================================
|
|
# Cmdlet: unlock_link
|
|
# ============================================================================
|
|
|
|
def unlock_link_cmdlet(result: Any, args: Sequence[str], config: Dict[str, Any]) -> int:
|
|
"""Unlock a restricted link using AllDebrid.
|
|
|
|
Converts free hosters and restricted links to direct download URLs.
|
|
|
|
Usage:
|
|
unlock-link <link>
|
|
unlock-link # Uses URL from pipeline result
|
|
|
|
Requires:
|
|
- AllDebrid API key in config under Debrid.All-debrid
|
|
|
|
Args:
|
|
result: Pipeline result object
|
|
args: Command arguments
|
|
config: Configuration dictionary
|
|
|
|
Returns:
|
|
0 on success, 1 on failure
|
|
"""
|
|
try:
|
|
from .link_utils import (
|
|
extract_link,
|
|
get_api_key,
|
|
add_direct_link_to_result,
|
|
)
|
|
except ImportError as e:
|
|
log(f"Required modules unavailable: {e}", file=sys.stderr)
|
|
return 1
|
|
|
|
# Get link from args or result
|
|
link = extract_link(result, args)
|
|
|
|
if not link:
|
|
log("No valid URL provided", file=sys.stderr)
|
|
return 1
|
|
|
|
# Get AllDebrid API key from config
|
|
api_key = get_api_key(config, "AllDebrid", "Debrid.All-debrid")
|
|
|
|
if not api_key:
|
|
log("AllDebrid API key not configured in Debrid.All-debrid", file=sys.stderr)
|
|
return 1
|
|
|
|
# Try to unlock the link
|
|
debug(f"Unlocking: {link}")
|
|
direct_link = convert_link_with_debrid(link, api_key)
|
|
|
|
if direct_link:
|
|
debug(f"✓ Direct link: {direct_link}")
|
|
|
|
# Update result with direct link
|
|
add_direct_link_to_result(result, direct_link, link)
|
|
|
|
# Return the updated result via pipeline context
|
|
# Note: The cmdlet wrapper will handle emitting to pipeline
|
|
return 0
|
|
else:
|
|
log(f"❌ Failed to unlock link or already unrestricted", file=sys.stderr)
|
|
return 1
|
|
|
|
|
|
# ============================================================================
|
|
# Cmdlet Registration
|
|
# ============================================================================
|
|
|
|
def _register_unlock_link():
|
|
"""Register unlock-link command with cmdlet registry if available."""
|
|
try:
|
|
from cmdlets import register
|
|
|
|
@register(["unlock-link"])
|
|
def unlock_link_wrapper(result: Any, args: Sequence[str], config: Dict[str, Any]) -> int:
|
|
"""Wrapper to make unlock_link_cmdlet available as cmdlet."""
|
|
import pipeline as ctx
|
|
|
|
ret_code = unlock_link_cmdlet(result, args, config)
|
|
|
|
# If successful, emit the result
|
|
if ret_code == 0:
|
|
ctx.emit(result)
|
|
|
|
return ret_code
|
|
|
|
return unlock_link_wrapper
|
|
except ImportError:
|
|
# If cmdlets module not available, just return None
|
|
return None
|
|
|
|
|
|
# Register when module is imported
|
|
_unlock_link_registration = _register_unlock_link()
|