"""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 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: """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); * 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 = 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}") # --- 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 = sheets.get_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_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.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, ) -> 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! 🎉" )