chemenu/tools/chemenu/collect.py

"""Pulling material into the quarantine from the channels that are not pushed
(Gitea #213): a folder a sync client fills (`folder`), a program of this
checkout (`command`).

A handler is only a source. It hands over files unchanged; converting or
deriving stays with `raw derive`, after a human has accepted the submission.
Everything it takes ends in `chemenu.upload.store`, so nothing reaches
`incoming/` or `raw/` any other way than `upload accept`.

This module is deliberately not part of `chemenu.upload`: it reads outside the
quarantine, deletes in the source folder and starts programs, none of which the
MCP server process may do. Like `upload accept`, it is not importable from there.

What a pull guarantees:

- A source file is removed only after its copy in the quarantine is fully
  written and the source still has the size and sha256 that were read.
- Outside the quarantine it changes one thing: it removes files it took, directly
  in the channel's `path`. It writes nothing there, deletes no directory and
  follows no symbolic link.
- Whatever it does not take stays where it is and is reported with its reason.
- It never empties data of the backup or of an export: a folder that holds a
  backup or export manifest or is a bare repository, or lies in one, is the
  channel's error - nothing read, nothing removed.

No `typer`, no `rich`: the CLI renders a `ChannelResult`.
"""
from __future__ import annotations

import hashlib
import os
import stat
import subprocess
import time
from dataclasses import dataclass, field
from pathlib import Path
from typing import Callable, Optional

from chemenu import backup, page_export, upload
from chemenu.errors import ValidationError

COMMAND_TIMEOUT_SECONDS = 300
_STDERR_TAIL_CHARS = 2000
_HIDDEN_PREFIXES = (".", "~$")
_WINDOWS_HIDDEN = getattr(stat, "FILE_ATTRIBUTE_HIDDEN", 0) | getattr(stat, "FILE_ATTRIBUTE_SYSTEM", 0)


@dataclass
class ChannelResult:
    """What one channel's pull did. `taken` and `duplicates` are
    `(name, submission id)`, `skipped` is `(name, reason)`, `failed` is
    `(name, reason)` for a file that should have been taken and was not.
    `error` is set when the channel as a whole could not be pulled."""

    channel: str
    handler: str
    taken: list = field(default_factory=list)
    duplicates: list = field(default_factory=list)
    skipped: list = field(default_factory=list)
    failed: list = field(default_factory=list)
    error: Optional[str] = None

    @property
    def ok(self) -> bool:
        return self.error is None and not self.failed


def _tail(stderr: bytes) -> str:
    text = stderr.decode("utf-8", errors="replace").strip()
    if len(text) > _STDERR_TAIL_CHARS:
        text = "..." + text[-_STDERR_TAIL_CHARS:]
    return text or "(nothing on stderr)"


def _digest(path: Path) -> tuple[int, str]:
    sha = hashlib.sha256()
    size = 0
    with path.open("rb") as handle:
        for block in iter(lambda: handle.read(1024 * 1024), b""):
            sha.update(block)
            size += len(block)
    return size, sha.hexdigest()


def source_matches(path: Path, size: int, digest: str) -> bool:
    """Whether `path` still is the file that was read: a regular file of the
    same size and sha256. The check that gates removing a source."""
    try:
        if path.is_symlink() or not path.is_file():
            return False
        return _digest(path) == (size, digest)
    except OSError:
        return False


def _hidden(name: str, mode_attributes: int) -> bool:
    return name.startswith(_HIDDEN_PREFIXES) or bool(mode_attributes & _WINDOWS_HIDDEN)


def _moving(info: os.stat_result, settle_seconds: int, now: float) -> bool:
    """Whether a file may still be arriving: its modification time or, where
    the platform has it, its creation time is younger than `settle_seconds`."""
    stamps = [info.st_mtime]
    born = getattr(info, "st_birthtime", None)
    if born is not None:
        stamps.append(born)
    return any(now - stamp < settle_seconds for stamp in stamps)


def _take(
    root: Path, channel: upload.Channel, result: ChannelResult, path: Path, name: str, origin: str,
    *, after: Optional[Callable[[Path], None]] = None, guard_source: bool,
) -> None:
    """Read `path`, store it, and call `after(path)` once it is safe in the
    quarantine. Reports into `result`; never raises for a problem with this
    one file."""
    try:
        before = path.stat()
        if before.st_size > channel.max_bytes:
            result.skipped.append((name, f"larger than max_bytes ({channel.max_bytes})"))
            return
        data = path.read_bytes()
    except OSError as exc:
        result.failed.append((name, f"could not be read ({exc})"))
        return
    if len(data) != before.st_size:
        result.skipped.append((name, "changed while it was read - left for the next run"))
        return
    if not data:
        result.skipped.append((name, "empty"))
        return
    digest = hashlib.sha256(data).hexdigest()
    waiting = upload.digest_to_pending_id(root, digest)
    if waiting is not None:
        if after is not None and not source_matches(path, len(data), digest):
            result.skipped.append((name, "changed while it was read - left for the next run"))
            return
        result.duplicates.append((name, waiting))
        _finish(result, path, name, after)
        return
    verify = (lambda: source_matches(path, len(data), digest)) if guard_source else None
    try:
        manifest = upload.store(
            root, channel, filename=name, data=data,
            submitter=upload.UNKNOWN_SUBMITTER, submitter_source=channel.handler,
            origin=origin, verify=verify,
        )
    except upload.SourceChanged:
        result.skipped.append((name, "changed while it was copied - nothing was kept, left for the next run"))
        return
    except (ValidationError, OSError) as exc:
        result.failed.append((name, str(exc)))
        return
    result.taken.append((name, manifest["id"]))
    _finish(result, path, name, after, manifest["id"])


def _finish(result: ChannelResult, path: Path, name: str, after, submission_id: str = "") -> None:
    if after is None:
        return
    try:
        after(path)
    except OSError as exc:
        kept = f" (stored as {submission_id})" if submission_id else ""
        result.failed.append((name, f"stays in the source, which could not be emptied{kept}: {exc}"))


def collect_folder(root: "Path | str", channel: upload.Channel, now: Optional[float] = None) -> ChannelResult:
    """Pull the top level of the channel's `path` into the quarantine."""
    root = Path(root)
    result = ChannelResult(channel.name, channel.handler)
    folder = channel.path
    if folder is None or not folder.is_dir():
        result.error = f"the folder {folder} does not exist or is not a directory"
        return result
    marker = backup.data_marker(folder, (page_export.MANIFEST_NAME,))
    if marker:
        result.error = (f"the folder {folder} lies in backup or export data ({marker}) - a channel empties "
                        "its folder, so nothing was read or removed")
        return result
    clock = time.time() if now is None else now
    try:
        entries = sorted(os.scandir(folder), key=lambda entry: entry.name)
    except OSError as exc:
        result.error = f"the folder {folder} could not be read ({exc})"
        return result
    for entry in entries:
        name = entry.name
        path = folder / name
        try:
            info = entry.stat(follow_symlinks=False)
        except OSError as exc:
            result.failed.append((name, f"could not be inspected ({exc})"))
            continue
        if entry.is_symlink():
            result.skipped.append((name, "symbolic link"))
        elif stat.S_ISDIR(info.st_mode):
            result.skipped.append((name, "subdirectory"))
        elif _hidden(name, getattr(info, "st_file_attributes", 0)):
            result.skipped.append((name, "hidden"))
        elif not stat.S_ISREG(info.st_mode):
            result.skipped.append((name, "not a regular file"))
        elif Path(name).suffix.lower() not in channel.allowed_extensions:
            result.skipped.append((name, "extension not allowed"))
        elif info.st_size > channel.max_bytes:
            result.skipped.append((name, f"larger than max_bytes ({channel.max_bytes})"))
        elif _moving(info, channel.settle_seconds, clock):
            result.skipped.append((name, f"still in motion (younger than {channel.settle_seconds} s)"))
        else:
            try:
                upload.sanitize_filename(name)
            except ValidationError as exc:
                result.skipped.append((name, f"name refused ({exc})"))
                continue
            _take(root, channel, result, path, name, str(path), after=lambda p: p.unlink(), guard_source=True)
    return result


def collect_command(root: "Path | str", channel: upload.Channel) -> ChannelResult:
    """Run the channel's program with an empty staging directory as its last
    argument; take each regular file it left at the top of it."""
    root = Path(root)
    result = ChannelResult(channel.name, channel.handler)
    origin = f"command:{channel.command_name}"
    staging = upload.staging_dir(root)
    try:
        where = f"the command {channel.command_name!r}"
        try:
            ran = subprocess.run(
                [*channel.command, str(staging)], capture_output=True, stdin=subprocess.DEVNULL,
                timeout=COMMAND_TIMEOUT_SECONDS, check=False,
            )
        except FileNotFoundError:
            result.error = f"{where} could not be started: {channel.command[0]} does not exist"
            return result
        except subprocess.TimeoutExpired:
            result.error = f"{where} did not finish within {COMMAND_TIMEOUT_SECONDS} s"
            return result
        except OSError as exc:
            result.error = f"{where} could not be started ({exc})"
            return result
        if ran.returncode != 0:
            result.error = f"{where} ended with exit {ran.returncode}. Its stderr:\n{_tail(ran.stderr)}"
            return result
        for entry in sorted(os.scandir(staging), key=lambda e: e.name):
            name = entry.name
            path = staging / name
            if entry.is_symlink():
                result.skipped.append((name, "symbolic link"))
            elif not entry.is_file(follow_symlinks=False):
                result.skipped.append((name, "not a regular file"))
            elif Path(name).suffix.lower() not in channel.allowed_extensions:
                result.skipped.append((name, "extension not allowed"))
            else:
                try:
                    upload.sanitize_filename(name)
                except ValidationError as exc:
                    result.skipped.append((name, f"name refused ({exc})"))
                    continue
                _take(root, channel, result, path, name, origin, guard_source=False)
    finally:
        upload.drop_staging(root, staging)
    return result


def collect(root: "Path | str", channel: upload.Channel) -> ChannelResult:
    """One channel, whatever its handler. Never raises for a failure of the
    channel: that is the result's `error`, so the others can still run."""
    try:
        if channel.handler == "folder":
            return collect_folder(root, channel)
        return collect_command(root, channel)
    except (ValidationError, OSError) as exc:
        failed = ChannelResult(channel.name, channel.handler)
        failed.error = str(exc)
        return failed