Spaces:

SkillArbitrage
/

UpworkAutomation

Sleeping

UpworkAutomation / app /services /dossier_reader.py
Fix audited pipeline, session isolation and proposal grounding issues
357ee36 29 days ago
34.6 kB
"""Dossier Reader.
Reads every supported file in the user-supplied local folder and returns
a list of extracted chunks. Original files never leave the machine and
raw extracted text is never logged.
Supported file types and their handlers:
* ``.pdf``  -> ``pdfplumber`` (one chunk per page; headings and paragraph
  breaks rebuilt from the layout).
* ``.docx`` -> ``python-docx`` (paragraphs + tables in document order, single
  chunk per file; headings marked).
* ``.txt`` / ``.md`` -> read as text: UTF-8, the UTF-16/32 a byte-order mark
  names, or Windows-1252 (see ``folder_validator.decode_text``).
* ``.json`` -> parsed via ``json``. The structured payload is preserved.
* ``.csv`` -> read as text plus a small ``json_data`` preview.
* ``.png`` / ``.jpg`` / ``.jpeg`` -> metadata only (no OCR yet).
Failures in a single file do not stop processing. The bad file is marked
``failed`` and reading continues.
"""

from __future__ import annotations

import csv
import hashlib
import io
import json
import re
import statistics
from dataclasses import dataclass, field
from pathlib import Path
from typing import Iterable, Optional

from app.models.schemas import (
    SOURCE_PRIORITY,
    ChunkRecord,
    ExtractionStatus,
    SourceType,
)
from app.services.folder_validator import (
    IMAGE_EXTENSIONS,
    SUPPORTED_EXTENSIONS,
    TEXT_LIKE_EXTENSIONS,
    classify_source,
    decode_text,
    load_json_file,
)
from app.utils.file_utils import (
    MAX_DOSSIER_FILE_BYTES,
    iter_contained_files,
    is_within_size_limit,
)


_PDF_MAX_PAGES = 100
_WARNING_MAX_CHARS = 200


def _chunk_id(file_path: str, suffix: str) -> str:
    digest = hashlib.sha1(file_path.encode("utf-8")).hexdigest()[:10]
    return f"{digest}::{suffix}"


def _sanitize_warning(message: Optional[str]) -> Optional[str]:
    """Collapse a warning to a single safe line.
    Reader warnings only ever carry exception class names and short
    operational notes — never raw dossier text — but we still flatten
    newlines and cap the length so nothing large or multi-line can leak
    into a log line or the debug panel.
    """
    if not message:
        return None
    flat = " ".join(str(message).split())
    if len(flat) > _WARNING_MAX_CHARS:
        flat = flat[: _WARNING_MAX_CHARS - 1].rstrip() + "…"
    return flat or None


def _detect_section(text: str) -> Optional[str]:
    """Best-effort section name from a Markdown heading, if present."""
    for line in text.splitlines():
        stripped = line.strip()
        if stripped.startswith("#"):
            heading = stripped.lstrip("#").strip()
            if heading:
                return heading[:120]
        if stripped:
            # Only the first non-blank line can be the document heading.
            break
    return None


def _load_json(path: Path) -> tuple[Optional[object], Optional[str]]:
    try:
        return load_json_file(path), None
    except (OSError, ValueError) as exc:
        return None, f"{exc.__class__.__name__}: {exc}"


def _read_text(path: Path) -> tuple[str, Optional[str], Optional[str]]:
    """Return ``(text, fatal_error, decode_warning)``.
    ``fatal_error`` is set only when the file could not be opened/read at
    all. ``decode_warning`` is set (but the text is still usable) when some
    bytes fit no text encoding (see :func:`decode_text`) and had to be
    dropped — so a mis-encoded file degrades visibly instead of producing
    quietly-corrupted evidence text.
    """
    try:
        raw = path.read_bytes()
    except OSError as exc:
        return "", f"{exc.__class__.__name__}: {exc}", None
    text, lossy = decode_text(raw)
    warning = "some unreadable bytes were dropped while reading" if lossy else None
    return text, None, warning


def _read_csv_preview(path: Path, max_rows: int = 50) -> tuple[str, list[list[str]], Optional[str]]:
    try:
        raw = path.read_bytes()
    except OSError as exc:
        return "", [], f"{exc.__class__.__name__}: {exc}"
    # newline="" as the csv module asks of a file, so quoted line breaks stay.
    reader = csv.reader(io.StringIO(decode_text(raw)[0], newline=""))
    rows: list[list[str]] = []
    for i, row in enumerate(reader):
        if i >= max_rows:
            break
        rows.append(row)

    text_lines = ["\t".join(cell for cell in row) for row in rows]
    return "\n".join(text_lines), rows, None


# ---------------------------------------------------------------------------
# Document layout: headings and paragraphs
# ---------------------------------------------------------------------------
#
# The evidence scanner works paragraph by paragraph and reads a Markdown-style
# "## Heading" paragraph as the start of a section. PDF and Word files carry
# neither marker on their own — pdfplumber returns a page as single-spaced
# lines and Word paragraphs used to be joined with single line breaks — so a
# whole page or file reached the scanner as one paragraph. The helpers below
# rebuild both: headings come out as "## " (top level) or "### " (sub level)
# paragraphs, and body text is split where the layout shows a paragraph break.

# A line set this many points larger than the body text is a heading, and
# this many more is a top-level heading.
_HEADING_SIZE_STEP = 1.5
_TOP_HEADING_SIZE_STEP = 4.0
_HEADING_MAX_CHARS = 160
# A gap between lines this many times the usual line gap starts a paragraph.
_PARAGRAPH_GAP_FACTOR = 1.8
# Running headers and footers sit in this outer share of the page height.
_MARGIN_BAND = 0.08
# A transcript's speaker tag on a line of its own: "Speaker_A [00:12]:".
_SPEAKER_TAG = re.compile(r"^[A-Za-z][\w .'-]{0,40}?\s*[\[(]?\d{1,2}(?::\d{2}){1,2}[\])]?\s*:?$")
_LIST_ITEM = re.compile(r"^\s*(?:\d{1,2}[.)]|[•▪●◦]|\\?#)\s")
# A short line that opens on a figure ("3x ROAS", "40% more leads", "$12k
# MRR", "10+ clients"), or states a result anywhere ("SALES UP 40%").
_STAT_LINE = re.compile(r"^[$₹£€]?\s?\d[\d.,]*\s*(?:%|[x×](?![a-z0-9])|[km]\b|\+)", re.IGNORECASE)
_RESULT_FIGURE = re.compile(r"\d\s?(?:%|percent\b)|\d[x×](?![a-z0-9])", re.IGNORECASE)
_STAT_MAX_WORDS = 8


def _is_value_line(text: str) -> bool:
    """True for a line that states values rather than naming a section.
    A list of short items split by one separator ("EXCEL, SQL, POWER BI,
    TABLEAU", "HTML | CSS | JAVASCRIPT") or a short line built on a figure
    ("3x ROAS" in a big font) is content, however it is set: read as a
    heading it would never be a claim. A caps heading that only contains a
    comma ("WEEK 4 — PROPOSAL VOLUME, FIVERR ACTIVATION, AND") is not a
    list, because its items are not all short, and neither is a series in a
    sentence ("I Help Restaurants, Cafés, and Cloud Kitchens").
    """
    stripped = " ".join(text.split())
    for separator in ",|/•·;":
        items = stripped.split(separator)
        if (
            len(items) >= 3
            and all(0 < len(item.split()) <= 3 for item in items)
            and not any(item.split()[0].lower() in {"and", "or", "&"} for item in items[1:])
        ):
            return True
    return len(stripped.split()) <= _STAT_MAX_WORDS and bool(
        _STAT_LINE.match(stripped) or _RESULT_FIGURE.search(stripped)
    )


def _is_caps_heading(text: str) -> bool:
    """True for a short ALL-CAPS line such as "HOW THIS HELPS THE CLIENT".
    Five letters is the floor, so a lone acronym ("HSBC") in body text is not
    mistaken for a heading. Callers still rule out value lines (see
    :func:`_is_value_line`), such as a caps list of skills.
    """
    stripped = text.strip()
    letters = [c for c in stripped if c.isalpha()]
    return (
        len(letters) >= 5
        and len(stripped) <= 80
        and all(c.isupper() for c in letters)
        and not stripped.endswith((".", ","))
    )


def _heading_marker(level: int, text: str) -> str:
    return ("## " if level == 1 else "### ") + " ".join(text.split())[:_HEADING_MAX_CHARS]


def _pdf_line_size(line: dict) -> float:
    sizes = [float(c.get("size") or 0.0) for c in line.get("chars") or []]
    return max(sizes) if sizes else 0.0


def _pdf_line_bold(line: dict) -> bool:
    chars = [c for c in line.get("chars") or [] if str(c.get("text") or "").strip()]
    return bool(chars) and all("bold" in str(c.get("fontname") or "").lower() for c in chars)


def _pdf_body_size(pages_lines: list[list[dict]]) -> float:
    """The most common text size in the document, weighted by characters."""
    weights: dict[float, int] = {}
    for lines in pages_lines:
        for line in lines:
            size = round(_pdf_line_size(line), 1)
            weights[size] = weights.get(size, 0) + len(str(line.get("text") or ""))
    return max(weights, key=weights.get) if weights else 0.0


def _pdf_page_text(
    lines: list[dict], body_size: float, open_headings: list[str]
) -> str:
    """Rebuild one page's text with heading markers and paragraph breaks.
    ``open_headings`` holds the top-level and sub-level heading in force when
    the page starts; it is repeated at the top of the page (so a section that
    runs across a page break keeps its name) and updated in place.
    """
    gaps = [
        float(cur["top"]) - float(prev["bottom"])
        for prev, cur in zip(lines, lines[1:])
        if float(cur["top"]) >= float(prev["bottom"])
        and abs(_pdf_line_size(cur) - body_size) < _HEADING_SIZE_STEP
    ]
    usual_gap = statistics.median(gaps) if gaps else 0.0
    paragraph_gap = max(usual_gap * _PARAGRAPH_GAP_FACTOR, usual_gap + body_size * 0.5, 3.0)

    blocks: list[str] = [h for h in open_headings if h]
    paragraph: list[str] = []
    heading: Optional[tuple[int, list[str]]] = None
    prev: Optional[dict] = None

    def close_paragraph() -> None:
        if paragraph:
            blocks.append("\n".join(paragraph))
            paragraph.clear()

    def close_heading() -> None:
        nonlocal heading
        if heading:
            level, parts = heading
            marker = _heading_marker(level, " ".join(parts))
            blocks.append(marker)
            if level == 1:
                open_headings[:] = [marker, ""]
            else:
                open_headings[1] = marker
            heading = None

    for line in lines:
        text = str(line.get("text") or "").strip()
        if not text:
            continue
        size = _pdf_line_size(line)
        level = 0
        if size >= body_size + _TOP_HEADING_SIZE_STEP:
            level = 1
        elif size >= body_size + _HEADING_SIZE_STEP or _is_caps_heading(text):
            level = 2
        gap = float(line["top"]) - float(prev["bottom"]) if prev else 0.0
        # A heading that wraps onto a second line continues the same heading
        # rather than starting another one.
        continues = heading is not None and heading[0] == level and gap < paragraph_gap
        # A list or a figure ("3x ROAS") is content at any size, unless it is
        # the wrapped tail of a heading already open, set in the heading's
        # own type. A caps list right under "SKILLS" is not a wrap.
        wraps = (
            continues and prev is not None
            and abs(_pdf_line_size(prev) - size) < 0.5
            and _pdf_line_bold(prev) == _pdf_line_bold(line)
        )
        if level and not wraps and _is_value_line(text):
            level = 0
        if level:
            close_paragraph()
            if heading and continues:
                heading[1].append(text)
            else:
                close_heading()
                heading = (level, [text])
        else:
            close_heading()
            # A bold lead-in line ("14 Years of Sales-Driven…") titles the
            # paragraph under it, so the gap just after it is not a break.
            after_lead = (
                len(paragraph) == 1 and prev is not None
                and _pdf_line_bold(prev) and len(paragraph[0]) < 100
            )
            if paragraph and (
                (gap > paragraph_gap and not after_lead) or _LIST_ITEM.match(text)
            ):
                close_paragraph()
            paragraph.append(text)
        prev = line
    close_heading()
    close_paragraph()
    return "\n\n".join(blocks)


def _drop_running_lines(
    pages_lines: list[Optional[list[dict]]],
    page_heights: list[float],
) -> list[Optional[list[dict]]]:
    """Remove page numbers and the header/footer lines repeated on most pages.
    A footer such as "SKILL ARBITRAGE — UPWORK OPTIMISATION | 7" is set in
    capitals, so left in place it would read as a heading at the foot of
    every page and change the section the next page belongs to.
    Only lines in a page's top or bottom margin band are removed. A template
    repeats body lines on every page too ("Results", "Tools: Shopify,
    Klaviyo", "Result: 40 percent increase…" with only the figure changing),
    and those are the evidence. The one exception is a call transcript's
    speaker tag ("Speaker_A [00:12]:"), which says nothing itself and is set
    on every wrapped line, so it goes wherever it sits.
    """
    def key(line: dict) -> str:
        return re.sub(r"\d+", "#", " ".join(str(line.get("text") or "").lower().split()))

    def in_margin(line: dict, height: float) -> bool:
        return height > 0 and (
            float(line["bottom"]) <= height * _MARGIN_BAND
            or float(line["top"]) >= height * (1 - _MARGIN_BAND)
        )

    def is_tag(line: dict) -> bool:
        return bool(_SPEAKER_TAG.match(str(line.get("text") or "").strip()))

    counts: dict[str, int] = {}
    for lines, height in zip(pages_lines, page_heights):
        for k in {key(line) for line in lines or [] if in_margin(line, height) or is_tag(line)}:
            counts[k] = counts.get(k, 0) + 1
    readable = [lines for lines in pages_lines if lines]
    threshold = max(3, int(len(readable) * 0.3))
    running = {k for k, n in counts.items() if n >= threshold}

    def keep(line: dict, height: float) -> bool:
        if in_margin(line, height):
            return key(line) not in running and bool(key(line).strip("# "))
        return not (is_tag(line) and key(line) in running)

    cleaned: list[Optional[list[dict]]] = []
    for lines, height in zip(pages_lines, page_heights):
        if lines is None:
            cleaned.append(None)
            continue
        cleaned.append([line for line in lines if keep(line, height)])
    return cleaned


def _pdf_chunks(
    path: Path,
    base_meta: dict,
) -> tuple[list[ChunkRecord], Optional[str]]:
    try:
        import pdfplumber  # type: ignore
    except Exception as exc:  # pragma: no cover - dependency must be present
        return [], f"pdfplumber unavailable: {exc}"

    records: list[ChunkRecord] = []
    try:
        with pdfplumber.open(str(path)) as pdf:
            page_iter = pdf.pages[:_PDF_MAX_PAGES]
            # Read every page's lines first: the body text size, which tells
            # a heading from body text, is a property of the whole document.
            pages_lines: list[Optional[list[dict]]] = []
            page_heights: list[float] = []
            failures: dict[int, str] = {}
            for index, page in enumerate(page_iter, start=1):
                page_heights.append(float(page.height or 0.0))
                try:
                    pages_lines.append(page.extract_text_lines(return_chars=True) or [])
                except Exception as exc:  # noqa: BLE001 - per-page tolerance
                    pages_lines.append(None)
                    failures[index] = f"page extract failed: {exc.__class__.__name__}"
            pages_lines = _drop_running_lines(pages_lines, page_heights)
            body_size = _pdf_body_size([lines for lines in pages_lines if lines])
            open_headings = ["", ""]
            for index, lines in enumerate(pages_lines, start=1):
                if lines is None:
                    records.append(
                        ChunkRecord(
                            chunk_id=_chunk_id(base_meta["file_path"], f"p{index}"),
                            page_number=index,
                            extracted_text="",
                            extraction_status="failed",
                            extraction_warning=failures.get(index),
                            **base_meta,
                        )
                    )
                    continue
                section = (open_headings[1] or open_headings[0]).lstrip("# ") or None
                text = _pdf_page_text(lines, body_size, open_headings).strip()
                records.append(
                    ChunkRecord(
                        chunk_id=_chunk_id(base_meta["file_path"], f"p{index}"),
                        page_number=index,
                        section_name=section,
                        extracted_text=text,
                        extraction_status="ok" if text else "empty",
                        **base_meta,
                    )
                )
    except Exception as exc:  # noqa: BLE001 - one bad PDF must not crash run
        return records, f"pdfplumber failed: {exc.__class__.__name__}"
    return records, None


# A team document names itself on its cover page even when the file name
# does not ("Final.pdf"). A PDF or Word file its name could not place is
# placed by these title phrases instead; checked in order, first match wins.
_TITLE_RULES: tuple[tuple[SourceType, tuple[str, ...]], ...] = (
    ("upwork_profile", (
        "upwork profile optimisation", "upwork profile optimization",
        "upwork optimisation", "upwork optimization",
    )),
    ("offer_blueprint", ("offer bank", "offer blueprint")),
    ("linkedin_optimization", ("linkedin optimisation", "linkedin optimization")),
    ("skillarbitrage_dossier_roadmap", ("dossier", "roadmap")),
    ("discovery_call_transcript", ("transcript",)),
)
_UNPLACED_SOURCES = frozenset({"unknown_supported_file", "generic_profile_document"})
_TITLE_WINDOW_CHARS = 600
# The title is the few short lines that open the first page.
_TITLE_MAX_LINES = 6
_TITLE_LINE_MAX_WORDS = 10


def _letter_spaced(line: str) -> bool:
    """True for a title set one letter at a time ("S K I L L  A R B I T R A G E")."""
    tokens = line.split()
    return len(tokens) >= 4 and 2 * sum(len(t) == 1 for t in tokens) > len(tokens)


def _title_lines(text: str) -> list[str]:
    """The lines of the cover title: the short lines that open the first page.
    The first sentence or long line of body text ends the title, so words in
    the opening prose ("I offer bank reconciliation…", "owned the product
    roadmap") never name the document. A marked heading may run long.
    """
    lines: list[str] = []
    for raw in text[:_TITLE_WINDOW_CHARS].splitlines():
        heading = raw.lstrip().startswith("#")
        line = " ".join(raw.strip().lstrip("#").split())
        if not line:
            continue
        if line.endswith((".", "!", "?")) or (
            not heading
            and not _letter_spaced(line)
            and len(line.split()) > _TITLE_LINE_MAX_WORDS
        ):
            break
        lines.append(line)
        if len(lines) >= _TITLE_MAX_LINES:
            break
    return lines


def _names_title(line: str, phrase: str) -> bool:
    """True when ``phrase`` appears in a title line the way a title sets it.
    Whole words only ("transcript" is not "transcription"), starting with a
    capital ("Offer Blueprint", "OFFER BANK"). A letter-spaced title is
    compared with its spaces taken out.
    """
    target, pattern = line, r"\b" + re.escape(phrase) + r"\b"
    if _letter_spaced(line):
        target, pattern = line.replace(" ", ""), re.escape(phrase.replace(" ", ""))
    return any(m.group(0)[0].isupper() for m in re.finditer(pattern, target, re.IGNORECASE))


def _place_by_title(chunks: list[ChunkRecord]) -> list[ChunkRecord]:
    """Re-type an unplaced file's chunks from the title on its first page."""
    if not chunks or chunks[0].source_type not in _UNPLACED_SOURCES:
        return chunks
    lines = _title_lines(chunks[0].extracted_text)
    for source_type, phrases in _TITLE_RULES:
        if any(_names_title(line, p) for line in lines for p in phrases):
            for chunk in chunks:
                chunk.source_type = source_type
                chunk.source_priority = SOURCE_PRIORITY[source_type]
            break
    return chunks


def _docx_blocks(parent, element) -> Iterable:
    """The paragraphs and tables in a Word body or table cell, in order.
    Word's templates put their content in content controls (``w:sdt``),
    which python-docx does not list among the body's or a cell's own
    paragraphs and tables; those inside one are read as if they stood in
    its place.
    """
    from docx.oxml.ns import qn  # type: ignore
    from docx.table import Table  # type: ignore
    from docx.text.paragraph import Paragraph  # type: ignore

    for child in element:
        if child.tag == qn("w:p"):
            yield Paragraph(child, parent)
        elif child.tag == qn("w:tbl"):
            yield Table(child, parent)
        elif child.tag == qn("w:sdt"):
            content = child.find(qn("w:sdtContent"))
            if content is not None:
                yield from _docx_blocks(parent, content)


def _docx_runs(paragraph) -> list:
    """Every run whose text the paragraph shows, in order.
    ``paragraph.runs`` leaves out the runs inside a hyperlink, although
    ``paragraph.text`` includes their text, and neither reaches the runs
    inside an inline content control (a template's "[Your Name]" field).
    """
    from docx.oxml.ns import qn  # type: ignore
    from docx.text.run import Run  # type: ignore

    runs: list = []

    def walk(element) -> None:
        for child in element:
            if child.tag == qn("w:r"):
                runs.append(Run(child, paragraph))
            elif child.tag == qn("w:hyperlink"):
                walk(child)
            elif child.tag == qn("w:sdt"):
                content = child.find(qn("w:sdtContent"))
                if content is not None:
                    walk(content)

    walk(paragraph._p)
    return runs


def _docx_text(paragraph) -> str:
    return "".join(run.text for run in _docx_runs(paragraph))


def _docx_cell_text(cell) -> str:
    """A table cell's text, as ``cell.text`` reads it plus content controls."""
    return "\n".join(
        _docx_text(item) for item in _docx_blocks(cell, cell._tc) if not hasattr(item, "rows")
    )


def _docx_heading_level(paragraph) -> int:
    """0 for body text, 1 for a top-level heading, 2 for a sub-heading."""
    all_runs = _docx_runs(paragraph)
    text = "".join(r.text for r in all_runs).strip()
    style = str(getattr(getattr(paragraph, "style", None), "name", "") or "").lower()
    if style in {"title", "heading 1"}:
        return 1
    if style.startswith("heading"):
        return 2
    # A bold label before a link ("Case study: <link>") is not a bold line.
    runs = [r for r in all_runs if r.text.strip()]
    all_bold = bool(runs) and all(r.bold for r in runs)
    if (
        _is_caps_heading(text) or (all_bold and len(text) <= 80 and not text.endswith("."))
    ) and not _is_value_line(text):
        return 2
    return 0


def _docx_chunk(
    path: Path,
    base_meta: dict,
) -> tuple[list[ChunkRecord], Optional[str]]:
    try:
        from docx import Document  # type: ignore
    except Exception as exc:  # pragma: no cover
        return [], f"python-docx unavailable: {exc}"

    try:
        document = Document(str(path))
    except Exception as exc:  # noqa: BLE001
        return [], f"docx open failed: {exc.__class__.__name__}"

    # Paragraphs and tables are read in document order, so a table (such as a
    # key/value snapshot) stays inside the section its heading opens. Each
    # Word paragraph is its own paragraph here, and headings are marked.
    blocks: list[str] = []
    for item in _docx_blocks(document, document.element.body):
        if hasattr(item, "rows"):
            for row in item.rows:
                cells: list[str] = []
                for cell in row.cells:
                    value = _docx_cell_text(cell).strip()
                    if value and value not in cells:
                        cells.append(value)
                if cells:
                    blocks.append(" | ".join(cells))
            continue
        text = _docx_text(item).strip()
        if not text:
            continue
        level = _docx_heading_level(item)
        blocks.append(_heading_marker(level, text) if level else text)

    full_text = "\n\n".join(blocks).strip()
    status: ExtractionStatus = "ok" if full_text else "empty"
    return (
        [
            ChunkRecord(
                chunk_id=_chunk_id(base_meta["file_path"], "docx"),
                extracted_text=full_text,
                extraction_status=status,
                **base_meta,
            )
        ],
        None,
    )


def _failed_chunk(path: Path, base_meta: dict, message: str) -> ChunkRecord:
    return ChunkRecord(
        chunk_id=_chunk_id(base_meta["file_path"], "failed"),
        extracted_text="",
        extraction_status="failed",
        extraction_warning=_sanitize_warning(message) or "extraction failed",
        **base_meta,
    )


def _build_base_meta(
    path: Path,
    rel_path: Path,
    source_type: SourceType,
) -> dict:
    return {
        "file_name": path.name,
        "file_path": str(rel_path),
        "file_type": path.suffix.lower(),
        "source_type": source_type,
        "source_priority": SOURCE_PRIORITY[source_type],
    }


def _fallback_base_meta(path: Path, rel_path: Path) -> dict:
    """Base metadata for a file whose classification itself failed."""
    return _build_base_meta(path, rel_path, "unknown_supported_file")


def read_dossier(folder_path: str | Path) -> list[ChunkRecord]:
    """Walk the dossier folder and return extracted chunks.
    One file may produce multiple chunks (e.g., one per PDF page). Files
    that fail to open produce a single ``failed`` chunk with the warning
    captured in ``extraction_warning``. Unsupported files are skipped.
    """
    folder = Path(folder_path).expanduser()
    if not folder.exists() or not folder.is_dir():
        return []

    chunks: list[ChunkRecord] = []
    try:
        # Symlink-safe, containment-checked, count-bounded walk. A symlink
        # whose target lives outside the folder is never read, so files the
        # user did not put in the dossier can never reach the LLM.
        candidates = list(iter_contained_files(folder))
    except OSError:
        return []

    for path in candidates:
        ext = path.suffix.lower()
        if ext not in SUPPORTED_EXTENSIONS:
            continue

        rel_path = path.relative_to(folder)

        # Skip files larger than the per-file cap rather than reading them
        # fully into memory; record a visible failed chunk so the UI can
        # show the file was intentionally skipped.
        if not is_within_size_limit(path):
            base_meta = _fallback_base_meta(path, rel_path)
            mb = MAX_DOSSIER_FILE_BYTES // (1024 * 1024)
            chunks.append(
                _failed_chunk(
                    path,
                    base_meta,
                    f"file skipped: exceeds {mb}MB size limit",
                )
            )
            continue

        try:
            chunks.extend(_read_one_file(path, rel_path, ext))
        except Exception as exc:  # noqa: BLE001 - one bad file must not abort the run
            base_meta = _fallback_base_meta(path, rel_path)
            chunks.append(
                _failed_chunk(
                    path,
                    base_meta,
                    f"unexpected reader error: {exc.__class__.__name__}",
                )
            )

    return chunks


def _read_one_file(path: Path, rel_path: Path, ext: str) -> list[ChunkRecord]:
    """Extract one supported file into one or more chunks.
    Per-file failures are represented as ``failed`` chunks rather than
    raised exceptions; the caller wraps this in a final safety net so any
    truly unexpected error still degrades to a single failed chunk.
    """
    # Classification: read a small sample for JSON / text-like files so
    # the classifier can use content as a signal.
    text_sample = ""
    json_payload: object = None
    if ext in TEXT_LIKE_EXTENSIONS:
        try:
            with open(path, "rb") as fh:
                text_sample = decode_text(fh.read(8192))[0]
        except OSError:
            text_sample = ""
        if ext == ".json":
            json_payload, _ = _load_json(path)

    source_type, _note = classify_source(
        path, text_sample=text_sample, json_payload=json_payload
    )
    base_meta = _build_base_meta(path, rel_path, source_type)

    if ext == ".pdf":
        pdf_chunks, warning = _pdf_chunks(path, base_meta)
        if pdf_chunks:
            if warning:
                pdf_chunks[-1].extraction_warning = _sanitize_warning(warning)
            return _place_by_title(pdf_chunks)
        return [_failed_chunk(path, base_meta, warning or "no pages")]

    if ext == ".docx":
        docx_chunks, warning = _docx_chunk(path, base_meta)
        if docx_chunks:
            return _place_by_title(docx_chunks)
        return [_failed_chunk(path, base_meta, warning or "no content")]

    if ext in {".txt", ".md"}:
        text, err, decode_warning = _read_text(path)
        status: ExtractionStatus = "ok" if text.strip() else "empty"
        section = _detect_section(text) if ext == ".md" else None
        return [
            ChunkRecord(
                chunk_id=_chunk_id(base_meta["file_path"], "text"),
                section_name=section,
                extracted_text=text.strip(),
                extraction_status="failed" if err else status,
                extraction_warning=_sanitize_warning(err or decode_warning),
                **base_meta,
            )
        ]

    if ext == ".json":
        payload, err = _load_json(path)
        if err:
            return [_failed_chunk(path, base_meta, err)]
        try:
            pretty = json.dumps(payload, indent=2, ensure_ascii=False)
        except Exception:  # noqa: BLE001
            pretty = ""
        return [
            ChunkRecord(
                chunk_id=_chunk_id(base_meta["file_path"], "json"),
                extracted_text=pretty,
                json_data=payload,
                extraction_status="ok" if payload is not None else "empty",
                **base_meta,
            )
        ]

    if ext == ".csv":
        text, rows, err = _read_csv_preview(path)
        return [
            ChunkRecord(
                chunk_id=_chunk_id(base_meta["file_path"], "csv"),
                extracted_text=text,
                json_data={"rows": rows} if rows else None,
                extraction_status="failed" if err else ("ok" if rows else "empty"),
                extraction_warning=_sanitize_warning(err),
                **base_meta,
            )
        ]

    if ext in IMAGE_EXTENSIONS:
        return [
            ChunkRecord(
                chunk_id=_chunk_id(base_meta["file_path"], "image"),
                extracted_text="",
                extraction_status="metadata_only",
                extraction_warning="image stored as metadata only; OCR not run",
                **base_meta,
            )
        ]

    return []


def chunks_by_file(chunks: Iterable[ChunkRecord]) -> dict[str, list[ChunkRecord]]:
    by_file: dict[str, list[ChunkRecord]] = {}
    for chunk in chunks:
        by_file.setdefault(chunk.file_path, []).append(chunk)
    return by_file


# ---------------------------------------------------------------------------
# Read summary (UI- and log-safe view over the extracted chunks)
# ---------------------------------------------------------------------------


@dataclass
class FileReadSummary:
    """Per-file rollup used by the UI summary and debug panel.
    Carries metadata and a sanitized warning only — never raw text.
    """

    file_name: str
    file_path: str
    file_type: str
    source_type: str
    chunk_count: int
    status: str
    warning: Optional[str] = None


@dataclass
class DossierReadSummary:
    """Aggregate, text-free summary of a dossier read.
    Safe to log and to render: it counts files and chunks and lists
    per-file status, but contains none of the extracted dossier text.
    """

    files_processed: int = 0
    chunks_extracted: int = 0
    failed_files: int = 0
    files: list[FileReadSummary] = field(default_factory=list)


def _aggregate_status(statuses: list[str]) -> str:
    """Reduce a file's per-chunk statuses to one headline status."""
    if not statuses:
        return "empty"
    if all(s == "failed" for s in statuses):
        return "failed"
    for preferred in ("ok", "partial", "metadata_only", "empty"):
        if preferred in statuses:
            return preferred
    return statuses[0]


def summarize_chunks(chunks: Iterable[ChunkRecord]) -> DossierReadSummary:
    """Build a text-free summary of an extracted dossier.
    ``files_processed`` counts the distinct files that produced at least
    one chunk, ``chunks_extracted`` is the total chunk count, and
    ``failed_files`` counts files whose every chunk failed to extract.
    """
    by_file = chunks_by_file(chunks)
    file_summaries: list[FileReadSummary] = []
    failed_files = 0
    total_chunks = 0

    for file_path, file_chunks in by_file.items():
        total_chunks += len(file_chunks)
        statuses = [c.extraction_status for c in file_chunks]
        status = _aggregate_status(statuses)
        if status == "failed":
            failed_files += 1
        warning = next(
            (c.extraction_warning for c in file_chunks if c.extraction_warning),
            None,
        )
        first = file_chunks[0]
        file_summaries.append(
            FileReadSummary(
                file_name=first.file_name,
                file_path=file_path,
                file_type=first.file_type,
                source_type=first.source_type,
                chunk_count=len(file_chunks),
                status=status,
                warning=_sanitize_warning(warning),
            )
        )

    return DossierReadSummary(
        files_processed=len(by_file),
        chunks_extracted=total_chunks,
        failed_files=failed_files,
        files=file_summaries,
    )