d48a436e26
Divider placement matched the divider role by its exact Discord name, so renaming the role in Discord silently broke positioning. Switch the TEAM_DIVIDER_<SUFFIX> config to hold a role ID; resolve the ID to the role's current name in apply_team_role_positions and keep the existing name-based ordering maths downstream unchanged. - config._parse_team_dividers now parses values as ints (rejects non-ints) - resolve_divider / _team_divider cache / get_team_dividers return IDs - apply_team_role_positions resolves each ID via guild.get_role once - .env.example documents IDs and ships the CS2/LoL divider role IDs - resolve_divider tests updated to assert IDs Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01XPRsW4tazVtYi2jNzVQkre
483 lines
18 KiB
Python
483 lines
18 KiB
Python
"""Member synchronization logic - roles, nicknames, birthdays."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import calendar
|
|
import logging
|
|
from datetime import datetime, date
|
|
from dataclasses import dataclass, field
|
|
from zoneinfo import ZoneInfo
|
|
|
|
import discord
|
|
|
|
import config
|
|
from . import sheets
|
|
|
|
log = logging.getLogger(__name__)
|
|
_PLACEHOLDER = {"-", "x", "n/a", "none", "ei"}
|
|
_TZ = ZoneInfo("Europe/Tallinn")
|
|
|
|
|
|
def today_local() -> date:
|
|
"""Today's date in Europe/Tallinn — the bot's operational timezone, independent of host TZ."""
|
|
return datetime.now(_TZ).date()
|
|
|
|
|
|
def _shift_year_safe(d: date, year: int) -> date:
|
|
"""Move `d` to `year`; Feb 29 falls back to Feb 28 when the target year is non-leap."""
|
|
try:
|
|
return d.replace(year=year)
|
|
except ValueError:
|
|
return d.replace(year=year, day=28)
|
|
|
|
|
|
def _is_placeholder(val: str) -> bool:
|
|
"""Return True if a sheet cell value is an intentional empty marker."""
|
|
return not val.strip() or val.strip().lower() in _PLACEHOLDER
|
|
|
|
|
|
# Maps sheet role values to Discord role names where they differ
|
|
ROLE_NAME_MAP: dict[str, str] = {
|
|
"Juht": "Tiimijuht",
|
|
"Admin": "+",
|
|
}
|
|
|
|
|
|
@dataclass
|
|
class SyncResult:
|
|
"""Tracks what happened during a sync operation."""
|
|
nickname_changed: bool = False
|
|
roles_added: list[str] = field(default_factory=list)
|
|
birthday_soon: bool = False
|
|
birthday_today: bool = False
|
|
not_found: bool = False
|
|
errors: list[str] = field(default_factory=list)
|
|
synced: bool = False # True when no errors; caller writes this to the sheet
|
|
|
|
@property
|
|
def changed(self) -> bool:
|
|
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
|
|
errors: list[str] = field(default_factory=list)
|
|
|
|
@property
|
|
def changed(self) -> bool:
|
|
return bool(self.added or self.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
|
|
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:
|
|
"""Format a nickname from a full name: first name + last name initial.
|
|
|
|
Examples:
|
|
'Mari Tamm' -> 'Mari T'
|
|
'Mari-Liis Tamm' -> 'Mari-Liis T'
|
|
'Jaan' -> 'Jaan'
|
|
"""
|
|
parts = full_name.strip().split()
|
|
if len(parts) < 2:
|
|
return full_name.strip()
|
|
first = parts[0] # preserves hyphenated names like Mari-Liis
|
|
last_initial = parts[-1][0].upper()
|
|
return f"{first} {last_initial}"
|
|
|
|
|
|
def _parse_roles(roles_str: str) -> list[str]:
|
|
"""Parse comma-separated role names from the sheet."""
|
|
if not roles_str:
|
|
return []
|
|
return [r.strip().rstrip(".,;:!?") for r in str(roles_str).split(",") if r.strip()]
|
|
|
|
|
|
def _parse_birthday(raw: str) -> date | None:
|
|
"""Parse a birthday string robustly. Returns None for garbage/placeholder values."""
|
|
raw = str(raw).strip()
|
|
if _is_placeholder(raw):
|
|
return None
|
|
today = today_local()
|
|
for fmt, has_year in [("%d/%m/%Y", True), ("%Y-%m-%d", True), ("%m-%d", False)]:
|
|
try:
|
|
parsed = datetime.strptime(raw, fmt).date()
|
|
if has_year and not (1920 <= parsed.year <= today.year):
|
|
return None # unrealistic year (formula artefact, future year, etc.)
|
|
return parsed if has_year else parsed.replace(year=today.year)
|
|
except ValueError:
|
|
continue
|
|
return None
|
|
|
|
|
|
def _is_birthday_soon(birthday_str: str, window_days: int | None = None) -> bool:
|
|
"""Check if a birthday falls within the configured window."""
|
|
bday = _parse_birthday(birthday_str)
|
|
if bday is None:
|
|
return False
|
|
window = window_days or config.BIRTHDAY_WINDOW_DAYS
|
|
today = today_local()
|
|
this_year_bday = _shift_year_safe(bday, today.year)
|
|
if this_year_bday < today:
|
|
this_year_bday = _shift_year_safe(bday, today.year + 1)
|
|
delta = (this_year_bday - today).days
|
|
return 0 <= delta <= window
|
|
|
|
|
|
def is_birthday_today(birthday_str: str) -> bool:
|
|
"""Return True if today is the member's birthday (any supported date format).
|
|
|
|
Feb 29 babies are observed on Feb 28 in non-leap years.
|
|
"""
|
|
bday = _parse_birthday(birthday_str)
|
|
if bday is None:
|
|
return False
|
|
today = today_local()
|
|
if bday.month == today.month and bday.day == today.day:
|
|
return True
|
|
if (
|
|
bday.month == 2 and bday.day == 29
|
|
and today.month == 2 and today.day == 28
|
|
and not calendar.isleap(today.year)
|
|
):
|
|
return True
|
|
return False
|
|
|
|
|
|
async def sync_member(
|
|
member: discord.Member,
|
|
guild: discord.Guild,
|
|
) -> SyncResult:
|
|
"""Synchronize a single member's nickname and roles based on sheet data.
|
|
|
|
Also populates Discord ID in the sheet if missing, and updates username.
|
|
Returns a SyncResult describing what happened.
|
|
"""
|
|
result = SyncResult()
|
|
|
|
# Look up member in sheet cache
|
|
row = sheets.find_member(member.id, member.name)
|
|
if row is None:
|
|
result.not_found = True
|
|
return result
|
|
|
|
# --- Backfill User ID if missing ---
|
|
raw_id = str(row.get("User ID", "")).strip()
|
|
if not raw_id or raw_id == "0":
|
|
await sheets.set_user_id(member.name, member.id)
|
|
|
|
# --- Update Discord username in sheet if it changed ---
|
|
sheet_username = str(row.get("Discord", "")).strip()
|
|
if sheet_username.lower() != member.name.lower():
|
|
await sheets.update_username(member.id, member.name)
|
|
|
|
# --- Nickname (Nimi = real name, formatted as first name + last initial) ---
|
|
nimi = str(row.get("Nimi", "")).strip()
|
|
desired_nick = _format_nickname(nimi) if nimi else None
|
|
if desired_nick and member.nick != desired_nick:
|
|
try:
|
|
await member.edit(nick=desired_nick)
|
|
result.nickname_changed = True
|
|
except discord.Forbidden:
|
|
log.debug("No permission to set nickname for %s (likely admin), skipping", member)
|
|
except discord.HTTPException as e:
|
|
result.errors.append(f"Hüüdnime viga kasutajale {member}: {e}")
|
|
|
|
# --- Roles (from Organisatsioon, Valdkond, Roll columns; values may be comma-separated) ---
|
|
desired_role_names: list[str] = []
|
|
for col in ("Organisatsioon", "Valdkond", "Roll"):
|
|
val = str(row.get(col, "")).strip()
|
|
if not _is_placeholder(val):
|
|
desired_role_names.extend(
|
|
r for r in _parse_roles(val)
|
|
if not _is_placeholder(r)
|
|
)
|
|
desired_roles: list[discord.Role] = []
|
|
for rname in desired_role_names:
|
|
rname = ROLE_NAME_MAP.get(rname, rname)
|
|
role = discord.utils.get(guild.roles, name=rname)
|
|
if role:
|
|
desired_roles.append(role)
|
|
else:
|
|
result.errors.append(f"Rolli '{rname}' ei leitud serverist")
|
|
|
|
# Base roles that every synced member must have
|
|
for rid in config.BASE_ROLE_IDS:
|
|
role = guild.get_role(rid)
|
|
if role:
|
|
desired_roles.append(role)
|
|
else:
|
|
result.errors.append(f"Baasrolli ID {rid} ei leitud serverist")
|
|
|
|
# Roles to add (desired but member doesn't have)
|
|
to_add = [r for r in desired_roles if r not in member.roles]
|
|
# (we currently only ADD the desired roles, not remove extras - safe default)
|
|
|
|
if to_add:
|
|
try:
|
|
await member.add_roles(*to_add, reason="Sheet sync")
|
|
result.roles_added = [r.name for r in to_add]
|
|
except discord.Forbidden:
|
|
log.debug("No permission to add roles for %s (likely admin), skipping", member)
|
|
except discord.HTTPException as e:
|
|
result.errors.append(f"Rolli viga kasutajale {member}: {e}")
|
|
|
|
# --- Birthday check ---
|
|
birthday_str = str(row.get("Sünnipäev", "")).strip()
|
|
if not _is_placeholder(birthday_str):
|
|
result.birthday_today = is_birthday_today(birthday_str)
|
|
result.birthday_soon = _is_birthday_soon(birthday_str)
|
|
|
|
# --- Mark synced (caller is responsible for writing to sheet) ---
|
|
result.synced = not bool(result.errors)
|
|
|
|
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).
|
|
|
|
Only role NAMES present in the team sheet 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 = sheets.get_team_for_username(member.name)
|
|
all_teams = sheets.all_team_names()
|
|
if not all_teams:
|
|
return result # feature switched off (no team sheet loaded)
|
|
|
|
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}")
|
|
|
|
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_dividers = sheets.get_team_dividers() # {team: divider role ID}
|
|
if not team_dividers:
|
|
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_dividers.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}")
|
|
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.changed:
|
|
bits: list[str] = []
|
|
if res.added:
|
|
bits.append(f"+{res.added}")
|
|
if res.removed:
|
|
bits.append("-" + ", -".join(res.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,
|
|
) -> None:
|
|
"""Send a birthday-coming-soon announcement to the configured channel."""
|
|
channel = bot.get_channel(config.BIRTHDAY_CHANNEL_ID)
|
|
if channel is None:
|
|
log.warning("Birthday channel %s not found", config.BIRTHDAY_CHANNEL_ID)
|
|
return
|
|
|
|
row = sheets.find_member(member.id, member.name)
|
|
bday_str = str(row.get("Sünnipäev", "")) if row else "?"
|
|
|
|
await channel.send(
|
|
f"🎂 @here **{member.display_name}** sünnipäev on täna! 🥳 Palju-palju õnne sünnipäevaks! 🎉"
|
|
)
|