diff --git a/.env.example b/.env.example index d613a52..ff599b0 100644 --- a/.env.example +++ b/.env.example @@ -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 diff --git a/README.md b/README.md index f723d6e..9a76199 100644 --- a/README.md +++ b/README.md @@ -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 ` | 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: diff --git a/VERSION b/VERSION index 9084fa2..26aaba0 100644 --- a/VERSION +++ b/VERSION @@ -1 +1 @@ -1.1.0 +1.2.0 diff --git a/bot.py b/bot.py index a486916..8049060 100644 --- a/bot.py +++ b/bot.py @@ -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 photo, GIF, video, voice, music, sticker, or file.\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 <id> — remove a meme by ID\n" "/cancel — cancel current operation" ) + if ACCEPT_SUBMISSIONS: + return ( + "Use me inline: type @botname keyword 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 @botname keyword 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,15 +225,24 @@ 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() - await msg.answer(f"Saved with keywords: {', '.join(keywords)}", parse_mode="HTML") + if status == "pending": + await msg.answer( + f"Submitted with keywords: {', '.join(keywords)}\n" + "A moderator will review it before it becomes searchable.", + parse_mode="HTML", + ) + else: + await msg.answer(f"Saved with keywords: {', '.join(keywords)}", parse_mode="HTML") @dp.message(AddMedia.waiting_for_keywords, F.text) @@ -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: {', '.join(keywords)}", 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: {', '.join(keywords)}", 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: {', '.join(merged)}\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 #{item['id']} [{item['media_type']}]\n" + f"From user {item['owner_id']}\n" + f"Keywords: {item['keywords'].replace(',', ', ')}\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 #{item['id']}\n" + f"Current keywords: {item['keywords']}\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 = { diff --git a/db.py b/db.py index 863180b..c30391c 100644 --- a/db.py +++ b/db.py @@ -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 . 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 diff --git a/test_bot.py b/test_bot.py index 9db25a0..81ae185 100644 --- a/test_bot.py +++ b/test_bot.py @@ -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.") diff --git a/test_db.py b/test_db.py index fc1c41d..3fd38b8 100644 --- a/test_db.py +++ b/test_db.py @@ -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.")