516 lines
21 KiB
Python
516 lines
21 KiB
Python
"""Self-hosted X360 Web Queue: authenticated FastAPI + one resilient worker."""
|
|
import asyncio
|
|
import contextlib
|
|
import json
|
|
import os
|
|
import re
|
|
import secrets
|
|
import shutil
|
|
import time
|
|
import uuid
|
|
from contextlib import asynccontextmanager
|
|
from pathlib import Path
|
|
from urllib.parse import urlparse, unquote
|
|
|
|
import httpx
|
|
from fastapi import APIRouter, Depends, FastAPI, HTTPException, Request, status
|
|
from fastapi.responses import FileResponse, PlainTextResponse
|
|
from fastapi.security import HTTPBasic, HTTPBasicCredentials
|
|
from fastapi.staticfiles import StaticFiles
|
|
from pydantic import BaseModel, Field
|
|
|
|
from . import archive_auth, archive_download, bt, download, pipeline, store, torrent_meta, vimm
|
|
from .common import (FORMATS, archive_download_url, parse_archive_id, safe_import_file,
|
|
safe_title, validate_url)
|
|
|
|
|
|
basic = HTTPBasic()
|
|
|
|
|
|
def authorize(credentials: HTTPBasicCredentials = Depends(basic)):
|
|
u = os.getenv('WEB_USER', 'admin').encode()
|
|
p = os.getenv('WEB_PASSWORD', '').encode()
|
|
if not p or not (secrets.compare_digest(credentials.username.encode(), u)
|
|
and secrets.compare_digest(credentials.password.encode(), p)):
|
|
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED,
|
|
detail='Nieprawidlowy login lub haslo',
|
|
headers={'WWW-Authenticate': 'Basic'})
|
|
|
|
|
|
class AddJob(BaseModel):
|
|
url: str = Field(min_length=1, max_length=4000)
|
|
title: str = Field(default='', max_length=160)
|
|
format: str = 'xex'
|
|
|
|
|
|
class AddImports(BaseModel):
|
|
filename: str = Field(min_length=1, max_length=300)
|
|
format: str = 'xex'
|
|
|
|
|
|
class AddArchive(BaseModel):
|
|
identifier: str = Field(min_length=1, max_length=250)
|
|
files: list[str] = Field(min_length=1, max_length=150)
|
|
title: str = Field(default='', max_length=160)
|
|
format: str = 'xex'
|
|
|
|
|
|
class ItemLookup(BaseModel):
|
|
identifier: str = Field(min_length=1, max_length=250)
|
|
|
|
|
|
class ArchiveLogin(BaseModel):
|
|
email: str = Field(min_length=3, max_length=254)
|
|
password: str = Field(min_length=1, max_length=1024)
|
|
|
|
|
|
class MagnetInspect(BaseModel):
|
|
magnet: str = Field(min_length=20, max_length=12000)
|
|
|
|
|
|
class MagnetJob(BaseModel):
|
|
preview_id: str = Field(min_length=32, max_length=32)
|
|
files: list[int] = Field(min_length=1, max_length=20000)
|
|
title: str = Field(default='', max_length=160)
|
|
format: str = 'xex'
|
|
|
|
|
|
def new_job(source_type, source, title, fmt):
|
|
if fmt not in FORMATS:
|
|
raise HTTPException(status_code=400, detail='Format: xex, god albo raw')
|
|
jid = str(uuid.uuid4())
|
|
store.create({'id': jid, 'title': safe_title(title or 'Xbox 360'),
|
|
'source_type': source_type, 'source': source, 'format': fmt})
|
|
return store.get(jid)
|
|
|
|
|
|
async def worker():
|
|
while True:
|
|
job = store.next_job()
|
|
if not job:
|
|
await asyncio.sleep(2)
|
|
continue
|
|
jid = job['id']
|
|
try:
|
|
if job['state'] == 'preparing':
|
|
await pipeline.process_job(job)
|
|
elif job['source_type'] == 'import':
|
|
# Local import is copied into persistent raw dir, then converted.
|
|
src = safe_import_file(job['source'])
|
|
target = store.DATA / 'raw' / jid
|
|
target.mkdir(parents=True, exist_ok=True)
|
|
await asyncio.to_thread(shutil.copy2, src, target / src.name)
|
|
store.update(jid, state='preparing', progress=100)
|
|
elif job['source_type'] == 'vimm':
|
|
await vimm.browser_download(job)
|
|
elif job['source_type'] == 'torrent_selected':
|
|
if job['state'] == 'queued':
|
|
await bt.start_download(job)
|
|
else:
|
|
await bt.poll_download(job)
|
|
elif job['source_type'] == 'archive_many' or (job['source_type'] == 'url' and archive_auth.is_archive_download(job['source'])):
|
|
if job['gid']:
|
|
await download.cancel_download(job['gid'])
|
|
store.update(jid, gid='')
|
|
await archive_download.run_one(job)
|
|
elif job['state'] == 'queued':
|
|
# Resume if there is an aria2 GID instead of accidentally
|
|
# starting a duplicate download.
|
|
if job['gid']:
|
|
await download.resume_download(job['gid'])
|
|
store.update(jid, state='downloading')
|
|
else:
|
|
await download.begin(job)
|
|
else:
|
|
await download.poll(job)
|
|
except asyncio.CancelledError:
|
|
raise
|
|
except archive_download.ArchiveStopped:
|
|
pass
|
|
except Exception as ex:
|
|
store.update(jid, state='error', speed=0, error=str(ex)[:1800],
|
|
note='Błąd zadania: otwórz Log')
|
|
await asyncio.sleep(1.5)
|
|
|
|
|
|
@asynccontextmanager
|
|
async def lifespan(app):
|
|
for subdir in ('raw', 'ready', 'work', 'imports', 'logs', 'metadata', 'torrents'):
|
|
(store.DATA / subdir).mkdir(parents=True, exist_ok=True)
|
|
store.init_db()
|
|
task = asyncio.create_task(worker())
|
|
try:
|
|
yield
|
|
finally:
|
|
task.cancel()
|
|
with contextlib.suppress(asyncio.CancelledError):
|
|
await task
|
|
await bt.shutdown()
|
|
|
|
|
|
app = FastAPI(title='X360 Web Queue', version='0.1.0', lifespan=lifespan,
|
|
docs_url=None, redoc_url=None, openapi_url=None)
|
|
api = APIRouter(prefix='/api', dependencies=[Depends(authorize)])
|
|
MAX_INSPECT_SECONDS = 300
|
|
|
|
|
|
@app.middleware('http')
|
|
async def guard_cross_origin_requests(request: Request, call_next):
|
|
if request.method in ('POST', 'PUT', 'PATCH', 'DELETE'):
|
|
origin = request.headers.get('origin')
|
|
if origin and urlparse(origin).netloc != request.headers.get('host'):
|
|
return PlainTextResponse('Nieprawidlowa domena zrodla zadania', status_code=403)
|
|
return await call_next(request)
|
|
|
|
|
|
@app.get('/')
|
|
async def index(_: None = Depends(authorize)):
|
|
return FileResponse(Path(__file__).parent / 'static' / 'index.html')
|
|
|
|
|
|
@app.get('/healthz')
|
|
async def healthz():
|
|
return {'ok': True}
|
|
|
|
|
|
@api.get('/jobs')
|
|
async def jobs():
|
|
return store.list_jobs()
|
|
|
|
|
|
@api.post('/jobs')
|
|
async def add_download(body: AddJob):
|
|
try:
|
|
kind, url = validate_url(body.url)
|
|
except ValueError as ex:
|
|
raise HTTPException(status_code=400, detail=str(ex)) from ex
|
|
parsed = urlparse(url)
|
|
guessed = unquote(Path(parsed.path).name) if kind != 'magnet' else 'Torrent'
|
|
if kind == 'vimm':
|
|
guessed = 'Vimm-' + parsed.path.rstrip('/').split('/')[-1]
|
|
if kind == 'magnet':
|
|
raise HTTPException(400, 'Magnet wymaga najpierw wyboru plików (/api/magnet/inspect)')
|
|
return new_job(kind, url, body.title or guessed, body.format)
|
|
|
|
|
|
@api.get('/archive/session')
|
|
async def archive_session():
|
|
return archive_auth.session_status()
|
|
|
|
|
|
@api.post('/archive/session')
|
|
async def archive_login(body: ArchiveLogin):
|
|
if '@' not in body.email:
|
|
raise HTTPException(400, 'Niepoprawny adres e-mail')
|
|
try:
|
|
return await archive_auth.login(body.email.strip(), body.password)
|
|
except archive_auth.ArchiveAuthError as exc:
|
|
raise HTTPException(400, str(exc)) from exc
|
|
|
|
|
|
@api.delete('/archive/session')
|
|
async def archive_logout():
|
|
archive_auth.logout()
|
|
return archive_auth.session_status()
|
|
|
|
|
|
@app.api_route('/internal/archive/{job_id}/{part_index}', methods=['GET', 'HEAD'])
|
|
async def archive_proxy(job_id: str, part_index: int, request: Request):
|
|
if (not 0 <= part_index <= 149 or
|
|
not archive_auth.verify_token(job_id, part_index, request.headers.get('x-x360-proxy-token'))):
|
|
raise HTTPException(403, 'Odmowa dostępu')
|
|
job = store.get(job_id)
|
|
if not job or job['source_type'] not in ('url', 'archive_many'):
|
|
raise HTTPException(404, 'Nie znaleziono zadania')
|
|
urls = download.source_urls(job)
|
|
if part_index >= len(urls) or not archive_auth.is_archive_download(urls[part_index]):
|
|
raise HTTPException(404, 'Nie znaleziono pliku')
|
|
return await archive_auth.stream_archive(urls[part_index], request.method, request.headers)
|
|
|
|
|
|
@api.post('/magnet/inspect')
|
|
async def magnet_inspect(body: MagnetInspect):
|
|
try:
|
|
kind, magnet = validate_url(body.magnet)
|
|
if kind != 'magnet':
|
|
raise ValueError('Wklej link magnet')
|
|
except ValueError as exc:
|
|
raise HTTPException(400, str(exc)) from exc
|
|
await asyncio.to_thread(torrent_meta.cleanup_old)
|
|
root = store.DATA / 'metadata'
|
|
active_previews = sum(1 for item in root.iterdir() if item.is_dir() and not item.is_symlink() and
|
|
(item / 'request.json').is_file() and
|
|
not list(item.glob('*.torrent')) and
|
|
time.time() - item.stat().st_mtime < MAX_INSPECT_SECONDS)
|
|
if active_previews >= 3:
|
|
raise HTTPException(429, 'Trzy podglądy już pobierają metadane. Poczekaj na zakończenie.')
|
|
preview_id = uuid.uuid4().hex
|
|
directory = torrent_meta.inspect_dir(preview_id)
|
|
directory.mkdir(parents=True, mode=0o700)
|
|
try:
|
|
await bt.start_inspection(preview_id, magnet)
|
|
torrent_meta.write_request(preview_id, {'gid': '', 'magnet': magnet, 'engine': 'anacrolix'})
|
|
except (RuntimeError, httpx.HTTPError, OSError) as exc:
|
|
await bt.stop_inspection(preview_id)
|
|
await asyncio.to_thread(shutil.rmtree, directory, True)
|
|
raise HTTPException(502, f'Nie można rozpocząć odczytu metadanych: {exc}') from exc
|
|
return {'preview_id': preview_id, 'status': 'loading'}
|
|
|
|
|
|
@api.post('/magnet/upload')
|
|
async def torrent_upload(request: Request):
|
|
# No multipart dependency. Transfer raw .torrent bytes with a fixed size limit.
|
|
if request.headers.get('content-type', '').split(';', 1)[0].strip() != 'application/x-bittorrent':
|
|
raise HTTPException(415, 'Wymagany plik .torrent')
|
|
raw = bytearray()
|
|
async for chunk in request.stream():
|
|
if len(raw) + len(chunk) > torrent_meta.MAX_METADATA:
|
|
raise HTTPException(413, 'Plik .torrent przekracza limit 32 MiB')
|
|
raw.extend(chunk)
|
|
if not raw:
|
|
raise HTTPException(400, 'Pusty plik .torrent')
|
|
preview_id = uuid.uuid4().hex
|
|
directory = torrent_meta.inspect_dir(preview_id)
|
|
directory.mkdir(parents=True, exist_ok=False, mode=0o700)
|
|
target = directory / 'uploaded.torrent'
|
|
try:
|
|
# Validate before storing the metadata. No game payload is transferred.
|
|
decoded = torrent_meta.decode_bencode(bytes(raw))
|
|
if b'info' not in decoded:
|
|
raise torrent_meta.MetadataError('Plik nie zawiera pola info')
|
|
target.write_bytes(raw)
|
|
torrent_meta.read_torrent(target)
|
|
torrent_meta.write_request(preview_id, {'gid': '', 'magnet': ''})
|
|
except (torrent_meta.MetadataError, OSError) as exc:
|
|
await asyncio.to_thread(shutil.rmtree, directory, True)
|
|
raise HTTPException(400, str(exc)) from exc
|
|
return {'preview_id': preview_id, 'status': 'ready'}
|
|
|
|
|
|
@api.get('/magnet/inspect/{preview_id}')
|
|
async def magnet_inspect_status(preview_id: str):
|
|
try:
|
|
req = torrent_meta.load_request(preview_id)
|
|
path = torrent_meta.torrent_file(preview_id)
|
|
if path:
|
|
info = torrent_meta.read_torrent(path)
|
|
await bt.stop_inspection(preview_id)
|
|
info.update({'status': 'ready', 'suggested': torrent_meta.suggested_indices(req['magnet'], len(info['files']))})
|
|
return info
|
|
elapsed = max(0, int(time.time() - req['created']))
|
|
status = bt.inspect_state(preview_id)
|
|
if status.get('status') == 'error':
|
|
return {'status': 'error', 'error': status.get('error', 'Silnik BT zgłosił błąd')[:500]}
|
|
if elapsed > MAX_INSPECT_SECONDS:
|
|
await bt.stop_inspection(preview_id)
|
|
return {'status': 'error', 'error': 'Brak metadanych po 5 minutach. Sprawdź dostępność peerów i DHT/UDP.'}
|
|
return {'status': 'loading', 'elapsed': elapsed,
|
|
'connections': int(status.get('connections') or 0),
|
|
'seeders': 0, 'known_peers': int(status.get('known_peers') or 0),
|
|
'aria_status': 'anacrolix'}
|
|
except torrent_meta.MetadataError as exc:
|
|
raise HTTPException(400, str(exc)) from exc
|
|
except (RuntimeError, httpx.HTTPError) as exc:
|
|
raise HTTPException(502, 'aria2: ' + str(exc)[:200]) from exc
|
|
|
|
|
|
@api.post('/magnet/inspect/{preview_id}/cancel')
|
|
async def magnet_inspect_cancel(preview_id: str):
|
|
try:
|
|
req = torrent_meta.load_request(preview_id)
|
|
except torrent_meta.MetadataError as exc:
|
|
raise HTTPException(400, str(exc)) from exc
|
|
if req.get('engine') == 'anacrolix':
|
|
await bt.stop_inspection(preview_id)
|
|
else:
|
|
await download.cancel_download(req['gid'])
|
|
directory = torrent_meta.inspect_dir(preview_id)
|
|
await asyncio.to_thread(shutil.rmtree, directory, True)
|
|
return {'ok': True}
|
|
|
|
|
|
@api.post('/magnet/jobs')
|
|
async def magnet_selected_job(body: MagnetJob):
|
|
if body.format not in FORMATS:
|
|
raise HTTPException(400, 'Format: xex, god albo raw')
|
|
try:
|
|
torrent_meta.load_request(body.preview_id)
|
|
path = torrent_meta.torrent_file(body.preview_id)
|
|
if path is None:
|
|
raise torrent_meta.MetadataError('Metadane jeszcze nie są gotowe')
|
|
info = torrent_meta.read_torrent(path)
|
|
await bt.stop_inspection(body.preview_id)
|
|
valid = {item['index'] for item in info['files'] if not item['padding']}
|
|
if len(body.files) != len(set(body.files)) or not set(body.files).issubset(valid):
|
|
raise torrent_meta.MetadataError('Wybierz poprawne indeksy plików')
|
|
except torrent_meta.MetadataError as exc:
|
|
raise HTTPException(400, str(exc)) from exc
|
|
jid = str(uuid.uuid4())
|
|
dest = store.DATA / 'torrents' / (jid + '.torrent')
|
|
dest.parent.mkdir(parents=True, exist_ok=True)
|
|
await asyncio.to_thread(shutil.copyfile, path, dest)
|
|
selection = ','.join(str(i) for i in sorted(body.files))
|
|
data = json.dumps({'selection': selection, 'count': len(body.files), 'name': info['name']})
|
|
try:
|
|
store.create({'id': jid, 'title': safe_title(body.title or info['name']),
|
|
'source_type': 'torrent_selected', 'source': data, 'format': body.format})
|
|
except Exception:
|
|
dest.unlink(missing_ok=True)
|
|
raise
|
|
return store.get(jid)
|
|
|
|
|
|
@api.post('/archive/lookup')
|
|
async def archive_lookup(body: ItemLookup):
|
|
try:
|
|
identifier = parse_archive_id(body.identifier)
|
|
except ValueError as ex:
|
|
raise HTTPException(400, str(ex)) from ex
|
|
try:
|
|
async with httpx.AsyncClient(timeout=25, follow_redirects=True, cookies=archive_auth.session_cookies()) as client:
|
|
res = await client.get('https://archive.org/metadata/' + identifier)
|
|
res.raise_for_status()
|
|
data = res.json()
|
|
except (httpx.HTTPError, ValueError) as ex:
|
|
raise HTTPException(502, 'Archive.org metadata: ' + str(ex)) from ex
|
|
result = []
|
|
for entry in data.get('files', []):
|
|
name = entry.get('name', '')
|
|
if not name or entry.get('source') == 'metadata' or name.startswith('__'):
|
|
continue
|
|
# Show all file types with an extension; user chooses exact archive parts.
|
|
if '.' not in Path(name).name:
|
|
continue
|
|
try:
|
|
archive_download_url(identifier, name)
|
|
except ValueError:
|
|
continue
|
|
result.append({'name': name, 'size': int(entry.get('size') or 0),
|
|
'type': entry.get('format', '')})
|
|
return {'identifier': identifier, 'title': data.get('metadata', {}).get('title', identifier),
|
|
'files': result[:1000]}
|
|
|
|
|
|
@api.post('/archive/jobs')
|
|
async def add_archive(body: AddArchive):
|
|
try:
|
|
identifier = parse_archive_id(body.identifier)
|
|
if len(set(body.files)) != len(body.files):
|
|
raise ValueError('Pliki nie moga sie powtarzac')
|
|
urls = [archive_download_url(identifier, file) for file in body.files]
|
|
if len({Path(unquote(urlparse(u).path)).name for u in urls}) != len(urls):
|
|
raise ValueError('Pliki maja identyczne nazwy; wybierz jeden folder na zadanie')
|
|
except ValueError as ex:
|
|
raise HTTPException(400, str(ex)) from ex
|
|
# Preserving filename order is important for multipart archives.
|
|
return new_job('archive_many', json.dumps(urls), body.title or identifier, body.format)
|
|
|
|
|
|
@api.get('/imports')
|
|
async def imports():
|
|
root = store.DATA / 'imports'
|
|
result = []
|
|
for item in root.rglob('*'):
|
|
if len(result) >= 200:
|
|
break
|
|
if item.is_file() and not item.is_symlink() and item.resolve().is_relative_to(root.resolve()):
|
|
result.append({'name': str(item.relative_to(root)), 'size': item.stat().st_size})
|
|
return result
|
|
|
|
|
|
@api.post('/imports/jobs')
|
|
async def import_job(body: AddImports):
|
|
try:
|
|
path = safe_import_file(body.filename)
|
|
except ValueError as ex:
|
|
raise HTTPException(400, str(ex)) from ex
|
|
return new_job('import', body.filename, path.stem, body.format)
|
|
|
|
|
|
@api.post('/jobs/{job_id}/{action}')
|
|
async def manage(job_id: str, action: str):
|
|
job = store.get(job_id)
|
|
if not job:
|
|
raise HTTPException(404, 'Nie znaleziono zadania')
|
|
state = job['state']
|
|
if action == 'pause' and state in ('queued', 'downloading'):
|
|
if state == 'downloading':
|
|
try:
|
|
if job['source_type'] == 'torrent_selected':
|
|
await bt.stop_download(job_id)
|
|
else:
|
|
await download.pause_download(job['gid'])
|
|
except (RuntimeError, httpx.HTTPError) as ex:
|
|
raise HTTPException(502, str(ex)) from ex
|
|
store.update(job_id, state='paused', speed=0, note='Wstrzymano')
|
|
elif action == 'resume' and state == 'paused':
|
|
store.update(job_id, state='queued', note='Wznowienie zadania')
|
|
elif action == 'cancel' and state in ('queued', 'paused', 'downloading'):
|
|
if job['source_type'] == 'torrent_selected':
|
|
await bt.stop_download(job_id)
|
|
elif job['gid']:
|
|
await download.cancel_download(job['gid'])
|
|
store.update(job_id, state='cancelled', speed=0, note='Anulowano')
|
|
elif action == 'retry' and state in ('error', 'cancelled'):
|
|
# For already-downloaded input, retry conversion without redownloading.
|
|
root = store.DATA / 'raw' / job_id
|
|
good_files = ([p for p in root.rglob('*') if p.is_file() and not p.name.endswith('.aria2')]
|
|
if root.exists() else [])
|
|
complete = job['progress'] >= 99.9 and bool(good_files)
|
|
if job['source_type'] == 'archive_many':
|
|
# One finished file (or a .part segment) never means the whole
|
|
# multi-file archive is ready to convert.
|
|
complete = all((root / download.target_filename(url)).is_file()
|
|
for url in download.source_urls(job))
|
|
store.update(job_id, state='preparing' if complete else 'queued',
|
|
gid='' if not complete else job['gid'], error='',
|
|
note='Ponowienie zadania')
|
|
elif action == 'remove' and state in ('error','cancelled','done','paused'):
|
|
if job['source_type'] == 'torrent_selected':
|
|
await bt.stop_download(job_id)
|
|
elif state == 'paused' and job['gid']:
|
|
await download.cancel_download(job['gid'])
|
|
store.erase(job_id)
|
|
return {'ok': True, 'note': 'Usunieto wpis z kolejki, pliki pozostaly na dysku'}
|
|
else:
|
|
raise HTTPException(409, 'Ta akcja nie jest dostepna dla obecnego stanu zadania')
|
|
return store.get(job_id)
|
|
|
|
|
|
@api.get('/jobs/{job_id}/log')
|
|
async def job_log(job_id: str):
|
|
if not store.get(job_id):
|
|
raise HTTPException(404, 'Nie znaleziono zadania')
|
|
path = store.DATA / 'logs' / f'{job_id}.log'
|
|
job = store.get(job_id)
|
|
lines = [f"Stan: {job['state']}", f"Etap: {job['note'] or 'brak'}"]
|
|
if job['error']:
|
|
lines.append('Błąd pobierania / zadania: ' + job['error'])
|
|
bt_log = store.DATA / 'logs' / f'{job_id}.bt.log'
|
|
if bt_log.is_file():
|
|
lines.append('BitTorrent (ostatnie 12 KiB):')
|
|
lines.append(bt_log.read_bytes()[-12000:].decode('utf-8', errors='replace'))
|
|
download_log = store.DATA / 'logs' / f'{job_id}.download.log'
|
|
if download_log.is_file():
|
|
lines.append('Pobieranie (ostatnie 12 KiB):')
|
|
lines.append(download_log.read_bytes()[-12000:].decode('utf-8', errors='replace'))
|
|
lines.append('')
|
|
lines.append(path.read_bytes()[-24000:].decode('utf-8', errors='replace')
|
|
if path.is_file() else 'Konwersja jeszcze nie została uruchomiona.')
|
|
return PlainTextResponse('\n'.join(lines))
|
|
|
|
|
|
@api.get('/system')
|
|
async def system():
|
|
usage = shutil.disk_usage(store.DATA / 'raw')
|
|
host_dir = os.getenv('XBOX_HOST_DIR', '/xbox').rstrip('/') or '/xbox'
|
|
return {'free': usage.free, 'total': usage.total,
|
|
'host_data_dir': host_dir,
|
|
'formats': ['xex','god','raw'],
|
|
'vimm_enabled': os.getenv('ENABLE_VIMM_BROWSER', 'true').lower() in ('1','yes','true'),
|
|
'bt_engine': 'anacrolix (Go)', 'bt_port': os.getenv('BT_PORT', '6881')}
|
|
|
|
|
|
app.include_router(api)
|
|
app.mount('/static', StaticFiles(directory=str(Path(__file__).parent / 'static')), name='static')
|