symfonic.agent.attachments

attachments

symfonic.agent.attachments — text extraction helpers for Attachment objects.

Public surface (spec §1):

from symfonic.agent.attachments import extract_text, extract_texts, is_extractable

Error classes are re-exported so callers can except UnsupportedAttachmentError without importing from the internal errors submodule.

AttachmentExtractionError

Bases: Exception

Base for all attachment extractor errors.

ExtractionDependencyMissingError

Bases: AttachmentExtractionError

An extractor is registered but its optional dependency is not installed.

The exception message always names the pip install extra the caller should add, e.g. pip install symfonic-core[attachments-pdf].

ExtractionFailedError

Bases: AttachmentExtractionError

The dependency is present but extraction itself failed.

Common causes: corrupt or password-protected file, zero-confidence OCR, unexpected format inside the container.

__cause__ is always set to the underlying library exception so callers can inspect or log it.

UnsupportedAttachmentError

Bases: AttachmentExtractionError

No extractor is registered for this (kind, media_type) combination.

Also raised when source_type="url" is passed — the helper never fetches remote URLs (SSRF avoidance; spec §4).

extract_text

extract_text(
    att: Any,
    *,
    max_pages: int | None = None,
    page_markers: bool = True,
) -> str

Extract plain text from an attachment.

Parameters:

Name	Type	Description	Default
att	Any	Attachment instance OR a raw LangChain / provider content-block dict (Anthropic or OpenAI shape — see spec §4).	required
max_pages	int | None	Optional page limit for PDFs and multi-page documents. None (default) means all pages. Must be a positive integer; 0 and negative values raise ValueError.	None
page_markers	bool	When True (default), multi-page documents insert --- page N --- markers between pages (not before page 1). Set False for raw concatenation.	True

Returns:

Type	Description
str	UTF-8 text. Empty string only if the source is genuinely empty.

Raises:

Type	Description
UnsupportedAttachmentError	No extractor matches the kind/media_type, or the attachment is a URL (which this helper never fetches), or a data: URL without the ;base64 marker.
ExtractionDependencyMissingError	An extractor exists but its optional pip extra isn't installed. Message names the extra to install.
ExtractionFailedError	Dependency present but extraction failed (corrupt file, malformed base64, password-protected PDF, scan-only PDF, OCR confidence below threshold). __cause__ is set to the underlying exception.
ValueError	max_pages is not None or a positive integer.

Source code in src/symfonic/agent/attachments/extractors.py

def extract_text(
    att: Any,
    *,
    max_pages: int | None = None,
    page_markers: bool = True,
) -> str:
    """Extract plain text from an attachment.

    Args:
        att: ``Attachment`` instance OR a raw LangChain / provider content-block
            dict (Anthropic or OpenAI shape — see spec §4).
        max_pages: Optional page limit for PDFs and multi-page documents.
            ``None`` (default) means all pages. Must be a positive integer;
            ``0`` and negative values raise ``ValueError``.
        page_markers: When ``True`` (default), multi-page documents insert
            ``--- page N ---`` markers *between* pages (not before page 1).
            Set ``False`` for raw concatenation.

    Returns:
        UTF-8 text.  Empty string only if the source is genuinely empty.

    Raises:
        UnsupportedAttachmentError: No extractor matches the kind/media_type,
            or the attachment is a URL (which this helper never fetches), or
            a ``data:`` URL without the ``;base64`` marker.
        ExtractionDependencyMissingError: An extractor exists but its optional
            pip extra isn't installed.  Message names the extra to install.
        ExtractionFailedError: Dependency present but extraction failed
            (corrupt file, malformed base64, password-protected PDF,
            scan-only PDF, OCR confidence below threshold).  ``__cause__``
            is set to the underlying exception.
        ValueError: ``max_pages`` is not ``None`` or a positive integer.
    """
    # L8 fix: reject zero and negative values outright. Python slicing
    # ``pages[:0]`` == [] and ``pages[:-1]`` silently drops the last page —
    # both surprising. Callers should pass ``None`` for "all pages".
    if max_pages is not None and max_pages <= 0:
        raise ValueError(
            "max_pages must be a positive integer or None (all pages); "
            f"got {max_pages!r}"
        )

    kind, media_type, data = bytes_from_attachment(att)

    extractor = _EXTRACTORS.get((kind, media_type))
    if extractor is None:
        raise UnsupportedAttachmentError(
            f"No extractor registered for kind={kind!r}, "
            f"media_type={media_type!r}. "
            f"Supported combinations: {sorted(_EXTRACTORS.keys())}"
        )

    return extractor(data, max_pages=max_pages, page_markers=page_markers)

extract_texts

extract_texts(
    atts: Sequence[Any],
    *,
    max_pages: int | None = None,
    page_markers: bool = True,
    on_error: Literal["raise", "skip", "warn"] = "warn",
) -> list[str]

Batch extraction.

Parameters:

Name	Type	Description	Default
atts	Sequence[Any]	Sequence of Attachment instances or content-block dicts.	required
max_pages	int | None	Forwarded to :func:extract_text for each item.	None
page_markers	bool	Forwarded to :func:extract_text for each item.	True
on_error	Literal['raise', 'skip', 'warn']	Controls per-item failure handling. "raise" — stop on first failure (re-raises the exception). "skip" — silently omit failing items. No log output. "warn" (default) — emit a logging.WARNING via the symfonic.agent.attachments.extractors logger and omit the failing item. Intercept in tests via caplog.at_level(logging.WARNING) or a custom logging.Handler — NOT warnings.catch_warnings.	'warn'

Returns:

Type	Description
list[str]	A list of extracted text strings, one per successfully extracted
list[str]	item. The returned length may be less than len(atts) when
list[str]	on_error is "skip" or "warn" and any item fails; positional
list[str]	correspondence with atts is not preserved. Use "warn" (or
list[str]	"raise") if you need per-item failure signalling.

Source code in src/symfonic/agent/attachments/extractors.py

def extract_texts(
    atts: Sequence[Any],
    *,
    max_pages: int | None = None,
    page_markers: bool = True,
    on_error: Literal["raise", "skip", "warn"] = "warn",
) -> list[str]:
    """Batch extraction.

    Args:
        atts: Sequence of ``Attachment`` instances or content-block dicts.
        max_pages: Forwarded to :func:`extract_text` for each item.
        page_markers: Forwarded to :func:`extract_text` for each item.
        on_error: Controls per-item failure handling.

            * ``"raise"`` — stop on first failure (re-raises the exception).
            * ``"skip"`` — silently omit failing items. **No log output.**
            * ``"warn"`` (default) — emit a ``logging.WARNING`` via the
              ``symfonic.agent.attachments.extractors`` logger and omit the
              failing item. Intercept in tests via
              ``caplog.at_level(logging.WARNING)`` or a custom
              ``logging.Handler`` — NOT ``warnings.catch_warnings``.

    Returns:
        A list of extracted text strings, one per *successfully* extracted
        item. **The returned length may be less than** ``len(atts)`` when
        ``on_error`` is ``"skip"`` or ``"warn"`` and any item fails; positional
        correspondence with ``atts`` is **not preserved**. Use ``"warn"`` (or
        ``"raise"``) if you need per-item failure signalling.
    """
    results: list[str] = []
    for i, att in enumerate(atts):
        try:
            results.append(
                extract_text(att, max_pages=max_pages, page_markers=page_markers)
            )
        except AttachmentExtractionError as exc:
            if on_error == "raise":
                raise
            if on_error == "warn":
                logger.warning(
                    "extract_texts: item %d failed — %s", i, exc
                )
            # "skip" silently omits; "warn" logs then omits.
    return results

is_extractable

is_extractable(att: Any) -> bool

Return True when the (kind, media_type) combination is supported AND the optional dependency is installed.

This is a capability check, not a payload validator. It answers three questions:

1. Is the input shape recognised (Attachment or known content-block dict)?
2. Is there an extractor registered for (kind, media_type)?
3. Is the extractor's optional Python dependency importable? For image OCR, also: is the tesseract binary on PATH?

Does NOT validate payload contents. A True result does not guarantee :func:extract_text will succeed on a specific payload — the call may still fail with UnsupportedAttachmentError (e.g., a non-base64 data: URL that carries a supported media_type), or ExtractionFailedError (e.g., corrupt PDF, malformed base64). The capability check is deliberately cheap and payload-agnostic (M5 + N2).

Callers that need to know whether extraction would actually succeed should call :func:extract_text and catch AttachmentExtractionError.

Never raises. Safe to use for runtime routing decisions (spec §1).

Source code in src/symfonic/agent/attachments/extractors.py

def is_extractable(att: Any) -> bool:
    """Return ``True`` when the (kind, media_type) combination is supported
    AND the optional dependency is installed.

    This is a **capability check**, not a payload validator. It answers three
    questions:

    1. Is the input shape recognised (Attachment or known content-block dict)?
    2. Is there an extractor registered for ``(kind, media_type)``?
    3. Is the extractor's optional Python dependency importable?
       For image OCR, also: is the ``tesseract`` binary on PATH?

    **Does NOT validate payload contents.** A ``True`` result does not
    guarantee :func:`extract_text` will succeed on a specific payload — the
    call may still fail with ``UnsupportedAttachmentError`` (e.g., a
    non-base64 ``data:`` URL that carries a supported media_type), or
    ``ExtractionFailedError`` (e.g., corrupt PDF, malformed base64). The
    capability check is deliberately cheap and payload-agnostic (M5 + N2).

    Callers that need to know whether extraction would actually succeed
    should call :func:`extract_text` and catch ``AttachmentExtractionError``.

    Never raises. Safe to use for runtime routing decisions (spec §1).
    """
    # Kind / media_type discovery. We need just those two fields, not the
    # decoded bytes — M5 fix. Read them directly to avoid decoding the
    # payload (which would raise on malformed base64).
    kind, media_type = _kind_and_media_type(att)
    if kind is None:
        return False

    extractor = _EXTRACTORS.get((kind, media_type))
    if extractor is None:
        return False

    return _dependency_available(extractor)