""" utils/helpers.py — small utilities + the async SQLite store for RepoRadar. Contains: * `truncate` / `discord_timestamp` formatting helpers used by the embeds. * The async SQLite `Store` for per-guild config, watches, seen-repo de-dup, posted drops (vote anchors), and community votes. """ from __future__ import annotations import logging from datetime import datetime, timezone from pathlib import Path from typing import Any import aiosqlite log = logging.getLogger("reporadar.helpers") # ────────────────────────────────────────────────────────────── # Formatting helpers # ────────────────────────────────────────────────────────────── def discord_timestamp(dt: datetime | None, style: str = "R") -> str: """Render a Discord dynamic timestamp tag, or 'N/A' if dt is None.""" if dt is None: return "N/A" return f"" def truncate(text: str | None, limit: int) -> str: """Trim text to `limit` chars (Discord field/desc safe), adding an ellipsis.""" if text is None: return "" return text if len(text) <= limit else text[: limit - 1].rstrip() + "…" # ────────────────────────────────────────────────────────────── # Async SQLite store # ────────────────────────────────────────────────────────────── class Store: """Async SQLite-backed persistence for RepoRadar. Tables: git_config(guild_id PK, drop_channel_id, highlight_channel_id, highlight_role_id, min_stars, highlight_keywords, enabled, updated_at) git_watch(id PK, guild_id, scope_type, scope_value, channel_id, added_by, created_at) git_seen(guild_id, repo_key, ... , first_seen_at, announced) -- newest-only de-dup git_drop(id PK, guild_id, repo_key, channel_id, message_id, dropped_at) -- vote anchor git_vote(repo_key, guild_id, user_id, value, comment, created_at) -- community votes """ def __init__(self, db_path: Path) -> None: self._db_path = str(db_path) async def init(self) -> None: async with aiosqlite.connect(self._db_path) as db: await db.execute( """ CREATE TABLE IF NOT EXISTS git_config ( guild_id INTEGER PRIMARY KEY, drop_channel_id INTEGER, highlight_channel_id INTEGER, highlight_role_id INTEGER, min_stars INTEGER DEFAULT 0, highlight_keywords TEXT DEFAULT '', enabled INTEGER DEFAULT 1, updated_at TEXT ) """ ) await db.execute( """ CREATE TABLE IF NOT EXISTS git_watch ( id INTEGER PRIMARY KEY AUTOINCREMENT, guild_id INTEGER NOT NULL, scope_type TEXT NOT NULL, -- 'user' | 'org' | 'server' scope_value TEXT NOT NULL DEFAULT '', channel_id INTEGER, -- override drop channel added_by INTEGER, created_at TEXT NOT NULL, UNIQUE(guild_id, scope_type, scope_value) ) """ ) await db.execute( """ CREATE TABLE IF NOT EXISTS git_seen ( guild_id INTEGER NOT NULL, repo_key TEXT NOT NULL, name TEXT, full_name TEXT, url TEXT, description TEXT, created_remote TEXT, first_seen_at TEXT NOT NULL, announced INTEGER DEFAULT 0, PRIMARY KEY (guild_id, repo_key) ) """ ) await db.execute( """ CREATE TABLE IF NOT EXISTS git_drop ( id INTEGER PRIMARY KEY AUTOINCREMENT, guild_id INTEGER NOT NULL, repo_key TEXT NOT NULL, full_name TEXT, url TEXT, channel_id INTEGER NOT NULL, message_id INTEGER NOT NULL, dropped_at TEXT NOT NULL, UNIQUE(guild_id, message_id) ) """ ) await db.execute( """ CREATE TABLE IF NOT EXISTS git_vote ( repo_key TEXT NOT NULL, guild_id INTEGER NOT NULL, user_id INTEGER NOT NULL, value INTEGER NOT NULL, -- +1 | -1 comment TEXT DEFAULT '', created_at TEXT NOT NULL, PRIMARY KEY (repo_key, guild_id, user_id) ) """ ) await db.commit() log.info("SQLite store initialised at %s", self._db_path) # ── per-guild config ───────────────────────────────────── async def set_git_config(self, guild_id: int, **fields: Any) -> None: """Upsert git config columns for a guild (only provided fields change).""" allowed = { "drop_channel_id", "highlight_channel_id", "highlight_role_id", "min_stars", "highlight_keywords", "enabled", } cols = {k: v for k, v in fields.items() if k in allowed} now = datetime.now(timezone.utc).isoformat() async with aiosqlite.connect(self._db_path) as db: # Ensure a row exists, then patch the requested columns. await db.execute( "INSERT OR IGNORE INTO git_config (guild_id, updated_at) VALUES (?, ?)", (guild_id, now), ) if cols: assignments = ", ".join(f"{k} = ?" for k in cols) await db.execute( f"UPDATE git_config SET {assignments}, updated_at = ? WHERE guild_id = ?", (*cols.values(), now, guild_id), ) await db.commit() async def get_git_config(self, guild_id: int) -> dict[str, Any] | None: async with aiosqlite.connect(self._db_path) as db: db.row_factory = aiosqlite.Row async with db.execute( "SELECT * FROM git_config WHERE guild_id = ?", (guild_id,) ) as cur: row = await cur.fetchone() return dict(row) if row else None # ── watches ────────────────────────────────────────────── async def add_git_watch( self, guild_id: int, scope_type: str, scope_value: str, channel_id: int | None, added_by: int | None, ) -> None: async with aiosqlite.connect(self._db_path) as db: await db.execute( """ INSERT INTO git_watch (guild_id, scope_type, scope_value, channel_id, added_by, created_at) VALUES (?, ?, ?, ?, ?, ?) ON CONFLICT(guild_id, scope_type, scope_value) DO UPDATE SET channel_id = excluded.channel_id """, ( guild_id, scope_type, scope_value, channel_id, added_by, datetime.now(timezone.utc).isoformat(), ), ) await db.commit() async def remove_git_watch(self, guild_id: int, scope_type: str, scope_value: str) -> int: async with aiosqlite.connect(self._db_path) as db: cur = await db.execute( "DELETE FROM git_watch WHERE guild_id = ? AND scope_type = ? AND scope_value = ?", (guild_id, scope_type, scope_value), ) await db.commit() return cur.rowcount async def get_git_watches(self, guild_id: int) -> list[dict[str, Any]]: async with aiosqlite.connect(self._db_path) as db: db.row_factory = aiosqlite.Row async with db.execute( "SELECT * FROM git_watch WHERE guild_id = ? ORDER BY created_at", (guild_id,), ) as cur: return [dict(r) for r in await cur.fetchall()] async def all_git_watches(self) -> list[dict[str, Any]]: async with aiosqlite.connect(self._db_path) as db: db.row_factory = aiosqlite.Row async with db.execute("SELECT * FROM git_watch ORDER BY guild_id") as cur: return [dict(r) for r in await cur.fetchall()] # ── seen repos (newest-only de-dup) ────────────────────── async def get_seen_keys(self, guild_id: int) -> set[str]: async with aiosqlite.connect(self._db_path) as db: async with db.execute( "SELECT repo_key FROM git_seen WHERE guild_id = ?", (guild_id,) ) as cur: return {r[0] for r in await cur.fetchall()} async def mark_seen( self, guild_id: int, repo_key: str, *, name: str = "", full_name: str = "", url: str = "", description: str = "", created_remote: str = "", announced: bool = False, ) -> None: async with aiosqlite.connect(self._db_path) as db: await db.execute( """ INSERT OR IGNORE INTO git_seen (guild_id, repo_key, name, full_name, url, description, created_remote, first_seen_at, announced) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?) """, ( guild_id, repo_key, name, full_name, url, description, created_remote, datetime.now(timezone.utc).isoformat(), 1 if announced else 0, ), ) await db.commit() async def has_any_seen(self, guild_id: int) -> bool: async with aiosqlite.connect(self._db_path) as db: async with db.execute( "SELECT 1 FROM git_seen WHERE guild_id = ? LIMIT 1", (guild_id,) ) as cur: return (await cur.fetchone()) is not None # ── drops (vote anchors) ───────────────────────────────── async def record_drop( self, guild_id: int, repo_key: str, full_name: str, url: str, channel_id: int, message_id: int, ) -> None: async with aiosqlite.connect(self._db_path) as db: await db.execute( """ INSERT OR REPLACE INTO git_drop (guild_id, repo_key, full_name, url, channel_id, message_id, dropped_at) VALUES (?, ?, ?, ?, ?, ?, ?) """, ( guild_id, repo_key, full_name, url, channel_id, message_id, datetime.now(timezone.utc).isoformat(), ), ) await db.commit() async def get_drop_by_message(self, guild_id: int, message_id: int) -> dict[str, Any] | None: async with aiosqlite.connect(self._db_path) as db: db.row_factory = aiosqlite.Row async with db.execute( "SELECT * FROM git_drop WHERE guild_id = ? AND message_id = ?", (guild_id, message_id), ) as cur: row = await cur.fetchone() return dict(row) if row else None async def get_latest_drop_for_repo(self, guild_id: int, repo_key: str) -> dict[str, Any] | None: async with aiosqlite.connect(self._db_path) as db: db.row_factory = aiosqlite.Row async with db.execute( """ SELECT * FROM git_drop WHERE guild_id = ? AND repo_key = ? ORDER BY dropped_at DESC LIMIT 1 """, (guild_id, repo_key), ) as cur: row = await cur.fetchone() return dict(row) if row else None async def find_drop_by_fullname(self, guild_id: int, full_name: str) -> dict[str, Any] | None: """Find the most recent drop whose full_name matches (case-insensitive).""" async with aiosqlite.connect(self._db_path) as db: db.row_factory = aiosqlite.Row async with db.execute( """ SELECT * FROM git_drop WHERE guild_id = ? AND LOWER(full_name) = LOWER(?) ORDER BY dropped_at DESC LIMIT 1 """, (guild_id, full_name), ) as cur: row = await cur.fetchone() return dict(row) if row else None # ── community votes ────────────────────────────────────── async def cast_vote( self, guild_id: int, repo_key: str, user_id: int, value: int, comment: str = "", ) -> None: """Insert/replace a single user's vote (+1/-1) for a repo.""" async with aiosqlite.connect(self._db_path) as db: await db.execute( """ INSERT INTO git_vote (repo_key, guild_id, user_id, value, comment, created_at) VALUES (?, ?, ?, ?, ?, ?) ON CONFLICT(repo_key, guild_id, user_id) DO UPDATE SET value = excluded.value, comment = excluded.comment, created_at = excluded.created_at """, ( repo_key, guild_id, user_id, 1 if value >= 0 else -1, comment, datetime.now(timezone.utc).isoformat(), ), ) await db.commit() async def clear_vote(self, guild_id: int, repo_key: str, user_id: int) -> int: async with aiosqlite.connect(self._db_path) as db: cur = await db.execute( "DELETE FROM git_vote WHERE guild_id = ? AND repo_key = ? AND user_id = ?", (guild_id, repo_key, user_id), ) await db.commit() return cur.rowcount async def get_vote_tally(self, guild_id: int, repo_key: str) -> dict[str, int]: """Return {'up', 'down', 'score'} for a repo in a guild.""" async with aiosqlite.connect(self._db_path) as db: async with db.execute( """ SELECT COALESCE(SUM(CASE WHEN value > 0 THEN 1 ELSE 0 END), 0) AS up, COALESCE(SUM(CASE WHEN value < 0 THEN 1 ELSE 0 END), 0) AS down FROM git_vote WHERE guild_id = ? AND repo_key = ? """, (guild_id, repo_key), ) as cur: row = await cur.fetchone() up = int(row[0] or 0) down = int(row[1] or 0) return {"up": up, "down": down, "score": up - down} async def get_user_vote(self, guild_id: int, repo_key: str, user_id: int) -> int | None: async with aiosqlite.connect(self._db_path) as db: async with db.execute( "SELECT value FROM git_vote WHERE guild_id = ? AND repo_key = ? AND user_id = ?", (guild_id, repo_key, user_id), ) as cur: row = await cur.fetchone() return int(row[0]) if row else None async def get_vote_leaderboard(self, guild_id: int, limit: int = 10) -> list[dict[str, Any]]: """Top repos by net vote score, joined with their latest drop metadata.""" async with aiosqlite.connect(self._db_path) as db: db.row_factory = aiosqlite.Row async with db.execute( """ SELECT v.repo_key AS repo_key, SUM(CASE WHEN v.value > 0 THEN 1 ELSE 0 END) AS up, SUM(CASE WHEN v.value < 0 THEN 1 ELSE 0 END) AS down, SUM(v.value) AS score, MAX(d.full_name) AS full_name, MAX(d.url) AS url FROM git_vote v LEFT JOIN git_drop d ON d.guild_id = v.guild_id AND d.repo_key = v.repo_key WHERE v.guild_id = ? GROUP BY v.repo_key ORDER BY score DESC, up DESC LIMIT ? """, (guild_id, limit), ) as cur: return [dict(r) for r in await cur.fetchall()]