File size: 5,177 Bytes
b68816f | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 | """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
|