refac

gsheets/sheets_helpers.py

@@ -7,11 +7,15 @@ conditional formatting helpers.
 
 import asyncio
 import json
+import logging
 import re
 from typing import List, Optional, Union
 
 from core.utils import UserInputError
 
+logger = logging.getLogger(__name__)
+
+MAX_GRID_METADATA_CELLS = 5000
 
 A1_PART_REGEX = re.compile(r"^([A-Za-z]*)(\d*)$")
 SHEET_TITLE_SAFE_RE = re.compile(r"^[A-Za-z0-9_]+$")
@@ -955,3 +959,94 @@ def _format_sheet_notes_section(
         else ""
     )
     return f"\n\nCell notes in range '{range_label}':\n" + "\n".join(lines) + suffix
+
+
+async def _fetch_grid_metadata(
+    service,
+    spreadsheet_id: str,
+    resolved_range: str,
+    values: List[List[object]],
+    include_hyperlinks: bool = False,
+    include_notes: bool = False,
+) -> tuple[str, str]:
+    """Fetch hyperlinks and/or notes for a range via a single spreadsheets.get call.
+
+    Computes tight range bounds, enforces the cell-count cap, builds a combined
+    ``fields`` selector so only one API round-trip is needed when both flags are
+    ``True``, then parses the response into formatted output sections.
+
+    Returns:
+        (hyperlink_section, notes_section) — each is an empty string when the
+        corresponding flag is ``False`` or no data was found.
+    """
+    if not include_hyperlinks and not include_notes:
+        return "", ""
+
+    tight_range = _a1_range_for_values(resolved_range, values)
+    if not tight_range:
+        logger.info(
+            "[read_sheet_values] Skipping grid metadata fetch for range '%s': "
+            "unable to determine tight bounds",
+            resolved_range,
+        )
+        return "", ""
+
+    cell_count = _a1_range_cell_count(tight_range) or sum(
+        len(row) for row in values
+    )
+    if cell_count > MAX_GRID_METADATA_CELLS:
+        logger.info(
+            "[read_sheet_values] Skipping grid metadata fetch for large range "
+            "'%s' (%d cells > %d limit)",
+            tight_range,
+            cell_count,
+            MAX_GRID_METADATA_CELLS,
+        )
+        return "", ""
+
+    # Build a combined fields selector so we hit the API at most once.
+    value_fields: list[str] = []
+    if include_hyperlinks:
+        value_fields.extend(["hyperlink", "textFormatRuns(format(link(uri)))"])
+    if include_notes:
+        value_fields.append("note")
+
+    fields = (
+        "sheets(properties(title),data(startRow,startColumn,"
+        f"rowData(values({','.join(value_fields)}))))"
+    )
+
+    try:
+        response = await asyncio.to_thread(
+            service.spreadsheets()
+            .get(
+                spreadsheetId=spreadsheet_id,
+                ranges=[tight_range],
+                includeGridData=True,
+                fields=fields,
+            )
+            .execute
+        )
+    except Exception as exc:
+        logger.warning(
+            "[read_sheet_values] Failed fetching grid metadata for range '%s': %s",
+            tight_range,
+            exc,
+        )
+        return "", ""
+
+    hyperlink_section = ""
+    if include_hyperlinks:
+        hyperlinks = _extract_cell_hyperlinks_from_grid(response)
+        hyperlink_section = _format_sheet_hyperlink_section(
+            hyperlinks=hyperlinks, range_label=tight_range
+        )
+
+    notes_section = ""
+    if include_notes:
+        notes = _extract_cell_notes_from_grid(response)
+        notes_section = _format_sheet_notes_section(
+            notes=notes, range_label=tight_range
+        )
+
+    return hyperlink_section, notes_section