217 lines
7.4 KiB
Python
217 lines
7.4 KiB
Python
"""Google Sheets integration - read/write member data via gspread."""
|
|
|
|
import gspread
|
|
from google.oauth2.service_account import Credentials
|
|
|
|
import config
|
|
|
|
# Scopes needed: read + write to Sheets
|
|
SCOPES = [
|
|
"https://www.googleapis.com/auth/spreadsheets",
|
|
"https://www.googleapis.com/auth/drive",
|
|
]
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Expected sheet columns (header row):
|
|
# Nimi | Organisatsioon | Meil | Discord | User ID | Sünnipäev |
|
|
# Telefon | Valdkond | Roll | Discordis synced? | Groupi lisatud?
|
|
#
|
|
# - Nimi : member's real name (used as Discord nickname)
|
|
# - Organisatsioon : organisation value - maps to a Discord role
|
|
# - Meil : email address (read-only for bot)
|
|
# - Discord : Discord username for initial matching
|
|
# - User ID : numeric Discord user ID (bot can populate this)
|
|
# - Sünnipäev : birthday date string (YYYY-MM-DD or MM-DD)
|
|
# - Telefon : phone number (read-only for bot)
|
|
# - Valdkond : field/area value - maps to a Discord role
|
|
# - Roll : role value - maps to a Discord role
|
|
# - Discordis synced? : TRUE/FALSE - bot writes this after confirming sync
|
|
# - Groupi lisatud? : group membership flag (managed externally)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
EXPECTED_HEADERS = [
|
|
"Nimi",
|
|
"Organisatsioon",
|
|
"Meil",
|
|
"Discord",
|
|
"User ID",
|
|
"Sünnipäev",
|
|
"Telefon",
|
|
"Valdkond",
|
|
"Roll",
|
|
"Discordis synced?",
|
|
"Groupi lisatud?",
|
|
]
|
|
|
|
_client: gspread.Client | None = None
|
|
_worksheet: gspread.Worksheet | None = None
|
|
|
|
# In-memory cache: list of dicts (one per row)
|
|
_cache: list[dict] = []
|
|
|
|
|
|
def _get_worksheet() -> gspread.Worksheet:
|
|
"""Authenticate and return the first worksheet of the configured sheet."""
|
|
global _client, _worksheet
|
|
creds = Credentials.from_service_account_file(config.GOOGLE_CREDS_PATH, scopes=SCOPES)
|
|
_client = gspread.authorize(creds)
|
|
spreadsheet = _client.open_by_key(config.SHEET_ID)
|
|
_worksheet = spreadsheet.sheet1
|
|
return _worksheet
|
|
|
|
|
|
def _ensure_headers(ws: gspread.Worksheet) -> None:
|
|
"""If the sheet is empty or missing headers, write them (headers are in row 1)."""
|
|
existing = ws.row_values(1)
|
|
if existing != EXPECTED_HEADERS:
|
|
for col_idx, header in enumerate(EXPECTED_HEADERS, start=1):
|
|
if col_idx > len(existing) or existing[col_idx - 1] != header:
|
|
ws.update_cell(1, col_idx, header)
|
|
|
|
|
|
def refresh() -> list[dict]:
|
|
"""Pull all rows from the sheet into the in-memory cache.
|
|
|
|
Returns the cache (list of dicts keyed by header names).
|
|
"""
|
|
global _cache
|
|
ws = _get_worksheet()
|
|
_ensure_headers(ws)
|
|
# head=1: row 1 is the header; row 2 is a formula/stats row - skip it
|
|
records = ws.get_all_records(head=1)
|
|
_cache = records[1:] # drop the formula row (row 2) from the cache
|
|
return _cache
|
|
|
|
|
|
def get_cache() -> list[dict]:
|
|
"""Return the current in-memory cache without re-querying."""
|
|
return _cache
|
|
|
|
|
|
def find_member_by_id(discord_id: int) -> dict | None:
|
|
"""Look up a member row by Discord user ID."""
|
|
for row in _cache:
|
|
raw = row.get("User ID", "")
|
|
if str(raw).strip() == str(discord_id):
|
|
return row
|
|
return None
|
|
|
|
|
|
def find_member_by_username(username: str) -> dict | None:
|
|
"""Look up a member row by Discord username (case-insensitive)."""
|
|
for row in _cache:
|
|
if str(row.get("Discord", "")).strip().lower() == username.lower():
|
|
return row
|
|
return None
|
|
|
|
|
|
def find_member(discord_id: int, username: str) -> dict | None:
|
|
"""Try ID first, fall back to username."""
|
|
return find_member_by_id(discord_id) or find_member_by_username(username)
|
|
|
|
|
|
# ---- Write helpers --------------------------------------------------------
|
|
|
|
def _row_index_for_member(discord_id: int | None = None, username: str | None = None) -> int | None:
|
|
"""Return the 1-based sheet row index for a member (header = row 2, data from row 3)."""
|
|
for idx, row in enumerate(_cache):
|
|
if discord_id and str(row.get("User ID", "")).strip() == str(discord_id):
|
|
return idx + 3 # +3 because header is row 2, data starts row 3, idx is 0-based
|
|
if username and str(row.get("Discord", "")).strip().lower() == username.lower():
|
|
return idx + 3
|
|
return None
|
|
|
|
|
|
def update_cell_for_member(
|
|
discord_id: int | None,
|
|
username: str | None,
|
|
column_name: str,
|
|
value: str,
|
|
) -> bool:
|
|
"""Write a value to a specific column for a member row.
|
|
|
|
Returns True if the write succeeded.
|
|
"""
|
|
ws = _worksheet or _get_worksheet()
|
|
row_idx = _row_index_for_member(discord_id=discord_id, username=username)
|
|
if row_idx is None:
|
|
return False
|
|
|
|
try:
|
|
col_idx = EXPECTED_HEADERS.index(column_name) + 1
|
|
except ValueError:
|
|
return False
|
|
|
|
ws.update([[value]], gspread.utils.rowcol_to_a1(row_idx, col_idx),
|
|
value_input_option="USER_ENTERED")
|
|
|
|
# Keep cache in sync
|
|
cache_idx = row_idx - 3
|
|
if 0 <= cache_idx < len(_cache):
|
|
_cache[cache_idx][column_name] = value
|
|
|
|
return True
|
|
|
|
|
|
def batch_set_synced(updates: list[tuple[int, bool]]) -> None:
|
|
"""Batch-write 'Discordis synced?' for multiple members in a single API call."""
|
|
ws = _worksheet or _get_worksheet()
|
|
col_idx = EXPECTED_HEADERS.index("Discordis synced?") + 1
|
|
cells = []
|
|
for discord_id, synced in updates:
|
|
row_idx = _row_index_for_member(discord_id=discord_id)
|
|
if row_idx is None:
|
|
continue
|
|
cells.append(gspread.Cell(row_idx, col_idx, "TRUE" if synced else "FALSE"))
|
|
cache_idx = row_idx - 3
|
|
if 0 <= cache_idx < len(_cache):
|
|
_cache[cache_idx]["Discordis synced?"] = "TRUE" if synced else "FALSE"
|
|
if cells:
|
|
ws.update_cells(cells, value_input_option="USER_ENTERED")
|
|
|
|
|
|
def set_user_id(username: str, discord_id: int) -> bool:
|
|
"""Write a Discord user ID for a row matched by Discord username."""
|
|
return update_cell_for_member(
|
|
discord_id=None,
|
|
username=username,
|
|
column_name="User ID",
|
|
value=str(discord_id),
|
|
)
|
|
|
|
|
|
def set_synced(discord_id: int, synced: bool) -> bool:
|
|
"""Mark a member as synced (TRUE) or not (FALSE)."""
|
|
return update_cell_for_member(
|
|
discord_id=discord_id,
|
|
username=None,
|
|
column_name="Discordis synced?",
|
|
value="TRUE" if synced else "FALSE",
|
|
)
|
|
|
|
|
|
def update_username(discord_id: int, new_username: str) -> bool:
|
|
"""Update the Discord column for a member (keeps sheet in sync with Discord)."""
|
|
return update_cell_for_member(
|
|
discord_id=discord_id,
|
|
username=None,
|
|
column_name="Discord",
|
|
value=new_username,
|
|
)
|
|
|
|
|
|
def add_new_member_row(username: str, discord_id: int) -> None:
|
|
"""Append a new row to the sheet with Discord username and User ID pre-filled.
|
|
|
|
All other columns are left empty for manual entry by an admin.
|
|
"""
|
|
ws = _worksheet or _get_worksheet()
|
|
row = [""] * len(EXPECTED_HEADERS)
|
|
row[EXPECTED_HEADERS.index("Discord")] = username
|
|
row[EXPECTED_HEADERS.index("User ID")] = str(discord_id)
|
|
row[EXPECTED_HEADERS.index("Discordis synced?")] = "FALSE"
|
|
ws.append_row(row, value_input_option="USER_ENTERED")
|
|
# Add to local cache so subsequent find_member() calls work in the same session
|
|
new_entry = {h: row[i] for i, h in enumerate(EXPECTED_HEADERS)}
|
|
_cache.append(new_entry)
|