SubINaclS/RepoRadar
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Initial sanitary RepoRadar release

Browse Source
This commit is contained in:
SINS
2026-06-27 21:20:43 -04:00
commit b90ec9cdd0
15 changed files with 2435 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
utils/helpers.py
Executable
+430
View File
@@ -0,0 +1,430 @@
"""
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"<t:{int(dt.timestamp())}:{style}>"
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()]