from __future__ import annotations

import hashlib
import os
import shutil
import stat
import time
from typing import Callable

from qgis.core import Qgis, QgsMessageLog
from qgis.PyQt.QtCore import QUrl
from qgis.PyQt.QtNetwork import QNetworkRequest

from .cache_paths import PLUGIN_CACHE_DIR
from .model_config import (
    CHECKPOINT_FILENAME,
    CHECKPOINT_SHA256,
    CHECKPOINT_URL,
    USE_SAM2,
)
from .qt_compat import NoLessSafeRedirectPolicy, RedirectPolicyAttribute


def tr(text: str) -> str:
    """Translate one user-facing line, falling back to English.

    The translator is imported inside the call, so this module keeps importing
    with no translator loaded and a lookup can never break a download.
    """
    try:
        from .i18n import tr as translate

        return translate(text)
    except Exception:  # noqa: BLE001 -- English is a fine answer here
        return text


CHECKPOINTS_DIR = os.path.join(PLUGIN_CACHE_DIR, "checkpoints")
FEATURES_DIR = os.path.join(PLUGIN_CACHE_DIR, "features")

SAM_CHECKPOINT_URL = CHECKPOINT_URL
SAM_CHECKPOINT_FILENAME = CHECKPOINT_FILENAME
SAM_CHECKPOINT_SHA256 = CHECKPOINT_SHA256
OLD_CHECKPOINT_FILENAME = "sam_vit_b_01ec64.pth"

# Shipped download settings: five attempts, two minutes of silence tolerated,
# one hour of ceiling on a live but slow connection.
DOWNLOAD_MAX_RETRIES = 5
DOWNLOAD_IDLE_TIMEOUT_MS = 120_000
DOWNLOAD_HARD_TIMEOUT_MS = 3_600_000
# Shipped retry ladder for moving the finished file into place past an
# antivirus lock.
REPLACE_ATTEMPTS = 5
REPLACE_DELAY_S = 2.0
# How long an unresumed partial download is kept before the startup sweep drops
# it. Long enough to cover a user who retries next week, short enough that a
# few hundred megabytes do not sit in the cache for good.
PARTIAL_DOWNLOAD_MAX_AGE_S = 7 * 24 * 60 * 60


def resolved_checkpoint() -> tuple[str, str, tuple[str, ...]]:
    """Where the weights come from, what they must hash to, and any mirror.

    The shipped address is a third-party host we do not run. When it goes
    away, every NEW install stops here until a plugin release clears the
    plugin repository, so the address, the digest and a list of alternates are
    all correctable from the product configuration. The digest is never
    optional: whichever address answers, the bytes are verified exactly as
    they are today, and a file that does not match is deleted and re-fetched.
    """
    try:
        from .install_config import checkpoint_source

        return checkpoint_source(
            SAM_CHECKPOINT_FILENAME, SAM_CHECKPOINT_URL, SAM_CHECKPOINT_SHA256)
    except Exception:  # noqa: BLE001 -- a bad config must never block an install
        return SAM_CHECKPOINT_URL, SAM_CHECKPOINT_SHA256, ()


def get_checkpoints_dir() -> str:
    os.makedirs(CHECKPOINTS_DIR, exist_ok=True)
    return CHECKPOINTS_DIR


def get_checkpoint_path() -> str:
    return os.path.join(get_checkpoints_dir(), SAM_CHECKPOINT_FILENAME)


def checkpoint_exists() -> bool:
    """Whether the model file is on disk. Creates nothing.

    Joins the directory rather than asking for it: the getter above makes the
    directory, and this question is asked on the click path and on every
    refresh of the review panel, so going through it turned a read into a
    filesystem write on a folder that may sit on a synced or network home.
    """
    return os.path.exists(os.path.join(CHECKPOINTS_DIR, SAM_CHECKPOINT_FILENAME))


def verify_checkpoint_hash(filepath: str) -> bool:
    expected_sha256 = resolved_checkpoint()[1]
    if not expected_sha256:
        # Fail closed: with no expected hash we cannot prove integrity, so we
        # refuse to treat the file as verified rather than silently trusting it.
        # (Both shipped checkpoints define a hash; an empty value here would be
        # a packaging mistake that must surface, not pass through.)
        QgsMessageLog.logMessage(
            "No expected checkpoint hash is configured; cannot verify integrity.",
            "AI Segmentation", level=Qgis.MessageLevel.Warning)
        return False
    if not os.path.isfile(filepath):
        QgsMessageLog.logMessage(
            f"Checkpoint file not found for hash verification: {filepath}",
            "AI Segmentation", level=Qgis.MessageLevel.Warning)
        return False
    try:
        sha256_hash = hashlib.sha256()
        with open(filepath, "rb") as f:
            # Checkpoints are 150-320 MB; a 4 KiB block would take tens of
            # thousands of Python-level loop iterations, so read in 1 MiB
            # chunks instead.
            for byte_block in iter(lambda: f.read(1024 * 1024), b""):
                sha256_hash.update(byte_block)
        return sha256_hash.hexdigest() == expected_sha256
    except OSError as e:
        QgsMessageLog.logMessage(
            f"Failed to verify checkpoint hash: {e}",
            "AI Segmentation", level=Qgis.MessageLevel.Warning)
        return False


def is_corrupt_checkpoint_error(error_message: str | None) -> bool:
    """Return True when an error indicates a corrupt checkpoint file.

    The checkpoint is a torch ``.pt`` (a zip archive). When the file on disk
    is truncated or damaged, ``torch.load`` raises errors whose text contains
    one of the signatures below. We match on these so the plugin can delete
    the bad file and re-download it instead of failing forever.
    """
    if not error_message:
        return False
    lowered = error_message.lower()
    signatures = (
        "pytorchstreamreader",
        "failed finding central directory",
        "checkpoint file is corrupted",
        "central directory",
        "not a zip archive",
        "invalid load key",
    )
    return any(sig in lowered for sig in signatures)


def delete_checkpoint() -> bool:
    """Delete the on-disk checkpoint (e.g. when found corrupt).

    Returns True if the path is clear afterwards (deleted, moved aside, or
    already absent), False if the bad file is still where the loader looks.

    This is called the moment an encode reports a corrupt checkpoint, while the
    predictor subprocess may still hold the file open, so a single unlink is
    not enough: clear the read-only bit, retry the delete, then move the file
    aside as a last resort. A rename succeeds on a file a delete cannot touch,
    and it frees the name so the re-download has somewhere to land. Anything
    else dead-ends the user in "delete this folder and restart QGIS".
    """
    path = get_checkpoint_path()
    if not os.path.exists(path):
        return True
    # Windows refuses to delete a read-only file, so clear that bit before the
    # retries rather than after them. Adding the write bit rather than setting
    # the whole mode leaves a POSIX file's permissions alone.
    try:
        os.chmod(path, os.stat(path).st_mode | stat.S_IWRITE)
    except OSError:
        pass  # nosec B110 - the retries below report the real problem
    if _remove_with_retry(path):
        QgsMessageLog.logMessage(
            "Removed corrupt checkpoint, will re-download",
            "AI Segmentation", level=Qgis.MessageLevel.Warning)
        return True
    try:
        os.replace(path, path + ".corrupt")
    except OSError as e:
        QgsMessageLog.logMessage(
            f"Could not remove corrupt checkpoint: {e}",
            "AI Segmentation", level=Qgis.MessageLevel.Warning)
        return False
    QgsMessageLog.logMessage(
        "Moved the corrupt checkpoint aside, will re-download",
        "AI Segmentation", level=Qgis.MessageLevel.Warning)
    return True


#: Shipped headroom the model download asks for, in MB. The weights are a few
#: hundred MB and the partial file sits beside them until the final swap, so
#: the figure covers both. Retunable from the product configuration.
CHECKPOINT_MIN_FREE_MB = 1024.0


def resolved_checkpoint_min_free_mb() -> float:
    """The free space the model download requires, server value when served."""
    try:
        from .server_dials import dial_in_range

        return float(dial_in_range(
            "install.disk.min_free_mb_checkpoint",
            CHECKPOINT_MIN_FREE_MB, 256.0, 20480.0))
    except Exception:  # noqa: BLE001 -- configuration is best-effort
        return CHECKPOINT_MIN_FREE_MB


def _disk_space_preflight_hint(
        dest_dir: str, min_free_mb: float | None = None) -> str | None:
    """Return an error message if dest_dir has less than min_free_mb free.

    Mirrors the venv install's own preflight (venv_manager.py) so a full disk
    fails clearly here too instead of an opaque mid-write error.
    """
    if min_free_mb is None:
        min_free_mb = resolved_checkpoint_min_free_mb()
    try:
        os.makedirs(dest_dir, exist_ok=True)
        free_mb = shutil.disk_usage(dest_dir).free / (1024 ** 2)
    except OSError:
        # Let the actual write report the real error instead of guessing here.
        return None
    if free_mb < min_free_mb:
        return (
            tr("Not enough free disk space to download the AI model: "
               "{free} MB available, at least {needed} MB is required.").format(
                   free=f"{free_mb:.0f}", needed=f"{min_free_mb:.0f}")
            + "\n\n"
            + tr("Free up disk space, or set the AI_SEGMENTATION_CACHE_DIR "
                 "environment variable to a directory on a larger drive, "
                 "then restart QGIS.")
        )
    return None


# Cancel request for an in-flight checkpoint download. The download runs in a
# worker thread with its own event loop, so the request travels through this
# module-level flag instead of a call into the worker: the flag can be set from
# the GUI thread (Cancel button) and from teardown, where the worker object may
# already be unreachable. One download runs at a time, so one flag is enough.
_cancel_requested = False

DOWNLOAD_CANCELLED_MESSAGE = "Download cancelled"


def request_download_cancel() -> None:
    """Ask an in-flight checkpoint download to stop as soon as it can.

    Never raises: it is called from the Cancel button and from plugin
    teardown, where nothing may block QGIS from closing.
    """
    global _cancel_requested
    _cancel_requested = True


def download_cancel_requested() -> bool:
    """True while a cancel has been requested and not yet consumed."""
    return _cancel_requested


def _consume_download_cancel() -> bool:
    """True when a cancel was pending, and clear it on the way out.

    The flag has to be cleared where it is acted on. Left armed, it stops the
    next download too: the user presses Cancel, then Install again, and the
    second run ends before it starts with nothing on screen to explain it.
    """
    global _cancel_requested
    if not _cancel_requested:
        return False
    _cancel_requested = False
    return True


def _wait_or_cancel(seconds: float) -> bool:
    """Sleep in short slices, returning False as soon as a cancel arrives.

    A plain time.sleep of up to 2 minutes between retries would keep the
    worker thread alive long after the user pressed Cancel or QGIS started
    tearing down.
    """
    deadline = time.monotonic() + seconds
    while True:
        if _cancel_requested:
            return False
        remaining = deadline - time.monotonic()
        if remaining <= 0:
            return True
        time.sleep(min(0.25, remaining))


def _resolved_retry_ladder(max_attempts: int | None,
                           delay: float | None) -> tuple[int, float]:
    """The shipped attempts and delay, with any served override applied.

    How long to fight a lock is a server dial: the machines that need more
    patience are the ones we cannot reach to ask. A bad configuration must
    never stop a file from being moved or removed, so it falls back to the
    shipped pair.
    """
    if max_attempts is not None and delay is not None:
        return max_attempts, delay
    try:
        from .install_config import replace_attempts, replace_delay_s

        served_attempts = replace_attempts(REPLACE_ATTEMPTS)
        served_delay = replace_delay_s(REPLACE_DELAY_S)
    except Exception:  # noqa: BLE001 -- a bad config must never block a cleanup
        served_attempts, served_delay = REPLACE_ATTEMPTS, REPLACE_DELAY_S
    return (
        served_attempts if max_attempts is None else max_attempts,
        served_delay if delay is None else delay,
    )


def _replace_with_retry(src: str, dst: str, max_attempts: int | None = None,
                        delay: float | None = None) -> bool:
    """Rename src to dst, retrying on PermissionError (Windows antivirus lock).

    True once the move landed, False when a cancel arrived mid-wait. Raises the
    last PermissionError when every attempt lost to the lock.
    """
    import gc

    max_attempts, delay = _resolved_retry_ladder(max_attempts, delay)
    gc.collect()
    for attempt in range(1, max_attempts + 1):
        try:
            os.replace(src, dst)
            return True
        except PermissionError:
            if attempt == max_attempts:
                raise
            QgsMessageLog.logMessage(
                f"File locked, retry {attempt}/{max_attempts} in {delay}s...",
                "AI Segmentation", level=Qgis.MessageLevel.Warning)
            # Every other wait in this module is cancellable, and this one can
            # run while QGIS is closing: a plain sleep would hold the worker
            # thread and freeze the window.
            if not _wait_or_cancel(delay):
                return False
    return False


def _remove_with_retry(path: str, max_attempts: int | None = None,
                       delay: float | None = None) -> bool:
    """Delete path, retrying past a transient lock. True once it is gone.

    Same ladder as _replace_with_retry, for the same reason: a file that was
    just read end to end is the one a scanner is most likely to still hold, and
    a raw unlink surfaces as an access error the user can do nothing about.
    Never raises: every caller has a way to carry on without the delete.
    """
    max_attempts, delay = _resolved_retry_ladder(max_attempts, delay)
    last_error: OSError | None = None
    for attempt in range(1, max_attempts + 1):
        try:
            os.remove(path)
            return True
        except FileNotFoundError:
            return True
        except OSError as err:
            last_error = err
            if attempt == max_attempts:
                break
            if not _wait_or_cancel(delay):
                return False
    QgsMessageLog.logMessage(
        f"Could not remove {os.path.basename(path)}: {last_error}",
        "AI Segmentation", level=Qgis.MessageLevel.Warning)
    return False


def _adopt_legacy_partial(legacy_path: str, temp_path: str) -> None:
    """Carry a partial left by an older version into the name used to resume.

    The partial is named after the digest it is being fetched for. One written
    before it carried that name would otherwise sit on disk untouched while
    the whole file came down again. What it actually holds is settled by the
    checksum at the end, exactly like any other resumed download.
    """
    if os.path.exists(temp_path) or not os.path.exists(legacy_path):
        return
    try:
        os.replace(legacy_path, temp_path)
    except OSError:
        return  # nosec B110 -- a partial that will not move only costs time
    QgsMessageLog.logMessage(
        "Resuming a partial model download left by an earlier version",
        "AI Segmentation", level=Qgis.MessageLevel.Info)


def _discard_partial_download(temp_path: str) -> None:
    """Drop a partial download so the next attempt starts from zero.

    Truncating in place is the fallback because it succeeds where an unlink
    does not. The resume offset is read from the file size, so an empty file
    restarts the download exactly as a missing one does, while a survivor with
    bytes in it would send a Range past its end on every future attempt and
    never recover.
    """
    if _remove_with_retry(temp_path):
        return
    try:
        with open(temp_path, "wb"):
            pass
    except OSError as err:
        QgsMessageLog.logMessage(
            f"Could not clear the partial download: {err}",
            "AI Segmentation", level=Qgis.MessageLevel.Warning)


#: The download takes a lock of its own rather than the install lock: a model
#: download and a dependency install are different jobs and must not block
#: each other, while two QGIS windows fetching the same weights write into the
#: same partial file and would interleave their bytes.
_CHECKPOINT_LOCK_BASENAME = "checkpoint.lock"


def checkpoint_lock_path() -> str:
    """Where the one-download-at-a-time lock file lives."""
    return os.path.join(CHECKPOINTS_DIR, _CHECKPOINT_LOCK_BASENAME)


def download_checkpoint(
    progress_callback: Callable[[int, str], None] | None = None
) -> tuple[bool, str]:
    """Download the model weights, one process at a time.

    The cache directory is shared by every QGIS window and profile on the
    machine, so this holds a lock for the whole transfer. A second window that
    finds the lock taken stops here with a message instead of writing into the
    same partial file.
    """
    from .install_lock import InstallBusyError, acquire_install_lock

    try:
        os.makedirs(CHECKPOINTS_DIR, exist_ok=True)
    except OSError:
        pass  # nosec B110 - the download below reports the real problem
    try:
        lock = acquire_install_lock(checkpoint_lock_path())
    except InstallBusyError:
        QgsMessageLog.logMessage(
            "Another process is already downloading the model",
            "AI Segmentation", level=Qgis.MessageLevel.Warning)
        return False, tr(
            "Another QGIS window is downloading the AI model. Wait for it to "
            "finish, then try again.")
    try:
        return _download_checkpoint(progress_callback)
    finally:
        lock.release()


def _download_checkpoint(
    progress_callback: Callable[[int, str], None] | None = None
) -> tuple[bool, str]:
    """
    Download SAM checkpoint using QGIS network manager with progress reporting.

    Uses QNetworkAccessManager with a local event loop to provide real-time
    download progress updates. Supports resuming partial downloads via HTTP
    Range requests (issue #129).
    """
    from qgis.core import QgsNetworkAccessManager
    from qgis.PyQt.QtCore import QByteArray, QEventLoop, QTimer
    from qgis.PyQt.QtNetwork import QNetworkReply

    # Clear any cancel left over from a previous download, so a new one is not
    # stopped before it starts.
    global _cancel_requested
    _cancel_requested = False

    checkpoint_path = get_checkpoint_path()

    if checkpoint_exists():
        QgsMessageLog.logMessage(
            "Checkpoint already exists, verifying...",
            "AI Segmentation",
            level=Qgis.MessageLevel.Info
        )
        if verify_checkpoint_hash(checkpoint_path):
            return True, "Checkpoint verified"
        QgsMessageLog.logMessage(
            "Checkpoint hash mismatch, re-downloading...",
            "AI Segmentation",
            level=Qgis.MessageLevel.Warning
        )
        # The file was just read end to end by the hash check, which is exactly
        # when a scanner still holds it. The delete only frees disk space for
        # the temp file: the good download lands through an atomic replace at
        # the end, so a lock here must not stop the re-download.
        _remove_with_retry(checkpoint_path)

    disk_hint = _disk_space_preflight_hint(os.path.dirname(checkpoint_path))
    if disk_hint:
        QgsMessageLog.logMessage(disk_hint, "AI Segmentation", level=Qgis.MessageLevel.Critical)
        return False, disk_hint

    if progress_callback:
        progress_callback(0, tr("Connecting to download server..."))

    from .install_config import (
        checkpoint_hard_timeout_ms,
        checkpoint_idle_timeout_ms,
        checkpoint_max_retries,
    )

    download_url, expected_sha256, mirrors = resolved_checkpoint()
    # The partial download carries the digest it is being fetched for. A
    # served change of file leaves the old partial on disk under the old
    # name, so the resume cannot append the new file onto the old bytes and
    # hand the user a mixture that fails its checksum for good.
    temp_path = f"{checkpoint_path}.{expected_sha256[:12]}.tmp"
    _adopt_legacy_partial(f"{checkpoint_path}.tmp", temp_path)
    # The primary address first, then any mirror, one per attempt. A mirror
    # serves the same file, so the digest check after the download is what
    # decides, exactly as before.
    download_urls = (download_url,) + mirrors
    max_retries = max(checkpoint_max_retries(DOWNLOAD_MAX_RETRIES), len(download_urls))
    idle_timeout_ms = checkpoint_idle_timeout_ms(DOWNLOAD_IDLE_TIMEOUT_MS)
    hard_timeout_ms = checkpoint_hard_timeout_ms(DOWNLOAD_HARD_TIMEOUT_MS)
    last_error = ""

    digest_mismatches = 0
    # Addresses that answered "not there" or "not for you". A refusal belongs
    # to the address that gave it, not to the file, so the others are still
    # worth asking.
    refused_urls: set[str] = set()

    for attempt in range(1, max_retries + 1):
        if _consume_download_cancel():
            return False, DOWNLOAD_CANCELLED_MESSAGE

        # The address this attempt uses, read before anything can fail: the
        # refusal branch below has to know which one answered.
        attempt_url = download_urls[(attempt - 1) % len(download_urls)]

        # Check for existing partial download to resume
        resume_offset = 0
        if os.path.exists(temp_path):
            resume_offset = os.path.getsize(temp_path)
            if 0 < resume_offset < 1024 * 1024:
                # Tiny partials are more likely an aborted handshake or a
                # proxy error page than real data: restart from scratch.
                try:
                    os.remove(temp_path)
                except OSError:
                    pass  # nosec B110
                resume_offset = 0
            elif resume_offset > 0:
                QgsMessageLog.logMessage(
                    f"Resuming download from {resume_offset / (1024 * 1024):.1f} MB",
                    "AI Segmentation", level=Qgis.MessageLevel.Info)

        # State container for progress tracking
        # Rebuilt every attempt, which is what makes it the right place to
        # keep the reply: the slots below read it from here instead of closing
        # over a name the next attempt rebinds, so a late signal from an
        # abandoned reply can never act on the live one.
        download_state = {
            "bytes_received": 0,
            "bytes_total": 0,
            "error": None,
            "file": None,
            "reply": None,
            "resume_offset": resume_offset,
            "start_time": time.monotonic(),
        }

        def on_download_progress(received, total):
            download_state["bytes_received"] = received
            download_state["bytes_total"] = total
            idle_timer = download_state.get("idle_timer")
            if idle_timer is not None:
                idle_timer.start()
            if not progress_callback:
                return
            # Read the base from the state dict, not the closure: a resume the
            # server answered with 200 resets it to 0 (see on_ready_read).
            base = download_state["resume_offset"]
            actual_received = base + received
            actual_total = base + total if total > 0 else 0
            elapsed = max(0.1, time.monotonic() - download_state["start_time"])
            speed_mbs = (received / (1024 * 1024)) / elapsed if received > 0 else 0.0
            retry_suffix = ""
            if attempt > 1:
                retry_suffix = " " + tr("(retry {done}/{total})").format(
                    done=attempt, total=max_retries)

            if actual_total > 0:
                percent = int((actual_received / actual_total) * 90) + 5
                mb_recv = actual_received / (1024 * 1024)
                mb_tot = actual_total / (1024 * 1024)
                remaining_bytes = max(0, (total - received))
                eta_s = int(remaining_bytes / max(1.0, received / elapsed)) if received > 0 else 0
                if eta_s >= 60:
                    eta_str = tr("~{minutes}m {seconds}s left").format(
                        minutes=eta_s // 60, seconds=eta_s % 60)
                else:
                    eta_str = tr("~{seconds}s left").format(seconds=eta_s)
                progress_callback(
                    min(percent, 95),
                    tr("Downloading: {done} / {total} MB ({speed} MB/s, {eta})").format(
                        done=f"{mb_recv:.1f}", total=f"{mb_tot:.1f}",
                        speed=f"{speed_mbs:.1f}", eta=eta_str) + retry_suffix)
            elif actual_received > 0:
                mb_recv = actual_received / (1024 * 1024)
                progress_callback(
                    50,
                    tr("Downloading: {done} MB ({speed} MB/s)").format(
                        done=f"{mb_recv:.1f}", speed=f"{speed_mbs:.1f}") + retry_suffix)

        def on_ready_read():
            attempt_reply = download_state["reply"]
            if attempt_reply is None or download_state["file"] is None:
                return
            data = attempt_reply.readAll()
            # A resume answered with HTTP 200 (not 206 Partial Content) means
            # the server or a proxy ignored the Range header and is sending
            # the FULL body; appending it to the partial corrupts the file,
            # caught only by the SHA256 check after a wasted full download
            # (common behind caching proxies that strip Range). Restart the
            # temp file as a fresh write instead, once, on first data.
            if download_state["resume_offset"] > 0 and not download_state.get("status_checked"):
                download_state["status_checked"] = True
                try:
                    status = attempt_reply.attribute(
                        QNetworkRequest.Attribute.HttpStatusCodeAttribute)
                except (RuntimeError, AttributeError):
                    status = None
                if status == 200:
                    QgsMessageLog.logMessage(
                        "Server ignored the resume range (HTTP 200): "
                        "restarting the file from scratch",
                        "AI Segmentation", level=Qgis.MessageLevel.Warning)
                    try:
                        download_state["file"].close()
                        download_state["file"] = open(temp_path, "wb")
                    except OSError as reset_err:
                        download_state["error"] = (
                            f"Cannot restart download file: {reset_err}")
                        download_state["file"] = None
                        try:
                            attempt_reply.abort()
                        except (RuntimeError, AttributeError):
                            pass
                        return
                    download_state["resume_offset"] = 0
            # This runs as a Qt slot: an exception raised here crosses back
            # into C++, which kills the whole application. A disk that fills
            # mid-download, or a network drive that drops, must end the
            # attempt through the normal retry path instead.
            try:
                download_state["file"].write(data.data())
            except OSError as write_err:
                download_state["error"] = f"Cannot write download file: {write_err}"
                try:
                    download_state["file"].close()
                except OSError:
                    pass  # nosec B110
                download_state["file"] = None
                try:
                    attempt_reply.abort()
                except (RuntimeError, AttributeError):
                    pass  # nosec B110

        def on_error(_error_code):
            attempt_reply = download_state["reply"]
            if attempt_reply is None:
                return
            download_state["error"] = attempt_reply.errorString()

        try:
            manager = QgsNetworkAccessManager.instance()
            qurl = QUrl(attempt_url)
            request = QNetworkRequest(qurl)
            # Follow a redirect, as long as it does not downgrade the
            # connection. Without this the request comes back as a 30x with an
            # empty body, which the checksum then rejects as a corrupt file.
            request.setAttribute(RedirectPolicyAttribute, NoLessSafeRedirectPolicy)

            # Resume from partial download using HTTP Range header
            if resume_offset > 0:
                range_header = f"bytes={resume_offset}-"
                request.setRawHeader(
                    QByteArray(b"Range"),
                    QByteArray(range_header.encode("ascii")))

            # Open temp file in append mode for resume, write mode for fresh
            try:
                if resume_offset > 0:
                    download_state["file"] = open(temp_path, "ab")
                else:
                    download_state["file"] = open(temp_path, "wb")
            except OSError as file_err:
                last_error = f"Cannot open download file: {file_err}"
                QgsMessageLog.logMessage(
                    last_error, "AI Segmentation", level=Qgis.MessageLevel.Warning)
                if attempt < max_retries:
                    _wait_or_cancel(min(5 * (2 ** (attempt - 1)), 120))
                continue

            reply = manager.get(request)
            download_state["reply"] = reply

            reply.downloadProgress.connect(on_download_progress)
            reply.readyRead.connect(on_ready_read)
            reply.errorOccurred.connect(on_error)

            loop = QEventLoop()
            reply.finished.connect(loop.quit)

            # Two watchdogs instead of one absolute 20 min cutoff: a stalled
            # server (connection open, zero bytes) aborts after 2 minutes of
            # silence, while a slow but live connection gets up to 60 minutes.
            def on_idle_timeout():
                download_state["timeout_reason"] = (
                    f"no data received for {idle_timeout_ms // 60000} minutes")
                loop.quit()

            def on_hard_timeout():
                download_state["timeout_reason"] = (
                    f"exceeded {hard_timeout_ms // 60000} minutes")
                loop.quit()

            idle_timeout = QTimer()
            idle_timeout.setSingleShot(True)
            idle_timeout.setInterval(idle_timeout_ms)
            idle_timeout.timeout.connect(on_idle_timeout)
            download_state["idle_timer"] = idle_timeout
            idle_timeout.start()

            hard_timeout = QTimer()
            hard_timeout.setSingleShot(True)
            hard_timeout.setInterval(hard_timeout_ms)
            hard_timeout.timeout.connect(on_hard_timeout)
            hard_timeout.start()

            # Cancel poll: a stalled connection sends no data, so waiting for
            # the progress signal to notice the flag could hold the thread for
            # the whole idle window. Polling leaves the loop within half a
            # second of a Cancel click or a plugin teardown.
            def on_cancel_poll():
                if _cancel_requested:
                    loop.quit()

            cancel_timer = QTimer()
            cancel_timer.setInterval(500)
            cancel_timer.timeout.connect(on_cancel_poll)
            cancel_timer.start()

            if progress_callback:
                retry_msg = ""
                if attempt > 1:
                    retry_msg = " " + tr("(retry {done}/{total})").format(
                        done=attempt, total=max_retries)
                if resume_offset > 0:
                    progress_callback(5, tr("Resuming download...") + retry_msg)
                else:
                    progress_callback(5, tr("Download started...") + retry_msg)

            loop.exec()

            idle_timeout.stop()
            hard_timeout.stop()
            cancel_timer.stop()
            download_state["idle_timer"] = None

            if _consume_download_cancel():
                # Every step here must survive a half-torn-down QGIS: the
                # partial file stays on disk so a later run can resume it.
                try:
                    reply.abort()
                except (RuntimeError, AttributeError):
                    pass  # nosec B110
                try:
                    reply.deleteLater()
                except (RuntimeError, AttributeError):
                    pass  # nosec B110
                if download_state["file"] is not None:
                    try:
                        download_state["file"].close()
                    except OSError:
                        pass  # nosec B110
                    download_state["file"] = None
                QgsMessageLog.logMessage(
                    "Model download cancelled",
                    "AI Segmentation", level=Qgis.MessageLevel.Info)
                return False, DOWNLOAD_CANCELLED_MESSAGE

            if download_state.get("timeout_reason"):
                reply.abort()
                reply.deleteLater()
                if download_state["file"] is not None:
                    download_state["file"].close()
                download_state["file"] = None
                last_error = f"Download timed out ({download_state['timeout_reason']})"
                if attempt < max_retries:
                    _wait_or_cancel(min(5 * (2 ** (attempt - 1)), 120))
                continue

            status_code = reply.attribute(QNetworkRequest.Attribute.HttpStatusCodeAttribute)

            # Server returned 416 (range not satisfiable)
            if status_code == 416:
                if download_state["file"] is not None:
                    download_state["file"].close()
                download_state["file"] = None
                reply.deleteLater()
                # A range at/after the file end means the temp file is already
                # complete. If it verifies, only the final rename is left (a
                # prior _replace_with_retry lost to a file lock); retry that
                # instead of discarding and re-downloading a good, verified file.
                if os.path.exists(temp_path) and verify_checkpoint_hash(temp_path):
                    try:
                        finalized = _replace_with_retry(temp_path, checkpoint_path)
                    except OSError as replace_err:
                        last_error = f"Could not finalize checkpoint: {replace_err}"
                        if attempt < max_retries:
                            _wait_or_cancel(min(5 * (2 ** (attempt - 1)), 120))
                        continue
                    if not finalized:
                        _consume_download_cancel()
                        return False, DOWNLOAD_CANCELLED_MESSAGE
                    if progress_callback:
                        progress_callback(100, tr("Model downloaded."))
                    QgsMessageLog.logMessage(
                        f"Checkpoint downloaded to: {checkpoint_path}",
                        "AI Segmentation", level=Qgis.MessageLevel.Success)
                    return True, "Checkpoint downloaded and verified"
                # The partial file is invalid: restart from scratch.
                QgsMessageLog.logMessage(
                    "Server rejected range request, restarting download",
                    "AI Segmentation", level=Qgis.MessageLevel.Warning)
                _discard_partial_download(temp_path)
                if attempt < max_retries:
                    _wait_or_cancel(1)
                continue

            # Check for errors
            if reply.error() != QNetworkReply.NetworkError.NoError:
                last_error = download_state["error"] or reply.errorString()
                QgsMessageLog.logMessage(
                    f"Checkpoint download attempt {attempt}/{max_retries} failed: {last_error}",
                    "AI Segmentation", level=Qgis.MessageLevel.Warning)
                reply.deleteLater()
                if download_state["file"] is not None:
                    download_state["file"].close()
                download_state["file"] = None
                # An answer of "not there" or "not for you" is final. Waiting
                # and asking again gets the same answer, and the ladder of
                # backoffs spends over a minute doing it before the user is
                # told anything.
                if status_code in (403, 404, 410):
                    _discard_partial_download(temp_path)
                    refused_urls.add(attempt_url)
                    refusal = {
                        403: tr("the server refused access to the model file"),
                        404: tr("the model file is not at that address"),
                        410: tr("the model file has been removed from that address"),
                    }[status_code]
                    if len(refused_urls) < len(download_urls):
                        # Another address serves the same file, and the digest
                        # check after the download decides either way. Only
                        # once they have all refused is the answer final.
                        QgsMessageLog.logMessage(
                            f"Model download: {refusal} (HTTP {status_code}); "
                            "trying another address",
                            "AI Segmentation", level=Qgis.MessageLevel.Warning)
                        continue
                    QgsMessageLog.logMessage(
                        f"Model download stopped: {refusal} (HTTP {status_code})",
                        "AI Segmentation", level=Qgis.MessageLevel.Critical)
                    return False, tr(
                        "Model download failed: {reason}. Retrying will not "
                        "help. Update the plugin, or ask your IT administrator "
                        "whether the download is being filtered."
                    ).format(reason=refusal)
                if attempt < max_retries:
                    wait = min(5 * (2 ** (attempt - 1)), 120)
                    if progress_callback:
                        progress_callback(
                            5, tr("Retry {done}/{total} in {seconds}s...").format(
                                done=attempt + 1, total=max_retries, seconds=wait))
                    _wait_or_cancel(wait)
                continue

            # Flush remaining data
            remaining = reply.readAll()
            if remaining and download_state["file"] is not None:
                download_state["file"].write(remaining.data())
            if download_state["file"] is not None:
                download_state["file"].close()
            download_state["file"] = None
            reply.deleteLater()

            # Verify file is not empty
            file_size = os.path.getsize(temp_path) if os.path.exists(temp_path) else 0
            if file_size == 0:
                last_error = "Download failed: empty file"
                if attempt < max_retries:
                    _wait_or_cancel(min(5 * (2 ** (attempt - 1)), 120))
                continue

            if progress_callback:
                mb_total = file_size / (1024 * 1024)
                progress_callback(
                    95, tr("Checking the {size} MB download...").format(
                        size=f"{mb_total:.1f}"))

            if not verify_checkpoint_hash(temp_path):
                # Hash mismatch after full download: partial file was
                # corrupted, delete and retry from scratch
                QgsMessageLog.logMessage(
                    "Hash mismatch, deleting partial file and retrying",
                    "AI Segmentation", level=Qgis.MessageLevel.Warning)
                _discard_partial_download(temp_path)
                last_error = "Download verification failed - hash mismatch"
                declared_total = download_state.get("bytes_total") or 0
                base = download_state["resume_offset"]
                if declared_total <= 0 or file_size >= base + declared_total:
                    # A response that ran to the end and still does not match
                    # says the bytes themselves are wrong. A host that names no
                    # length counts the same way: the reply finished, and
                    # nothing else here can tell it from a complete one.
                    digest_mismatches += 1
                else:
                    # A short file is a cut connection, which is worth another
                    # attempt.
                    digest_mismatches = 0
                if digest_mismatches >= 2:
                    return False, tr(
                        "The model file arrived complete twice and did not "
                        "match its checksum either time. Something between "
                        "this computer and the download is altering the file, "
                        "usually a proxy or a security appliance. Ask your IT "
                        "administrator to let the download through untouched."
                    )
                if attempt < max_retries:
                    _wait_or_cancel(min(5 * (2 ** (attempt - 1)), 120))
                continue

            if not _replace_with_retry(temp_path, checkpoint_path):
                _consume_download_cancel()
                return False, DOWNLOAD_CANCELLED_MESSAGE

            # Clean up old SAM1 checkpoint if present (only on SAM2 path)
            if USE_SAM2:
                old_checkpoint = os.path.join(
                    get_checkpoints_dir(), OLD_CHECKPOINT_FILENAME)
                if os.path.exists(old_checkpoint):
                    try:
                        os.remove(old_checkpoint)
                        QgsMessageLog.logMessage(
                            f"Removed old checkpoint: {OLD_CHECKPOINT_FILENAME}",
                            "AI Segmentation", level=Qgis.MessageLevel.Info)
                    except OSError:
                        pass

            if progress_callback:
                progress_callback(100, tr("Model downloaded."))

            QgsMessageLog.logMessage(
                f"Checkpoint downloaded to: {checkpoint_path}",
                "AI Segmentation", level=Qgis.MessageLevel.Success)
            return True, "Checkpoint downloaded and verified"

        except Exception as e:
            last_error = str(e)
            QgsMessageLog.logMessage(
                f"Checkpoint download attempt {attempt}/{max_retries} exception: {last_error}",
                "AI Segmentation", level=Qgis.MessageLevel.Warning)
            if download_state.get("file") is not None:
                try:
                    download_state["file"].close()
                except Exception:
                    pass  # nosec B110
                download_state["file"] = None
            if attempt < max_retries:
                _wait_or_cancel(min(5 * (2 ** (attempt - 1)), 120))

    # All retries exhausted - keep partial file for next resume attempt
    partial_mb = 0
    if os.path.exists(temp_path):
        partial_mb = os.path.getsize(temp_path) / (1024 * 1024)
    firewall_hint = " " + tr(
        "A firewall or proxy may be blocking the download. Check your network "
        "settings in QGIS (Settings > Options > Network).")
    failed = tr("Download failed after {attempts} attempts: {reason}").format(
        attempts=max_retries, reason=last_error)
    if partial_mb > 0:
        resume = tr(
            "Partial file ({size} MB) saved, it will resume on the next try."
        ).format(size=f"{partial_mb:.1f}")
        return False, f"{failed}. {resume}{firewall_hint}"
    return False, f"{failed}{firewall_hint}"


def cleanup_legacy_sam1_data():
    """Remove legacy SAM1 data: old checkpoint and features cache.

    Called once at plugin startup. Errors are logged silently
    to avoid disturbing the user.

    On Python 3.9 (USE_SAM2=False), the SAM1 checkpoint is still needed,
    so only features cache is cleaned up.
    """
    import shutil

    # Remove old SAM1 checkpoint only when using SAM2 (Python 3.10+)
    if USE_SAM2:
        old_checkpoint = os.path.join(CHECKPOINTS_DIR, OLD_CHECKPOINT_FILENAME)
        if os.path.exists(old_checkpoint):
            try:
                os.remove(old_checkpoint)
                QgsMessageLog.logMessage(
                    f"Removed old SAM1 checkpoint: {OLD_CHECKPOINT_FILENAME}",
                    "AI Segmentation", level=Qgis.MessageLevel.Info)
            except OSError as e:
                QgsMessageLog.logMessage(
                    f"Could not remove old checkpoint: {e}",
                    "AI Segmentation", level=Qgis.MessageLevel.Warning)

    # Remove old features cache (SAM2 does on-demand encoding, no cache needed)
    if os.path.exists(FEATURES_DIR):
        # ignore_errors, because rmtree stops at the first entry it cannot
        # delete and leaves everything after it. This runs at every start, so
        # without it one read-only leftover means the same half-deleted tree is
        # walked and re-logged forever.
        shutil.rmtree(FEATURES_DIR, ignore_errors=True)
        if os.path.exists(FEATURES_DIR):
            QgsMessageLog.logMessage(
                "Legacy features cache only partly removed",
                "AI Segmentation", level=Qgis.MessageLevel.Warning)
        else:
            QgsMessageLog.logMessage(
                "Removed legacy features cache",
                "AI Segmentation", level=Qgis.MessageLevel.Info)

    cleanup_stale_partial_downloads()


def cleanup_stale_partial_downloads() -> None:
    """Drop partial checkpoint downloads nobody came back to resume.

    A failed download deliberately leaves its ".tmp" behind so the next attempt
    can pick it up, and nothing else ever deletes it. Past
    PARTIAL_DOWNLOAD_MAX_AGE_S that is a few hundred megabytes of dead weight
    in the user's cache. One pass at startup, never raises, and a file it
    cannot delete is simply left for the next start.
    """
    try:
        names = os.listdir(CHECKPOINTS_DIR)
    except OSError:
        return
    cutoff = time.time() - PARTIAL_DOWNLOAD_MAX_AGE_S
    for name in names:
        if not name.endswith(".tmp"):
            continue
        path = os.path.join(CHECKPOINTS_DIR, name)
        try:
            if not os.path.isfile(path) or os.path.getmtime(path) > cutoff:
                continue
        except OSError:
            continue
        # One attempt with no wait: a startup sweep must never hold the GUI on
        # a locked file, and the next start tries again.
        if _remove_with_retry(path, max_attempts=1, delay=0.0):
            QgsMessageLog.logMessage(
                f"Removed an abandoned partial download: {name}",
                "AI Segmentation", level=Qgis.MessageLevel.Info)
