"""Execute in the full a-Shell app from the Spoti transfer shortcut.""" import argparse import base64 import ctypes import hashlib import hmac import json import os from pathlib import Path import re import ssl import subprocess import sys import tempfile import time from urllib.error import HTTPError, URLError from urllib.parse import parse_qs, urlsplit from urllib.request import HTTPRedirectHandler, Request, build_opener CHUNK = 1024 * 1024 LIMIT = 512 * CHUNK TOKEN = re.compile(r'[A-Za-z0-9_-]{43}\Z') VIDEO = re.compile(r'[A-Za-z0-9_-]{11}\Z') PLAYLIST = re.compile(r'[A-Za-z0-9_-]{2,100}\Z') RELAY = re.compile(r'https://[a-z0-9][a-z0-9.-]+\.workers\.dev\Z') DOCUMENTS = Path.home() / 'Documents' COMMAND_PREFIX = 'python3 ~/Documents/spoti_transfer.py ' def b64decode(value): return base64.urlsafe_b64decode(value + '=' * (-len(value) % 4)) def b64encode(value): return base64.urlsafe_b64encode(value).decode('ascii').rstrip('=') def youtube_url(value, playlist): url = urlsplit(value) args = parse_qs(url.query) if url.scheme != 'https' or url.netloc not in { 'youtube.com', 'www.youtube.com', 'm.youtube.com', 'music.youtube.com', 'youtu.be' }: raise ValueError('El enlace no es de YouTube.') if playlist: identifier = args.get('list', [''])[0] if not PLAYLIST.fullmatch(identifier): raise ValueError('Identificador de lista inválido.') return 'https://www.youtube.com/playlist?list=' + identifier if url.netloc == 'youtu.be' or url.path.startswith(('/shorts/', '/live/')): identifier = url.path.rstrip('/').split('/')[-1] else: identifier = args.get('v', [''])[0] if url.path == '/watch' else '' if not VIDEO.fullmatch(identifier): raise ValueError('Identificador de vídeo inválido.') return 'https://www.youtube.com/watch?v=' + identifier def commoncrypto(): """Use Apple's public CommonCrypto API without an iOS Python wheel.""" library = ctypes.CDLL('/usr/lib/libSystem.B.dylib') encrypt = library.CCCrypt v = ctypes.c_void_p n = ctypes.c_size_t encrypt.argtypes = [ctypes.c_uint32, ctypes.c_uint32, ctypes.c_uint32, v, n, v, v, n, v, n, ctypes.POINTER(n)] encrypt.restype = ctypes.c_int32 def cbc(data, key, iv): output = ctypes.create_string_buffer(len(data) + 16) written = n() result = encrypt(0, 0, 1, key, len(key), iv, data, len(data), output, len(output), ctypes.byref(written)) if result: raise RuntimeError('CommonCrypto devolvió un error.') return output.raw[:written.value] return cbc def cipher(): if sys.platform in ('ios', 'darwin'): try: return commoncrypto() except (OSError, AttributeError): pass from yt_dlp.dependencies import Cryptodome if Cryptodome.AES: def cbc(data, key, iv): count = 16 - len(data) % 16 return Cryptodome.AES.new(key, Cryptodome.AES.MODE_CBC, iv).encrypt(data + bytes([count]) * count) return cbc from yt_dlp.aes import aes_cbc_encrypt_bytes return aes_cbc_encrypt_bytes class NoRedirect(HTTPRedirectHandler): def redirect_request(self, req, fp, code, msg, headers, newurl): return None class Transfer: def __init__(self, descriptor, allow_loopback=False): if descriptor.get('v') != 1: raise ValueError('Versión de transferencia incorrecta.') relay = descriptor.get('relay', '') loopback = allow_loopback and re.fullmatch(r'http://127\.0\.0\.1:[0-9]{1,5}', relay) if not RELAY.fullmatch(relay) and not loopback: raise ValueError('Dirección de transferencia inválida.') self.job = descriptor['job'] self.token = descriptor['send'] if not TOKEN.fullmatch(self.job) or not TOKEN.fullmatch(self.token): raise ValueError('Código de transferencia inválido.') self.key = b64decode(descriptor['key']) if len(self.key) != 64: raise ValueError('Clave de cifrado inválida.') self.url = relay + '/v1/jobs/' + self.job self.http = build_opener(NoRedirect()) self.cbc = cipher() self.bytes = 0 def request(self, method, suffix, value): body = value if isinstance(value, bytes) else json.dumps(value).encode() headers = { 'Authorization': 'Bearer ' + self.token, 'User-Agent': 'Spoti-Transfer/1.0', 'Content-Type': 'application/octet-stream' if isinstance(value, bytes) else 'application/json' } for attempt in range(4): try: request = Request(self.url + suffix, body, headers, method=method) with self.http.open(request, timeout=60) as response: if response.status != 204: raise RuntimeError('Respuesta de transferencia inesperada.') return except HTTPError as error: status = error.code error.close() if status not in (429, 500, 502, 503, 504) or attempt == 3: raise RuntimeError(f'Cloudflare rechazó la transferencia (HTTP {status}).') from None except (URLError, TimeoutError, ssl.SSLError): if attempt == 3: raise RuntimeError('No se pudo conectar con Cloudflare.') from None time.sleep(2 ** attempt) def seal(self, label, data): iv = os.urandom(16) ciphertext = self.cbc(data, self.key[:32], iv) payload = iv + ciphertext mac = hmac.digest(self.key[32:], (self.job + ':' + label).encode() + b'\x00' + payload, 'sha256') return payload + mac def send(self, index, path, title, playlist=None): size = path.stat().st_size parts = (size + CHUNK - 1) // CHUNK if not size or self.bytes + size + 64 * parts > LIMIT: raise ValueError('El archivo supera el espacio disponible de la transferencia (512 MiB).') root = '/files/' + str(index) with path.open('rb') as stream: for part in range(parts): packet = self.seal(f'file:{index}:part:{part}', stream.read(CHUNK)) self.request('PUT', root + '/parts/' + str(part), packet) self.bytes += len(packet) metadata = {'title': title[:240], 'ext': path.suffix[1:], 'playlist': playlist[:120] if playlist else None} sealed = self.seal(f'file:{index}:metadata', json.dumps(metadata, ensure_ascii=False).encode()) self.request('PUT', root, {'parts': parts, 'metadata': b64encode(sealed)}) def jobs(descriptor): from yt_dlp import YoutubeDL raw_url = descriptor['url'] options = {'quiet': True, 'no_warnings': True, 'socket_timeout': 40} adapter = DOCUMENTS / 'yt-dlp-ios.git' if not adapter.is_dir(): adapter = DOCUMENTS / 'yt-dlp-ios' if adapter.is_dir(): sys.path.insert(0, str(adapter)) import ashell_jsc # Registers a-Shell's JS runtime with yt-dlp. if not descriptor['playlist']: yield youtube_url(raw_url, False), None return with YoutubeDL({**options, 'extract_flat': 'in_playlist', 'playlistend': 101}) as ydl: info = ydl.extract_info(youtube_url(raw_url, True), download=False) title = info.get('title') or 'Lista de YouTube' for entry in info.get('entries') or []: identifier = (entry or {}).get('id', '') yield 'https://www.youtube.com/watch?v=' + identifier if VIDEO.fullmatch(identifier) else None, title def download(url, fmt, folder): from yt_dlp import YoutubeDL options = { 'outtmpl': str(folder / 'media.%(ext)s'), 'noplaylist': True, 'socket_timeout': 40, 'retries': 3, 'fragment_retries': 3, 'max_filesize': LIMIT, 'merge_output_format': 'mp4' } if fmt == 'mp3': options['format'] = 'bestaudio/best' options['postprocessors'] = [{'key': 'FFmpegExtractAudio', 'preferredcodec': 'mp3', 'preferredquality': '0'}] else: options['format'] = 'bestvideo[ext=mp4][vcodec^=avc1]+bestaudio[ext=m4a]/best[ext=mp4][vcodec^=avc1]' with YoutubeDL(options) as ydl: info = ydl.extract_info(url, download=True) result = folder / ('media.' + fmt) if not result.is_file(): raise RuntimeError(f'No se obtuvo un archivo {fmt.upper()} compatible.') return result, info.get('title') or 'Vídeo de YouTube' def read_descriptor(argument): if not argument and sys.platform in ('ios', 'darwin'): try: command = subprocess.check_output(['pbpaste'], text=True, timeout=5).strip() except (OSError, subprocess.SubprocessError): command = '' if command.startswith(COMMAND_PREFIX): argument = command[len(COMMAND_PREFIX):] if not argument: raise ValueError('No llegó el código de descarga. En Spoti, toca «Copiar comando para a‑Shell» y pégalo en a‑Shell.') if not re.fullmatch(r'[A-Za-z0-9_-]{1,4096}', argument): raise ValueError('Código de transferencia inválido. Copia el comando de la descarga actual desde Spoti.') try: descriptor = json.loads(b64decode(argument)) except (ValueError, UnicodeError): raise ValueError('No se pudo leer el código de descarga.') from None if not isinstance(descriptor, dict): raise ValueError('Código de transferencia inválido.') return descriptor def main(): parser = argparse.ArgumentParser() parser.add_argument('descriptor', nargs='?', help='Código que genera Spoti') parser.add_argument('--test-file', type=Path, action='append', help='Prueba el transporte con un archivo local') args = parser.parse_args() try: descriptor = read_descriptor(args.descriptor) except ValueError as error: print(error, file=sys.stderr) return 2 transfer = Transfer(descriptor, allow_loopback=bool(args.test_file)) added = failed = 0 try: if args.test_file: title = 'Lista de prueba' if descriptor.get('playlist') else None for path in args.test_file: if path.suffix not in ('.mp3', '.mp4'): raise ValueError('El archivo de prueba debe ser MP3 o MP4.') transfer.send(added, path, path.stem, title) added += 1 else: fmt = 'mp3' if descriptor['playlist'] else descriptor['format'] if fmt not in ('mp3', 'mp4'): raise ValueError('Formato inválido.') for position, (source, playlist) in enumerate(jobs(descriptor)): if position == 100: failed += 1 print('Límite de 100 entradas por transferencia.') break if source is None: failed += 1 continue with tempfile.TemporaryDirectory(prefix='spoti-') as directory: try: path, title = download(source, fmt, Path(directory)) except Exception as error: failed += 1 print(f'No se pudo procesar la entrada {position + 1}: {error}') continue try: transfer.send(added, path, title, playlist) except (ValueError, RuntimeError) as error: failed += 1 print(error) break added += 1 print(f'Enviadas: {added}.') except Exception as error: failed += 1 print(error) state = 'done' if added and not failed else 'partial' if added else 'failed' transfer.request('PATCH', '', {'state': state}) print(f'Enviadas: {added}. Fallos: {failed}. Vuelve a Spoti para terminar la importación.') if __name__ == '__main__': sys.exit(main())