forked from sass/tipibot
Merge branch 'master' of ssh://git.lapikud.ee:202/renkar/tipibot
# Conflicts: # bot.py
This commit is contained in:
Rene Arumetsa
@@ -32,7 +32,7 @@ QUESTS_DAILY: dict[str, QuestDef] = {
|
||||
"fish2": {"stat": "total_fish_caught", "goal": 2, "coins": 150, "exp": 20},
|
||||
"crime1": {"stat": "crimes_succeeded", "goal": 1, "coins": 200, "exp": 25},
|
||||
"earn1000": {"stat": "lifetime_earned", "goal": 1000, "coins": 150, "exp": 20},
|
||||
"give200": {"stat": "total_given", "goal": 200, "coins": 100, "exp": 15},
|
||||
"give200": {"stat": "total_given", "goal": 200, "coins": 250, "exp": 20},
|
||||
}
|
||||
|
||||
QUESTS_WEEKLY: dict[str, QuestDef] = {
|
||||
|
||||
190
core/fienta.py
Normal file
190
core/fienta.py
Normal file
@@ -0,0 +1,190 @@
|
||||
"""Fienta ticketing integration - authoritative Discord -> team mapping.
|
||||
|
||||
The tournament registration on Fienta collects, per competitor ticket, the
|
||||
player's Discord username (and sometimes Discord user ID), their team name
|
||||
(order-level, echoed onto each attendee), and the ticket type (which names the
|
||||
game). That gives a reliable Discord-identity -> team mapping the nickname-only
|
||||
registration sheet cannot, so this is the PRIMARY source for team-role sync,
|
||||
with :mod:`core.sheets` kept as a fallback.
|
||||
|
||||
Enabled by ``FIENTA_API_TOKEN`` + ``FIENTA_EVENT_ID``; when either is unset all
|
||||
caches stay empty and every getter is a no-op, so team sync silently falls back
|
||||
to the sheet. Only real team members get roles - competitor, coach/manager and
|
||||
substitute tickets are included; visitors, supporters, LAN-access, early-bird
|
||||
and waiting-list tickets are excluded.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
|
||||
import aiohttp
|
||||
|
||||
import config
|
||||
|
||||
log = logging.getLogger(__name__)
|
||||
|
||||
_API_BASE = "https://fienta.com/api/v1"
|
||||
_PAGE_SIZE = 1000 # Fienta's max per page; one page covers a whole tournament
|
||||
|
||||
# Ticket-type titles that represent an actual team member who should get a role.
|
||||
_INCLUDED_KEYWORDS = ("competitor", "coach", "manager", "substitute")
|
||||
# ...unless the title also matches one of these (visitors etc. are never players).
|
||||
_EXCLUDED_KEYWORDS = (
|
||||
"visitor", "supporter", "lan area", "early bird", "waiting list", "waitlist",
|
||||
)
|
||||
|
||||
# Attendee custom-field keys. Fienta appends the field id to the machine name;
|
||||
# these come from GET /events/{id}/custom-fields for event 176532.
|
||||
_F_DISCORD_USERNAME = "discord_username_134871"
|
||||
_F_DISCORD_USERID = "discord_user_id_135840"
|
||||
_F_TEAM_NAME = "team_name_134821"
|
||||
|
||||
_GAME_KEYWORDS: list[tuple[str, tuple[str, ...]]] = [
|
||||
("CS2", ("counter-strike", "counter strike", "cs2", "csgo", "cs:go")),
|
||||
("LoL", ("league of legends", "league", "lol")),
|
||||
]
|
||||
|
||||
# Caches, rebuilt by refresh_teams()/parse_tickets().
|
||||
_by_username: dict[str, str] = {} # discord username (lower) -> team name
|
||||
_by_userid: dict[str, str] = {} # discord user id (str) -> team name
|
||||
_team_game: dict[str, str] = {} # team name -> "CS2" / "LoL"
|
||||
_team_names: set[str] = set()
|
||||
|
||||
|
||||
def _norm_username(name: str) -> str:
|
||||
"""Normalise a Discord username for matching: lowercased, no leading @."""
|
||||
return name.strip().lower().lstrip("@")
|
||||
|
||||
|
||||
def _detect_game(ticket_type_title: str) -> str | None:
|
||||
"""Return the game code for a ticket-type title, or None if unrecognised."""
|
||||
title = ticket_type_title.lower()
|
||||
for game, keywords in _GAME_KEYWORDS:
|
||||
if any(k in title for k in keywords):
|
||||
return game
|
||||
return None
|
||||
|
||||
|
||||
def _is_included(ticket_type_title: str) -> bool:
|
||||
"""True when this ticket type is an actual team member (not a visitor etc.)."""
|
||||
title = ticket_type_title.lower()
|
||||
if any(k in title for k in _EXCLUDED_KEYWORDS):
|
||||
return False
|
||||
return any(k in title for k in _INCLUDED_KEYWORDS)
|
||||
|
||||
|
||||
def _game_divider_ids() -> dict[str, int]:
|
||||
"""Map game code -> divider role id, derived from ``config.TEAM_DIVIDERS``.
|
||||
|
||||
Reuses the same ``TEAM_DIVIDER_<SUFFIX>`` role IDs the sheet path uses: a
|
||||
suffix like ``cs2_2026`` contributes its id to game ``CS2``.
|
||||
"""
|
||||
out: dict[str, int] = {}
|
||||
for suffix, rid in config.TEAM_DIVIDERS.items():
|
||||
parts = suffix.split("_")
|
||||
if any(p in ("cs2", "cs", "csgo") for p in parts):
|
||||
out.setdefault("CS2", rid)
|
||||
if any(p in ("lol", "league") for p in parts):
|
||||
out.setdefault("LoL", rid)
|
||||
return out
|
||||
|
||||
|
||||
def parse_tickets(tickets: list[dict]) -> None:
|
||||
"""Rebuild the caches from a list of Fienta ticket objects.
|
||||
|
||||
Pure/synchronous so it can be unit-tested without hitting the API.
|
||||
"""
|
||||
global _by_username, _by_userid, _team_game, _team_names
|
||||
by_username: dict[str, str] = {}
|
||||
by_userid: dict[str, str] = {}
|
||||
team_game: dict[str, str] = {}
|
||||
for ticket in tickets:
|
||||
rows = ticket.get("rows") or []
|
||||
if not rows:
|
||||
continue
|
||||
row = rows[0]
|
||||
title = (row.get("ticket_type") or {}).get("title", "")
|
||||
if not _is_included(title):
|
||||
continue
|
||||
attendee = row.get("attendee") or {}
|
||||
team = (attendee.get(_F_TEAM_NAME) or "").strip()
|
||||
if not team:
|
||||
continue
|
||||
game = _detect_game(title)
|
||||
# Keep the first non-None game seen for a team (all its tickets agree).
|
||||
team_game[team] = game or team_game.get(team)
|
||||
uname = _norm_username(attendee.get(_F_DISCORD_USERNAME) or "")
|
||||
uid = (attendee.get(_F_DISCORD_USERID) or "").strip()
|
||||
if uname:
|
||||
by_username[uname] = team
|
||||
if uid.isdigit():
|
||||
by_userid[uid] = team
|
||||
_by_username = by_username
|
||||
_by_userid = by_userid
|
||||
_team_game = team_game
|
||||
_team_names = set(team_game)
|
||||
|
||||
|
||||
async def refresh_teams() -> set[str]:
|
||||
"""Fetch competitor tickets from Fienta and rebuild the caches.
|
||||
|
||||
No-op returning an empty set when ``FIENTA_API_TOKEN`` / ``FIENTA_EVENT_ID``
|
||||
are unset, so the caller transparently falls back to the sheet.
|
||||
"""
|
||||
if not config.FIENTA_API_TOKEN or not config.FIENTA_EVENT_ID:
|
||||
parse_tickets([])
|
||||
return set()
|
||||
|
||||
url = f"{_API_BASE}/events/{config.FIENTA_EVENT_ID}/tickets"
|
||||
headers = {"Authorization": f"Bearer {config.FIENTA_API_TOKEN}"}
|
||||
tickets: list[dict] = []
|
||||
async with aiohttp.ClientSession() as session:
|
||||
page = 1
|
||||
while True:
|
||||
params = {"attendees": "true", "per_page": str(_PAGE_SIZE), "page": str(page)}
|
||||
async with session.get(url, headers=headers, params=params) as resp:
|
||||
resp.raise_for_status()
|
||||
data = await resp.json()
|
||||
batch = data.get("tickets") or []
|
||||
tickets.extend(batch)
|
||||
if len(batch) < _PAGE_SIZE:
|
||||
break
|
||||
page += 1
|
||||
|
||||
parse_tickets(tickets)
|
||||
log.info(
|
||||
"Fienta: %d tickets -> %d teams, %d discord usernames, %d discord ids",
|
||||
len(tickets), len(_team_names), len(_by_username), len(_by_userid),
|
||||
)
|
||||
return set(_team_names)
|
||||
|
||||
|
||||
def get_team_for_username(username: str) -> str | None:
|
||||
"""Team the given Discord username is registered on, or None."""
|
||||
return _by_username.get(_norm_username(username))
|
||||
|
||||
|
||||
def get_team_for_userid(user_id: int) -> str | None:
|
||||
"""Team the given Discord user ID is registered on, or None (IDs are sparse)."""
|
||||
return _by_userid.get(str(user_id))
|
||||
|
||||
|
||||
def all_team_names() -> set[str]:
|
||||
"""Every team name seen in the included Fienta tickets."""
|
||||
return set(_team_names)
|
||||
|
||||
|
||||
def get_team_game(team: str) -> str | None:
|
||||
"""Game code ("CS2"/"LoL") for a team, or None."""
|
||||
return _team_game.get(team)
|
||||
|
||||
|
||||
def get_team_dividers() -> dict[str, int]:
|
||||
"""{team name -> divider role id}, via each team's game and config dividers."""
|
||||
game_div = _game_divider_ids()
|
||||
return {
|
||||
team: game_div[game]
|
||||
for team, game in _team_game.items()
|
||||
if game and game in game_div
|
||||
}
|
||||
@@ -11,9 +11,32 @@ from zoneinfo import ZoneInfo
|
||||
import discord
|
||||
|
||||
import config
|
||||
from . import sheets
|
||||
from . import fienta, sheets
|
||||
|
||||
log = logging.getLogger(__name__)
|
||||
|
||||
|
||||
def resolve_team(member: discord.Member) -> str | None:
|
||||
"""Team a member is registered on: Fienta first (by ID, then username),
|
||||
then the sheet by username. Fienta is authoritative; the sheet is fallback."""
|
||||
return (
|
||||
fienta.get_team_for_userid(member.id)
|
||||
or fienta.get_team_for_username(member.name)
|
||||
or sheets.get_team_for_username(member.name)
|
||||
)
|
||||
|
||||
|
||||
def all_managed_team_names() -> set[str]:
|
||||
"""Union of every team name from Fienta and the sheet - the only role names
|
||||
team sync ever adds or removes."""
|
||||
return fienta.all_team_names() | sheets.all_team_names()
|
||||
|
||||
|
||||
def team_dividers() -> dict[str, int]:
|
||||
"""{team -> divider role id} merged from both sources; Fienta wins on overlap."""
|
||||
merged = dict(sheets.get_team_dividers())
|
||||
merged.update(fienta.get_team_dividers())
|
||||
return merged
|
||||
_PLACEHOLDER = {"-", "x", "n/a", "none", "ei"}
|
||||
_TZ = ZoneInfo("Europe/Tallinn")
|
||||
|
||||
@@ -48,7 +71,6 @@ class SyncResult:
|
||||
"""Tracks what happened during a sync operation."""
|
||||
nickname_changed: bool = False
|
||||
roles_added: list[str] = field(default_factory=list)
|
||||
roles_removed: list[str] = field(default_factory=list)
|
||||
birthday_soon: bool = False
|
||||
birthday_today: bool = False
|
||||
not_found: bool = False
|
||||
@@ -57,7 +79,36 @@ class SyncResult:
|
||||
|
||||
@property
|
||||
def changed(self) -> bool:
|
||||
return self.nickname_changed or self.roles_added or self.roles_removed
|
||||
return self.nickname_changed or self.roles_added
|
||||
|
||||
|
||||
@dataclass
|
||||
class TeamSyncResult:
|
||||
"""What happened when syncing one member's tournament team role."""
|
||||
added: str | None = None # team role name granted, if any
|
||||
removed: list[str] = field(default_factory=list) # stale team roles taken away
|
||||
created: str | None = None # team role name auto-created in the guild, if any
|
||||
divider_added: str | None = None # game divider role granted as participant tag
|
||||
divider_removed: list[str] = field(default_factory=list) # stale divider roles taken away
|
||||
errors: list[str] = field(default_factory=list)
|
||||
|
||||
@property
|
||||
def changed(self) -> bool:
|
||||
return bool(self.added or self.removed or self.divider_added or self.divider_removed)
|
||||
|
||||
|
||||
@dataclass
|
||||
class TeamSyncSummary:
|
||||
"""Aggregate outcome of a whole-guild team-role sync."""
|
||||
scanned: int = 0
|
||||
assigned: int = 0
|
||||
removed: int = 0
|
||||
created: list[str] = field(default_factory=list)
|
||||
positioned: int = 0 # team roles moved under a divider
|
||||
divider_assigned: int = 0 # members given their game divider role
|
||||
divider_removed: int = 0 # stale game divider roles taken away
|
||||
changes: list[str] = field(default_factory=list) # human-readable per-member lines
|
||||
errors: list[str] = field(default_factory=list)
|
||||
|
||||
|
||||
def _format_nickname(full_name: str) -> str:
|
||||
@@ -224,6 +275,272 @@ async def sync_member(
|
||||
return result
|
||||
|
||||
|
||||
async def sync_team_role(
|
||||
member: discord.Member,
|
||||
guild: discord.Guild,
|
||||
) -> TeamSyncResult:
|
||||
"""Give one member their tournament team role from the registration sheet.
|
||||
|
||||
Roster-INDEPENDENT: unlike :func:`sync_member` this does not touch the
|
||||
internal member sheet at all. It matches the member's Discord username
|
||||
against the team sheet caches (populated by ``sheets.refresh_teams``) and:
|
||||
|
||||
* grants the role for the team they're registered on (auto-creating that
|
||||
role in the guild when it does not exist yet);
|
||||
* removes any *other* team role they still carry (left / switched teams);
|
||||
* grants their game's divider role as a participant tag (and strips any
|
||||
other configured divider role they still carry, i.e. switched game).
|
||||
|
||||
Only team role NAMES present in the team sheet and the configured divider
|
||||
role IDs (``config.TEAM_DIVIDERS``) are ever added or removed, so no
|
||||
unrelated role is ever at risk. When ``TEAM_SHEET_ID`` is unset the caches
|
||||
are empty and this is a no-op returning an unchanged result.
|
||||
"""
|
||||
result = TeamSyncResult()
|
||||
|
||||
team_name = resolve_team(member)
|
||||
all_teams = all_managed_team_names()
|
||||
if not all_teams:
|
||||
return result # feature switched off (no Fienta token and no team sheet)
|
||||
|
||||
desired: discord.Role | None = None
|
||||
if team_name:
|
||||
desired = discord.utils.get(guild.roles, name=team_name)
|
||||
if desired is None:
|
||||
try:
|
||||
desired = await guild.create_role(name=team_name, reason="Team sync: uus tiim")
|
||||
result.created = team_name
|
||||
log.info("Created team role %r for %s", team_name, member)
|
||||
except discord.Forbidden:
|
||||
result.errors.append(f"Tiimirolli '{team_name}' loomiseks puudub õigus")
|
||||
except discord.HTTPException as e:
|
||||
result.errors.append(f"Tiimirolli '{team_name}' loomine ebaõnnestus: {e}")
|
||||
|
||||
# Team roles held but no longer registered for (switched teams / dropped out).
|
||||
to_remove = [r for r in member.roles if r.name in all_teams and r.name != team_name]
|
||||
|
||||
if desired is not None and desired not in member.roles:
|
||||
try:
|
||||
await member.add_roles(desired, reason="Team sync")
|
||||
result.added = desired.name
|
||||
except discord.Forbidden:
|
||||
log.debug("No permission to add team role for %s, skipping", member)
|
||||
except discord.HTTPException as e:
|
||||
result.errors.append(f"Tiimirolli viga kasutajale {member}: {e}")
|
||||
|
||||
if to_remove:
|
||||
try:
|
||||
await member.remove_roles(*to_remove, reason="Team sync: tiim vahetus")
|
||||
result.removed = [r.name for r in to_remove]
|
||||
except discord.Forbidden:
|
||||
log.debug("No permission to remove team roles for %s, skipping", member)
|
||||
except discord.HTTPException as e:
|
||||
result.errors.append(f"Tiimirolli eemaldamise viga kasutajale {member}: {e}")
|
||||
|
||||
# --- Participant divider role (the game's divider role doubles as a tag) ---
|
||||
# Grant the divider role for the member's team's game, and strip any other
|
||||
# configured divider role (switched game / dropped out). Matched by ID, so
|
||||
# only the roles named in config.TEAM_DIVIDERS are ever touched.
|
||||
divider_ids = set(config.TEAM_DIVIDERS.values())
|
||||
want_divider_id = team_dividers().get(team_name) if team_name else None
|
||||
want_divider = guild.get_role(want_divider_id) if want_divider_id else None
|
||||
|
||||
if want_divider is not None and want_divider not in member.roles:
|
||||
try:
|
||||
await member.add_roles(want_divider, reason="Team sync: mänguosaleja")
|
||||
result.divider_added = want_divider.name
|
||||
except discord.Forbidden:
|
||||
log.debug("No permission to add divider role for %s, skipping", member)
|
||||
except discord.HTTPException as e:
|
||||
result.errors.append(f"Eraldajarolli viga kasutajale {member}: {e}")
|
||||
|
||||
stale_dividers = [
|
||||
r for r in member.roles if r.id in divider_ids and r.id != want_divider_id
|
||||
]
|
||||
if stale_dividers:
|
||||
try:
|
||||
await member.remove_roles(*stale_dividers, reason="Team sync: mäng vahetus")
|
||||
result.divider_removed = [r.name for r in stale_dividers]
|
||||
except discord.Forbidden:
|
||||
log.debug("No permission to remove divider roles for %s, skipping", member)
|
||||
except discord.HTTPException as e:
|
||||
result.errors.append(f"Eraldajarolli eemaldamise viga kasutajale {member}: {e}")
|
||||
|
||||
return result
|
||||
|
||||
|
||||
def plan_team_positions(
|
||||
ordered_names: list[str],
|
||||
managed: set[str],
|
||||
placements: dict[str, list[str]],
|
||||
) -> dict[str, int]:
|
||||
"""Work out the new position of every team role that needs to move.
|
||||
|
||||
``ordered_names`` is every role name ascending by Discord position, so index
|
||||
0 is the bottom of the role list (@everyone) and the last entry is the top.
|
||||
``placements`` maps a divider role name to the team roles that belong under
|
||||
it. Roles in ``managed`` are never moved (integration-managed roles, and
|
||||
anything at or above the bot's own role, cannot be repositioned).
|
||||
|
||||
Teams are pulled out of the list and re-inserted immediately below their
|
||||
divider, sorted so they read alphabetically top-to-bottom in the Discord UI.
|
||||
Everything else keeps its relative order; a role only appears in the result
|
||||
when its position actually changed. Roles are matched by name, mirroring the
|
||||
rest of the team sync - with duplicate role names the lowest one wins.
|
||||
"""
|
||||
present = set(ordered_names)
|
||||
placeable: dict[str, list[str]] = {}
|
||||
for divider, teams in placements.items():
|
||||
if divider not in present or divider in managed:
|
||||
continue
|
||||
# Descending here because the list is bottom-up: reversing it renders
|
||||
# alphabetically downwards from the divider.
|
||||
block = sorted({t for t in teams if t in present and t not in managed}, reverse=True)
|
||||
if block:
|
||||
placeable[divider] = block
|
||||
if not placeable:
|
||||
return {}
|
||||
|
||||
movable = {t for block in placeable.values() for t in block}
|
||||
remaining = [n for n in ordered_names if n not in movable]
|
||||
for divider, block in placeable.items():
|
||||
idx = remaining.index(divider)
|
||||
remaining[idx:idx] = block
|
||||
|
||||
old_pos = {name: i for i, name in enumerate(ordered_names)}
|
||||
return {
|
||||
name: i
|
||||
for i, name in enumerate(remaining)
|
||||
if name not in managed and old_pos.get(name) != i
|
||||
}
|
||||
|
||||
|
||||
async def apply_team_role_positions(
|
||||
guild: discord.Guild,
|
||||
log: logging.Logger = log,
|
||||
) -> tuple[int, list[str]]:
|
||||
"""Move every team role directly beneath its configured divider role.
|
||||
|
||||
Driven by the ``TEAM_DIVIDER_*`` config: teams whose sheet section matched
|
||||
one are placed under that divider role, the rest are left exactly where they
|
||||
are. Dividers are configured by role ID, so a rename never breaks placement;
|
||||
the ID is resolved to the role's current name here, and the ordering maths
|
||||
downstream is name-based. Returns ``(roles_moved, errors)``; a no-op returns
|
||||
``(0, [])``.
|
||||
"""
|
||||
team_divider_ids = team_dividers() # {team: divider role ID}, both sources
|
||||
if not team_divider_ids:
|
||||
return 0, [] # no dividers configured, or nothing matched
|
||||
|
||||
errors: list[str] = []
|
||||
# Resolve each configured divider ID to its role once, then key placements by
|
||||
# that role's current name for the name-based positioning maths below.
|
||||
placements: dict[str, list[str]] = {}
|
||||
resolved: dict[int, discord.Role | None] = {}
|
||||
for team, divider_id in team_divider_ids.items():
|
||||
if divider_id not in resolved:
|
||||
resolved[divider_id] = guild.get_role(divider_id)
|
||||
if resolved[divider_id] is None:
|
||||
errors.append(f"Eraldajarolli ID {divider_id} ei leitud serverist")
|
||||
divider = resolved[divider_id]
|
||||
if divider is not None:
|
||||
placements.setdefault(divider.name, []).append(team)
|
||||
if not placements:
|
||||
return 0, errors
|
||||
|
||||
by_name: dict[str, discord.Role] = {}
|
||||
for role in sorted(guild.roles, key=lambda r: r.position):
|
||||
by_name.setdefault(role.name, role)
|
||||
|
||||
# The bot can only reorder roles strictly below its own highest role.
|
||||
bot_top = max((r.position for r in guild.me.roles), default=0)
|
||||
for divider in list(placements):
|
||||
role = by_name.get(divider)
|
||||
if role is None:
|
||||
errors.append(f"Eraldajarolli '{divider}' ei leitud serverist")
|
||||
placements.pop(divider)
|
||||
elif role.position >= bot_top:
|
||||
errors.append(f"Eraldaja '{divider}' on boti rollist kõrgemal - ei saa liigutada")
|
||||
placements.pop(divider)
|
||||
if not placements:
|
||||
return 0, errors
|
||||
|
||||
ordered = sorted(guild.roles, key=lambda r: r.position)
|
||||
ordered_names = [r.name for r in ordered]
|
||||
managed = {r.name for r in ordered if r.managed or r.position >= bot_top}
|
||||
plan = plan_team_positions(ordered_names, managed, placements)
|
||||
if not plan:
|
||||
return 0, errors # already in the right place
|
||||
|
||||
positions = {by_name[n]: p for n, p in plan.items() if n in by_name}
|
||||
try:
|
||||
await guild.edit_role_positions(positions=positions)
|
||||
except discord.Forbidden:
|
||||
errors.append("Tiimirollide järjestamiseks puudub õigus")
|
||||
return 0, errors
|
||||
except discord.HTTPException as e:
|
||||
errors.append(f"Tiimirollide järjestamine ebaõnnestus: {e}")
|
||||
# 50013 here despite Manage Roles usually means a role in the batch sits
|
||||
# at/above the bot's top role. Log the batch vs bot_top to pinpoint it.
|
||||
log.warning(
|
||||
"edit_role_positions failed (%s); bot_top=%d; batch=%s",
|
||||
e, bot_top,
|
||||
sorted(
|
||||
((r.name, r.position, target) for r, target in positions.items()),
|
||||
key=lambda x: -x[1],
|
||||
),
|
||||
)
|
||||
return 0, errors
|
||||
log.info("Positioned %d team role(s) under their dividers", len(positions))
|
||||
return len(positions), errors
|
||||
|
||||
|
||||
async def sync_all_team_roles(
|
||||
guild: discord.Guild,
|
||||
log: logging.Logger = log,
|
||||
) -> TeamSyncSummary:
|
||||
"""Run :func:`sync_team_role` for every human member of ``guild``.
|
||||
|
||||
Assumes the team caches are already fresh (caller runs ``refresh_teams``
|
||||
first). Returns an aggregate summary for reporting.
|
||||
"""
|
||||
summary = TeamSyncSummary()
|
||||
for member in guild.members:
|
||||
if member.bot:
|
||||
continue
|
||||
summary.scanned += 1
|
||||
res = await sync_team_role(member, guild)
|
||||
if res.created:
|
||||
summary.created.append(res.created)
|
||||
if res.errors:
|
||||
summary.errors.extend(res.errors)
|
||||
if res.added:
|
||||
summary.assigned += 1
|
||||
if res.removed:
|
||||
summary.removed += len(res.removed)
|
||||
if res.divider_added:
|
||||
summary.divider_assigned += 1
|
||||
if res.divider_removed:
|
||||
summary.divider_removed += len(res.divider_removed)
|
||||
if res.changed:
|
||||
bits: list[str] = []
|
||||
if res.added:
|
||||
bits.append(f"+{res.added}")
|
||||
if res.removed:
|
||||
bits.append("-" + ", -".join(res.removed))
|
||||
if res.divider_added:
|
||||
bits.append(f"+[{res.divider_added}]")
|
||||
if res.divider_removed:
|
||||
bits.append("-[" + "], -[".join(res.divider_removed) + "]")
|
||||
summary.changes.append(f"{member.display_name}: {', '.join(bits)}")
|
||||
|
||||
# Placement runs after the grant/remove pass so roles created this run are
|
||||
# positioned in the same sweep rather than waiting for the next one.
|
||||
summary.positioned, position_errors = await apply_team_role_positions(guild, log)
|
||||
summary.errors.extend(position_errors)
|
||||
return summary
|
||||
|
||||
|
||||
async def announce_birthday(
|
||||
member: discord.Member,
|
||||
bot: discord.Client,
|
||||
|
||||
247
core/sheets.py
247
core/sheets.py
@@ -8,6 +8,8 @@ Pure-cache helpers (get_cache, find_*) remain sync.
|
||||
|
||||
import asyncio
|
||||
import logging
|
||||
import re
|
||||
from dataclasses import dataclass, field
|
||||
|
||||
import gspread
|
||||
from google.oauth2.service_account import Credentials
|
||||
@@ -250,3 +252,248 @@ def _add_new_member_row_sync(username: str, discord_id: int) -> None:
|
||||
async def add_new_member_row(username: str, discord_id: int) -> None:
|
||||
"""Append a new row pre-filled with Discord username and User ID (non-blocking)."""
|
||||
await asyncio.to_thread(_add_new_member_row_sync, username, discord_id)
|
||||
|
||||
|
||||
# ===========================================================================
|
||||
# Team registration sheet (a SEPARATE spreadsheet, config.TEAM_SHEET_ID)
|
||||
# ---------------------------------------------------------------------------
|
||||
# Unlike the member roster this sheet is NOT a single clean table: it stacks
|
||||
# several game sections (CS2, LoL, ...) - each with merged title/description
|
||||
# rows, its own header row, and a block of team rows - across one or more tabs.
|
||||
# So we read raw cell values (get_all_values) and scan for header rows rather
|
||||
# than relying on get_all_records, which requires one rectangular table.
|
||||
#
|
||||
# Each team's players live in one "Lineup" cell, comma-joined, every nickname
|
||||
# suffixed with a citizenship marker like "(EST)". Those nicknames ARE the
|
||||
# players' Discord usernames; the citizenship is used only as a delimiter
|
||||
# (a nickname may itself contain commas) and then discarded.
|
||||
# ===========================================================================
|
||||
|
||||
# Matches a citizenship marker such as "(EST)" / "(LAT)". Used to split a
|
||||
# lineup cell into individual players, then thrown away.
|
||||
_CITIZENSHIP_RE = re.compile(r"\(\s*[A-Za-z]{2,4}\s*\)")
|
||||
|
||||
_TEAM_NAME_HEADER = "team name"
|
||||
_LINEUP_HEADER_PREFIX = "lineup"
|
||||
|
||||
# Team-sheet caches (mirrors the member-roster cache above)
|
||||
_team_roster: dict[str, list[str]] = {} # team name -> [Discord username, ...]
|
||||
_team_by_username: dict[str, str] = {} # normalized username -> team name
|
||||
_team_names: set[str] = set() # universe of all team names
|
||||
_team_divider: dict[str, int] = {} # team name -> divider role ID
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class TeamSection:
|
||||
"""One game's block of teams within a tab, plus the title row above it.
|
||||
|
||||
The title ("TipiLAN 2026 CS2 Registration Log") is the only thing that says
|
||||
which game and year a block belongs to - the header and team rows below it
|
||||
carry neither - so it is what :func:`resolve_divider` matches against.
|
||||
"""
|
||||
title: str = ""
|
||||
rosters: dict[str, list[str]] = field(default_factory=dict)
|
||||
|
||||
|
||||
def parse_lineup(cell: str) -> list[str]:
|
||||
"""Split one 'Lineup' cell into player nicknames (Discord usernames).
|
||||
|
||||
Players are delimited by their trailing citizenship marker, e.g.
|
||||
'TFT (EST), nqmm (EST), sn1rk (EST)' -> ['TFT', 'nqmm', 'sn1rk']
|
||||
Splitting on the marker (not on commas) keeps nicknames that themselves
|
||||
contain commas or semicolons intact. Falls back to comma-splitting when a
|
||||
cell carries no citizenship markers at all.
|
||||
"""
|
||||
cell = str(cell).strip()
|
||||
if not cell:
|
||||
return []
|
||||
names: list[str] = []
|
||||
last = 0
|
||||
matched = False
|
||||
for m in _CITIZENSHIP_RE.finditer(cell):
|
||||
matched = True
|
||||
chunk = cell[last:m.start()].strip().strip(",;").strip()
|
||||
if chunk:
|
||||
names.append(chunk)
|
||||
last = m.end()
|
||||
if not matched:
|
||||
return [p.strip() for p in cell.split(",") if p.strip()]
|
||||
tail = cell[last:].strip().strip(",;").strip() # stray name after last marker
|
||||
if tail:
|
||||
names.append(tail)
|
||||
return names
|
||||
|
||||
|
||||
def _find_col(row: list, matches) -> int | None:
|
||||
for idx, cell in enumerate(row):
|
||||
if matches(str(cell)):
|
||||
return idx
|
||||
return None
|
||||
|
||||
|
||||
def _cell(row: list, idx: int) -> str:
|
||||
return str(row[idx]) if 0 <= idx < len(row) else ""
|
||||
|
||||
|
||||
def _merged_title(row: list) -> str | None:
|
||||
"""Return a row's lone non-empty cell - a merged section title - else None.
|
||||
|
||||
A fully blank row returns None rather than "", so blank separators between
|
||||
sections do not wipe the title we are holding for the next header row.
|
||||
"""
|
||||
values = [str(c).strip() for c in row if str(c).strip()]
|
||||
return values[0] if len(values) == 1 else None
|
||||
|
||||
|
||||
def parse_team_sections(rows: list[list]) -> list[TeamSection]:
|
||||
"""Extract each game's block of teams from a tab's raw rows, with its title.
|
||||
|
||||
Scans for every header row that has both a 'Team Name' and a 'Lineup...'
|
||||
column, then reads the rows beneath it (using that section's own column
|
||||
positions) until the team-name column goes blank or a new header appears.
|
||||
Handles multiple stacked sections with differing layouts in one tab, and
|
||||
tags each with the most recent merged title row seen above it.
|
||||
"""
|
||||
sections: list[TeamSection] = []
|
||||
title = ""
|
||||
i, n = 0, len(rows)
|
||||
while i < n:
|
||||
name_col = _find_col(rows[i], lambda c: c.strip().lower() == _TEAM_NAME_HEADER)
|
||||
lineup_col = _find_col(rows[i], lambda c: c.strip().lower().startswith(_LINEUP_HEADER_PREFIX))
|
||||
if name_col is None or lineup_col is None:
|
||||
if (text := _merged_title(rows[i])) is not None:
|
||||
# Accumulate every single-cell row above the header, not just the
|
||||
# last one: sheets stack the "TipiLAN 2026 CS2" title above notice
|
||||
# rows ("If a team withdraws..."), and the later notices must not
|
||||
# clobber the title whose keywords resolve_divider needs.
|
||||
title = f"{title} {text}".strip() if title else text
|
||||
i += 1
|
||||
continue
|
||||
rosters: dict[str, list[str]] = {}
|
||||
i += 1 # move past the header into the data block
|
||||
while i < n:
|
||||
team = _cell(rows[i], name_col).strip()
|
||||
if not team or team.lower() == _TEAM_NAME_HEADER:
|
||||
break # blank team-name (or a new header) ends this section
|
||||
players = parse_lineup(_cell(rows[i], lineup_col))
|
||||
if players:
|
||||
rosters.setdefault(team, []).extend(players)
|
||||
i += 1
|
||||
if rosters:
|
||||
sections.append(TeamSection(title=title, rosters=rosters))
|
||||
title = "" # consumed - do not leak it onto the next section
|
||||
return sections
|
||||
|
||||
|
||||
def parse_team_rosters(rows: list[list]) -> dict[str, list[str]]:
|
||||
"""Flatten every section in a tab into {team_name: [nickname, ...]}."""
|
||||
rosters: dict[str, list[str]] = {}
|
||||
for section in parse_team_sections(rows):
|
||||
for team, players in section.rosters.items():
|
||||
rosters.setdefault(team, []).extend(players)
|
||||
return rosters
|
||||
|
||||
|
||||
def resolve_divider(title: str, dividers: dict[str, int] | None = None) -> int | None:
|
||||
"""Return the divider role ID configured for a section title, if any.
|
||||
|
||||
A ``TEAM_DIVIDER_<SUFFIX>`` entry matches when every underscore-separated
|
||||
part of its suffix appears as a whole word in the title, so ``CS2`` matches
|
||||
a CS2 section from any year while ``CS2_2026`` matches only the 2026 one.
|
||||
The most specific match (most parts) wins, which lets a year-scoped entry
|
||||
override a general one for the same game.
|
||||
"""
|
||||
if dividers is None:
|
||||
dividers = config.TEAM_DIVIDERS
|
||||
haystack = title.lower()
|
||||
best_id: int | None = None
|
||||
best_parts = 0
|
||||
for suffix, role_id in dividers.items():
|
||||
parts = [p for p in suffix.split("_") if p]
|
||||
if not parts or len(parts) <= best_parts:
|
||||
continue
|
||||
if all(re.search(rf"\b{re.escape(p)}\b", haystack) for p in parts):
|
||||
best_id, best_parts = role_id, len(parts)
|
||||
return best_id
|
||||
|
||||
|
||||
def build_username_index(rosters: dict[str, list[str]]) -> dict[str, str]:
|
||||
"""Invert {team: [usernames]} into {normalized username: team}.
|
||||
|
||||
If the same username appears on two teams the last one wins and a warning
|
||||
is logged (a person is expected to be on exactly one team).
|
||||
"""
|
||||
index: dict[str, str] = {}
|
||||
for team, players in rosters.items():
|
||||
for player in players:
|
||||
key = player.strip().lower()
|
||||
if not key:
|
||||
continue
|
||||
if key in index and index[key] != team:
|
||||
log.warning(
|
||||
"Player %r appears on both %r and %r; using %r",
|
||||
player, index[key], team, team,
|
||||
)
|
||||
index[key] = team
|
||||
return index
|
||||
|
||||
|
||||
def _refresh_teams_sync() -> dict[str, list[str]]:
|
||||
global _team_roster, _team_by_username, _team_names, _team_divider
|
||||
creds = Credentials.from_service_account_file(config.GOOGLE_CREDS_PATH, scopes=SCOPES)
|
||||
client = gspread.authorize(creds)
|
||||
spreadsheet = client.open_by_key(config.TEAM_SHEET_ID)
|
||||
|
||||
rosters: dict[str, list[str]] = {}
|
||||
dividers: dict[str, int] = {}
|
||||
for ws in spreadsheet.worksheets():
|
||||
for section in parse_team_sections(ws.get_all_values()):
|
||||
# Match on the tab name too ("CS2"/"LoL"): it names the game reliably
|
||||
# even when the game/year title row is shadowed by notice rows, while
|
||||
# the section title still supplies the year for year-scoped dividers.
|
||||
divider = resolve_divider(f"{ws.title} {section.title}")
|
||||
for team, players in section.rosters.items():
|
||||
rosters.setdefault(team, []).extend(players)
|
||||
if divider:
|
||||
dividers[team] = divider
|
||||
|
||||
_team_roster = rosters
|
||||
_team_by_username = build_username_index(rosters)
|
||||
_team_names = set(rosters)
|
||||
_team_divider = dividers
|
||||
return rosters
|
||||
|
||||
|
||||
async def refresh_teams() -> dict[str, list[str]]:
|
||||
"""Reload the team registration sheet into the in-memory team caches.
|
||||
|
||||
No-op returning {} when TEAM_SHEET_ID is not configured, so the whole
|
||||
feature can be left switched off without touching sync behaviour.
|
||||
"""
|
||||
if not config.TEAM_SHEET_ID:
|
||||
return {}
|
||||
return await asyncio.to_thread(_refresh_teams_sync)
|
||||
|
||||
|
||||
def get_team_for_username(username: str) -> str | None:
|
||||
"""Return the team a Discord username is registered on, or None."""
|
||||
return _team_by_username.get(str(username).strip().lower())
|
||||
|
||||
|
||||
def all_team_names() -> set[str]:
|
||||
"""Every team name known from the registration sheet (the role universe)."""
|
||||
return set(_team_names)
|
||||
|
||||
|
||||
def get_team_rosters() -> dict[str, list[str]]:
|
||||
"""Current {team: [usernames]} cache (mainly for diagnostics/tests)."""
|
||||
return _team_roster
|
||||
|
||||
|
||||
def get_team_dividers() -> dict[str, int]:
|
||||
"""Current {team: divider role ID} cache.
|
||||
|
||||
Only teams whose section title matched a configured TEAM_DIVIDER_* entry
|
||||
appear here, so an empty dict means positioning is switched off.
|
||||
"""
|
||||
return dict(_team_divider)
|
||||
|
||||
Reference in New Issue
Block a user