"""Bulk import — the volume intake a brain needs to get rich.

Every other ingestion path in Alexandria is one item at a time, in a
conversation. That is a *curator*: careful, reviewable, and the right design for
filing today's meeting. It is the wrong design for the first day of a brain's
life, when the owner already has a folder of documents, a mail archive, or an
export from whatever they used before. A curator cannot close that gap — a new
brain stays empty one approved paste at a time.

This module is the intake pipe. The design points that matter:

- **Deterministic routing first.** ``route()`` is free (stdlib, no model spend)
  and returns a computed ``high|med|low`` confidence band. Only items that clear
  the band are handed to a model; low-confidence items stage in ``Agent/inbox/``
  as pending queue items — with their route evidence and a re-checkable plan
  attached — rather than being force-filed somewhere wrong.
- **The index is hoisted, and it learns.** :class:`~alexandria_vault.ingest.IngestIndex`
  is built once for the whole run instead of per item (which would re-walk the
  vault N times), and a note filed at item 3 is visible to item 40. Without that
  overlay a single import would manufacture exactly the duplicate people this
  whole change exists to prevent.
- **Batched commits.** A 500-document import is not one commit, and not 500.
- **``plan_only`` costs nothing.** It reports what would happen — counts, bands,
  projected model passes — before a single token is spent.
"""
import os

from .. import extract as _extract
from .. import ingest as _ingest

# Commit every N filed items: small enough that a failure loses little and the
# history stays readable, large enough that 500 items don't make 500 commits.
COMMIT_EVERY = 25

# Bands that earn a model pass. 'low' means route could not tell where this goes;
# spending a frontier call to guess is how a vault fills with confident mistakes.
AUTOFILE_BANDS = ("high", "med")



class ImportError_(RuntimeError):
    """A failure the caller should surface verbatim."""


def walk_dir(root, *, suffixes=None):
    """(name, bytes) for every readable file under a host directory."""
    root = os.path.abspath(os.path.expanduser(root))
    if not os.path.isdir(root):
        raise ImportError_(f"not a directory: {root}")
    for dirpath, dirnames, filenames in os.walk(root):
        dirnames[:] = sorted(d for d in dirnames if not d.startswith("."))
        for fname in sorted(filenames):
            if fname.startswith("."):
                continue
            suffix = os.path.splitext(fname)[1].lower()
            if suffixes and suffix not in suffixes:
                continue
            if suffix not in _extract._HANDLERS:
                continue
            full = os.path.join(dirpath, fname)
            try:
                with open(full, "rb") as fh:
                    yield os.path.relpath(full, root), fh.read()
            except OSError:
                continue          # unreadable file is skipped, never fatal


def walk_filestore(store, prefix="", *, limit=2000):
    """(name, bytes) from the brain's file space — the hosted-brain source.

    Orion uploads with ``file_put`` and then imports, so a tenant with no shell
    can still bulk-load, and nothing has to travel as one enormous tool payload.
    """
    # Normalize the prefix: `FileStore.list` appends the separator itself, so a
    # caller-supplied trailing slash reaches `safe_key` as an empty final
    # component and is refused outright ("must not contain empty, '.' or '..'").
    # Both forms are the obvious thing to type — and `import_start`'s own schema
    # used to show the failing one — so accept either rather than teaching the
    # difference. Found on a live hosted brain, not in tests: the unit path
    # stubs the store, and a fake cannot reproduce its key validation.
    prefix = str(prefix or "").strip().strip("/")
    listing = store.list(prefix, limit=limit)
    for row in listing.get("files", listing.get("items", []) or []):
        rel = row.get("path") or row.get("key") or ""
        if not rel or os.path.splitext(rel)[1].lower() not in _extract._HANDLERS:
            continue
        try:
            yield rel, store.get(rel)
        except Exception:
            continue


def _stage(name, text, routed):
    """Build the queue item for an import the router could not place.

    Delegates to `ingest.stage_item` so a hand capture and an import produce the
    same shape — the drain globs one thing, not two. The route evidence rides in
    the frontmatter and the plan goes in `## Manifest`, which is exactly what the
    drain re-validates.
    """
    cands = routed.get("candidates") or []
    plan = ("Router could not place this confidently "
            f"(band {routed.get('band')}, confidence {routed.get('confidence')}).")
    if cands:
        plan += " Nearest candidates: " + ", ".join(f"[[{c['rel']}]]" for c in cands[:3]) + "."
    new = (routed.get("entities") or {}).get("unresolved") or []
    if new:
        plan += " Possibly-new entities: " + ", ".join(new[:5]) + "."
    plan += " Re-check all of this against the vault before applying it."
    return _ingest.stage_item(
        content=text, label=os.path.splitext(os.path.basename(name))[0],
        manifest=plan, source=name, author="vault-import", route=routed)


def run(svc, items, *, seat_id="import", instructions=None, plan_only=False,
        progress=None, cancelled=None, limit=None, commit_every=COMMIT_EVERY):
    """Drive a bulk import. ``items`` is an iterable of ``(name, bytes)``.

    Returns a report dict; raises only on a failure that stops the whole run.
    A per-item failure is recorded and the run continues — one unreadable PDF
    must never abandon 400 good documents.
    """
    from ..mcp.tools import _commit, _writable_rel

    index = _ingest.IngestIndex(svc.vault)
    report = {"scanned": 0, "filed": 0, "staged": 0, "skipped": 0,
              "notes_written": [], "staged_paths": [], "failures": [],
              "bands": {"high": 0, "med": 0, "low": 0}, "commits": [],
              "plan_only": bool(plan_only)}
    pending = 0

    def _flush(force=False):
        nonlocal pending
        if pending and (force or pending >= commit_every):
            sha = _commit(svc, _Seat(seat_id), "import",
                          f"{report['filed']} filed, {report['staged']} staged")
            if sha:
                report["commits"].append(sha)
            pending = 0

    for name, data in items:
        if cancelled is not None and cancelled():
            report["cancelled"] = True
            break
        if limit and report["scanned"] >= limit:
            report["truncated_at_limit"] = limit
            break
        report["scanned"] += 1
        if progress:
            progress({"scanned": report["scanned"], "filed": report["filed"],
                      "staged": report["staged"], "current": name})
        try:
            text, meta = _extract.extract_text(data, name)
        except _extract.ExtractError as e:
            report["skipped"] += 1
            report["failures"].append({"name": name, "error": str(e)})
            continue

        routed = index.route(text[:20000])       # routing reads the head, not the tail
        band = routed.get("band", "low")
        report["bands"][band] = report["bands"].get(band, 0) + 1

        if plan_only:
            continue

        if band in AUTOFILE_BANDS:
            try:
                planned = _ingest.plan(svc, text, instructions, index=index,
                                       tier="gather")
            except _ingest.IngestError as e:
                report["failures"].append({"name": name, "error": str(e)})
                report["skipped"] += 1
                continue
            with svc.writer.lock:
                for it in planned["items"]:
                    try:
                        rel = _writable_rel(svc, it.get("path") or "")
                    except Exception as e:
                        report["failures"].append({"name": name, "error": str(e)})
                        continue
                    body = it.get("content") or ""
                    exists = (svc.vault / rel).is_file()
                    if exists:                    # never clobber; append like apply_ingest
                        old = (svc.vault / rel).read_text(encoding="utf-8")
                        body = old + ("" if old.endswith("\n") or not old else "\n") + body
                    svc.writer.atomic_write(rel, body if body.endswith("\n") else body + "\n")
                    report["notes_written"].append(rel)
                    if not exists:
                        # The overlay is what stops item 40 re-creating item 3's note.
                        index.note_created(rel, body=it.get("content") or "")
                    pending += 1
            report["filed"] += 1
        else:
            rel, body = _stage(name, text, routed)
            with svc.writer.lock:
                rel = _writable_rel(svc, rel)
                svc.writer.atomic_write(rel, body)
            index.note_created(rel, type_="inbox-item")
            report["staged_paths"].append(rel)
            report["staged"] += 1
            pending += 1
        _flush()

    if not plan_only:
        _flush(force=True)
    report["tldr"] = _tldr(report)
    return report


def _tldr(r):
    if r["plan_only"]:
        b = r["bands"]
        return (f"plan: {r['scanned']} readable item(s) — {b['high'] + b['med']} would "
                f"file, {b['low']} would stage to {_ingest.INBOX_REL}/, "
                f"{r['skipped']} unreadable")
    parts = [f"{r['filed']} filed",
             f"{r['staged']} staged to {_ingest.INBOX_REL}/"]
    if r["skipped"]:
        parts.append(f"{r['skipped']} skipped")
    if r.get("cancelled"):
        parts.append("cancelled mid-run")
    return ", ".join(parts) + f" ({len(r['notes_written'])} notes, "\
                              f"{len(r['commits'])} commit(s))"


class _Seat:
    """Commit attribution for a run that isn't driven by a live seat (the CLI)."""

    def __init__(self, sid):
        self.id = sid
