121 lines
4.7 KiB
Python
121 lines
4.7 KiB
Python
import base64
|
|
import binascii
|
|
import json
|
|
import logging
|
|
import re
|
|
from functools import lru_cache
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
import gspread
|
|
from gspread.utils import InsertDataOption, ValueInputOption
|
|
|
|
from app import config
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
# Таймаут HTTP-запросов (сек): защита от зависания при проблемах с сетью/Google API
|
|
_SHEETS_HTTP_TIMEOUT = 30
|
|
|
|
|
|
def _load_service_account_info(raw: str) -> dict[str, Any]:
|
|
"""Разобрать ключ из переменной окружения: чистый JSON или его base64."""
|
|
raw = raw.strip()
|
|
if not raw.startswith("{"):
|
|
try:
|
|
raw = base64.b64decode(raw, validate=True).decode("utf-8")
|
|
except (binascii.Error, UnicodeDecodeError) as e:
|
|
raise ValueError(
|
|
"GOOGLE_SERVICE_ACCOUNT_JSON: ожидается JSON ключа сервисного "
|
|
"аккаунта или его base64"
|
|
) from e
|
|
return json.loads(raw)
|
|
|
|
|
|
@lru_cache(maxsize=1)
|
|
def get_client() -> gspread.Client:
|
|
"""Клиент Google Sheets. Создаётся при первом обращении, не при импорте."""
|
|
if config.GOOGLE_SERVICE_ACCOUNT_JSON:
|
|
info = _load_service_account_info(config.GOOGLE_SERVICE_ACCOUNT_JSON)
|
|
client = gspread.service_account_from_dict(info)
|
|
logger.info("Google Sheets: ключ загружен из GOOGLE_SERVICE_ACCOUNT_JSON")
|
|
else:
|
|
key_path = Path(config.GOOGLE_SERVICE_ACCOUNT_FILE)
|
|
if not key_path.is_file():
|
|
raise FileNotFoundError(
|
|
f"Ключ сервисного аккаунта Google не найден: {key_path}. "
|
|
"Задайте GOOGLE_SERVICE_ACCOUNT_JSON или положите файл по пути "
|
|
"GOOGLE_SERVICE_ACCOUNT_FILE"
|
|
)
|
|
client = gspread.service_account(filename=str(key_path))
|
|
logger.info("Google Sheets: ключ загружен из файла %s", key_path)
|
|
client.set_timeout(_SHEETS_HTTP_TIMEOUT)
|
|
return client
|
|
|
|
# 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 = get_client().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()
|