chemenu/tools/chemenu/commands/raw/derive.py

"""`wikitool raw derive` - the sanctioned intake for a document that is not text: a
PDF, a mail, an Office file. Like `raw fetch` it ends in `incoming/`, never in
`raw/`. The formats live in `chemenu.filters`; this adapter decides names and
folders and what the caller is told. Everything is read, derived and checked
before the first write, so a refusal leaves `incoming/` as it was.

`raw derive` is the way in for a document that is not text (PDF, mail, Office
file): it turns one file in `incoming/` into a folder `<stem>/` holding the
untouched `original.<ext>` and a `derived.md` made by the registered filter
(`chemenu.filters`), so `raw accept` takes the folder as one source. Everything
is checked before the first write; the original moves in last.
"""
from __future__ import annotations

import os
import shutil
from dataclasses import dataclass
from pathlib import Path
from typing import Annotated, Optional

import typer
from rich.markup import escape

from chemenu import cli_contract, config, filters, repo_capture
from chemenu.commands._util import fail, path_budget_problem_for, rel_path, success
from chemenu.errors import ValidationError
from chemenu.commands.raw._common import (
    _Refused,
    _check_directly_in_incoming,
    _folder_collision_message,
    _incoming_dir,
    _occupied_stems,
    _refusals_fail,
    _resolve,
    _shard_dir,
    _unusable_name,
    _walk_folder,
    _write_exclusive,
)


DERIVE_MAX_BYTES = 100 * 1024 * 1024
ORIGINAL_STEM = "original"
DERIVED_NAME = "derived.md"


@dataclass(frozen=True)
class _Derived:
    """One finished derivation, not yet written."""

    filter: filters.Filter
    derivation: filters.Derivation
    source: Path  # the file in incoming/
    markdown: bytes
    target: Path  # where the .md goes
    extras: tuple[tuple[Path, bytes], ...] = ()  # the derivation's further files, beside the .md


def _read_for_derive(path: Path) -> bytes:
    size = path.stat().st_size
    if size > DERIVE_MAX_BYTES:
        raise _Refused(
            f"{rel_path(path)} is {size // (1024 * 1024)} MiB; raw derive reads at most "
            f"{DERIVE_MAX_BYTES // (1024 * 1024)} MiB."
        )
    return path.read_bytes()


def _run_filter(candidate: filters.Filter, data: bytes, path: Path) -> filters.Derivation:
    try:
        filters.require_available(candidate)
        return candidate.derive(data, path.name)
    except ValidationError as exc:
        raise _Refused(f"{rel_path(path)}: {exc}") from None


def _collision_warning(name: str) -> Optional[str]:
    holder = _occupied_stems(config.RAW_DIR).get(name)
    if holder is None:
        return None
    return _folder_collision_message(name, holder)


@dataclass(frozen=True)
class _FilePlan:
    """A single file turned into a folder, not yet written."""

    folder: Path
    stem: str
    candidate: filters.Filter
    original: Optional[Path]  # where the source moves to; None when the source is only unpacked
    derived: Optional[Path]
    markdown: bytes
    container: Optional[filters.Container]
    derivations: list[filters.Derivation]
    extras: tuple[tuple[str, bytes], ...] = ()  # (path in the folder, bytes) of the derivation's further files


def _plan_derive_file(path: Path, name: Optional[str]) -> _FilePlan:
    """The single-file form: `incoming/<stem>/` with the original moved in as
    `original.<ext>` and the derived text beside it - or, for a container (a ZIP,
    a mail), the members under their own names."""
    incoming = _incoming_dir()
    if not path.is_file():
        raise _Refused(f"{rel_path(path)} does not exist or is not a file.")
    _check_directly_in_incoming(path, incoming)
    if path.name.startswith(".") or path.is_symlink():
        raise _Refused(f"{rel_path(path)} is hidden or a symlink - raw derive does not take it.")
    candidate = filters.filter_for(path.name)
    if candidate is None:
        listed = ", ".join(filters.supported_extensions())
        raise _Refused(
            f"No filter reads {filters.extension_of(path.name) or 'a file without an extension'} "
            f"({rel_path(path)}). raw derive reads: {listed}. Any other file is accepted as it is "
            "- note what it shows."
        )
    stem = name if name is not None else path.stem
    if _unusable_name(stem):
        raise _Refused(
            f"{stem!r} is not a usable folder name: one directory name - no path separator, not "
            f"hidden, not {repo_capture.MANIFEST_NAME}. Pass --name <stem>."
        )
    folder = incoming / stem
    if os.path.lexists(folder):
        raise _Refused(
            f"{rel_path(folder)} already exists in incoming/ - raw derive never overwrites. Accept "
            "or remove what is there, or pass --name <stem>."
        )
    extension = filters.extension_of(path.name)
    data = _read_for_derive(path)
    container = None
    try:
        filters.require_available(candidate)
        container = candidate.container(data, path.name)
    except ValidationError as exc:
        raise _Refused(f"{rel_path(path)}: {exc}") from None
    if container is not None and container.text is None:
        original = derived = None
        markdown = b""
        targets = [rel for rel, _data in container.files]
        derivations = list(container.derivations)
        extras = ()
    else:
        original = folder / f"{ORIGINAL_STEM}{extension}"
        derived = folder / DERIVED_NAME
        derivation = container.text if container is not None else _run_filter(candidate, data, path)
        assert derivation is not None
        companion = filters.companion_name(DERIVED_NAME) if derivation.files else None
        markdown = filters.render(
            candidate, derivation, derived_from=original.name, original_name=path.name,
            companion=companion,
        )
        extras = tuple((f"{companion}/{name}", content) for name, content in derivation.files)
        targets = [original.name, derived.name] + [rel for rel, _data in extras] + (
            [rel for rel, _data in container.files] if container is not None else []
        )
        derivations = [derivation] + (list(container.derivations) if container is not None else [])
    for target in targets:
        problem = path_budget_problem_for(_shard_dir() / stem / target)
        if problem:
            raise _Refused(
                f"Cannot write {problem}. The folder name comes from the file's name: pass "
                "--name <shorter stem>."
            )
    return _FilePlan(folder, stem, candidate, original, derived, markdown, container, derivations, extras)


def _write_folder(source: Path, plan: _FilePlan) -> None:
    """Create the folder and write everything into it, then move the original in
    (or, for an unpacked archive, remove it) - last, so a failure before it has
    taken nothing from `incoming/`. A failure part-way removes the folder this
    call made and puts the original back."""
    moved = False
    try:
        plan.folder.mkdir()
        if plan.derived is not None:
            with plan.derived.open("xb") as handle:
                handle.write(plan.markdown)
        for rel, data in plan.extras:
            target = plan.folder / rel
            target.parent.mkdir(parents=True, exist_ok=True)
            with target.open("xb") as handle:
                handle.write(data)
        if plan.container is not None:
            for rel, data in plan.container.files:
                target = plan.folder / rel
                target.parent.mkdir(parents=True, exist_ok=True)
                with target.open("xb") as handle:
                    handle.write(data)
        if plan.original is not None:
            source.rename(plan.original)
            moved = True
        else:
            source.unlink()
    except OSError as exc:
        if moved and plan.original is not None:
            plan.original.rename(source)
        shutil.rmtree(plan.folder, ignore_errors=True)
        fail(escape(f"Could not write {rel_path(plan.folder)}: {exc}. incoming/ was left as it was."))


def _plan_derive_folder(folder: Path):
    """The folder form: a `<stem>.md` beside every file that has a filter."""
    incoming = _incoming_dir()
    try:
        rel = folder.relative_to(incoming)
    except ValueError:
        raise _Refused(
            f"{rel_path(folder)} is not under incoming/ - raw derive only takes material from there."
        ) from None
    if len(rel.parts) != 1:
        raise _Refused(
            f"incoming/{rel.as_posix()} is not directly in incoming/ - a folder is derived from "
            f"the top: raw derive incoming/{rel.parts[0]}"
        )
    if folder.name.startswith(".") or folder.is_symlink():
        raise _Refused(f"{rel_path(folder)} is hidden or a symlink - raw derive does not take it.")
    if (folder / repo_capture.MANIFEST_NAME).is_file():
        raise _Refused(f"{rel_path(folder)}/ is a captured bundle; a captured bundle is not derived.")
    files, _dirs, refused = _walk_folder(folder)
    if refused:
        listed = "\n".join(f"  - {entry}" for entry in refused)
        raise _Refused(
            f"{rel_path(folder)}/ holds entries raw accept would not take - hidden entries, "
            f"symlinks and special files:\n{listed}\n  Remove or replace them in incoming/, then "
            "derive the folder again."
        )
    plan: list[tuple[Path, filters.Filter, Path]] = []
    untouched: list[Path] = []
    for file in files:
        candidate = filters.filter_for(file.name)
        if candidate is None or candidate.container_only:
            untouched.append(file)
        else:
            plan.append((file, candidate, file.with_suffix(".md")))
    if not plan:
        raise _Refused(
            f"No file in {rel_path(folder)}/ has a filter. raw derive reads: "
            f"{', '.join(filters.supported_extensions())}."
        )
    targets: dict[Path, Path] = {}
    for file, _candidate, md in plan:
        if md in targets:
            raise _Refused(
                f"{rel_path(targets[md])} and {rel_path(file)} would both derive to {rel_path(md)}. "
                "Rename one of them in incoming/ and derive the folder again."
            )
        targets[md] = file
        if os.path.lexists(md):
            raise _Refused(
                f"{rel_path(md)} already exists - raw derive never overwrites. Remove it or "
                "rename it, then derive the folder again."
            )
        problem = path_budget_problem_for(_shard_dir() / folder.name / md.relative_to(folder))
        if problem:
            raise _Refused(
                f"Cannot write {problem}. The path comes from the folder in incoming/: shorten the "
                "folder's name or the names inside it."
            )
    derived: list[_Derived] = []
    for file, candidate, md in plan:
        derivation = _run_filter(candidate, _read_for_derive(file), file)
        companion = md.with_name(filters.companion_name(md.name))
        extras: tuple[tuple[Path, bytes], ...] = ()
        if derivation.files:
            if os.path.lexists(companion):
                raise _Refused(
                    f"{rel_path(companion)} already exists - raw derive never overwrites. Remove it "
                    "or rename it, then derive the folder again."
                )
            extras = tuple((companion / name, content) for name, content in derivation.files)
            for target, _content in extras:
                problem = path_budget_problem_for(_shard_dir() / folder.name / target.relative_to(folder))
                if problem:
                    raise _Refused(
                        f"Cannot write {problem}. The path comes from the folder in incoming/: "
                        "shorten the folder's name or the names inside it."
                    )
        markdown = filters.render(
            candidate, derivation, derived_from=file.name, companion=companion.name
        )
        derived.append(_Derived(candidate, derivation, file, markdown, md, extras))
    return derived, untouched


def _echo_notes(
    derivations: list[filters.Derivation], container: Optional[filters.Container] = None
) -> None:
    warnings = [w for d in derivations for w in d.warnings]
    checks = [c for d in derivations for c in d.checks]
    if container is not None:
        warnings += container.warnings
        checks += container.checks
    for warning in filters.unique(warnings):
        typer.echo(f"WARN {warning}")
    for line in filters.unique(checks):
        typer.echo(f"CHECK {line}")


def _container_report(container: filters.Container) -> list[str]:
    lines = [f"  {src} -> {md} ({name})" for src, md, name in container.derived]
    if container.plain:
        lines.append("  No filter for, written as they are: " + ", ".join(container.plain))
    if container.renamed:
        lines.append(
            "  Renamed (a name was taken or reserved): "
            + ", ".join(f"{old} -> {new}" for old, new in container.renamed)
        )
    if container.hidden:
        lines.append("  Hidden entries, named but not written: " + ", ".join(container.hidden))
    return lines


def _derive_next(target: Path) -> str:
    return (
        "  Next (after the commitment question in wiki-ingest):\n"
        f"  tools/wikitool raw accept --fidelity <value> --authority <value> {rel_path(target)}"
    )


@cli_contract.record(cli_contract.CommandRecord(
    path="raw derive",
    summary="Derive reading text from a PDF, e-mail, Office file or web page in `incoming/`, or "
    "unpack a ZIP, and leave the original beside it, for `raw accept` to promote.",
    synopsis=(
        cli_contract.Variant(
            usage="raw derive incoming/<file> [--name <stem>]",
            notes="Move the file into the new folder `incoming/<stem>/` as `original.<ext>` and "
            "write `incoming/<stem>/derived.md`",
        ),
        cli_contract.Variant(
            usage="raw derive incoming/<folder>",
            notes="Write a `<stem>.md` beside every file in the folder that has a filter",
        ),
    ),
    properties=cli_contract.Properties(
        effect=cli_contract.Effect.WRITE,
        idempotent=cli_contract.Idempotent.NO,
        atomic="Yes for what it leaves behind - everything is read, derived and checked before "
        "the first write; the files are created exclusively, and a failure part-way puts back "
        "what this call changed",
        budget=cli_contract.Budget.COUNTED,
        network=cli_contract.Network.NO,
    ),
    notes=(
        "Writes into `incoming/` only, never into `raw/`, and reads no network: `raw accept` "
        "promotes the folder afterwards as one source under `raw/<YYYY>/<MM>/<stem>/`.",
        "A single file directly in `incoming/` becomes a folder named after its stem (or "
        "`--name`) holding `original.<ext>` - the original bytes, moved - and `derived.md`. The "
        "fixed names keep the path inside the path budget; the header of `derived.md` records "
        "the file's own name as `original_name`.",
        "A folder is derived in place: each file with a filter gets `<stem>.md` beside it, "
        "nothing is renamed, and the files without a filter are named in the output.",
        "`derived.md` starts with a fixed header (`derived_by`, `filter`, `engine`, `derived`, "
        "`derived_from`, `original_name` for a renamed file, `title`, then the fields of the "
        "format) followed by the derived text. The same bytes always give the same text; only "
        "`derived` differs between runs. `engine` names the tool version and, where a package "
        "reads the format, its version.",
        "Formats: web pages (`.html`), Word (`.docx`), PowerPoint (`.pptx`), Excel (`.xlsx`, "
        "`.xlsm`), e-mail (`.eml`) and ZIP archives (`.zip`), read with the standard library "
        "only; PDF (`.pdf`, `pypdf` and `pdfminer.six`) and Outlook messages (`.msg`, `olefile`), each "
        "read by packages in `tools/requirements.txt` that are imported only when such a file is "
        "derived. A missing package refuses that format alone, with the preflight hint; "
        "`wikitool doctor` shows one `import-filter-<name>` line per package-backed format. A "
        "PDF gets `<!-- page N -->` markers and a WARN when it yields almost no text (a scan; "
        "there is no OCR); an encrypted PDF is refused. A `.msg` is a container like an `.eml`; an "
        "embedded message inside it is named, not read. The text of a mail (`.eml` and `.msg` "
        "alike) is its plain text, unless that is under 10% of the length of the HTML text - a "
        "\"plain text version not available\" placeholder - in which case it is the HTML text "
        "and a WARN names both lengths. The other version of the text in a "
        "`multipart/alternative` is not an attachment.",
        "A PDF also gets, where the document has them: an `outline` header field (one entry per "
        "outline item: title, level, page and the line of `derived.md` where that page's "
        "marker is), and markings in the text that follow the visible form only - `<mark "
        "data-farbe=\"<name>\">…</mark>` for text over a coloured background, `*…*` for text in a "
        "colour other than the running text that is not under a link, `~~…~~` for strike-through. "
        "A `markings` header field is the legend (form, colour, places, pages); removing the "
        "markings gives the text the PDF has without them. A CHECK names the pages with images, "
        "which the text does not contain. A PDF without outline, markings and images is derived "
        "as before.",
        "A workbook (`.xlsx`, `.xlsm`) gets its cells as files: beside its `.md` a folder "
        "`<md stem>.sheets/` (`derived.sheets/` for a single file) holds one `NN-<sheet>.tsv` "
        "per sheet - tab-separated per RFC 4180, UTF-8, row 1 of the file is row 1 of the sheet "
        "- and, for a sheet with formulas, `NN-<sheet>.formulas.tsv` listing cell, formula and "
        "last stored value. The `.md` is the overview: one section per sheet with its range, "
        "the file names and a preview. Formulas are not recalculated; a shared formula is "
        "written out per cell, and one that cannot be is named `shared:<master cell>` with a "
        "WARN. The folder counts against the path budget and is accepted with the rest; an "
        "existing one is never overwritten.",
        "A ZIP and a mail are containers. A ZIP at the top of `incoming/` becomes the folder "
        "`incoming/<stem>/` holding its files under their own names, each with a `<stem>.md` "
        "where a filter reads it; the ZIP is removed and the output names it with its sha256. A "
        "mail becomes a folder with `original.eml`, `derived.md` and its attachments under their "
        "own names; a ZIP attached to it is unpacked once into a folder of its own, and an "
        "attached mail is quoted in `derived.md`, not unpacked. A ZIP inside a ZIP, an "
        "absolute or `..` path, a symbolic link, an encrypted entry, more than 1000 entries, a "
        "file over 25 MiB or more than 250 MiB in total refuse the whole call. Hidden entries "
        "(`__MACOSX/`, `.DS_Store`, anything starting with a dot) are named, not written. A name "
        "taken twice, or reserved (`original.*` and `derived.md` at the top of a mail's folder, "
        "`_capture.json` anywhere), gets `-2`, `-3` before its extension, and the output lists "
        "each rename.",
        "A name already taken under `raw/` is a WARN, not a refusal: whether the folder is a "
        "later edition of that source or a separate one is the human's call. A later edition of "
        "a derived source is replaced file by file with `raw accept --replaces`; there is no "
        "`--replaces-bundle` for it.",
        "After a success the output prints the warnings the filters raised, at most three "
        "`CHECK` lines per format that was derived in this run - what to look at before the "
        "text is trusted - and the `raw accept` line for the result.",
        "Never edit `derived.md` by hand: a better derivation is a new `raw derive` of the original.",
    ),
    failures=(
        cli_contract.Failure(
            cause="The path is not directly in `incoming/`, does not exist, or no filter reads "
            "the file's extension",
            reaction="Nothing was written. Fix the argument and retry once - a file in a "
            "subfolder is moved up, or its folder is derived whole",
        ),
        cli_contract.Failure(
            cause="The target folder, a `.md` or a workbook's `.sheets` folder already exists, or the path would be over the "
            "path budget",
            reaction="Nothing was written. Remove what is there or pass `--name <shorter stem>`, "
            "then retry once. An existing `.md` is never overwritten",
        ),
        cli_contract.Failure(
            cause="The document cannot be read (damaged, encrypted), a container breaks one of its "
            "rules (a ZIP inside a ZIP, an unsafe path, a limit) or the filter's package is not "
            "installed",
            reaction="Nothing was written. A missing package is reported with the preflight "
            "hint - show it to the user and stop, never install it. Any other filter keeps "
            "working",
        ),
    ),
    examples=(
        "tools/wikitool raw derive incoming/report.docx",
        "tools/wikitool raw derive incoming/Angebot-XYZ.xlsx --name angebot-xyz",
        "tools/wikitool raw derive incoming/meeting-files",
        "tools/wikitool raw derive incoming/scans.zip",
    ),
    never=(
        "Never edit `derived.md` or its header by hand.",
        "Never derive a file the user did not put into `incoming/` for this purpose.",
    ),
    see_also=(
        "`raw/CONTRACT.md` \"Getting a document in: `raw derive`\" - the rules and why",
        "`wikitool raw accept` - promotes the folder into `raw/`",
        "`instructions/wiki-ingest/SKILL.md` - where a PDF, mail or Office file to ingest starts",
    ),
))
def raw_derive_command(
    path: Path = typer.Argument(..., help="A file or a folder directly in incoming/"),
    name: Annotated[Optional[str], typer.Option(
        "--name", help="Folder stem for a single file instead of the file's own stem"
    )] = None,
):
    """Derive reading text from a document in incoming/ and keep the original.
    See raw/CONTRACT.md "Getting a document in: `raw derive`"."""
    resolved = _resolve(path)
    if resolved.is_dir():
        if name is not None:
            fail("--name applies to a single file; a folder keeps its own name.")
        derived, untouched = _refusals_fail(_plan_derive_folder, resolved)
        _write_exclusive([write for d in derived for write in [(d.target, d.markdown), *d.extras]])
        lines = [f"Derived {len(derived)} file(s) in {rel_path(resolved)}/:"]
        for d in derived:
            more = f" + {len(d.extras)} file(s) in {rel_path(d.extras[0][0].parent)}/" if d.extras else ""
            lines.append(f"  {rel_path(d.source)} -> {rel_path(d.target)} ({d.filter.name}){more}")
        if untouched:
            lines.append(
                "  No filter for, left as they are: " + ", ".join(rel_path(p) for p in untouched)
            )
        success(escape("\n".join(lines)))
        warning = _collision_warning(resolved.name)
        if warning:
            typer.echo("WARN " + warning)
        _echo_notes([d.derivation for d in derived])
        typer.echo(_derive_next(resolved))
        return

    plan = _refusals_fail(_plan_derive_file, resolved, name)
    _write_folder(resolved, plan)
    container = plan.container
    if plan.original is None:
        success(escape(
            f"Unpacked {resolved.name} into {rel_path(plan.folder)}/ ({plan.candidate.name}); "
            f"{resolved.name} is removed."
        ))
    elif container is not None and container.files:
        success(escape(
            f"Derived {rel_path(plan.derived)} from {resolved.name} ({plan.candidate.name}) with "
            f"{len(container.files)} file(s) from its attachments; the original is now "
            f"{rel_path(plan.original)}."
        ))
    else:
        success(escape(
            f"Derived {rel_path(plan.derived)} from {resolved.name} ({plan.candidate.name}); the "
            f"original is now {rel_path(plan.original)}."
        ))
    if plan.extras:
        lines = f"  + {len(plan.extras)} file(s) beside it in {rel_path(plan.folder)}/{plan.extras[0][0].rsplit('/', 1)[0]}/"
        typer.echo(escape(lines))
    if container is not None:
        report = _container_report(container)
        if report:
            typer.echo("\n".join(report))
    warning = _collision_warning(plan.stem)
    if warning:
        typer.echo("WARN " + warning)
    _echo_notes(plan.derivations, container)
    typer.echo(_derive_next(plan.folder))