"""A robust, typed Python wrapper around the rclone command-line tool.

The public entry point is :class:`Rclone`.  Every method builds an argument
*list* and runs it without a shell, so paths and remote specs are never
interpreted by the shell.  Common subcommands have explicit, typed methods;
anything else is reachable through :meth:`Rclone.run`.
"""

from __future__ import annotations

import json
import logging
import os
import shlex
import shutil
import subprocess
import threading
from pathlib import Path
from typing import Any, Callable, Dict, List, Optional, Sequence, Union

from .exceptions import RcloneCommandError, RcloneError, RcloneNotFoundError
from .models import Entry, RcloneResult, SizeResult, Stats

try:  # pragma: no cover - exercised indirectly
    from tqdm import tqdm as _tqdm
except ImportError:  # pragma: no cover
    _tqdm = None

logger = logging.getLogger("rclone")

#: A path on the local filesystem or a remote spec such as ``"gdrive:/dir"``.
PathLike = Union[str, "os.PathLike[str]"]

#: ``False``/``None`` for none, ``True`` for a tqdm bar, or a callback that
#: receives a :class:`~rclone.models.Stats` on every stats tick.
Progress = Union[bool, None, Callable[[Stats], None]]


def _parse_stats_line(line: str) -> Optional[Stats]:
    """Return a :class:`Stats` from one rclone JSON log line, or ``None``."""
    line = line.strip()
    if not line or '"stats"' not in line:
        return None
    try:
        entry = json.loads(line)
    except json.JSONDecodeError:
        return None
    stats = entry.get("stats")
    if not isinstance(stats, dict):
        return None
    return Stats.from_dict(stats)


class Rclone:
    """Wrapper around the ``rclone`` binary.

    Parameters
    ----------
    binary:
        Path to (or name of) the rclone executable.  When omitted the binary
        is looked up via the ``RCLONE_BINARY`` environment variable and then
        on ``PATH``.
    config:
        Optional path to an rclone config file (passed as ``--config``).
    flags:
        Global flags applied to *every* invocation (e.g. ``["--fast-list"]``).
    debug:
        Log each command at ``DEBUG`` level on the ``"rclone"`` logger.
    timeout:
        Default timeout, in seconds, for non-streaming commands.
    stats_interval:
        How often rclone emits progress stats while streaming (default
        ``"100ms"``).
    """

    def __init__(
        self,
        binary: Optional[PathLike] = None,
        *,
        config: Optional[PathLike] = None,
        flags: Optional[Sequence[str]] = None,
        debug: bool = False,
        timeout: Optional[float] = None,
        stats_interval: str = "100ms",
    ) -> None:
        self.binary = self._resolve_binary(binary)
        self.config = config
        self.flags: List[str] = list(flags or [])
        self.debug = debug
        self.timeout = timeout
        self.stats_interval = stats_interval

    # ------------------------------------------------------------------ #
    # Binary discovery
    # ------------------------------------------------------------------ #
    @staticmethod
    def _resolve_binary(explicit: Optional[PathLike] = None) -> str:
        for candidate in (explicit, os.environ.get("RCLONE_BINARY"), "rclone"):
            if not candidate:
                continue
            found = shutil.which(str(candidate))
            if found:
                return found
            path = Path(str(candidate)).expanduser()
            if path.is_file():
                return str(path)
        raise RcloneNotFoundError(
            "Could not find the rclone binary. Install rclone "
            "(https://rclone.org/downloads/) and add it to your PATH, set the "
            "RCLONE_BINARY environment variable, or pass binary=... to Rclone()."
        )

    @classmethod
    def installed(cls, binary: Optional[PathLike] = None) -> bool:
        """Return ``True`` if an rclone binary can be located."""
        try:
            cls._resolve_binary(binary)
        except RcloneNotFoundError:
            return False
        return True

    # ------------------------------------------------------------------ #
    # Low-level command execution
    # ------------------------------------------------------------------ #
    @property
    def _global_flags(self) -> List[str]:
        flags: List[str] = []
        if self.config is not None:
            flags += ["--config", str(self.config)]
        flags += self.flags
        return flags

    def _command(self, *parts: Any) -> List[str]:
        return [self.binary, *self._global_flags, *(str(p) for p in parts)]

    def _run(
        self,
        *parts: Any,
        check: bool = True,
        input_data: Union[str, bytes, None] = None,
        timeout: Optional[float] = None,
    ) -> RcloneResult:
        command = self._command(*parts)
        if self.debug:
            logger.debug("running: %s", " ".join(shlex.quote(c) for c in command))

        is_bytes = isinstance(input_data, (bytes, bytearray))
        try:
            completed = subprocess.run(
                command,
                input=input_data,
                stdout=subprocess.PIPE,
                stderr=subprocess.PIPE,
                encoding=None if is_bytes else "utf-8",
                errors=None if is_bytes else "replace",
                timeout=self.timeout if timeout is None else timeout,
            )
        except FileNotFoundError as exc:  # binary vanished between calls
            raise RcloneNotFoundError(
                f"rclone binary not found at {self.binary!r}"
            ) from exc

        stdout = completed.stdout or ("" if not is_bytes else b"")
        stderr = completed.stderr or ("" if not is_bytes else b"")
        if is_bytes:
            stdout = stdout.decode("utf-8", "replace")
            stderr = stderr.decode("utf-8", "replace")

        result = RcloneResult(command, completed.returncode, stdout, stderr)
        if check and not result.success:
            raise RcloneCommandError(command, result.returncode, stdout, stderr)
        return result

    def _run_with_progress(
        self,
        *parts: Any,
        progress: Progress,
        check: bool = True,
    ) -> RcloneResult:
        callback = progress if callable(progress) else None
        use_bar = progress is True
        if use_bar and _tqdm is None:
            logger.warning(
                "tqdm is not installed; install 'rclone[progress]' for a "
                "progress bar. Running without one."
            )
            use_bar = False

        command = self._command(
            *parts,
            "--use-json-log",
            "--stats",
            self.stats_interval,
            "--stats-log-level",
            "NOTICE",
            "-v",
        )
        if self.debug:
            logger.debug("running: %s", " ".join(shlex.quote(c) for c in command))

        try:
            proc = subprocess.Popen(
                command,
                stdout=subprocess.PIPE,
                stderr=subprocess.PIPE,
                encoding="utf-8",
                errors="replace",
                bufsize=1,
            )
        except FileNotFoundError as exc:
            raise RcloneNotFoundError(
                f"rclone binary not found at {self.binary!r}"
            ) from exc

        # Drain stdout on a thread so a full pipe buffer can never deadlock us
        # while we read progress from stderr.
        stdout_chunks: List[str] = []
        drainer = threading.Thread(
            target=lambda: stdout_chunks.append(proc.stdout.read() if proc.stdout else "")
        )
        drainer.start()

        bar = None
        stderr_lines: List[str] = []
        last_stats: Optional[Stats] = None
        try:
            assert proc.stderr is not None
            for line in proc.stderr:
                stderr_lines.append(line)
                stats = _parse_stats_line(line)
                if stats is None:
                    continue
                last_stats = stats
                if callback is not None:
                    callback(stats)
                if use_bar:
                    if bar is None:
                        bar = _tqdm(
                            total=stats.total_bytes or None,
                            unit="B",
                            unit_scale=True,
                            unit_divisor=1024,
                        )
                    if stats.total_bytes and bar.total != stats.total_bytes:
                        bar.total = stats.total_bytes
                        bar.refresh()
                    delta = stats.bytes - bar.n
                    if delta > 0:
                        bar.update(delta)
        finally:
            if bar is not None:
                bar.close()
            if proc.stderr is not None:
                proc.stderr.close()
            drainer.join()
            returncode = proc.wait()

        result = RcloneResult(
            command,
            returncode,
            stdout_chunks[0] if stdout_chunks else "",
            "".join(stderr_lines),
            stats=last_stats,
        )
        if check and not result.success:
            raise RcloneCommandError(
                command, returncode, result.stdout, result.stderr
            )
        return result

    def _transfer(
        self,
        subcommand: str,
        source: PathLike,
        dest: PathLike,
        flags: Sequence[str],
        progress: Progress,
        check: bool,
    ) -> RcloneResult:
        if progress:
            return self._run_with_progress(
                subcommand, source, dest, *flags, progress=progress, check=check
            )
        return self._run(subcommand, source, dest, *flags, check=check)

    # ------------------------------------------------------------------ #
    # Transfer commands (support progress reporting)
    # ------------------------------------------------------------------ #
    def copy(
        self,
        source: PathLike,
        dest: PathLike,
        *flags: str,
        progress: Progress = False,
        check: bool = True,
    ) -> RcloneResult:
        """Copy ``source`` to ``dest`` (``rclone copy``)."""
        return self._transfer("copy", source, dest, flags, progress, check)

    def copyto(
        self,
        source: PathLike,
        dest: PathLike,
        *flags: str,
        progress: Progress = False,
        check: bool = True,
    ) -> RcloneResult:
        """Copy ``source`` to ``dest``, file-to-file (``rclone copyto``)."""
        return self._transfer("copyto", source, dest, flags, progress, check)

    def move(
        self,
        source: PathLike,
        dest: PathLike,
        *flags: str,
        progress: Progress = False,
        check: bool = True,
    ) -> RcloneResult:
        """Move ``source`` to ``dest`` (``rclone move``)."""
        return self._transfer("move", source, dest, flags, progress, check)

    def moveto(
        self,
        source: PathLike,
        dest: PathLike,
        *flags: str,
        progress: Progress = False,
        check: bool = True,
    ) -> RcloneResult:
        """Move ``source`` to ``dest``, file-to-file (``rclone moveto``)."""
        return self._transfer("moveto", source, dest, flags, progress, check)

    def sync(
        self,
        source: PathLike,
        dest: PathLike,
        *flags: str,
        progress: Progress = False,
        check: bool = True,
    ) -> RcloneResult:
        """Make ``dest`` identical to ``source`` (``rclone sync``).

        Destructive: extra files at ``dest`` are deleted.
        """
        return self._transfer("sync", source, dest, flags, progress, check)

    def bisync(
        self,
        path1: PathLike,
        path2: PathLike,
        *flags: str,
        progress: Progress = False,
        check: bool = True,
    ) -> RcloneResult:
        """Two-way synchronise ``path1`` and ``path2`` (``rclone bisync``)."""
        return self._transfer("bisync", path1, path2, flags, progress, check)

    def copyurl(
        self,
        url: str,
        dest: PathLike,
        *flags: str,
        progress: Progress = False,
        check: bool = True,
    ) -> RcloneResult:
        """Download ``url`` and copy it to ``dest`` (``rclone copyurl``)."""
        return self._transfer("copyurl", url, dest, flags, progress, check)

    # ------------------------------------------------------------------ #
    # Destructive commands
    # ------------------------------------------------------------------ #
    def delete(self, path: PathLike, *flags: str, check: bool = True) -> RcloneResult:
        """Delete the contents of ``path`` (``rclone delete``)."""
        return self._run("delete", path, *flags, check=check)

    def deletefile(self, path: PathLike, *flags: str, check: bool = True) -> RcloneResult:
        """Delete a single file (``rclone deletefile``)."""
        return self._run("deletefile", path, *flags, check=check)

    def purge(self, path: PathLike, *flags: str, check: bool = True) -> RcloneResult:
        """Remove ``path`` and all of its contents (``rclone purge``)."""
        return self._run("purge", path, *flags, check=check)

    def rmdir(self, path: PathLike, *flags: str, check: bool = True) -> RcloneResult:
        """Remove an empty directory (``rclone rmdir``)."""
        return self._run("rmdir", path, *flags, check=check)

    def rmdirs(self, path: PathLike, *flags: str, check: bool = True) -> RcloneResult:
        """Remove empty directories under ``path`` (``rclone rmdirs``)."""
        return self._run("rmdirs", path, *flags, check=check)

    def cleanup(self, path: PathLike, *flags: str, check: bool = True) -> RcloneResult:
        """Clean up the remote's trash/old versions (``rclone cleanup``)."""
        return self._run("cleanup", path, *flags, check=check)

    # ------------------------------------------------------------------ #
    # Creation commands
    # ------------------------------------------------------------------ #
    def mkdir(self, path: PathLike, *flags: str, check: bool = True) -> RcloneResult:
        """Create ``path`` (``rclone mkdir``)."""
        return self._run("mkdir", path, *flags, check=check)

    def touch(self, path: PathLike, *flags: str, check: bool = True) -> RcloneResult:
        """Create or update the mod-time of a file (``rclone touch``)."""
        return self._run("touch", path, *flags, check=check)

    # ------------------------------------------------------------------ #
    # Listing commands
    # ------------------------------------------------------------------ #
    def ls(self, path: PathLike, *flags: str) -> List[str]:
        """List names in ``path`` (``rclone lsf``). Returns a list of names."""
        out = self._run("lsf", path, *flags).stdout
        return [line for line in out.splitlines() if line]

    def lsjson(self, path: PathLike, *flags: str) -> List[Dict[str, Any]]:
        """List ``path`` as parsed JSON objects (``rclone lsjson``)."""
        out = self._run("lsjson", path, *flags).stdout
        return json.loads(out or "[]")

    def lsl(self, path: PathLike, *flags: str) -> List[Entry]:
        """List files with size and mod-time as :class:`Entry` objects (``rclone lsl``)."""
        out = self._run("lsl", path, *flags).stdout
        return [Entry.from_lsl_line(line) for line in out.splitlines() if line.strip()]

    def lsd(self, path: PathLike, *flags: str) -> List[Entry]:
        """List directories as :class:`Entry` objects (``rclone lsd``)."""
        out = self._run("lsd", path, *flags).stdout
        return [Entry.from_lsd_line(line) for line in out.splitlines() if line.strip()]

    def tree(self, path: PathLike, *flags: str) -> str:
        """Return a tree view of ``path`` (``rclone tree``)."""
        return self._run("tree", path, *flags).stdout

    # ------------------------------------------------------------------ #
    # Info commands
    # ------------------------------------------------------------------ #
    def size(self, path: PathLike, *flags: str) -> SizeResult:
        """Return object count and total size for ``path`` (``rclone size``)."""
        out = self._run("size", path, "--json", *flags).stdout
        data = json.loads(out)
        return SizeResult(
            count=int(data.get("count", 0)),
            bytes=int(data.get("bytes", 0)),
            sizeless=int(data.get("sizeless", 0)),
        )

    def about(self, remote: str, *flags: str) -> Dict[str, Any]:
        """Return usage info for ``remote`` (``rclone about --json``)."""
        out = self._run("about", remote, "--json", *flags).stdout
        return json.loads(out)

    def hashsum(self, hash_type: str, path: PathLike, *flags: str) -> Dict[str, str]:
        """Return ``{name: hash}`` for ``path`` (``rclone hashsum``)."""
        out = self._run("hashsum", hash_type, path, *flags).stdout
        result: Dict[str, str] = {}
        for line in out.splitlines():
            parts = line.strip().split(None, 1)
            if len(parts) == 2:
                digest, name = parts
                result[name.strip()] = digest
        return result

    def md5sum(self, path: PathLike, *flags: str) -> Dict[str, str]:
        """Return ``{name: md5}`` for ``path``."""
        return self.hashsum("md5", path, *flags)

    def sha1sum(self, path: PathLike, *flags: str) -> Dict[str, str]:
        """Return ``{name: sha1}`` for ``path``."""
        return self.hashsum("sha1", path, *flags)

    def check(
        self,
        source: PathLike,
        dest: PathLike,
        *flags: str,
        raise_on_error: bool = False,
    ) -> RcloneResult:
        """Compare ``source`` and ``dest`` (``rclone check``).

        Differences make rclone exit non-zero; by default this is reported in
        the returned :class:`RcloneResult` rather than raised.  Pass
        ``raise_on_error=True`` to raise on any non-zero exit.
        """
        return self._run("check", source, dest, *flags, check=raise_on_error)

    # ------------------------------------------------------------------ #
    # Content commands
    # ------------------------------------------------------------------ #
    def cat(self, path: PathLike, *flags: str) -> str:
        """Return the contents of a remote file as text (``rclone cat``)."""
        return self._run("cat", path, *flags).stdout

    def rcat(
        self, dest: PathLike, data: Union[str, bytes], *flags: str, check: bool = True
    ) -> RcloneResult:
        """Write ``data`` to ``dest``, reading from stdin (``rclone rcat``)."""
        return self._run("rcat", dest, *flags, input_data=data, check=check)

    # ------------------------------------------------------------------ #
    # Remote / config / misc
    # ------------------------------------------------------------------ #
    def listremotes(self, *flags: str, strip_colon: bool = False) -> List[str]:
        """List configured remotes (``rclone listremotes``).

        Names keep their trailing ``":"`` unless ``strip_colon=True``.
        """
        out = self._run("listremotes", *flags).stdout
        remotes = [line for line in out.splitlines() if line]
        if strip_colon:
            remotes = [r[:-1] if r.endswith(":") else r for r in remotes]
        return remotes

    def obscure(self, password: str) -> str:
        """Return the obscured form of ``password`` (``rclone obscure``)."""
        return self._run("obscure", password).stdout.strip()

    def version(self) -> str:
        """Return the rclone version string, e.g. ``"v1.74.3"``."""
        out = self._run("version").stdout
        first = out.splitlines()[0] if out else ""
        return first.split()[-1] if first else ""

    def config(self, *args: str, check: bool = True) -> RcloneResult:
        """Run an ``rclone config`` subcommand (e.g. ``config("dump")``)."""
        return self._run("config", *args, check=check)

    # ------------------------------------------------------------------ #
    # Generic escape hatch
    # ------------------------------------------------------------------ #
    def run(
        self,
        *args: Any,
        check: bool = True,
        input_data: Union[str, bytes, None] = None,
        timeout: Optional[float] = None,
    ) -> RcloneResult:
        """Run an arbitrary rclone command, e.g. ``run("ls", "remote:dir")``.

        Arguments are passed verbatim as a list (no shell), so values
        containing spaces or special characters are safe.
        """
        if not args:
            raise RcloneError("run() requires at least a subcommand")
        return self._run(
            *args, check=check, input_data=input_data, timeout=timeout
        )

    def execute(self, command: Union[str, Sequence[str]], check: bool = True) -> str:
        """Run a command and return its stdout (legacy convenience helper).

        A string is split with :func:`shlex.split`; a sequence is used as-is.
        Prefer :meth:`run` for new code.
        """
        args = shlex.split(command) if isinstance(command, str) else list(command)
        return self._run(*args, check=check).stdout.rstrip()

    def __repr__(self) -> str:
        return f"Rclone(binary={self.binary!r})"