src/knowledge_ingest/selector.py

"""Batch membership and tenant scope — what one click actually covers.

Two questions, both settled 2026-09-07 and both recorded in
KNOWLEDGE_SURFACE_PLAN.md:

1. **Which documents?** The caller's own (`documents.user_id = caller`), filtered
   to the types the pipeline can actually parse. The company-wide alternative is
   deferred, not dropped — hence `ingest_scope()` below, which is the single
   place that answer lives.
2. **Whose knowledge is it?** The company's. Rows carry `scope_id` (the tenant)
   beside `user_id` (the uploader), so switching membership to company-wide
   later needs a different WHERE and no backfill.

⚠️ `scope_id` must never enter the `entity_id` derivation. `ids.py` hashes
`doc_id` + content and nothing else; keeping it that way is what makes a future
re-point an `UPDATE … SET scope_id` — a value migration — rather than a re-key
that would lose every expert approval.
"""

from __future__ import annotations

import re
from typing import Literal

from sqlalchemy import select

from src.db.postgres.connection import AsyncSessionLocal
from src.db.postgres.models import Document, User
from src.middlewares.logging import get_logger

logger = get_logger("knowledge_selector")

# Only these reach the pipeline. Six of the eight live `documents` rows are
# xlsx/csv — those belong to the structured-data path and are skipped with a
# reason rather than silently ignored.
PARSEABLE_TYPES = frozenset({"pdf", "docx"})

IngestScope = Literal["user", "company"]

# The single flip point (N10f). Batch membership and the diff baseline must move
# together: selecting per user while diffing company-wide would classify a
# user's candidates against approvals they cannot see — an asymmetry that is
# invisible when testing with one user.
INGEST_SCOPE: IngestScope = "user"


def ingest_scope() -> IngestScope:
    """Whose documents one click covers. `user` today, `company` later."""
    return INGEST_SCOPE


def normalise_company(company: str | None) -> str | None:
    """Fold a free-text company name into a usable tenant key.

    Interim only: dedorch has no companies table, so the tenant key is derived
    from `users.company`, which is nullable free text with no id. Casefolding
    and collapsing whitespace absorbs the common variations; it does not save
    us from a genuine typo forking a knowledge base, which is why the contract
    asks Harry for a real `companies` table (Q6).
    """
    if not company:
        return None
    folded = re.sub(r"\s+", " ", company).strip().casefold()
    return folded or None


async def resolve_scope_id(user_id: str) -> str:
    """The tenant key for a user: their normalised company, else the user.

    Falls back to `user:<user_id>` rather than raising, so a user with no
    company recorded still gets a private, well-formed scope instead of
    sharing one with every other company-less user.
    """
    async with AsyncSessionLocal() as session:
        result = await session.execute(select(User.company).where(User.id == user_id))
        row = result.first()

    company = normalise_company(row[0] if row else None)
    if company:
        return company

    logger.info(
        "scope_id_fallback",
        extra={"user_id": user_id, "reason": "no company on user row"},
    )
    return f"user:{user_id}"


async def select_documents(
    user_id: str, document_ids: list[str] | None = None
) -> tuple[list[Document], list[tuple[Document, str]]]:
    """Pick the batch. Returns (selected, [(document, skip_reason), …]).

    An explicit `document_ids` list is still filtered by owner and type — a
    caller naming someone else's document, or a spreadsheet, gets it skipped
    with a reason rather than honoured.

    `documents.status` is deliberately NOT filtered on: it tracks the separate
    pgvector processing path, and a row exists only once `blob_name` is set, so
    it says nothing about whether this pipeline can read the file. A blob that
    turns out to be unreadable fails that one document and settles the job
    `partial`.
    """
    async with AsyncSessionLocal() as session:
        stmt = select(Document).where(Document.user_id == user_id)
        if document_ids:
            stmt = stmt.where(Document.id.in_(document_ids))
        result = await session.execute(stmt)
        rows = list(result.scalars().all())

    selected: list[Document] = []
    skipped: list[tuple[Document, str]] = []
    for doc in rows:
        file_type = (doc.file_type or "").strip().lower().lstrip(".")
        if file_type not in PARSEABLE_TYPES:
            skipped.append((doc, f"unsupported_type:{file_type or 'unknown'}"))
            continue
        selected.append(doc)

    # A named id that matched nothing is a caller error worth surfacing: it is
    # either a typo or another user's document, and both look like silence.
    if document_ids:
        found = {d.id for d in rows}
        for missing in [d for d in document_ids if d not in found]:
            logger.info(
                "ingest_document_not_found",
                extra={"user_id": user_id, "document_id": missing},
            )

    return selected, skipped