"""Direct, authenticated Internet Archive downloader.

Avoids the old aria2 -> localhost HTTP relay (which hid origin errors behind
HTTP 500). Supports Range resumes, transient 5xx/429 retries and pause/cancel.
Only forwards IA cookies via httpx's domain-scoped cookie jar.
"""
import hashlib
import json
import os
import re
import shutil
import time
from collections import deque
from concurrent.futures import FIRST_COMPLETED, ThreadPoolExecutor, wait
from contextlib import closing
from threading import Event, Lock
from pathlib import Path
from urllib.parse import urljoin

import httpx

from . import archive_auth, store
from .download import source_urls, target_filename

RETRY_CODES = {408, 425, 429, 500, 502, 503, 504}
MAX_ATTEMPTS = 5
MAX_CONNECTIONS = 50
MIN_SEGMENT_BYTES = 8 * 1024 * 1024
CHUNK_BYTES = 256 * 1024
RANGE = re.compile(r'^bytes (\d+)-(\d+)/(\d+|\*)$', re.I)
FINISH_RANGE = re.compile(r'^bytes \*/(\d+)$', re.I)
LOG_LOCK = Lock()


class ArchiveDownloadError(RuntimeError):
    pass


class ArchiveStopped(Exception):
    """Pause or cancellation was requested, not a transfer error."""


def _client():
    return httpx.Client(timeout=httpx.Timeout(90, connect=25), cookies=archive_auth.session_cookies(),
                        follow_redirects=False, trust_env=True)


def _check_state(jid, stop=None):
    if stop is not None and stop.is_set():
        raise ArchiveStopped()
    current = store.get(jid)
    if not current or current['state'] in ('paused', 'cancelled', 'error'):
        raise ArchiveStopped()


def _response(client, url, headers):
    current = url
    for _ in range(8):
        if not archive_auth.allowed_redirect(current):
            raise ArchiveDownloadError('Archive.org: nieprawidłowe przekierowanie poza domenę Archive.org')
        request = client.build_request('GET', current, headers=headers)
        response = client.send(request, stream=True)
        if response.status_code in (301, 302, 303, 307, 308) and 'location' in response.headers:
            next_url = urljoin(current, response.headers['location'])
            response.close()
            current = next_url
            continue
        return response
    raise ArchiveDownloadError('Archive.org: zbyt wiele przekierowań')


def _log(job_id, message):
    target = store.DATA / 'logs' / (job_id + '.download.log')
    target.parent.mkdir(parents=True, exist_ok=True)
    with LOG_LOCK:
        with target.open('a', encoding='utf-8') as fp:
            fp.write(time.strftime('%Y-%m-%d %H:%M:%S') + ' ' + message + '\n')


def _transfer_one(job, *, batch=None, connections=None, stop=None):
    jid = job['id']
    ix = job['part_index']
    urls = source_urls(job)
    if ix >= len(urls):
        store.update(jid, state='preparing')
        return
    uri = urls[ix]
    if not archive_auth.is_archive_download(uri):
        raise ArchiveDownloadError('Nieprawidłowy URL pobierania Archive.org')
    root = store.DATA / 'raw' / jid
    root.mkdir(parents=True, exist_ok=True)
    filename = target_filename(uri)
    if not filename:
        raise ArchiveDownloadError('Brak nazwy pliku Archive.org')
    destination = root / filename
    partial = root / (filename + '.part')
    _check_state(jid, stop)
    if batch is None:
        store.update(jid, state='downloading', gid='', note=f'Archive.org: plik {ix+1}/{len(urls)}')
    _log(jid, f'Rozpoczęto plik {ix+1}/{len(urls)}: {filename}')

    if connections is None:
        connections = _connection_limit()
    # An existing manifest may contain many segments even if the new budget
    # only allows one active connection. Keep those downloaded bytes.
    if (connections > 1 or _segment_dir(partial).exists()) and _parallel_transfer(
            job, ix, urls, uri, filename, partial, destination, connections, batch=batch, stop=stop):
        return
    # If the server cannot serve ranges (or parallel transfer was disabled),
    # keep the verified single-stream prefix, not an unfinished assembly.
    _discard_parallel_state(partial)
    # Existing aria2 control files are not compatible with this downloader.
    # Start from our own .part file rather than assuming an aria2 file is complete.
    with _client() as client:
        for attempt in range(MAX_ATTEMPTS):
            _check_state(jid, stop)
            start = partial.stat().st_size if partial.exists() else 0
            headers = {'User-Agent': 'X360-Web-Queue/2.0', 'Accept-Encoding': 'identity'}
            if start:
                headers['Range'] = f'bytes={start}-'
            try:
                with closing(_response(client, uri, headers)) as response:
                    code = response.status_code
                    if code in RETRY_CODES:
                        _log(jid, f'HTTP {code} z Archive.org, próba {attempt+1}/{MAX_ATTEMPTS}')
                        reason = f'Archive.org HTTP {code}'
                        if attempt == MAX_ATTEMPTS - 1:
                            raise ArchiveDownloadError(f'{reason} po {MAX_ATTEMPTS} próbach; możliwy problem serwera Archive.org')
                    elif code in (401, 403):
                        raise ArchiveDownloadError(f'Archive.org HTTP {code}: brak dostępu. Sprawdź sesję i uprawnienia do pliku.')
                    elif code == 404:
                        raise ArchiveDownloadError('Archive.org HTTP 404: nie znaleziono wybranego pliku')
                    elif code == 416 and start:
                        m = FINISH_RANGE.fullmatch(response.headers.get('Content-Range', '').strip())
                        if m and int(m.group(1)) == start:
                            _check_state(jid, stop)
                            partial.replace(destination)
                            _finish(job, ix, urls, batch=batch)
                            _log(jid, f'Ukończono {filename} (HTTP 416 potwierdziło rozmiar)')
                            return
                        partial.unlink(missing_ok=True)
                        continue
                    elif code not in (200, 206):
                        raise ArchiveDownloadError(f'Archive.org HTTP {code}: odmowa pobrania pliku')
                    else:
                        if 'text/html' in response.headers.get('Content-Type', '').lower():
                            raise ArchiveDownloadError('Archive.org zwróciło HTML zamiast pliku. Sprawdź dostęp i logowanie.')
                        if code == 206:
                            match = RANGE.fullmatch(response.headers.get('Content-Range', '').strip())
                            if not match or int(match.group(1)) != start:
                                raise ArchiveDownloadError('Archive.org zwróciło niezgodny zakres HTTP Range')
                            total = int(match.group(3)) if match.group(3) != '*' else 0
                        else:
                            # Server ignored Range: do not append duplicate bytes.
                            start = 0
                            total = int(response.headers.get('Content-Length') or 0)
                        written = start
                        last_update = 0.0
                        prev = written
                        prev_at = time.monotonic()
                        mode = 'ab' if start else 'wb'
                        with partial.open(mode) as fp:
                            for chunk in response.iter_raw(chunk_size=256 * 1024):
                                _check_state(jid, stop)
                                if not chunk:
                                    continue
                                fp.write(chunk)
                                written += len(chunk)
                                now = time.monotonic()
                                if now - last_update >= 0.8:
                                    elapsed = max(0.001, now - prev_at)
                                    speed = int((written - prev) / elapsed)
                                    prev, prev_at, last_update = written, now, now
                                    portion = written / total if total else 0
                                    if batch is not None:
                                        batch.report(ix, written, total)
                                    else:
                                        store.update(jid, progress=round((ix + min(portion, 1)) / len(urls) * 100, 2),
                                                     speed=speed, total=total, completed=written,
                                                     note=f'Archive.org: plik {ix+1}/{len(urls)}, próba {attempt+1}')
                        if total and written != total:
                            raise httpx.ReadError(f'Przerwany transfer: pobrano {written} z {total} bajtów')
                        _check_state(jid, stop)
                        partial.replace(destination)
                        _finish(job, ix, urls, batch=batch)
                        _log(jid, f'Ukończono {filename}: {written} bajtów')
                        return
            except ArchiveStopped:
                raise
            except ArchiveDownloadError:
                raise
            except (httpx.HTTPError, OSError) as exc:
                _log(jid, f'Błąd transferu {type(exc).__name__}: {str(exc)[:180]}')
                if attempt == MAX_ATTEMPTS - 1:
                    raise ArchiveDownloadError(f'Archive.org: przerwane połączenie po {MAX_ATTEMPTS} próbach: {type(exc).__name__}: {exc}') from exc
            # Applies to transient HTTP codes and connection exceptions.
            if batch is None:
                store.update(jid, note=f'Archive.org: ponawianie po HTTP/błędzie sieci ({attempt+2}/{MAX_ATTEMPTS})', speed=0)
            for _ in range(min(2 ** attempt, 12) * 2):
                _check_state(jid, stop)
                time.sleep(0.5)


def _finish(job, ix, urls, *, batch=None):
    jid = job['id']
    if batch is not None:
        batch.complete(ix)
        return
    if ix + 1 < len(urls):
        store.update(jid, part_index=ix+1, state='queued', gid='', speed=0,
                     progress=round((ix+1)/len(urls) * 100, 2), note='Pobrano część archiwum')
    else:
        store.update(jid, state='preparing', progress=100, speed=0,
                     note='Pobieranie Archive.org ukończone')



class _RangeUnsupported(Exception):
    """A server stopped honoring byte ranges; caller can use the serial path."""
    def __init__(self, message, reset_prefix=False):
        super().__init__(message)
        self.reset_prefix = reset_prefix


def _connection_limit():
    try:
        return max(1, min(MAX_CONNECTIONS, int(os.getenv('ARCHIVE_CONNECTIONS', '50'))))
    except (ValueError, OverflowError):
        return 50


def _segment_dir(partial):
    return Path(str(partial) + '.segments')


def _read_manifest(folder):
    try:
        item = json.loads((folder / 'manifest.json').read_text(encoding='utf-8'))
        if not isinstance(item, dict):
            return None
        return item
    except (OSError, ValueError):
        return None


def _write_manifest(folder, manifest):
    folder.mkdir(mode=0o700, parents=True, exist_ok=True)
    temp = folder / 'manifest.json.tmp'
    with temp.open('w', encoding='utf-8') as f:
        json.dump(manifest, f, sort_keys=True)
        f.flush()
        os.fsync(f.fileno())
    temp.replace(folder / 'manifest.json')


def _discard_parallel_state(partial, reset_prefix=False):
    """Discard stale segment files without interpreting an assembled prefix as a resume."""
    folder = _segment_dir(partial)
    if not folder.exists():
        return
    manifest = _read_manifest(folder)
    if reset_prefix:
        partial.unlink(missing_ok=True)
    elif manifest and manifest.get('assembling') and partial.exists():
        prefix = manifest.get('prefix')
        if isinstance(prefix, int) and 0 <= prefix <= partial.stat().st_size:
            with partial.open('r+b') as f:
                f.truncate(prefix)
        else:
            partial.unlink(missing_ok=True)
    shutil.rmtree(folder)


def _validate_ranges(ranges, prefix, total):
    if not isinstance(ranges, list) or len(ranges) < 2 or len(ranges) > MAX_CONNECTIONS:
        return False
    cursor = prefix
    for item in ranges:
        if (not isinstance(item, list) or len(item) != 2 or
                not all(isinstance(x, int) and not isinstance(x, bool) for x in item)):
            return False
        start, end = item
        if start != cursor or end < start or end >= total:
            return False
        cursor = end + 1
    return cursor == total


def _prepare_segments(partial, uri, total, validator, connections):
    folder = _segment_dir(partial)
    source_hash = hashlib.sha256(uri.encode('utf-8')).hexdigest()
    manifest = _read_manifest(folder)
    valid = False
    if manifest:
        prefix = manifest.get('prefix')
        numeric_prefix = isinstance(prefix, int) and not isinstance(prefix, bool)
        prefix_present = numeric_prefix and (prefix == 0 or
                                             (partial.exists() and partial.stat().st_size >= prefix))
        valid = (manifest.get('version') == 1 and manifest.get('source_hash') == source_hash and
                 manifest.get('total') == total and manifest.get('validator') == validator and
                 isinstance(prefix, int) and not isinstance(prefix, bool) and 0 <= prefix < total and
                 _validate_ranges(manifest.get('ranges'), prefix, total) and prefix_present)
        if valid and partial.exists() and partial.stat().st_size != prefix:
            # A crash during assembly is safe to recover from: parts remain intact.
            if manifest.get('assembling') and partial.stat().st_size > prefix:
                with partial.open('r+b') as f:
                    f.truncate(prefix)
            else:
                valid = False
    if valid:
        return folder, manifest

    if folder.exists():
        # Invalid or mismatched metadata cannot establish which bytes are safe.
        _discard_parallel_state(partial, reset_prefix=True)
    prefix = partial.stat().st_size if partial.exists() else 0
    if prefix > total:
        partial.unlink(missing_ok=True)
        prefix = 0
    if prefix >= total:
        return None, {'prefix': prefix, 'total': total}
    remaining = total - prefix
    n = min(connections, max(1, remaining // MIN_SEGMENT_BYTES))
    if n < 2:
        return None, None
    quotient, extra = divmod(remaining, n)
    cursor = prefix
    ranges = []
    for i in range(n):
        end = cursor + quotient + (1 if i < extra else 0) - 1
        ranges.append([cursor, end])
        cursor = end + 1
    manifest = {'version': 1, 'source_hash': source_hash, 'total': total,
                'validator': validator, 'prefix': prefix, 'ranges': ranges,
                'assembling': False}
    _write_manifest(folder, manifest)
    return folder, manifest


def _retry_wait(job_id, attempt, stop, retry_after=None, external_stop=None):
    delay = min(2 ** attempt, 16)
    if retry_after and retry_after.isdigit():
        delay = min(max(1, int(retry_after)), 60)
    for _ in range(delay * 2):
        if stop.is_set():
            raise ArchiveStopped()
        _check_state(job_id, external_stop)
        time.sleep(0.5)


def _check_content(response):
    if 'text/html' in response.headers.get('Content-Type', '').lower():
        raise ArchiveDownloadError('Archive.org zwróciło HTML zamiast pliku. Sprawdź dostęp i logowanie.')
    if response.headers.get('Content-Encoding', 'identity').lower() not in ('', 'identity'):
        raise ArchiveDownloadError('Archive.org zignorowało Accept-Encoding: identity; transfer w tym kodowaniu jest niebezpieczny')


def _probe_range(job_id, uri, stop=None):
    """Only a real 206 with a numeric total enables parallel mode."""
    headers = {'User-Agent': 'X360-Web-Queue/2.1', 'Accept-Encoding': 'identity', 'Range': 'bytes=0-0'}
    with _client() as client:
        for attempt in range(MAX_ATTEMPTS):
            _check_state(job_id, stop)
            try:
                with closing(_response(client, uri, headers)) as response:
                    code = response.status_code
                    if code in RETRY_CODES:
                        _log(job_id, f'Archive.org: sprawdzenie Range HTTP {code}, próba {attempt+1}/{MAX_ATTEMPTS}')
                        if attempt == MAX_ATTEMPTS - 1:
                            raise ArchiveDownloadError(f'Archive.org HTTP {code} podczas sprawdzania Range po {MAX_ATTEMPTS} próbach')
                        delay = response.headers.get('Retry-After')
                    elif code in (401, 403):
                        raise ArchiveDownloadError(f'Archive.org HTTP {code}: brak dostępu. Sprawdź sesję i uprawnienia.')
                    elif code == 404:
                        raise ArchiveDownloadError('Archive.org HTTP 404: nie znaleziono pliku')
                    elif code == 200:
                        _check_content(response)
                        return None
                    elif code == 206:
                        _check_content(response)
                        match = RANGE.fullmatch(response.headers.get('Content-Range', '').strip())
                        if not match or (int(match.group(1)), int(match.group(2))) != (0, 0) or match.group(3) == '*':
                            return None
                        total = int(match.group(3))
                        if total <= 0:
                            return None
                        etag = response.headers.get('ETag', '').strip()
                        if etag.startswith('W/'):
                            etag = ''
                        validator = etag or response.headers.get('Last-Modified', '').strip()
                        if not validator or len(validator) > 256 or '\n' in validator:
                            validator = ''
                        return total, validator
                    else:
                        raise ArchiveDownloadError(f'Archive.org HTTP {code}: odmowa pobrania pliku')
            except ArchiveDownloadError:
                raise
            except (httpx.HTTPError, OSError) as exc:
                _log(job_id, f'Archive.org: sprawdzenie Range - {type(exc).__name__}: {str(exc)[:150]}')
                if attempt == MAX_ATTEMPTS - 1:
                    raise ArchiveDownloadError(f'Archive.org: brak odpowiedzi na żądanie Range po {MAX_ATTEMPTS} próbach') from exc
                delay = None
            _retry_wait(job_id, attempt, Event(), delay, stop)
    return None


def _segment_worker(job_id, uri, folder, manifest, index, stop, external_stop=None):
    begin, end = manifest['ranges'][index]
    total = manifest['total']
    limit = end - begin + 1
    target = folder / f'{index:02d}.part'
    if target.exists() and target.stat().st_size > limit:
        target.unlink()
    for attempt in range(MAX_ATTEMPTS):
        if stop.is_set():
            raise ArchiveStopped()
        _check_state(job_id, external_stop)
        present = target.stat().st_size if target.exists() else 0
        if present == limit:
            return
        headers = {'User-Agent': 'X360-Web-Queue/2.1', 'Accept-Encoding': 'identity',
                   'Range': f'bytes={begin + present}-{end}'}
        if manifest['validator']:
            headers['If-Range'] = manifest['validator']
        try:
            with _client() as client:
                with closing(_response(client, uri, headers)) as response:
                    code = response.status_code
                    if code in RETRY_CODES:
                        _log(job_id, f'Segment {index+1}/{len(manifest["ranges"])}: HTTP {code}, próba {attempt+1}/{MAX_ATTEMPTS}')
                        if attempt == MAX_ATTEMPTS - 1:
                            raise ArchiveDownloadError(f'Segment {index+1}/{len(manifest["ranges"])}: HTTP {code} po {MAX_ATTEMPTS} próbach')
                        delay = response.headers.get('Retry-After')
                    elif code in (401, 403):
                        raise ArchiveDownloadError(f'Segment {index+1}: Archive.org HTTP {code}: brak dostępu')
                    elif code == 200:
                        raise _RangeUnsupported('Archive.org zignorowało zakres HTTP Range / If-Range', reset_prefix=True)
                    elif code != 206:
                        raise ArchiveDownloadError(f'Segment {index+1}: Archive.org HTTP {code}')
                    else:
                        _check_content(response)
                        match = RANGE.fullmatch(response.headers.get('Content-Range', '').strip())
                        if not match or match.group(3) == '*' or (int(match.group(1)), int(match.group(2)), int(match.group(3))) != (begin + present, end, total):
                            raise _RangeUnsupported('Archive.org zwróciło niezgodny Content-Range', reset_prefix=True)
                        if manifest['validator']:
                            header = 'ETag' if manifest['validator'].startswith('"') else 'Last-Modified'
                            received_validator = response.headers.get(header, '').strip()
                            if received_validator and received_validator != manifest['validator']:
                                raise _RangeUnsupported('Archive.org zmieniło plik podczas pobierania', reset_prefix=True)
                        last_check = time.monotonic()
                        with target.open('ab') as f:
                            for chunk in response.iter_raw(chunk_size=CHUNK_BYTES):
                                if stop.is_set():
                                    raise ArchiveStopped()
                                if time.monotonic() - last_check >= 1:
                                    _check_state(job_id, external_stop)
                                    last_check = time.monotonic()
                                if not chunk:
                                    continue
                                if f.tell() + len(chunk) > limit:
                                    raise _RangeUnsupported('Archive.org wysłało zbyt dużo danych w zakresie', reset_prefix=True)
                                f.write(chunk)
                            f.flush()
                            os.fsync(f.fileno())
                        if target.stat().st_size != limit:
                            raise httpx.ReadError('Przerwana transmisja segmentu')
                        return
        except (ArchiveStopped, ArchiveDownloadError, _RangeUnsupported):
            raise
        except (httpx.HTTPError, OSError) as exc:
            _log(job_id, f'Segment {index+1}: {type(exc).__name__}: {str(exc)[:150]}, próba {attempt+1}/{MAX_ATTEMPTS}')
            if attempt == MAX_ATTEMPTS - 1:
                raise ArchiveDownloadError(f'Segment {index+1}/{len(manifest["ranges"])}: przerwany transfer po {MAX_ATTEMPTS} próbach: {type(exc).__name__}') from exc
            delay = None
        _retry_wait(job_id, attempt, stop, delay, external_stop)


def _report_parallel(job_id, ix, url_count, partial, folder, manifest, last, batch=None):
    prefix = manifest['prefix']
    downloaded = prefix
    for index, (start, end) in enumerate(manifest['ranges']):
        path = folder / f'{index:02d}.part'
        if path.exists():
            downloaded += min(end - start + 1, path.stat().st_size)
    now = time.monotonic()
    prev_size, prev_at = last
    speed = max(0, int((downloaded - prev_size) / max(0.001, now - prev_at)))
    total = manifest['total']
    if batch is None:
        store.update(job_id, completed=downloaded, total=total, speed=speed,
                     progress=round((ix + min(downloaded / total, 1)) / url_count * 100, 2),
                     note=f'Archive.org: plik {ix+1}/{url_count}, {len(manifest["ranges"])} połączeń')
    else:
        batch.report(ix, downloaded, total)
    return downloaded, now


def _assemble_segments(job_id, partial, destination, folder, manifest, stop=None):
    prefix = manifest['prefix']
    total = manifest['total']
    for index, (start, end) in enumerate(manifest['ranges']):
        part = folder / f'{index:02d}.part'
        if not part.is_file() or part.stat().st_size != end - start + 1:
            raise ArchiveDownloadError('Niekompletne segmenty; nie można złożyć pliku')
    manifest['assembling'] = True
    _write_manifest(folder, manifest)
    with partial.open('r+b' if partial.exists() else 'w+b') as output:
        output.truncate(prefix)
        output.seek(prefix)
        for index in range(len(manifest['ranges'])):
            with (folder / f'{index:02d}.part').open('rb') as source:
                while True:
                    _check_state(job_id, stop)
                    block = source.read(1024 * 1024)
                    if not block:
                        break
                    output.write(block)
        output.flush()
        os.fsync(output.fileno())
    if partial.stat().st_size != total:
        raise ArchiveDownloadError('Błąd scalania pliku: niezgodny rozmiar końcowy')
    partial.replace(destination)
    shutil.rmtree(folder)


def _parallel_transfer(job, ix, urls, uri, filename, partial, destination, connections, *, batch=None, stop=None):
    jid = job['id']
    batch_stop = stop
    probe = _probe_range(jid, uri, stop)
    if probe is None:
        _log(jid, 'Archive.org: serwer nie obsługuje Range; pobieranie pojedynczym połączeniem')
        return False
    total, validator = probe
    folder, manifest = _prepare_segments(partial, uri, total, validator, connections)
    if folder is None and manifest is None:
        _log(jid, 'Archive.org: mały plik lub końcówka; pobieranie pojedynczym połączeniem')
        return False
    if folder is None and manifest['prefix'] == total:
        partial.replace(destination)
        _finish(job, ix, urls, batch=batch)
        return True
    _log(jid, f'Archive.org: {len(manifest["ranges"])} połączeń HTTP Range, rozmiar {total} B, wznowienie od {manifest["prefix"]} B')
    local_stop = Event()
    current = (manifest['prefix'] + sum(min((folder / f'{idx:02d}.part').stat().st_size, end - start + 1)
                                        for idx, (start, end) in enumerate(manifest['ranges'])
                                        if (folder / f'{idx:02d}.part').exists()), time.monotonic())
    try:
        with ThreadPoolExecutor(max_workers=min(connections, len(manifest['ranges']))) as pool:
            pending = {pool.submit(_segment_worker, jid, uri, folder, manifest, i, local_stop, batch_stop)
                       for i in range(len(manifest['ranges']))}
            try:
                while pending:
                    _check_state(jid, batch_stop)
                    done, pending = wait(pending, timeout=0.8, return_when=FIRST_COMPLETED)
                    current = _report_parallel(jid, ix, len(urls), partial, folder, manifest, current, batch)
                    for task in done:
                        task.result()
            finally:
                local_stop.set()
                for task in pending:
                    task.cancel()
    except _RangeUnsupported as exc:
        _log(jid, 'Archive.org: niezgodna obsługa Range, przełączam na jedno połączenie')
        _discard_parallel_state(partial, reset_prefix=exc.reset_prefix)
        return False
    _check_state(jid, batch_stop)
    _report_parallel(jid, ix, len(urls), partial, folder, manifest, current, batch)
    _log(jid, 'Archive.org: scalanie segmentów')
    if batch is None:
        store.update(jid, note='Archive.org: scalanie segmentów', speed=0)
    if batch_stop is None:
        _assemble_segments(jid, partial, destination, folder, manifest)
    else:
        _assemble_segments(jid, partial, destination, folder, manifest, batch_stop)
    _finish(job, ix, urls, batch=batch)
    _log(jid, f'Ukończono {filename}: {total} bajtów, {len(manifest["ranges"])} połączeń')
    return True


class _BatchProgress:
    """Collect progress from file threads; only the scheduler updates SQLite.

    A single stream/segment never overwrites the total progress or total speed
    from another file. No extra database table is needed for old installations.
    """

    def __init__(self, job_id, filenames, limit):
        self.job_id = job_id
        self.filenames = filenames
        self.limit = limit
        self.lock = Lock()
        self.sizes = [0] * len(filenames)
        self.totals = [0] * len(filenames)
        self.seen = [False] * len(filenames)
        self.done = set()
        self.active = set()
        self.transferred = 0
        self.last_publish = time.monotonic()

    def start(self, index):
        with self.lock:
            self.active.add(index)

    def report(self, index, downloaded, total):
        with self.lock:
            downloaded = max(0, int(downloaded))
            if self.seen[index]:
                self.transferred += max(0, downloaded - self.sizes[index])
            else:
                # Existing .part and .segments from an earlier run are not
                # counted as freshly transferred bytes per second.
                self.seen[index] = True
            self.sizes[index] = downloaded
            self.totals[index] = max(0, int(total))

    def complete(self, index, size=None):
        if size is None:
            size = (store.DATA / 'raw' / self.job_id / self.filenames[index]).stat().st_size
        with self.lock:
            if self.seen[index]:
                self.transferred += max(0, size - self.sizes[index])
            self.seen[index] = True
            self.sizes[index] = size
            self.totals[index] = size
            self.done.add(index)
            self.active.discard(index)

    def publish(self):
        now = time.monotonic()
        with self.lock:
            elapsed = max(0.001, now - self.last_publish)
            speed = max(0, int(self.transferred / elapsed))
            self.transferred = 0
            self.last_publish = now
            total_files = len(self.filenames)
            fractions = []
            for index in range(total_files):
                if index in self.done:
                    fractions.append(1.0)
                elif self.totals[index]:
                    fractions.append(min(self.sizes[index] / self.totals[index], 1))
                else:
                    fractions.append(0.0)
            completed_bytes = sum(self.sizes)
            total_bytes = sum(self.totals)
            if total_bytes and all(self.totals):
                # A 20 KiB NFO must not count the same as an 8 GiB ISO.
                progress = round(100 * min(completed_bytes, total_bytes) / total_bytes, 2)
            else:
                # Until all content lengths are known, show the fraction of
                # completed files/known segments as a provisional estimate.
                progress = round(sum(fractions) / total_files * 100, 2)
            finished = len(self.done)
            active = len(self.active)
            contiguous = next((i for i in range(total_files) if i not in self.done), total_files)
        store.update(self.job_id, part_index=contiguous, progress=progress,
                     completed=completed_bytes, total=total_bytes, speed=speed,
                     note=f'Archive.org: {finished}/{total_files} gotowe, {active} równolegle, limit {self.limit} połączeń')


def _transfer_many(job):
    """Download multiple Archive.org files at once with ONE connection budget.

    Each slot owns a fixed connection allowance. Reusing the slot for the next
    file preserves the global cap even with 50-segment old resume manifests.
    """
    jid = job['id']
    urls = source_urls(job)
    if not urls:
        raise ArchiveDownloadError('Brak plików Archive.org do pobrania')
    names = []
    for uri in urls:
        if not archive_auth.is_archive_download(uri):
            raise ArchiveDownloadError('Nieprawidłowy URL pobierania Archive.org')
        name = target_filename(uri)
        if not name:
            raise ArchiveDownloadError('Brak nazwy pliku Archive.org')
        names.append(name)
    if len({name.casefold() for name in names}) != len(names):
        raise ArchiveDownloadError('Wybrane pliki mają powtarzające się nazwy. Wybierz pliki o różnych nazwach.')

    root = store.DATA / 'raw' / jid
    root.mkdir(parents=True, exist_ok=True)
    _check_state(jid)
    limit = _connection_limit()
    tracker = _BatchProgress(jid, names, limit)
    waiting = deque()
    for ix, name in enumerate(names):
        dest = root / name
        if dest.is_file() and not Path(str(dest) + '.aria2').exists():
            # A completed destination was atomically renamed by the old or
            # new downloader. A legacy aria2 control file means its sibling
            # destination may only contain an incomplete preallocated file.
            tracker.complete(ix, dest.stat().st_size)
            _log(jid, f'Plik {ix+1}/{len(urls)} już gotowy: {name}')
        else:
            waiting.append(ix)

    store.update(jid, state='downloading', gid='', note=f'Archive.org: start równoległego pobierania {len(waiting)} plików')
    tracker.publish()
    if not waiting:
        _check_state(jid)
        store.update(jid, state='preparing', progress=100, speed=0,
                     note='Pobieranie Archive.org ukończone')
        return

    slots = min(limit, len(waiting))
    budgets = [limit // slots + (1 if i < limit % slots else 0) for i in range(slots)]
    _log(jid, f'Archive.org: start {slots} plików równolegle; budżet {limit} połączeń HTTP '
              f'(przydział na plik: {", ".join(map(str, budgets))})')
    stopped = Event()
    first_error = None

    def launch(pool, slot):
        ix = waiting.popleft()
        item = dict(job, part_index=ix)
        tracker.start(ix)
        return pool.submit(_transfer_one, item, batch=tracker, connections=budgets[slot], stop=stopped)

    with ThreadPoolExecutor(max_workers=slots, thread_name_prefix='archive-file') as pool:
        running = {launch(pool, slot): slot for slot in range(slots)}
        try:
            while running:
                _check_state(jid, stopped)
                finished, _ = wait(running, timeout=0.8, return_when=FIRST_COMPLETED)
                tracker.publish()
                for future in finished:
                    slot = running.pop(future)
                    try:
                        future.result()
                    except Exception as exc:
                        first_error = first_error or exc
                        stopped.set()
                    else:
                        if waiting and not stopped.is_set():
                            running[launch(pool, slot)] = slot
                if first_error is not None:
                    break
        finally:
            stopped.set()
            for future in running:
                future.cancel()

    if first_error is not None:
        tracker.publish()
        raise first_error
    _check_state(jid)  # Especially important if a pause arrives at the last byte.
    tracker.publish()
    store.update(jid, state='preparing', progress=100, speed=0,
                 part_index=len(urls), note='Pobieranie Archive.org ukończone')
    _log(jid, f'Archive.org: pobrano wszystkie {len(urls)} pliki')


async def run_one(job):
    import asyncio
    urls = source_urls(job)
    if job['source_type'] == 'archive_many' and len(urls) > 1:
        await asyncio.to_thread(_transfer_many, job)
    else:
        await asyncio.to_thread(_transfer_one, job)