"""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')