Add a moderation queue for user submissions

Completes TODO #3. Submissions are gated behind ACCEPT_SUBMISSIONS,
which defaults to off: is_allowed() treats everyone as allowed when
ALLOWED_USERS is empty, so opening submissions implicitly would have
accepted media from anyone who found the bot.

- Schema v3 adds status, reviewed_by and reviewed_at. Existing rows
  default to approved, so upgrading changes nothing for a running bot.
- Search, /list and counts only return approved items; callers that
  need everything pass status=None.
- /review shows the oldest pending item with approve, reject, fix
  keywords and stop. No cursor is stored anywhere: each action just
  asks for the next pending item, which is what lets a moderator stop
  and resume later, and lets several moderators share one queue.
- set_status only updates rows still pending, so if two moderators open
  the same item the first decision wins and the second is told so.
- A failed send no longer wedges the queue; the item's details are
  shown anyway so it can still be rejected.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
goyban
2026-09-01 21:32:35 +00:00
parent efc2be5f8a
commit 0a8c81db90
7 changed files with 331 additions and 27 deletions
+3
View File
@@ -1,3 +1,6 @@
BOT_TOKEN=your_telegram_bot_token_here
ALLOWED_USERS=123456789,987654321
ADMIN_USERS=123456789
# Let non-admin users submit media for moderator approval (default: off)
ACCEPT_SUBMISSIONS=false
+11
View File
@@ -65,10 +65,21 @@ Telegram suggests these commands as you type `/`, with a short description besid
| `/start` | Start the bot and show usage instructions |
| `/menu` | Show the available commands |
| `/list` | Show all saved media |
| `/review` | Review pending submissions one by one (admin only) |
| `/edit [search]` | Select a meme and edit its keywords (admin only) |
| `/delete <id>` | Remove an entry (get ID from `/list`; admin only) |
| `/cancel` | Cancel current add operation |
### Accepting submissions
Set `ACCEPT_SUBMISSIONS=true` in `.env` to let non-admin users send media too. Their submissions are stored but stay invisible — they don't appear in inline search or `/list` until a moderator approves them. Admins' own media is still saved directly.
Run `/review` to work through the queue. Each item is shown with its keywords and four buttons: approve, reject, fix keywords, or stop.
The queue keeps no cursor. `/review` always fetches the oldest pending item, so you can stop after three and come back tomorrow, and several moderators can review at the same time without stepping on each other — if two open the same item, the first decision wins and the second is told it was already handled.
> **Leave this off unless you mean it.** With `ALLOWED_USERS` empty, everyone is an allowed user, so turning submissions on would accept media from anyone who finds the bot.
### Moving to another server
The only files you need to transfer are:
+1 -1
View File
@@ -1 +1 @@
1.1.0
1.2.0
+159 -6
View File
@@ -41,6 +41,10 @@ ADMIN_USERS: set[int] = {
int(uid.strip()) for uid in os.environ.get("ADMIN_USERS", "").split(",") if uid.strip()
}
# Off by default: with ALLOWED_USERS unset, is_allowed() lets everyone in, so
# opening submissions implicitly would accept media from the whole world.
ACCEPT_SUBMISSIONS = os.environ.get("ACCEPT_SUBMISSIONS", "").strip().lower() in {"1", "true", "yes"}
bot = Bot(token=BOT_TOKEN)
dp = Dispatcher(storage=MemoryStorage())
@@ -52,6 +56,7 @@ USER_COMMANDS = [
]
ADMIN_COMMANDS = [
*USER_COMMANDS,
BotCommand(command="review", description="Review pending submissions"),
BotCommand(command="edit", description="Edit meme keywords"),
BotCommand(command="delete", description="Remove a meme by ID"),
]
@@ -98,10 +103,17 @@ def menu_text(user_id: int) -> str:
"Send me any <b>photo, GIF, video, voice, music, sticker, or file</b>.\n"
"I'll ask you for keywords next.\n\n"
"/list — show all saved memes\n"
"/review — review pending submissions\n"
"/edit [search] — edit keywords of a meme\n"
"/delete &lt;id&gt; — remove a meme by ID\n"
"/cancel — cancel current operation"
)
if ACCEPT_SUBMISSIONS:
return (
"Use me inline: type <code>@botname keyword</code> in any chat.\n\n"
"Send me media to submit it — a moderator will review it before it goes live.\n\n"
"/list — browse all saved memes"
)
return (
"Use me inline: type <code>@botname keyword</code> in any chat.\n\n"
"/list — browse all saved memes"
@@ -141,13 +153,15 @@ async def cmd_cancel(msg: Message, state: FSMContext):
# ── step 1: receive media (admin only) ───────────────────────────────────────
async def _receive_media(msg: Message, state: FSMContext, file_id: str, file_unique_id: str, media_type: str):
async def _receive_media(msg: Message, state: FSMContext, file_id: str, file_unique_id: str,
media_type: str, status: str = "approved"):
await state.set_state(AddMedia.waiting_for_keywords)
await state.update_data(
file_id=file_id,
file_unique_id=file_unique_id,
media_type=media_type,
keywords=[],
status=status,
)
await msg.reply(
"Got it! Now send keywords — one per message or comma-separated.\n"
@@ -188,13 +202,18 @@ def extract_media(msg: Message) -> tuple[str, str, str] | None:
@dp.message(MEDIA_FILTER, StateFilter(None))
async def on_media(msg: Message, state: FSMContext):
if not is_admin(msg.from_user.id):
user_id = msg.from_user.id
if is_admin(user_id):
status = "approved"
elif ACCEPT_SUBMISSIONS and is_allowed(user_id):
status = "pending"
else:
return
found = extract_media(msg)
if found is None:
return
media_type, file_id, file_unique_id = found
await _receive_media(msg, state, file_id, file_unique_id, media_type)
await _receive_media(msg, state, file_id, file_unique_id, media_type, status)
# ── step 2: collect keywords ─────────────────────────────────────────────────
@@ -206,14 +225,23 @@ async def cmd_done(msg: Message, state: FSMContext):
if not keywords:
await msg.answer("You haven't added any keywords yet. Send some, then /done.")
return
status = data.get("status", "approved")
await db.add_media(
msg.from_user.id,
data["file_id"],
data["file_unique_id"],
data["media_type"],
keywords,
status=status,
)
await state.clear()
if status == "pending":
await msg.answer(
f"Submitted with keywords: <b>{', '.join(keywords)}</b>\n"
"A moderator will review it before it becomes searchable.",
parse_mode="HTML",
)
else:
await msg.answer(f"Saved with keywords: <b>{', '.join(keywords)}</b>", parse_mode="HTML")
@@ -357,11 +385,20 @@ async def cmd_edit_done(msg: Message, state: FSMContext):
await msg.answer("You haven't entered any keywords yet. Send some, then /done.")
return
updated = await db.update_media_keywords(data["media_id"], keywords)
from_review = data.get("from_review", False)
await state.clear()
if updated:
await msg.answer(f"Updated keywords: <b>{', '.join(keywords)}</b>", parse_mode="HTML")
else:
if not updated:
await msg.answer("Failed to update (meme may have been deleted).")
return
if from_review:
# The item is still pending, so the queue hands the same one back with
# its new keywords — ready to approve.
await send_next_pending(msg.chat.id, "Keywords updated.")
return
await msg.answer(f"Updated keywords: <b>{', '.join(keywords)}</b>", parse_mode="HTML")
@dp.message(EditMedia.waiting_for_new_keywords, Command("cancel"))
@@ -382,6 +419,122 @@ async def on_edit_keyword(msg: Message, state: FSMContext):
await msg.reply(f"Keywords so far: <b>{', '.join(merged)}</b>\nSend more or /done.", parse_mode="HTML")
# ── /review ──────────────────────────────────────────────────────────────────
# Sending a stored item back needs one call per media type. Stickers cannot
# carry a caption, so the item and its review buttons are sent as two messages
# rather than special-casing captions per type.
MEDIA_SENDERS = {
"photo": lambda chat_id, fid: bot.send_photo(chat_id, fid),
"gif": lambda chat_id, fid: bot.send_animation(chat_id, fid),
"video": lambda chat_id, fid: bot.send_video(chat_id, fid),
"voice": lambda chat_id, fid: bot.send_voice(chat_id, fid),
"audio": lambda chat_id, fid: bot.send_audio(chat_id, fid),
"sticker": lambda chat_id, fid: bot.send_sticker(chat_id, fid),
"document": lambda chat_id, fid: bot.send_document(chat_id, fid),
}
def review_keyboard(media_id: int) -> InlineKeyboardMarkup:
return InlineKeyboardMarkup(inline_keyboard=[
[
InlineKeyboardButton(text="✅ Approve", callback_data=f"review:approve:{media_id}"),
InlineKeyboardButton(text="❌ Reject", callback_data=f"review:reject:{media_id}"),
],
[
InlineKeyboardButton(text="✏️ Fix keywords", callback_data=f"review:edit:{media_id}"),
InlineKeyboardButton(text="⏸ Stop", callback_data="review:stop:0"),
],
])
async def send_next_pending(chat_id: int, notice: str = "") -> None:
"""Show the next item awaiting review, or report the queue is empty.
Nothing tracks where a moderator got to — each call asks the database for
the oldest pending item. Stopping and resuming later needs no state, and
two moderators working at once simply take items off the same queue.
"""
item = await db.next_pending()
prefix = f"{notice}\n\n" if notice else ""
if item is None:
await bot.send_message(chat_id, f"{prefix}Nothing left to review.")
return
remaining = await db.count_pending()
send = MEDIA_SENDERS.get(item["media_type"])
if send is not None:
try:
await send(chat_id, item["file_id"])
except Exception:
# A stale or unusable file_id must not wedge the queue — show the
# details anyway so the moderator can still reject it.
log.exception("could not send pending item %s for review", item["id"])
await bot.send_message(chat_id, "⚠️ Could not load this media.")
else:
log.warning("no sender for media_type %r (item %s)", item["media_type"], item["id"])
await bot.send_message(chat_id, f"⚠️ Unsupported media type: {item['media_type']}")
await bot.send_message(
chat_id,
f"{prefix}Pending <b>#{item['id']}</b> [{item['media_type']}]\n"
f"From user <code>{item['owner_id']}</code>\n"
f"Keywords: <b>{item['keywords'].replace(',', ', ')}</b>\n\n"
f"{remaining} awaiting review.",
parse_mode="HTML",
reply_markup=review_keyboard(item["id"]),
)
@dp.message(Command("review"))
async def cmd_review(msg: Message):
if not is_admin(msg.from_user.id):
await msg.answer("Only moderators can review submissions.")
return
await send_next_pending(msg.chat.id)
@dp.callback_query(F.data.startswith("review:"))
async def on_review_action(callback: CallbackQuery, state: FSMContext):
if not is_admin(callback.from_user.id):
await callback.answer("Not allowed.", show_alert=True)
return
_, action, raw_id = callback.data.split(":", 2)
await callback.message.edit_reply_markup(reply_markup=None)
if action == "stop":
await callback.message.answer("Review paused. Send /review to pick up where you left off.")
await callback.answer()
return
media_id = int(raw_id)
if action == "edit":
item = await db.get_media_by_id(media_id)
if not item:
await send_next_pending(callback.message.chat.id, "That item is gone.")
await callback.answer()
return
await state.set_state(EditMedia.waiting_for_new_keywords)
await state.update_data(media_id=media_id, keywords=[], from_review=True)
await callback.message.answer(
f"Editing <b>#{item['id']}</b>\n"
f"Current keywords: <b>{item['keywords']}</b>\n\n"
"Send the new keywords, then /done. /cancel stops reviewing.",
parse_mode="HTML",
)
await callback.answer()
return
status = "approved" if action == "approve" else "rejected"
changed = await db.set_status(media_id, status, callback.from_user.id)
notice = f"#{media_id} {status}." if changed else f"#{media_id} was already handled."
await send_next_pending(callback.message.chat.id, notice)
await callback.answer()
# Cached inline results, keyed by the media_type stored in the database.
# Audio carries its own title metadata; video and document require one.
INLINE_BUILDERS = {
+69 -13
View File
@@ -4,7 +4,7 @@ import os
DB_PATH = os.getenv("DB_PATH", "/data/media.db")
# Bump this and add a matching _migrate_to_vN() when the schema changes.
SCHEMA_VERSION = 2
SCHEMA_VERSION = 3
# ── schema / migrations ──────────────────────────────────────────────────────
@@ -60,8 +60,18 @@ async def _migrate_to_v2(db) -> None:
await db.execute("CREATE INDEX IF NOT EXISTS idx_owner ON media(owner_id)")
async def _migrate_to_v3(db) -> None:
"""Add the moderation queue. Existing rows default to approved, so an
upgraded deployment behaves exactly as it did before."""
await db.execute("ALTER TABLE media ADD COLUMN status TEXT NOT NULL DEFAULT 'approved'")
await db.execute("ALTER TABLE media ADD COLUMN reviewed_by INTEGER")
await db.execute("ALTER TABLE media ADD COLUMN reviewed_at TIMESTAMP")
await db.execute("CREATE INDEX IF NOT EXISTS idx_status ON media(status)")
MIGRATIONS = {
2: _migrate_to_v2,
3: _migrate_to_v3,
}
@@ -90,21 +100,23 @@ async def init_db():
# ── writes ───────────────────────────────────────────────────────────────────
async def add_media(owner_id: int, file_id: str, file_unique_id: str, media_type: str, keywords: list[str]):
async def add_media(owner_id: int, file_id: str, file_unique_id: str, media_type: str,
keywords: list[str], status: str = "approved"):
kw = ",".join(k.strip().lower() for k in keywords if k.strip())
async with aiosqlite.connect(DB_PATH) as db:
# Upsert rather than INSERT OR REPLACE: replacing would delete and
# reinsert the row, handing it a new id and breaking /delete <id>.
await db.execute(
"""
INSERT INTO media (owner_id, file_id, file_unique_id, media_type, keywords)
VALUES (?,?,?,?,?)
INSERT INTO media (owner_id, file_id, file_unique_id, media_type, keywords, status)
VALUES (?,?,?,?,?,?)
ON CONFLICT(owner_id, file_unique_id) DO UPDATE SET
file_id = excluded.file_id,
media_type = excluded.media_type,
keywords = excluded.keywords
keywords = excluded.keywords,
status = excluded.status
""",
(owner_id, file_id, file_unique_id, media_type, kw),
(owner_id, file_id, file_unique_id, media_type, kw, status),
)
await db.commit()
@@ -129,15 +141,20 @@ async def delete_media(media_id: int) -> bool:
# ── reads ────────────────────────────────────────────────────────────────────
COLUMNS = "id, file_id, file_unique_id, media_type, keywords"
COLUMNS = "id, owner_id, file_id, file_unique_id, media_type, keywords, status"
async def search_media(query: str, owner_id: int | None = None) -> list[dict]:
async def search_media(query: str, owner_id: int | None = None,
status: str | None = "approved") -> list[dict]:
"""Search by keyword prefix. owner_id=None searches every owner, which is
the current shared-library behaviour; public mode will pass a scope."""
the current shared-library behaviour; public mode will pass a scope.
status=None includes items still awaiting moderation."""
query = query.strip().lower()
where = []
params: list = []
if status is not None:
where.append("status = ?")
params.append(status)
if query:
where.append("(',' || keywords || ',') LIKE ?")
params.append(f"%,{query}%")
@@ -156,27 +173,32 @@ async def search_media(query: str, owner_id: int | None = None) -> list[dict]:
return [dict(r) for r in rows]
async def list_media(page: int, page_size: int) -> list[dict]:
async def list_media(page: int, page_size: int, status: str | None = "approved") -> list[dict]:
"""Return one page of saved media, newest first."""
offset = page * page_size
clause = "WHERE status = ?" if status is not None else ""
params = ([status] if status is not None else []) + [page_size, offset]
async with aiosqlite.connect(DB_PATH) as db:
db.row_factory = aiosqlite.Row
cursor = await db.execute(
f"""
SELECT {COLUMNS}
FROM media
{clause}
ORDER BY added_at DESC, id DESC
LIMIT ? OFFSET ?
""",
(page_size, offset),
params,
)
rows = await cursor.fetchall()
return [dict(r) for r in rows]
async def count_media() -> int:
async def count_media(status: str | None = "approved") -> int:
clause = "WHERE status = ?" if status is not None else ""
params = [status] if status is not None else []
async with aiosqlite.connect(DB_PATH) as db:
cursor = await db.execute("SELECT COUNT(*) FROM media")
cursor = await db.execute(f"SELECT COUNT(*) FROM media {clause}", params)
row = await cursor.fetchone()
return row[0]
@@ -190,3 +212,37 @@ async def get_media_by_id(media_id: int) -> dict | None:
)
row = await cursor.fetchone()
return dict(row) if row else None
# ── moderation queue ─────────────────────────────────────────────────────────
async def next_pending() -> dict | None:
"""Oldest item awaiting review. The queue keeps no cursor: each call just
asks for the next one, so a moderator can stop and resume at any time and
several moderators can work at once."""
async with aiosqlite.connect(DB_PATH) as db:
db.row_factory = aiosqlite.Row
cursor = await db.execute(
f"SELECT {COLUMNS} FROM media WHERE status = 'pending' ORDER BY added_at ASC, id ASC LIMIT 1"
)
row = await cursor.fetchone()
return dict(row) if row else None
async def count_pending() -> int:
return await count_media(status="pending")
async def set_status(media_id: int, status: str, reviewer_id: int) -> bool:
"""Approve or reject a pending item. Returns False if it was not pending
any more, which is how two moderators reviewing at once are kept apart."""
async with aiosqlite.connect(DB_PATH) as db:
cur = await db.execute(
"""
UPDATE media SET status = ?, reviewed_by = ?, reviewed_at = CURRENT_TIMESTAMP
WHERE id = ? AND status = 'pending'
""",
(status, reviewer_id, media_id),
)
await db.commit()
return cur.rowcount > 0
+9 -1
View File
@@ -75,6 +75,13 @@ def test_every_stored_type_renders_inline():
print(" every stored media_type has a matching inline builder")
def test_every_stored_type_can_be_sent_back():
stored = {t for t, _ in B.MEDIA_EXTRACTORS}
missing = stored - set(B.MEDIA_SENDERS)
assert not missing, f"stored but not sendable for review: {sorted(missing)}"
print(" every stored media_type can be sent back for review")
def test_builders_construct():
for media_type, build in B.INLINE_BUILDERS.items():
build("rid", f"FID_{media_type}", "cat, funny")
@@ -86,7 +93,8 @@ def test_builders_construct():
def main():
for test in (test_every_type_extracts, test_animation_beats_document,
test_video_note_stored_as_video, test_non_media_ignored,
test_every_stored_type_renders_inline, test_builders_construct):
test_every_stored_type_renders_inline,
test_every_stored_type_can_be_sent_back, test_builders_construct):
test()
print("\nAll media tests passed.")
+78 -5
View File
@@ -47,10 +47,13 @@ async def test_legacy_upgrade(tmp: str):
db = load_db(path)
await db.init_db()
assert user_version(path) == 2
assert user_version(path) == 3
rows = sorted((r["id"], r["keywords"]) for r in await db.search_media(""))
assert rows == [(7, "cat,funny"), (9, "dog")], f"rows not preserved: {rows}"
print(" legacy v1 database upgrades to v2 with ids and keywords intact")
with sqlite3.connect(path) as c:
statuses = {r[0] for r in c.execute("SELECT status FROM media")}
assert statuses == {"approved"}, f"upgraded rows must stay live: {statuses}"
print(" legacy v1 database upgrades with ids, keywords and visibility intact")
async def test_per_owner_uniqueness(tmp: str):
@@ -88,18 +91,88 @@ async def test_fresh_and_idempotent(tmp: str):
with sqlite3.connect(path) as c:
sql = c.execute("SELECT sql FROM sqlite_master WHERE name='media'").fetchone()[0]
assert "UNIQUE(owner_id, file_unique_id)" in sql
assert user_version(path) == 2
assert user_version(path) == 3
await db.init_db()
await db.init_db()
print(" fresh install lands on v2; init_db is idempotent")
print(" fresh install lands on the current version; init_db is idempotent")
async def test_pending_is_hidden(tmp: str):
path = os.path.join(tmp, "pending.db")
db = load_db(path)
await db.init_db()
await db.add_media(111, "F1", "U1", "photo", ["live"])
await db.add_media(222, "F2", "U2", "photo", ["waiting"], status="pending")
assert len(await db.search_media("")) == 1, "pending items must not be searchable"
assert await db.count_media() == 1
assert len(await db.list_media(0, 30)) == 1, "pending items must not appear in /list"
assert len(await db.search_media("", status=None)) == 2, "status=None should see everything"
assert await db.count_pending() == 1
print(" pending items stay out of search, /list and counts until approved")
async def test_queue_is_oldest_first(tmp: str):
path = os.path.join(tmp, "queue.db")
db = load_db(path)
await db.init_db()
for i in range(3):
await db.add_media(100 + i, f"F{i}", f"U{i}", "photo", [f"kw{i}"], status="pending")
first = await db.next_pending()
assert first["keywords"] == "kw0", f"queue must be oldest-first, got {first['keywords']}"
# Approving removes it from the queue and the next one steps up.
assert await db.set_status(first["id"], "approved", 999) is True
second = await db.next_pending()
assert second["keywords"] == "kw1"
assert await db.count_pending() == 2
print(" queue serves oldest first and advances as items are handled")
async def test_two_moderators_cannot_double_review(tmp: str):
path = os.path.join(tmp, "race.db")
db = load_db(path)
await db.init_db()
await db.add_media(111, "F1", "U1", "photo", ["kw"], status="pending")
item = await db.next_pending()
assert await db.set_status(item["id"], "approved", 111) is True
assert await db.set_status(item["id"], "rejected", 222) is False, \
"a second moderator must not overwrite a completed review"
row = await db.get_media_by_id(item["id"])
assert row["status"] == "approved", "first review must win"
print(" a second moderator cannot overturn an already-reviewed item")
async def test_rejected_stays_hidden(tmp: str):
path = os.path.join(tmp, "rejected.db")
db = load_db(path)
await db.init_db()
await db.add_media(111, "F1", "U1", "photo", ["nope"], status="pending")
item = await db.next_pending()
await db.set_status(item["id"], "rejected", 999)
assert await db.search_media("nope") == [], "rejected items must never surface"
assert await db.count_pending() == 0
assert await db.next_pending() is None
print(" rejected items leave the queue and never surface in search")
async def main():
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
with tempfile.TemporaryDirectory() as tmp:
for test in (test_legacy_upgrade, test_per_owner_uniqueness,
test_readd_keeps_id, test_fresh_and_idempotent):
test_readd_keeps_id, test_fresh_and_idempotent,
test_pending_is_hidden, test_queue_is_oldest_first,
test_two_moderators_cannot_double_review,
test_rejected_stays_hidden):
await test(tmp)
print("\nAll database tests passed.")