текущая версия (рабочая)

app/services/sheets.py

@@ -0,0 +1,79 @@
+import logging
+import re
+from functools import lru_cache
+from typing import Any
+
+import gspread
+from gspread.utils import InsertDataOption, ValueInputOption
+
+logger = logging.getLogger(__name__)
+sheets_api = gspread.service_account(filename="google-auth/key.json")
+# Таймаут HTTP-запросов (сек): защита от зависания при проблемах с сетью/Google API
+sheets_api.set_timeout(30)
+
+# HTTP коды, при которых стоит сбросить кэш листа (устаревший объект)
+_CACHE_INVALIDATE_CODES = frozenset({401, 403, 404, 429})
+
+
+def parse_google_sheets_url(url: str) -> tuple[str, int | None]:
+    spreadsheet_match = re.search(r"/d/([a-zA-Z0-9-_]+)", url)
+    if not spreadsheet_match:
+        raise ValueError("Invalid Google Sheets URL: Could not find spreadsheet ID")
+    spreadsheet_id = spreadsheet_match.group(1)
+
+    worksheet_match = re.search(r"[?&]gid=(\d+)", url)
+    worksheet_id = int(worksheet_match.group(1)) if worksheet_match else None
+
+    return spreadsheet_id, worksheet_id
+
+
+@lru_cache
+def get_sheet(sheet_link: str) -> gspread.Worksheet:
+    spreadsheet_id, worksheet_id = parse_google_sheets_url(sheet_link)
+
+    spreadsheet = sheets_api.open_by_key(spreadsheet_id)
+    if worksheet_id is not None:
+        return spreadsheet.get_worksheet_by_id(worksheet_id)
+    return spreadsheet.sheet1
+
+
+def add_rows_to_sheet(sheet_link: str, rows: list[list]) -> None:
+    """Append rows to Google Sheet. Raises on failure so caller can retry."""
+    sheet = get_sheet(sheet_link)
+    try:
+        sheet.append_rows(
+            rows,
+            value_input_option=ValueInputOption.user_entered,
+            insert_data_option=InsertDataOption.insert_rows,
+            table_range="A1",
+        )
+    except gspread.exceptions.APIError as e:
+        http_code = getattr(e, "response", None)
+        status_code = http_code.status_code if http_code else None
+        error_detail = getattr(e, "error", {})
+        error_msg = error_detail.get("message", str(e)) if isinstance(error_detail, dict) else str(e)
+        logger.exception(
+            "SHEETS_API_ERROR: type=APIError http_code=%s message=%r error=%s",
+            status_code,
+            error_msg,
+            error_detail,
+            extra={"sheet_link": sheet_link, "rows_count": len(rows)},
+        )
+        if status_code in _CACHE_INVALIDATE_CODES:
+            get_sheet.cache_clear()
+            logger.info("SHEETS_CACHE_CLEARED: сброшен кэш get_sheet после ошибки %s", status_code)
+        raise
+    except Exception as e:
+        logger.exception(
+            "SHEETS_ERROR: type=%s message=%s sheet_link=%s rows_count=%s",
+            type(e).__name__,
+            str(e),
+            sheet_link,
+            len(rows),
+        )
+        raise
+
+
+def read_sheet(sheet_link: str) -> list[dict[str, Any]]:
+    sheet = get_sheet(sheet_link)
+    return sheet.get_all_records()