Bot & Automation
jorgasisten
/root/hermes-projects/jorgasisten
apps/jorgasisten/services/sheets.py
text
import asyncio
import base64
import copy
import json
import logging
import math
import time
from collections import defaultdict
from dataclasses import dataclass
from difflib import SequenceMatcher
from functools import cached_property
from pathlib import Path
from time import monotonic
from typing import Any, Literal
import httpx
from google.oauth2.credentials import Credentials as AuthorizedUserCredentials
from google.oauth2.service_account import Credentials
from googleapiclient.discovery import build
from googleapiclient.errors import HttpError
from jorgasisten.core.config import Settings, get_settings
from jorgasisten.schemas import SheetRow, SheetSchema
from jorgasisten.services.text import (
compact_jsonable_rows,
compact_search_token_sequence,
normalize_compact_search_key,
normalize_key,
normalize_search_key,
safe_string,
search_tokens,
)
logger = logging.getLogger(__name__)
SCOPES = ["https://www.googleapis.com/auth/spreadsheets"]
STATUS_COLUMNS = {"status", "deleted", "dihapus", "hapus", "aktif"}
HEADER_SCAN_EXTRA_ROWS = 15
METADATA_CACHE_TTL_SECONDS = 20.0
SPREADSHEET_TITLE_CACHE_TTL_SECONDS = 20.0
CATALOG_CACHE_TTL_SECONDS = 20.0
ROWS_CACHE_TTL_SECONDS = 15.0
SCHEMA_DETAILS_CACHE_TTL_SECONDS = 30.0
_SPREADSHEET_TITLE_CACHE: dict[str, tuple[float, str]] = {}
_SPREADSHEET_METADATA_CACHE: dict[str, tuple[float, dict[str, Any]]] = {}
_SPREADSHEET_CATALOG_CACHE: dict[str, tuple[float, list[SheetSchema]]] = {}
_SPREADSHEET_ROWS_CACHE: dict[str, tuple[float, list[SheetRow]]] = {}
_SPREADSHEET_SCHEMA_DETAILS_CACHE: dict[str, tuple[float, SheetSchema]] = {}
_SHEET_SEARCH_INDEX_CACHE: dict[str, tuple[float, "SheetSearchIndex"]] = {}
CredentialPayloadKind = Literal["service_account", "authorized_user", "oauth_client", "unknown"]
@dataclass(frozen=True, slots=True)
class SheetSearchIndex:
postings: dict[str, tuple[int, ...]]
vocabulary: tuple[str, ...]
row_count: int
def build_sheet_search_index(rows: list[SheetRow]) -> SheetSearchIndex:
postings: dict[str, list[int]] = defaultdict(list)
for row_index, row in enumerate(rows):
searchable_text = " ".join(safe_string(value) for value in row.values.values())
for token in search_tokens(searchable_text):
postings[token].append(row_index)
frozen_postings = {token: tuple(indexes) for token, indexes in postings.items()}
return SheetSearchIndex(
postings=frozen_postings,
vocabulary=tuple(frozen_postings),
row_count=len(rows),
)
def candidate_row_indexes(index: SheetSearchIndex, filters: dict[str, Any]) -> list[int]:
expected_tokens: set[str] = set()
for value in filters.values():
expected_tokens.update(search_tokens(safe_string(value)))
if not expected_tokens:
return list(range(index.row_count))
match_counts: dict[int, int] = defaultdict(int)
searchable_token_count = 0
for expected_token in expected_tokens:
alternatives = search_token_alternatives(expected_token, index)
if not alternatives:
continue
searchable_token_count += 1
matching_indexes: set[int] = set()
for alternative in alternatives:
matching_indexes.update(index.postings.get(alternative, ()))
for row_index in matching_indexes:
match_counts[row_index] += 1
if searchable_token_count == 0:
return list(range(index.row_count))
minimum_overlap = max(1, math.ceil(searchable_token_count * 0.6))
return [row_index for row_index, count in match_counts.items() if count >= minimum_overlap]
def search_token_alternatives(token: str, index: SheetSearchIndex) -> set[str]:
if token in index.postings:
return {token}
if len(token) < 4 or token.isdigit():
return set()
ranked: list[tuple[float, str]] = []
for candidate in index.vocabulary:
if len(candidate) < 4 or candidate[0] != token[0]:
continue
ratio = SequenceMatcher(None, token, candidate).ratio()
if ratio >= 0.84:
ranked.append((ratio, candidate))
ranked.sort(reverse=True)
return {candidate for _, candidate in ranked[:3]}
def get_cached_payload(cache: dict[str, tuple[float, Any]], key: str) -> Any:
cached = cache.get(key)
if cached is None:
return None
expires_at, payload = cached
if monotonic() > expires_at:
cache.pop(key, None)
return None
return copy.deepcopy(payload)
def set_cached_payload(cache: dict[str, tuple[float, Any]], key: str, payload: Any, ttl_seconds: float) -> Any:
cache[key] = (monotonic() + ttl_seconds, copy.deepcopy(payload))
return payload
def quote_sheet_name(sheet_name: str) -> str:
return "'" + sheet_name.replace("'", "''") + "'"
def column_letter(column_number: int) -> str:
if column_number < 1:
raise ValueError("column_number must be greater than zero")
result = ""
current = column_number
while current:
current, remainder = divmod(current - 1, 26)
result = chr(65 + remainder) + result
return result
class GoogleSheetsClient:
TRANSIENT_HTTP_STATUS_CODES = {429, 500, 502, 503, 504}
GOOGLE_RETRY_DELAYS_SECONDS = (0.6, 1.2, 2.0)
def __init__(self, settings: Settings | None = None, spreadsheet_id: str = "") -> None:
self.settings = settings or get_settings()
self._spreadsheet_id = safe_string(spreadsheet_id)
@property
def spreadsheet_id(self) -> str:
return self._spreadsheet_id or safe_string(self.settings.google_sheets_spreadsheet_id)
def _sheet_cache_key(self, sheet_name: str) -> str:
return f"{self.spreadsheet_id}:{sheet_name}"
@cached_property
def service(self) -> Any:
return build("sheets", "v4", credentials=self._credentials(), cache_discovery=False)
def _execute(self, request: Any) -> Any:
last_error: HttpError | None = None
for attempt, delay_seconds in enumerate((*self.GOOGLE_RETRY_DELAYS_SECONDS, None), start=1):
try:
return request.execute()
except HttpError as exc:
status_code = int(getattr(getattr(exc, "resp", None), "status", 0) or 0)
if status_code not in self.TRANSIENT_HTTP_STATUS_CODES or delay_seconds is None:
raise
last_error = exc
logger.warning(
"Google Sheets transient error status=%s on attempt %s. Retrying in %.1fs",
status_code,
attempt,
delay_seconds,
)
time.sleep(delay_seconds)
if last_error is not None:
raise last_error
raise RuntimeError("Google Sheets request gagal tanpa detail error.")
def _credentials(self) -> Any:
if self.settings.google_service_account_json_base64:
decoded = base64.b64decode(self.settings.google_service_account_json_base64).decode("utf-8")
payload = json.loads(decoded)
return build_google_credentials_from_payload(payload, scopes=SCOPES)
if self.settings.google_oauth_authorized_user_json_base64:
decoded = base64.b64decode(self.settings.google_oauth_authorized_user_json_base64).decode("utf-8")
payload = json.loads(decoded)
return build_google_credentials_from_payload(payload, scopes=SCOPES)
if self.settings.google_application_credentials:
credential_path = Path(self.settings.google_application_credentials)
payload = json.loads(credential_path.read_text(encoding="utf-8"))
return build_google_credentials_from_payload(payload, scopes=SCOPES, source=str(credential_path))
raise RuntimeError("Google Sheets credential belum dikonfigurasi.")
def is_configured(self) -> bool:
has_credentials = bool(
self.settings.google_service_account_json_base64
or self.settings.google_oauth_authorized_user_json_base64
or self.settings.google_application_credentials
)
return bool(self.spreadsheet_id and has_credentials)
def get_spreadsheet_title(self) -> str:
cache_key = self.spreadsheet_id
cached_title = get_cached_payload(_SPREADSHEET_TITLE_CACHE, cache_key)
if cached_title is not None:
return cached_title
metadata = self._execute(
self.service.spreadsheets().get(
spreadsheetId=self.spreadsheet_id,
fields="properties(title)",
)
)
title = safe_string(metadata.get("properties", {}).get("title"))
return set_cached_payload(_SPREADSHEET_TITLE_CACHE, cache_key, title, SPREADSHEET_TITLE_CACHE_TTL_SECONDS)
def _metadata(self) -> dict[str, Any]:
cache_key = self.spreadsheet_id
cached_metadata = get_cached_payload(_SPREADSHEET_METADATA_CACHE, cache_key)
if cached_metadata is not None:
return cached_metadata
metadata = self._execute(
self.service.spreadsheets().get(
spreadsheetId=self.spreadsheet_id,
fields="sheets.properties(sheetId,title,gridProperties(rowCount,columnCount))",
)
)
return set_cached_payload(_SPREADSHEET_METADATA_CACHE, cache_key, metadata, METADATA_CACHE_TTL_SECONDS)
def _sheet_grid_snapshot(self, sheet_name: str, end_row: int) -> dict[str, Any]:
return self._execute(
self.service.spreadsheets().get(
spreadsheetId=self.spreadsheet_id,
ranges=[f"{quote_sheet_name(sheet_name)}!A1:ZZ{max(end_row, 1)}"],
includeGridData=True,
fields=(
"sheets.data.rowData.values("
"formattedValue,dataValidation,userEnteredValue,effectiveValue)"
),
)
)
def _sheet_values_snapshot(self, sheet_name: str, end_column: str) -> list[list[Any]]:
response = self._execute(
self.service.spreadsheets().values().get(
spreadsheetId=self.spreadsheet_id,
range=f"{quote_sheet_name(sheet_name)}!A:{end_column}",
)
)
return response.get("values", [])
def get_catalog(self) -> list[SheetSchema]:
cache_key = self.spreadsheet_id
cached_catalog = get_cached_payload(_SPREADSHEET_CATALOG_CACHE, cache_key)
if cached_catalog is not None:
return cached_catalog
metadata = self._metadata()
catalog: list[SheetSchema] = []
for sheet in metadata.get("sheets", []):
properties = sheet.get("properties", {})
title = safe_string(properties.get("title"))
if not title:
continue
grid = properties.get("gridProperties", {})
column_count = max(int(grid.get("columnCount", 0)), 1)
end_column = column_letter(column_count)
values = self._sheet_values_snapshot(title, end_column)
header_row_index, headers, sample_rows, context_lines = detect_sheet_table(
values,
sample_limit=self.settings.max_sample_rows,
)
catalog.append(
SheetSchema(
title=title,
headers=headers,
sample_rows=compact_jsonable_rows(sample_rows, limit=self.settings.max_sample_rows),
context_lines=context_lines,
field_options={},
field_validation_messages={},
header_row_number=header_row_index + 1,
row_count=count_detected_data_rows(values, header_row_index, headers),
)
)
return set_cached_payload(_SPREADSHEET_CATALOG_CACHE, cache_key, catalog, CATALOG_CACHE_TTL_SECONDS)
def get_schema(self, sheet_name: str) -> SheetSchema:
cache_key = self._sheet_cache_key(sheet_name)
cached_schema = get_cached_payload(_SPREADSHEET_SCHEMA_DETAILS_CACHE, cache_key)
if cached_schema is not None:
return cached_schema
for schema in self.get_catalog():
if schema.title == sheet_name:
detailed_schema = self._enrich_schema_with_validation(schema)
return set_cached_payload(
_SPREADSHEET_SCHEMA_DETAILS_CACHE,
cache_key,
detailed_schema,
SCHEMA_DETAILS_CACHE_TTL_SECONDS,
)
raise ValueError(f"Sheet tidak ditemukan: {sheet_name}")
def _enrich_schema_with_validation(self, schema: SheetSchema) -> SheetSchema:
if not schema.headers:
return schema
field_options, field_validation_messages = self._extract_validation_metadata(
schema.title,
schema.header_row_number - 1,
schema.headers,
end_row=max(
schema.header_row_number + HEADER_SCAN_EXTRA_ROWS,
self.settings.max_sample_rows + HEADER_SCAN_EXTRA_ROWS,
),
)
return schema.model_copy(
update={
"field_options": field_options,
"field_validation_messages": field_validation_messages,
}
)
def get_sheet_id(self, sheet_name: str) -> int:
return int(self._sheet_properties(sheet_name)["sheetId"])
def _sheet_properties(self, sheet_name: str) -> dict[str, Any]:
metadata = self._metadata()
for sheet in metadata.get("sheets", []):
properties = sheet.get("properties", {})
if properties.get("title") == sheet_name:
return properties
raise ValueError(f"Sheet tidak ditemukan: {sheet_name}")
def create_sheet(self, sheet_name: str, headers: list[str]) -> SheetSchema:
if any(schema.title == sheet_name for schema in self.get_catalog()):
raise ValueError(f"Sheet sudah ada: {sheet_name}")
self._execute(
self.service.spreadsheets().batchUpdate(
spreadsheetId=self.spreadsheet_id,
body={
"requests": [
{
"addSheet": {
"properties": {
"title": sheet_name,
"gridProperties": {
"rowCount": 1000,
"columnCount": max(len(headers), 1),
"frozenRowCount": 1,
},
}
}
}
]
},
)
)
self._invalidate_runtime_caches()
if headers:
self.update_headers(sheet_name, headers)
return self.get_schema(sheet_name)
def update_headers(self, sheet_name: str, headers: list[str]) -> SheetSchema:
if not headers:
raise ValueError("Header kolom tidak boleh kosong.")
schema = self.get_schema(sheet_name)
self.get_sheet_id(sheet_name)
self._execute(
self.service.spreadsheets().values().clear(
spreadsheetId=self.spreadsheet_id,
range=f"{quote_sheet_name(sheet_name)}!A{schema.header_row_number}:ZZ{schema.header_row_number}",
body={},
)
)
end_column = column_letter(len(headers))
self._execute(
self.service.spreadsheets().values().update(
spreadsheetId=self.spreadsheet_id,
range=(
f"{quote_sheet_name(sheet_name)}!"
f"A{schema.header_row_number}:{end_column}{schema.header_row_number}"
),
valueInputOption="USER_ENTERED",
body={"values": [headers]},
)
)
self._invalidate_runtime_caches()
return self.get_schema(sheet_name)
def rename_sheet(self, sheet_name: str, new_sheet_name: str) -> SheetSchema:
if any(schema.title == new_sheet_name for schema in self.get_catalog()):
raise ValueError(f"Sheet tujuan sudah ada: {new_sheet_name}")
self._execute(
self.service.spreadsheets().batchUpdate(
spreadsheetId=self.spreadsheet_id,
body={
"requests": [
{
"updateSheetProperties": {
"properties": {
"sheetId": self.get_sheet_id(sheet_name),
"title": new_sheet_name,
},
"fields": "title",
}
}
]
},
)
)
self._invalidate_runtime_caches()
return self.get_schema(new_sheet_name)
def delete_sheet_full(self, sheet_name: str) -> None:
catalog = self.get_catalog()
if len(catalog) <= 1:
raise ValueError("Spreadsheet harus punya minimal satu sheet, jadi sheet terakhir gak bisa dihapus.")
self._execute(
self.service.spreadsheets().batchUpdate(
spreadsheetId=self.spreadsheet_id,
body={
"requests": [
{
"deleteSheet": {
"sheetId": self.get_sheet_id(sheet_name),
}
}
]
},
)
)
self._invalidate_runtime_caches()
def read_rows(self, sheet_name: str) -> list[SheetRow]:
cache_key = self._sheet_cache_key(sheet_name)
cached_rows = get_cached_payload(_SPREADSHEET_ROWS_CACHE, cache_key)
if cached_rows is not None:
return cached_rows
response = self._execute(
self.service.spreadsheets().values().get(
spreadsheetId=self.spreadsheet_id,
range=f"{quote_sheet_name(sheet_name)}!A:ZZ",
)
)
values = response.get("values", [])
if not values:
return []
header_row_index, headers, _, _ = detect_sheet_table(values, sample_limit=self.settings.max_sample_rows)
if not headers:
return []
rows: list[SheetRow] = []
for row_offset, raw_row in enumerate(values[header_row_index + 1 :], start=header_row_index + 2):
row_values = self._row_to_dict(headers, raw_row)
if any(safe_string(value) for value in row_values.values()):
rows.append(SheetRow(sheet_name=sheet_name, row_number=row_offset, values=row_values))
_SHEET_SEARCH_INDEX_CACHE.pop(cache_key, None)
return set_cached_payload(_SPREADSHEET_ROWS_CACHE, cache_key, rows, ROWS_CACHE_TTL_SECONDS)
def append_row(self, sheet_name: str, data: dict[str, Any]) -> SheetRow:
schema = self.get_schema(sheet_name)
if not schema.headers:
raise ValueError("Sheet belum memiliki header kolom.")
next_row_number = self.next_empty_row_number(sheet_name)
self._prepare_target_row(sheet_name, schema, next_row_number)
row_values = [safe_string(data.get(header, "")) for header in schema.headers]
end_column = column_letter(len(schema.headers))
self._execute(
self.service.spreadsheets().values().update(
spreadsheetId=self.spreadsheet_id,
range=f"{quote_sheet_name(sheet_name)}!A{next_row_number}:{end_column}{next_row_number}",
valueInputOption="USER_ENTERED",
body={"values": [row_values]},
)
)
self._invalidate_runtime_caches()
values = dict(zip(schema.headers, row_values, strict=False))
return SheetRow(sheet_name=sheet_name, row_number=next_row_number, values=values)
def update_row(self, sheet_name: str, row_number: int, data: dict[str, Any]) -> SheetRow:
schema = self.get_schema(sheet_name)
current = self._get_row(sheet_name, row_number, schema.headers)
updated_values = current.values.copy()
for key, value in data.items():
header = resolve_header(key, schema.headers)
if header:
updated_values[header] = value
row_values = [safe_string(updated_values.get(header, "")) for header in schema.headers]
end_column = column_letter(len(schema.headers))
self._execute(
self.service.spreadsheets().values().update(
spreadsheetId=self.spreadsheet_id,
range=f"{quote_sheet_name(sheet_name)}!A{row_number}:{end_column}{row_number}",
valueInputOption="USER_ENTERED",
body={"values": [row_values]},
)
)
self._invalidate_runtime_caches()
values = dict(zip(schema.headers, row_values, strict=False))
return SheetRow(sheet_name=sheet_name, row_number=row_number, values=values)
def soft_delete_row(self, sheet_name: str, row_number: int) -> bool:
schema = self.get_schema(sheet_name)
status_header = next((header for header in schema.headers if normalize_key(header) in STATUS_COLUMNS), None)
if not status_header:
return False
self.update_row(sheet_name, row_number, {status_header: "deleted"})
return True
def delete_row(self, sheet_name: str, row_number: int) -> None:
self._execute(
self.service.spreadsheets().batchUpdate(
spreadsheetId=self.spreadsheet_id,
body={
"requests": [
{
"deleteDimension": {
"range": {
"sheetId": self.get_sheet_id(sheet_name),
"dimension": "ROWS",
"startIndex": row_number - 1,
"endIndex": row_number,
}
}
}
]
},
)
)
self._invalidate_runtime_caches()
def find_rows(self, sheet_name: str, filters: dict[str, Any], limit: int | None = None) -> list[SheetRow]:
logger.info("find_rows in sheet '%s' with filters: %s", sheet_name, filters)
rows = self.read_rows(sheet_name)
logger.info("Total rows read from sheet '%s': %s", sheet_name, len(rows))
if not filters:
result = rows[-(limit or self.settings.max_result_rows) :]
logger.info("No filters. Returning last %s rows.", len(result))
return result
index = self._get_search_index(sheet_name, rows)
candidate_indexes = candidate_row_indexes(index, filters)
logger.info(
"Search index narrowed sheet '%s' from %s to %s rows",
sheet_name,
len(rows),
len(candidate_indexes),
)
ranked_rows = [
(score_row_against_filters(rows[row_index], filters), rows[row_index])
for row_index in candidate_indexes
]
# Log top 5 scored rows for debugging search issues
debug_scores = sorted(
[(score, row.row_number, list(row.values.values())[:3]) for score, row in ranked_rows],
key=lambda item: item[0],
reverse=True,
)[:5]
logger.info("Top 5 matching scores: %s", debug_scores)
matched = [item for item in ranked_rows if item[0] > 0]
matched.sort(key=lambda item: (item[0], item[1].row_number), reverse=True)
if not matched:
logger.info("No rows matched the filters (score > 0)")
return []
top_score = matched[0][0]
second_score = matched[1][0] if len(matched) > 1 else 0
if top_score >= second_score + 40:
result = [matched[0][1]]
logger.info(
"Strong match found (top_score=%s, second_score=%s). Returning row %s",
top_score,
second_score,
result[0].row_number,
)
return result
minimum_score = top_score
filtered_rows = [row for score, row in matched if score == top_score]
result = filtered_rows[: (limit or self.settings.max_result_rows)]
logger.info("Returning %s rows with top score %s", len(result), minimum_score)
return result
def _get_search_index(self, sheet_name: str, rows: list[SheetRow]) -> SheetSearchIndex:
cache_key = self._sheet_cache_key(sheet_name)
cached = _SHEET_SEARCH_INDEX_CACHE.get(cache_key)
if cached is not None:
expires_at, index = cached
if monotonic() <= expires_at and index.row_count == len(rows):
return index
_SHEET_SEARCH_INDEX_CACHE.pop(cache_key, None)
index = build_sheet_search_index(rows)
_SHEET_SEARCH_INDEX_CACHE[cache_key] = (monotonic() + ROWS_CACHE_TTL_SECONDS, index)
return index
def _get_row(self, sheet_name: str, row_number: int, headers: list[str]) -> SheetRow:
end_column = column_letter(max(len(headers), 1))
response = self._execute(
self.service.spreadsheets().values().get(
spreadsheetId=self.spreadsheet_id,
range=f"{quote_sheet_name(sheet_name)}!A{row_number}:{end_column}{row_number}",
)
)
values = response.get("values", [[]])
return SheetRow(sheet_name=sheet_name, row_number=row_number, values=self._row_to_dict(headers, values[0]))
def next_empty_row_number(self, sheet_name: str) -> int:
schema = self.get_schema(sheet_name)
if not schema.headers:
return schema.header_row_number + 1
sheet_properties = self._sheet_properties(sheet_name)
total_row_count = int(sheet_properties.get("gridProperties", {}).get("rowCount", 0))
total_data_rows = max(total_row_count - schema.header_row_number, 1)
end_column = column_letter(len(schema.headers))
snapshot = self._execute(
self.service.spreadsheets().get(
spreadsheetId=self.spreadsheet_id,
ranges=[
(
f"{quote_sheet_name(sheet_name)}!"
f"A{schema.header_row_number + 1}:{end_column}{schema.header_row_number + total_data_rows}"
)
],
includeGridData=True,
fields="sheets.data.rowData.values(formattedValue,effectiveValue,userEnteredValue)",
)
)
row_data = snapshot.get("sheets", [{}])[0].get("data", [{}])[0].get("rowData", [])
return compute_next_empty_row_number_from_grid_rows(
row_data,
schema.header_row_number + 1,
total_data_rows=total_data_rows,
)
@staticmethod
def _row_to_dict(headers: list[str], raw_row: list[Any]) -> dict[str, Any]:
padded = list(raw_row) + [""] * max(0, len(headers) - len(raw_row))
return {header: padded[index] if index < len(padded) else "" for index, header in enumerate(headers)}
def _extract_validation_metadata(
self,
sheet_name: str,
header_row_index: int,
headers: list[str],
*,
end_row: int,
) -> tuple[dict[str, list[str]], dict[str, str]]:
if not headers:
return {}, {}
snapshot = self._sheet_grid_snapshot(sheet_name, end_row)
row_data = snapshot.get("sheets", [{}])[0].get("data", [{}])[0].get("rowData", [])
field_options: dict[str, list[str]] = {}
field_validation_messages: dict[str, str] = {}
for column_index, header in enumerate(headers):
rule = None
for row in row_data[header_row_index + 1 :]:
values = row.get("values", [])
if column_index >= len(values):
continue
candidate = values[column_index].get("dataValidation")
if candidate:
rule = candidate
break
if not rule:
continue
options = self._validation_options_from_rule(rule)
if options:
field_options[header] = options
input_message = safe_string(rule.get("inputMessage"))
if input_message:
field_validation_messages[header] = input_message
return field_options, field_validation_messages
def _validation_options_from_rule(self, rule: dict[str, Any]) -> list[str]:
condition = rule.get("condition", {})
condition_type = safe_string(condition.get("type"))
values = condition.get("values", [])
if condition_type == "ONE_OF_LIST":
return [
safe_string(item.get("userEnteredValue"))
for item in values
if safe_string(item.get("userEnteredValue"))
]
if condition_type == "ONE_OF_RANGE":
range_ref = normalize_validation_range_ref(safe_string(values[0].get("userEnteredValue")) if values else "")
if not range_ref:
return []
response = self._execute(
self.service.spreadsheets().values().get(
spreadsheetId=self.spreadsheet_id,
range=range_ref,
)
)
flattened: list[str] = []
for row in response.get("values", []):
flattened.extend(safe_string(cell) for cell in row if safe_string(cell))
return flattened
return []
def _prepare_target_row(self, sheet_name: str, schema: SheetSchema, target_row_number: int) -> None:
self._ensure_row_capacity(sheet_name, target_row_number)
if target_row_number <= schema.header_row_number + 1:
return
source_row_number = target_row_number - 1
sheet_id = self.get_sheet_id(sheet_name)
self._execute(
self.service.spreadsheets().batchUpdate(
spreadsheetId=self.spreadsheet_id,
body={
"requests": [
{
"copyPaste": {
"source": {
"sheetId": sheet_id,
"startRowIndex": source_row_number - 1,
"endRowIndex": source_row_number,
},
"destination": {
"sheetId": sheet_id,
"startRowIndex": target_row_number - 1,
"endRowIndex": target_row_number,
},
"pasteType": "PASTE_FORMAT",
"pasteOrientation": "NORMAL",
}
},
{
"copyPaste": {
"source": {
"sheetId": sheet_id,
"startRowIndex": source_row_number - 1,
"endRowIndex": source_row_number,
},
"destination": {
"sheetId": sheet_id,
"startRowIndex": target_row_number - 1,
"endRowIndex": target_row_number,
},
"pasteType": "PASTE_DATA_VALIDATION",
"pasteOrientation": "NORMAL",
}
},
]
},
)
)
def _ensure_row_capacity(self, sheet_name: str, target_row_number: int) -> None:
properties = self._sheet_properties(sheet_name)
current_row_count = int(properties.get("gridProperties", {}).get("rowCount", 0))
if target_row_number <= current_row_count:
return
rows_to_add = expanded_row_count(current_row_count, target_row_number) - current_row_count
self._execute(
self.service.spreadsheets().batchUpdate(
spreadsheetId=self.spreadsheet_id,
body={
"requests": [
{
"appendDimension": {
"sheetId": int(properties["sheetId"]),
"dimension": "ROWS",
"length": rows_to_add,
}
}
]
},
)
)
_SPREADSHEET_METADATA_CACHE.pop(self.spreadsheet_id, None)
def _invalidate_runtime_caches(self) -> None:
cache_key = self.spreadsheet_id
_SPREADSHEET_TITLE_CACHE.pop(cache_key, None)
_SPREADSHEET_METADATA_CACHE.pop(cache_key, None)
_SPREADSHEET_CATALOG_CACHE.pop(cache_key, None)
row_cache_prefix = f"{cache_key}:"
for row_cache_key in list(_SPREADSHEET_ROWS_CACHE):
if row_cache_key.startswith(row_cache_prefix):
_SPREADSHEET_ROWS_CACHE.pop(row_cache_key, None)
for schema_cache_key in list(_SPREADSHEET_SCHEMA_DETAILS_CACHE):
if schema_cache_key.startswith(row_cache_prefix):
_SPREADSHEET_SCHEMA_DETAILS_CACHE.pop(schema_cache_key, None)
for search_cache_key in list(_SHEET_SEARCH_INDEX_CACHE):
if search_cache_key.startswith(row_cache_prefix):
_SHEET_SEARCH_INDEX_CACHE.pop(search_cache_key, None)
def refresh_runtime_caches(self) -> None:
self._invalidate_runtime_caches()
async def export_sheet_bytes(self, sheet_name: str | None = None, file_format: str = "xlsx") -> bytes:
creds = self._credentials()
if creds and not creds.valid:
from google.auth.transport.requests import Request
await asyncio.to_thread(creds.refresh, Request())
token = creds.token
headers = {"Authorization": f"Bearer {token}"}
url = f"https://docs.google.com/spreadsheets/d/{self.spreadsheet_id}/export?format={file_format}"
if file_format == "pdf" and sheet_name:
try:
sheet_id = self.get_sheet_id(sheet_name)
url += f"&gid={sheet_id}"
url += "&size=A4&portrait=true&fitw=true&gridlines=true"
except Exception:
pass
async with httpx.AsyncClient(timeout=60.0) as client:
response = await client.get(url, headers=headers)
response.raise_for_status()
return response.content
def resolve_header(key: str, headers: list[str]) -> str | None:
normalized_key = normalize_key(key)
if not normalized_key:
return None
normalized_headers = {normalize_key(header): header for header in headers}
if normalized_key in normalized_headers:
return normalized_headers[normalized_key]
for normalized_header, header in normalized_headers.items():
if normalized_key in normalized_header or normalized_header in normalized_key:
return header
return None
def detect_google_credential_payload_kind(payload: dict[str, Any]) -> CredentialPayloadKind:
payload_type = safe_string(payload.get("type")).lower()
if payload_type == "service_account":
return "service_account"
if payload_type == "authorized_user":
return "authorized_user"
if {"refresh_token", "client_id", "client_secret", "token_uri"}.issubset(payload):
return "authorized_user"
if "installed" in payload or "web" in payload:
return "oauth_client"
return "unknown"
def build_google_credentials_from_payload(
payload: dict[str, Any],
*,
scopes: list[str],
source: str = "",
) -> Any:
payload_kind = detect_google_credential_payload_kind(payload)
if payload_kind == "service_account":
return Credentials.from_service_account_info(payload, scopes=scopes)
if payload_kind == "authorized_user":
return AuthorizedUserCredentials.from_authorized_user_info(payload, scopes=scopes)
if payload_kind == "oauth_client":
source_text = f" ({source})" if source else ""
raise RuntimeError(
"file Google credential yang dipakai masih OAuth client secret"
f"{source_text}. untuk bot headless, pakai service account atau authorized user token hasil login OAuth."
)
raise RuntimeError("format Google credential tidak dikenali.")
def row_matches_filters(row: SheetRow, filters: dict[str, Any]) -> bool:
return score_row_against_filters(row, filters) > 0
def score_row_against_filters(row: SheetRow, filters: dict[str, Any]) -> int:
headers = list(row.values.keys())
searchable_text = " ".join(
[*headers, *(safe_string(value) for value in row.values.values())]
).lower()
searchable_tokens = search_tokens(searchable_text)
total_score = 0
for key, expected in filters.items():
expected_text = safe_string(expected).lower()
if not expected_text:
continue
expected_tokens = search_tokens(expected_text)
header = resolve_header(key, headers)
if header:
actual_text = safe_string(row.values.get(header, "")).lower()
score = text_match_score(expected_text, expected_tokens, actual_text, search_tokens(actual_text))
if score <= 0:
return 0
total_score += score + 20
continue
score = text_match_score(expected_text, expected_tokens, searchable_text, searchable_tokens)
if score <= 0:
return 0
total_score += score
return total_score
def text_match_score(
expected_text: str,
expected_tokens: set[str],
actual_text: str,
actual_tokens: set[str],
) -> int:
if not expected_text or not actual_text:
return 0
normalized_expected = normalize_search_key(expected_text)
normalized_actual = normalize_search_key(actual_text)
if normalized_expected and normalized_expected == normalized_actual:
return 180
compact_expected = normalize_compact_search_key(expected_text)
compact_actual = normalize_compact_search_key(actual_text)
if compact_expected and compact_expected == compact_actual:
return 170
if compact_expected and compact_expected in compact_actual:
# A complete phrase found in the row is more precise than a token-set
# overlap. This keeps product variants such as "... 150 RACING" from
# tying with "... 150 STANDAR" just because both contain "RACING".
return 220
if expected_text in actual_text:
return 120
effective_expected_tokens = set(compact_search_token_sequence(expected_text)) or expected_tokens
effective_actual_tokens = set(compact_search_token_sequence(actual_text)) or actual_tokens
if not effective_expected_tokens:
return 0
overlap_count = fuzzy_token_overlap_count(effective_expected_tokens, effective_actual_tokens)
minimum_overlap = max(1, math.ceil(len(effective_expected_tokens) * 0.6))
if overlap_count < minimum_overlap:
return 0
score = overlap_count * 24
if overlap_count == len(effective_expected_tokens):
score += 40
return score
def fuzzy_token_overlap_count(expected_tokens: set[str], actual_tokens: set[str]) -> int:
unmatched_actual = set(actual_tokens)
overlap_count = 0
for expected in sorted(expected_tokens, key=len, reverse=True):
if expected in unmatched_actual:
unmatched_actual.remove(expected)
overlap_count += 1
continue
if len(expected) < 4 or expected.isdigit():
continue
best_match = ""
best_ratio = 0.0
for actual in unmatched_actual:
if len(actual) < 4 or actual[0] != expected[0]:
continue
ratio = SequenceMatcher(None, expected, actual).ratio()
if ratio > best_ratio:
best_ratio = ratio
best_match = actual
if best_match and best_ratio >= 0.84:
unmatched_actual.remove(best_match)
overlap_count += 1
return overlap_count
def normalize_validation_range_ref(range_ref: str) -> str:
cleaned = safe_string(range_ref)
if cleaned.startswith("="):
cleaned = cleaned[1:].strip()
return cleaned
def detect_sheet_table(
values: list[list[Any]],
*,
sample_limit: int,
) -> tuple[int, list[str], list[dict[str, Any]], list[str]]:
if not values:
return 0, [], [], []
candidate_scores = [
(index, score_header_candidate(values, index))
for index, row in enumerate(values)
if any(safe_string(cell) for cell in row)
]
best_index = 0
best_score = -1
for index, score in candidate_scores:
if score > best_score:
best_index = index
best_score = score
headers = sanitize_headers(values[best_index])
if not headers:
first_non_empty = next((index for index, row in enumerate(values) if any(safe_string(cell) for cell in row)), 0)
best_index = first_non_empty
headers = sanitize_headers(values[best_index])
all_rows: list[dict[str, Any]] = []
for raw_row in values[best_index + 1 :]:
row_values = GoogleSheetsClient._row_to_dict(headers, raw_row)
if any(safe_string(value) for value in row_values.values()):
all_rows.append(row_values)
context_lines = extract_context_lines(values[:best_index])
return best_index, headers, select_representative_sample_rows(all_rows, sample_limit), context_lines
def score_header_candidate(values: list[list[Any]], row_index: int) -> int:
row = values[row_index]
cells = [safe_string(cell) for cell in row]
non_empty = [cell for cell in cells if cell]
if len(non_empty) < 2:
return -100
normalized = [normalize_key(cell) for cell in non_empty if normalize_key(cell)]
unique_bonus = 6 if len(set(normalized)) == len(normalized) else 0
width_bonus = min(len(non_empty), 8) * 5
long_text_penalty = sum(8 for cell in non_empty if len(cell) > 45)
numeric_penalty = sum(3 for cell in non_empty if cell.replace("/", "").replace("-", "").isdigit())
data_bonus = 0
for next_row in values[row_index + 1 : row_index + 5]:
aligned = next_row[: len(cells)]
aligned_non_empty = sum(1 for cell in aligned[: len(non_empty)] if safe_string(cell))
if aligned_non_empty:
data_bonus += min(aligned_non_empty, 4) * 3
keyword_bonus = 0
header_keywords = {"no", "tanggal", "nama", "status", "catatan", "harga", "stok", "kode"}
if any(normalize_key(cell) in header_keywords for cell in non_empty):
keyword_bonus += 8
return width_bonus + unique_bonus + data_bonus + keyword_bonus - long_text_penalty - numeric_penalty
def sanitize_headers(raw_row: list[Any]) -> list[str]:
last_non_empty = -1
for index, cell in enumerate(raw_row):
if safe_string(cell):
last_non_empty = index
if last_non_empty == -1:
return []
headers: list[str] = []
seen: set[str] = set()
for index in range(last_non_empty + 1):
cell = raw_row[index]
header = safe_string(cell)
if not header:
header = f"Column_{index + 1}"
normalized = normalize_key(header) or f"column{index + 1}"
if normalized in seen:
header = f"{header}_{index + 1}"
normalized = normalize_key(header) or f"column{index + 1}"
seen.add(normalized)
headers.append(header)
return headers
def extract_context_lines(rows: list[list[Any]], limit: int = 3) -> list[str]:
context_lines: list[str] = []
for row in rows:
parts = [safe_string(cell) for cell in row if safe_string(cell)]
if not parts:
continue
context_lines.append(" | ".join(parts))
if len(context_lines) >= limit:
break
return context_lines
def select_representative_sample_rows(rows: list[dict[str, Any]], sample_limit: int) -> list[dict[str, Any]]:
if sample_limit <= 0 or not rows:
return []
if len(rows) <= sample_limit:
return rows
selected_indices: list[int] = []
if sample_limit == 1:
selected_indices = [0]
else:
max_index = len(rows) - 1
for position in range(sample_limit):
candidate_index = round(position * max_index / (sample_limit - 1))
if candidate_index not in selected_indices:
selected_indices.append(candidate_index)
selected_rows = [rows[index] for index in selected_indices]
if len(selected_rows) < sample_limit:
for row in rows:
if row not in selected_rows:
selected_rows.append(row)
if len(selected_rows) >= sample_limit:
break
return selected_rows[:sample_limit]
def count_detected_data_rows(values: list[list[Any]], header_row_index: int, headers: list[str]) -> int:
if not headers:
return 0
total = 0
for raw_row in values[header_row_index + 1 :]:
row_values = GoogleSheetsClient._row_to_dict(headers, raw_row)
if any(safe_string(value) for value in row_values.values()):
total += 1
return total
def compute_next_empty_row_number(values: list[list[Any]], start_row_number: int) -> int:
for index, row in enumerate(values):
if not any(safe_string(cell) for cell in row):
return start_row_number + index
return start_row_number + len(values)
def compute_next_empty_row_number_from_grid_rows(
row_data: list[dict[str, Any]],
start_row_number: int,
*,
total_data_rows: int | None = None,
) -> int:
total_rows = max(total_data_rows or len(row_data), len(row_data))
for index in range(total_rows):
row = row_data[index] if index < len(row_data) else {}
cells = row.get("values", [])
has_content = any(_grid_cell_has_value(cell) for cell in cells)
if not has_content:
return start_row_number + index
return start_row_number + total_rows
def _grid_cell_has_value(cell: dict[str, Any]) -> bool:
if safe_string(cell.get("formattedValue")):
return True
for payload_key in ("userEnteredValue", "effectiveValue"):
payload = cell.get(payload_key)
if not isinstance(payload, dict):
continue
if any(safe_string(value) for value in payload.values()):
return True
return False
def expanded_row_count(current_row_count: int, target_row_number: int, *, buffer_rows: int = 50) -> int:
if target_row_number <= current_row_count:
return current_row_count
minimum_required = max(target_row_number, 1)
return max(minimum_required, current_row_count + max(buffer_rows, 1))