"""Scope-compatible resolution of recorded document links.
Ledger evidence may start from a recorded
:class:`~domain.attachments.enums.AttachmentSource` link, but the link is not
stored as evidence by itself.
:func:`~adapters.outbound.google.document_link_resolver.resolve_document_link` fetches reachable
Drive content as bytes so the caller can persist those bytes through
:func:`~domain.attachments.service.add_attachment`; the original link remains
provenance metadata on that byte-bearing attachment.
:func:`~adapters.outbound.google.document_link_resolver.list_drive_folder_documents` extends the
same minimal-scope posture to a *folder*: it lists the PDF/image children of a
``drive.file``-reachable folder so a caller can bulk-fetch every invoice in one
sweep instead of resolving one document link at a time.
The resolver stays inside the integration's deliberate minimal-scope posture:
``drive.file`` can download Drive files the app created or the operator picked,
so a ``GOOGLE_DRIVE`` reference to such a file resolves. Operator-external
documents, arbitrary Drive files that require ``drive.readonly``, and Gmail
messages that require ``gmail.readonly`` are refused with
:exc:`~adapters.outbound.storage.errors.OutboundStoragePermissionError` instead
of being silently stored as links.
"""
from __future__ import annotations
import re
from collections.abc import Iterator, Mapping
from dataclasses import dataclass
from enum import StrEnum
from typing import TYPE_CHECKING, Any, Final, NoReturn, Protocol, cast
from pydantic import BaseModel, Field, ValidationError, field_validator
from ....core.errors.hierarchy import pydantic_validation_boundary
from ....core.external_constants import PDF_MIME_TYPE
from ....core.models import STRICT_FROZEN_CONFIG
from ....core.operator_action_enums import ActionEvidenceProvenance, NoRecoveryOutcome
from ....core.type_guards import is_object_list
from ....domain.attachments.enums import AttachmentSource
from ..storage.drive_pagination import next_drive_page_token
from ..storage.errors import (
OutboundStorageError,
OutboundStorageNetworkError,
OutboundStoragePermissionError,
OutboundStorageValidationError,
)
from ._preconditions import google_terminal_refusal
from .api import execute_request
if TYPE_CHECKING:
from google.auth.credentials import Credentials
# .../d/<ID>/... (file, spreadsheets, document) | ...?id=<ID> | bare <ID>.
# The bare form requires >=25 chars so a hyphenated English token (e.g.
# "not-a-drive-reference") is not mistaken for a file id and sent to the network;
# real Drive ids are ~28-44 chars. The URL-embedded forms are unambiguous from
# context so they accept the shorter >=10.
_DRIVE_ID_PATTERNS: Final[tuple[re.Pattern[str], ...]] = (
re.compile(r"/d/(?P<id>[A-Za-z0-9_-]{10,})"),
re.compile(r"[?&]id=(?P<id>[A-Za-z0-9_-]{10,})"),
re.compile(r"^(?P<id>[A-Za-z0-9_-]{25,})$"),
)
_GMAIL_SCOPE: Final[str] = "https://www.googleapis.com/auth/gmail.readonly"
_DRIVE_READONLY_SCOPE: Final[str] = "https://www.googleapis.com/auth/drive.readonly"
# The bulk folder sweep filters to the document kinds an invoice PDF/scan can
# take. This mirrors the CLI's ``_sniff_document_mime_type`` fallback set for
# consistency between the single-doclink and folder-sweep paths.
_INVOICE_MIME_TYPES: Final[frozenset[str]] = frozenset(
{PDF_MIME_TYPE, "image/png", "image/jpeg"},
)
_DRIVE_FOLDER_MIME_TYPE: Final[str] = "application/vnd.google-apps.folder"
_DRIVE_LIST_PAGE_SIZE: Final[int] = 100
[docs]
class DocumentLinkPreconditionCondition(StrEnum):
"""Closed terminal conditions owned by the document-link adapter."""
API_CLIENT_AVAILABLE = "google.document_link.api_client_available"
REMOTE_SOURCE_SUPPORTED = "google.document_link.remote_source_supported"
DRIVE_REFERENCE_IDENTIFIED = "google.document_link.drive_reference_identified"
SOURCE_SCOPE_SUFFICIENT = "google.document_link.source_scope_sufficient"
FILE_SCOPE_SUFFICIENT = "google.document_link.file_scope_sufficient"
MEDIA_TRANSPORT_AVAILABLE = "google.document_link.media_transport_available"
MEDIA_PAYLOAD_BYTES = "google.document_link.media_payload_bytes"
FOLDER_SCOPE_SUFFICIENT = "google.document_link.folder_scope_sufficient"
FOLDER_FILES_LIST_VALID = "google.document_link.folder_files_list_valid"
FOLDER_ENTRY_VALID = "google.document_link.folder_entry_valid"
FOLDER_PAGINATION_VALID = "google.document_link.folder_pagination_valid"
def _document_link_terminal_refusal(
error: OutboundStorageError,
condition: DocumentLinkPreconditionCondition,
*,
facts: Mapping[str, str | int | bool],
outcome: NoRecoveryOutcome,
) -> OutboundStorageError:
"""Return ``error``'s typed equivalent carrying this adapter's terminal verdict."""
return google_terminal_refusal(
error,
condition_id=condition.value,
facts=facts,
provenance=ActionEvidenceProvenance.RUNTIME_OBSERVATION,
outcome=outcome,
)
class _DriveMediaRequest(Protocol):
def execute(self) -> object: ...
class _DriveFilesListRequest(Protocol):
def execute(self, num_retries: int = ...) -> dict[str, Any]: ...
class _DriveFilesResource(Protocol):
def get_media(self, **request: object) -> _DriveMediaRequest: ...
def list(self, **request: object) -> _DriveFilesListRequest: ...
class _DriveService(Protocol):
def files(self) -> _DriveFilesResource: ...
[docs]
class DriveFolderDocument(BaseModel):
"""One PDF/image child of a ``drive.file``-reachable Drive folder.
Returned by :func:`list_drive_folder_documents`; carries just enough
metadata for a caller to fetch the file (:attr:`file_id`) and report it
to the operator (:attr:`name`, :attr:`mime_type`).
"""
model_config = STRICT_FROZEN_CONFIG
file_id: str = Field(validation_alias="id", min_length=1)
name: str = Field(min_length=1)
mime_type: str = Field(alias="mimeType", min_length=1)
@field_validator("file_id", "name", "mime_type")
@classmethod
@pydantic_validation_boundary
def _require_non_blank_wire_value(cls, value: str) -> str:
"""Reject blank Drive resource values before a listing can consume them."""
if not value.strip():
raise ValueError("must not be blank")
return value
[docs]
@dataclass(frozen=True)
class DriveFolderListing:
"""The result of listing one Drive folder's invoice-shaped children.
``documents`` carries the PDF/image files the sweep will fetch;
``skipped_non_document_count`` records how many folder children were
filtered out for carrying a non-invoice MIME type (e.g. a nested folder
or an unrelated file), so the caller can report an honest total.
"""
documents: tuple[DriveFolderDocument, ...]
skipped_non_document_count: int
[docs]
def parse_drive_file_id(reference: str) -> str | None:
"""Extract the Drive file id consumed by :func:`resolve_document_link`.
Args:
reference: A Drive URL, ``?id=...`` link, bare Drive file id, or
non-Drive reference.
Returns:
The parsed Drive file id, or ``None`` when ``reference`` does not carry
a recognisable Drive id.
"""
candidate = reference.strip()
for pattern in _DRIVE_ID_PATTERNS:
match = pattern.search(candidate)
if match:
drive_id = match.group("id")
if isinstance(drive_id, str):
return drive_id
return None
def _drive_service(credentials: Credentials) -> _DriveService:
try:
from googleapiclient.discovery import build
except ImportError as exc:
raise _document_link_terminal_refusal(
OutboundStorageNetworkError(
"googleapiclient is not importable",
context={"dependency": "google-api-python-client"},
),
DocumentLinkPreconditionCondition.API_CLIENT_AVAILABLE,
facts={"dependency": "google_api_python_client", "client_available": False},
outcome=NoRecoveryOutcome.SAFETY,
) from exc
# dynamic resource; the protocol pins the files().get_media().execute
# surface used below.
# CAST-RATIONALE-thirdparty: googleapiclient build() returns an untyped Resource
return cast(_DriveService, build("drive", "v3", credentials=credentials, cache_discovery=False))
def _resolved_drive_service(credentials: Credentials | None, service: _DriveService | None) -> _DriveService:
"""Return the injected Drive service, or build one from credentials.
Neither present is a caller defect rather than a recoverable condition:
absence is reported explicitly instead of being carried into a transport
call that would fail with an unrelated message.
"""
if service is not None:
return service
if credentials is None:
raise OutboundStorageValidationError(
"resolving a Drive document requires either credentials or an injected service",
)
return _drive_service(credentials)
[docs]
def resolve_document_link(
*,
source: AttachmentSource,
reference: str,
credentials: Credentials | None = None,
service: _DriveService | None = None,
) -> bytes:
"""Resolve a recorded :class:`~domain.attachments.enums.AttachmentSource` link to bytes.
Args:
source: The recorded link source.
reference: The link reference (Drive URL / file id, Gmail link, or URL).
credentials: Google OAuth credentials carrying the granted scopes.
Optional only because ``service`` may be injected instead; one
of the two must be supplied.
service: Optional pre-built Drive ``v3`` service. When ``None`` (the
production path) the service is built from ``credentials``; tests
inject a transport-only seam here so the fetch path runs without a
live network or real credentials.
Returns:
The fetched document bytes for ``GOOGLE_DRIVE`` links the ``drive.file``
scope can reach.
Raises:
:exc:`~adapters.outbound.storage.errors.OutboundStoragePermissionError`:
For Gmail links, arbitrary URLs, and Drive files outside the
``drive.file`` scope. The required sensitive scope is named in
``context["required_scope"]``.
:exc:`~adapters.outbound.storage.errors.OutboundStorageValidationError`:
For sources that are not remote documents, or a Drive reference
with no recognisable file id.
"""
if source is AttachmentSource.GMAIL:
raise _document_link_terminal_refusal(
OutboundStoragePermissionError(
"Gmail document-link resolution requires the sensitive gmail.readonly scope, "
"which this integration does not request",
context={"required_scope": _GMAIL_SCOPE, "source": source.value},
),
DocumentLinkPreconditionCondition.SOURCE_SCOPE_SUFFICIENT,
facts={"source": source.value, "required_scope": _GMAIL_SCOPE, "scope_sufficient": False},
outcome=NoRecoveryOutcome.SAFETY,
)
if source is AttachmentSource.GOOGLE_DRIVE:
file_id = parse_drive_file_id(reference)
if file_id is None:
raise _document_link_terminal_refusal(
OutboundStorageValidationError(
"Google Drive link does not contain a recognisable file id",
context={"reference": reference},
),
DocumentLinkPreconditionCondition.DRIVE_REFERENCE_IDENTIFIED,
facts={"source": source.value, "reference_identified": False},
outcome=NoRecoveryOutcome.OPERATOR_DECISION,
)
drive_service = _resolved_drive_service(credentials, service)
return _download_drive_file_from_service(file_id, drive_service)
if source is AttachmentSource.URL:
raise _document_link_terminal_refusal(
OutboundStoragePermissionError(
"resolving an arbitrary external URL is outside the granted drive.file scope; "
"it requires drive.readonly (for Drive content) or manual download",
context={"required_scope": _DRIVE_READONLY_SCOPE, "source": source.value},
),
DocumentLinkPreconditionCondition.SOURCE_SCOPE_SUFFICIENT,
facts={"source": source.value, "required_scope": _DRIVE_READONLY_SCOPE, "scope_sufficient": False},
outcome=NoRecoveryOutcome.SAFETY,
)
raise _document_link_terminal_refusal(
OutboundStorageValidationError(
f"document-link source {source.value!r} is not a resolvable remote document",
context={"source": source.value},
),
DocumentLinkPreconditionCondition.REMOTE_SOURCE_SUPPORTED,
facts={"source": source.value, "remote_source_supported": False},
outcome=NoRecoveryOutcome.OPERATOR_DECISION,
)
def _download_drive_file_from_service(file_id: str, service: _DriveService) -> bytes:
request = service.files().get_media(fileId=file_id)
try:
payload = request.execute()
except OutboundStorageError:
raise
except Exception as exc:
status = getattr(getattr(exc, "resp", None), "status", None)
# drive.file cannot see files the app did not create / the operator did
# not pick — Google returns 403/404. Surface the scope-upgrade path.
if status in (403, 404):
raise _document_link_terminal_refusal(
OutboundStoragePermissionError(
f"Drive file {file_id!r} is not reachable under the drive.file scope "
"(the app can only read files it created or the operator picked); "
"reading an arbitrary operator file requires drive.readonly",
context={"required_scope": _DRIVE_READONLY_SCOPE, "file_id": file_id},
),
DocumentLinkPreconditionCondition.FILE_SCOPE_SUFFICIENT,
facts={
"file_id": file_id,
"required_scope": _DRIVE_READONLY_SCOPE,
"status": str(status),
"scope_sufficient": False,
},
outcome=NoRecoveryOutcome.SAFETY,
) from exc
raise _document_link_terminal_refusal(
OutboundStorageNetworkError(
"Drive files.get_media failed",
context={"file_id": file_id, "status": str(status) if status is not None else "unknown"},
),
DocumentLinkPreconditionCondition.MEDIA_TRANSPORT_AVAILABLE,
facts={
"file_id": file_id,
"status": str(status) if status is not None else "unknown",
"transport_available": False,
},
outcome=NoRecoveryOutcome.SAFETY,
) from exc
if not isinstance(payload, (bytes, bytearray)):
raise _document_link_terminal_refusal(
OutboundStorageValidationError(
"Drive files.get_media returned a non-bytes payload",
context={"file_id": file_id},
),
DocumentLinkPreconditionCondition.MEDIA_PAYLOAD_BYTES,
facts={"file_id": file_id, "payload_bytes": False, "payload_type": type(payload).__name__},
outcome=NoRecoveryOutcome.OPERATOR_DECISION,
)
return bytes(payload)
[docs]
def list_drive_folder_documents(
*,
folder_id: str,
credentials: Credentials | None = None,
service: _DriveService | None = None,
) -> DriveFolderListing:
"""List the PDF/image children of a ``drive.file``-reachable Drive folder.
Stays inside the same minimal-scope posture as
:func:`resolve_document_link`: ``drive.file`` only lists files the app
created or the operator explicitly picked, so a folder the operator has
not shared with the app (or does not own under this scope) surfaces no
children rather than a permission escalation. A non-existent or
unreachable ``folder_id`` maps Google's 403/404 to the same
:exc:`~adapters.outbound.storage.errors.OutboundStoragePermissionError`
scope-named refusal :func:`resolve_document_link` uses, so the two
fetch surfaces read the same way.
Args:
folder_id: The Drive folder id (parsed the same way a document
reference is via :func:`parse_drive_file_id`, or passed as a bare id).
credentials: Google OAuth credentials carrying the granted scopes.
Optional only because ``service`` may be injected instead; one
of the two must be supplied.
service: Optional pre-built Drive ``v3`` service (test seam).
Returns:
A :class:`DriveFolderListing` naming every PDF/image child plus a
count of filtered-out non-document children.
Raises:
:exc:`~adapters.outbound.storage.errors.OutboundStoragePermissionError`:
When the folder is not reachable under the ``drive.file`` scope.
:exc:`~adapters.outbound.storage.errors.OutboundStorageNetworkError`:
On any other transport or unmapped Drive failure.
"""
drive_service = _resolved_drive_service(credentials, service)
documents: list[DriveFolderDocument] = []
skipped = 0
for document in _iter_drive_folder_files(drive_service, folder_id=folder_id):
mime_type = document.mime_type
if mime_type == _DRIVE_FOLDER_MIME_TYPE:
continue
if mime_type not in _INVOICE_MIME_TYPES:
skipped += 1
continue
documents.append(document)
return DriveFolderListing(documents=tuple(documents), skipped_non_document_count=skipped)
def _drive_folder_list_request(
files: _DriveFilesResource,
*,
query: str,
page_token: str | None,
) -> _DriveFilesListRequest:
"""Build one Drive list request, preserving the first-page call shape."""
if page_token is None:
return files.list(
q=query,
fields="nextPageToken, files(id, name, mimeType)",
pageSize=_DRIVE_LIST_PAGE_SIZE,
)
return files.list(
q=query,
fields="nextPageToken, files(id, name, mimeType)",
pageSize=_DRIVE_LIST_PAGE_SIZE,
pageToken=page_token,
)
def _raise_folder_scope_refusal(
error: OutboundStoragePermissionError,
*,
folder_id: str,
) -> NoReturn:
"""Map an unscoped Drive permission failure to the folder refusal contract."""
context = dict(error.context or {})
if "required_scope" in context:
raise error
raise _document_link_terminal_refusal(
OutboundStoragePermissionError(
f"Drive folder {folder_id!r} is not reachable under the drive.file scope "
"(the app can only list files/folders it created or the operator picked); "
"reading an arbitrary operator folder requires drive.readonly",
context={"required_scope": _DRIVE_READONLY_SCOPE, "folder_id": folder_id, **context},
),
DocumentLinkPreconditionCondition.FOLDER_SCOPE_SUFFICIENT,
facts={"folder_id": folder_id, "required_scope": _DRIVE_READONLY_SCOPE, "scope_sufficient": False},
outcome=NoRecoveryOutcome.SAFETY,
) from error
def _fetch_drive_folder_page(
drive_service: _DriveService,
*,
folder_id: str,
query: str,
page_token: str | None,
) -> dict[str, Any]:
"""Fetch one folder page and map permission failures at the folder boundary."""
request = _drive_folder_list_request(
drive_service.files(),
query=query,
page_token=page_token,
)
try:
response = execute_request(
# CAST-RATIONALE-thirdparty: Google Drive SDK request object is untyped at the client boundary
cast("Any", request),
action="drive.files.list",
)
except OutboundStoragePermissionError as exc:
_raise_folder_scope_refusal(exc, folder_id=folder_id)
# CAST-RATIONALE-thirdparty: Google Drive files-list is an untyped JSON API response
return cast(dict[str, Any], response)
def _folder_page_documents(
response: Mapping[str, Any],
*,
folder_id: str,
) -> Iterator[DriveFolderDocument]:
"""Validate and materialise each raw file row from one successful page."""
# CAST-RATIONALE-thirdparty: Google Drive files-list is an untyped JSON API response
raw_files: object = response.get("files", [])
if not is_object_list(raw_files):
raise _document_link_terminal_refusal(
OutboundStorageValidationError(
"Drive files.list returned a non-list files field",
context={"action": "drive.files.list", "folder_id": folder_id},
),
DocumentLinkPreconditionCondition.FOLDER_FILES_LIST_VALID,
facts={"folder_id": folder_id, "files_list_valid": False},
outcome=NoRecoveryOutcome.OPERATOR_DECISION,
)
for entry_index, raw_file in enumerate(raw_files):
yield _drive_folder_document_from_response(raw_file, folder_id=folder_id, entry_index=entry_index)
def _next_drive_folder_page_token(
response: Mapping[str, Any],
*,
folder_id: str,
seen_tokens: set[str],
) -> str | None:
"""Validate one folder page's continuation token and map malformed tokens."""
# CAST-RATIONALE-thirdparty: Google Drive nextPageToken is an untyped JSON API response
raw_page_token = response.get("nextPageToken")
try:
return next_drive_page_token(
raw_page_token,
seen_tokens=seen_tokens,
action="drive.files.list",
)
except OutboundStorageNetworkError as exc:
token_state = "repeated" if isinstance(raw_page_token, str) else "non_string"
raise _document_link_terminal_refusal(
OutboundStorageValidationError(
"Drive files.list returned an invalid nextPageToken",
context={
"action": "drive.files.list",
"folder_id": folder_id,
"page_token_state": token_state,
},
),
DocumentLinkPreconditionCondition.FOLDER_PAGINATION_VALID,
facts={"folder_id": folder_id, "page_token_state": token_state, "page_token_valid": False},
outcome=NoRecoveryOutcome.OPERATOR_DECISION,
) from exc
def _iter_drive_folder_files(drive_service: _DriveService, *, folder_id: str) -> Iterator[DriveFolderDocument]:
"""Yield every raw Drive child of ``folder_id`` across pages, mapping scope refusals.
A 403/404 (or any permission refusal without an already-named ``required_scope``)
is remapped to the same ``drive.file``-scope refusal :func:`resolve_document_link`
uses, so both fetch surfaces read identically.
"""
query = f"'{folder_id}' in parents and trashed = false"
page_token: str | None = None
seen_tokens: set[str] = set()
while True:
response = _fetch_drive_folder_page(
drive_service,
folder_id=folder_id,
query=query,
page_token=page_token,
)
yield from _folder_page_documents(response, folder_id=folder_id)
page_token = _next_drive_folder_page_token(response, folder_id=folder_id, seen_tokens=seen_tokens)
if not page_token:
break
def _drive_folder_document_from_response(
raw_file: object,
*,
folder_id: str,
entry_index: int,
) -> DriveFolderDocument:
"""Validate a successful ``files.list`` row before callers classify it."""
if not isinstance(raw_file, Mapping):
raise _document_link_terminal_refusal(
OutboundStorageValidationError(
"Drive files.list returned a malformed file entry",
context={"action": "drive.files.list", "folder_id": folder_id, "entry_index": str(entry_index)},
),
DocumentLinkPreconditionCondition.FOLDER_ENTRY_VALID,
facts={"folder_id": folder_id, "entry_index": entry_index, "entry_mapping": False},
outcome=NoRecoveryOutcome.OPERATOR_DECISION,
)
try:
return DriveFolderDocument.model_validate(raw_file)
except ValidationError as exc:
raise _document_link_terminal_refusal(
OutboundStorageValidationError(
"Drive files.list returned a malformed file entry",
context={"action": "drive.files.list", "folder_id": folder_id, "entry_index": str(entry_index)},
),
DocumentLinkPreconditionCondition.FOLDER_ENTRY_VALID,
facts={"folder_id": folder_id, "entry_index": entry_index, "entry_mapping": True},
outcome=NoRecoveryOutcome.OPERATOR_DECISION,
) from exc
__all__ = [
"DriveFolderDocument",
"DriveFolderListing",
"list_drive_folder_documents",
"parse_drive_file_id",
"resolve_document_link",
]