diff --git a/.gitignore b/.gitignore index 65a26f9..4843584 100644 --- a/.gitignore +++ b/.gitignore @@ -3,3 +3,4 @@ bot_db/ __pycache__/ *.pyc push.sh +TODO diff --git a/README.fa.md b/README.fa.md index 22b98a8..8cb5a7a 100644 --- a/README.fa.md +++ b/README.fa.md @@ -1,6 +1,6 @@ # Memes Bot -ربات شخصی تلگرام برای ذخیره و جستجوی میم (عکس، گیف، ویدیو، صدا) با کلیدواژه. +ربات شخصی تلگرام برای ذخیره و جستجوی میم (عکس، گیف، ویدیو، صدا، موزیک، استیکر، فایل) با کلیدواژه. [English](README.md) @@ -50,7 +50,7 @@ docker compose up -d --build ### نحوه استفاده -**افزودن میم** — یک عکس، گیف، ویدیو یا صدا برای ربات بفرستید. ربات از شما کلیدواژه می‌خواهد. کلیدواژه‌ها را یکی‌یکی یا با کاما جدا بفرستید، سپس `/done` بزنید. +**افزودن میم** — یک عکس، گیف، ویدیو، صدا، موزیک، استیکر یا هر فایلی برای ربات بفرستید. ربات از شما کلیدواژه می‌خواهد. کلیدواژه‌ها را یکی‌یکی یا با کاما جدا بفرستید، سپس `/done` بزنید. **جستجوی اینلاین** — در هر چتی بنویسید `@نام_ربات کلیدواژه` تا نتایج نمایش داده شود. diff --git a/README.md b/README.md index af2c9f1..f723d6e 100644 --- a/README.md +++ b/README.md @@ -1,6 +1,6 @@ # Memes Bot -A personal Telegram inline bot to store and search media (photos, GIFs, videos, voice) by keywords. +A personal Telegram inline bot to store and search media (photos, GIFs, videos, voice, music, stickers, files) by keywords. [فارسی](README.fa.md) @@ -50,7 +50,9 @@ The database will be saved in `./bot_db/media.db`. ### Usage -**Adding media** — send the bot a photo, GIF, video, or voice message, then type keywords when prompted (comma-separated or one per message), then `/done`. +**Adding media** — send the bot a photo, GIF, video, voice message, music file, sticker, or any document, then type keywords when prompted (comma-separated or one per message), then `/done`. + +Round video messages are stored as regular videos: Telegram has no inline result type for video notes, so keeping them as video is what makes them searchable. **Searching inline** — in any chat type `@yourbotusername keyword` to search and send. diff --git a/VERSION b/VERSION index 6d7de6e..9084fa2 100644 --- a/VERSION +++ b/VERSION @@ -1 +1 @@ -1.0.2 +1.1.0 diff --git a/bot.py b/bot.py index 9663bd7..a486916 100644 --- a/bot.py +++ b/bot.py @@ -15,6 +15,8 @@ from aiogram.types import ( InlineKeyboardButton, InlineKeyboardMarkup, InlineQuery, + InlineQueryResultCachedAudio, + InlineQueryResultCachedDocument, InlineQueryResultCachedMpeg4Gif, InlineQueryResultCachedPhoto, InlineQueryResultCachedSticker, @@ -93,7 +95,7 @@ class EditMedia(StatesGroup): def menu_text(user_id: int) -> str: if is_admin(user_id): return ( - "Send me any photo, GIF, video, voice, or sticker.\n" + "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" "/edit [search] — edit keywords of a meme\n" @@ -154,40 +156,45 @@ async def _receive_media(msg: Message, state: FSMContext, file_id: str, file_uni ) -@dp.message(F.photo, StateFilter(None)) -async def on_photo(msg: Message, state: FSMContext): +# Ordered: Telegram sets `document` alongside `animation` for backward +# compatibility, so animations must be matched before documents or every GIF +# would be filed as a file. Round videos are stored as plain "video" so they +# stay reachable from inline search, which has no cached video_note result. +MEDIA_EXTRACTORS = ( + ("photo", lambda m: m.photo[-1] if m.photo else None), + ("gif", lambda m: m.animation), + ("sticker", lambda m: m.sticker), + ("video", lambda m: m.video), + ("video", lambda m: m.video_note), + ("voice", lambda m: m.voice), + ("audio", lambda m: m.audio), + ("document", lambda m: m.document), +) + +MEDIA_FILTER = ( + F.photo | F.animation | F.sticker | F.video + | F.video_note | F.voice | F.audio | F.document +) + + +def extract_media(msg: Message) -> tuple[str, str, str] | None: + """Return (media_type, file_id, file_unique_id) for any supported media.""" + for media_type, getter in MEDIA_EXTRACTORS: + obj = getter(msg) + if obj is not None: + return media_type, obj.file_id, obj.file_unique_id + return None + + +@dp.message(MEDIA_FILTER, StateFilter(None)) +async def on_media(msg: Message, state: FSMContext): if not is_admin(msg.from_user.id): return - photo = msg.photo[-1] - await _receive_media(msg, state, photo.file_id, photo.file_unique_id, "photo") - - -@dp.message(F.animation, StateFilter(None)) -async def on_gif(msg: Message, state: FSMContext): - if not is_admin(msg.from_user.id): + found = extract_media(msg) + if found is None: return - await _receive_media(msg, state, msg.animation.file_id, msg.animation.file_unique_id, "gif") - - -@dp.message(F.video, StateFilter(None)) -async def on_video(msg: Message, state: FSMContext): - if not is_admin(msg.from_user.id): - return - await _receive_media(msg, state, msg.video.file_id, msg.video.file_unique_id, "video") - - -@dp.message(F.voice, StateFilter(None)) -async def on_voice(msg: Message, state: FSMContext): - if not is_admin(msg.from_user.id): - return - await _receive_media(msg, state, msg.voice.file_id, msg.voice.file_unique_id, "voice") - - -@dp.message(F.sticker, StateFilter(None)) -async def on_sticker(msg: Message, state: FSMContext): - if not is_admin(msg.from_user.id): - return - await _receive_media(msg, state, msg.sticker.file_id, msg.sticker.file_unique_id, "sticker") + media_type, file_id, file_unique_id = found + await _receive_media(msg, state, file_id, file_unique_id, media_type) # ── step 2: collect keywords ───────────────────────────────────────────────── @@ -222,7 +229,7 @@ async def on_keyword(msg: Message, state: FSMContext): await msg.reply(f"Keywords so far: {', '.join(merged)}\nSend more or /done.", parse_mode="HTML") -@dp.message(AddMedia.waiting_for_keywords, F.photo | F.animation | F.video | F.voice | F.sticker) +@dp.message(AddMedia.waiting_for_keywords, MEDIA_FILTER) async def on_media_during_keywords(msg: Message): await msg.reply("Still waiting for keywords. Send /done to save the current item, or /cancel to discard.") @@ -375,6 +382,26 @@ 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") +# 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 = { + "photo": lambda rid, fid, kw: InlineQueryResultCachedPhoto( + id=rid, photo_file_id=fid, title=kw, description=kw), + "gif": lambda rid, fid, kw: InlineQueryResultCachedMpeg4Gif( + id=rid, mpeg4_file_id=fid, title=kw), + "video": lambda rid, fid, kw: InlineQueryResultCachedVideo( + id=rid, video_file_id=fid, title=kw, description=kw), + "voice": lambda rid, fid, kw: InlineQueryResultCachedVoice( + id=rid, voice_file_id=fid, title=kw), + "sticker": lambda rid, fid, kw: InlineQueryResultCachedSticker( + id=rid, sticker_file_id=fid), + "audio": lambda rid, fid, kw: InlineQueryResultCachedAudio( + id=rid, audio_file_id=fid), + "document": lambda rid, fid, kw: InlineQueryResultCachedDocument( + id=rid, document_file_id=fid, title=kw or "file", description=kw), +} + + # ── inline mode ────────────────────────────────────────────────────────────── @dp.inline_query() @@ -392,28 +419,9 @@ async def inline_handler(query: InlineQuery): kw_display = item["keywords"].replace(",", ", ") mtype = item["media_type"] - if mtype == "photo": - results.append(InlineQueryResultCachedPhoto( - id=rid, photo_file_id=item["file_id"], - title=kw_display, description=kw_display, - )) - elif mtype == "gif": - results.append(InlineQueryResultCachedMpeg4Gif( - id=rid, mpeg4_file_id=item["file_id"], title=kw_display, - )) - elif mtype == "video": - results.append(InlineQueryResultCachedVideo( - id=rid, video_file_id=item["file_id"], - title=kw_display, description=kw_display, - )) - elif mtype == "voice": - results.append(InlineQueryResultCachedVoice( - id=rid, voice_file_id=item["file_id"], title=kw_display, - )) - elif mtype == "sticker": - results.append(InlineQueryResultCachedSticker( - id=rid, sticker_file_id=item["file_id"], - )) + builder = INLINE_BUILDERS.get(mtype) + if builder is not None: + results.append(builder(rid, item["file_id"], kw_display)) if not results: results = [InlineQueryResultArticle( diff --git a/test_bot.py b/test_bot.py new file mode 100644 index 0000000..9db25a0 --- /dev/null +++ b/test_bot.py @@ -0,0 +1,95 @@ +"""Media handling tests. Run with: python test_bot.py + +Covers the dispatch table that decides how an incoming message is stored and +how it comes back through inline search. +""" +import datetime +import os +import sys + +os.environ.setdefault("BOT_TOKEN", "123456789:AAHtestTokenPlaceholder_ABCDEFGHIJKLMNOP") +os.environ.setdefault("DB_PATH", ":memory:") +sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) + +from aiogram.types import (Animation, Audio, Chat, Document, Message, PhotoSize, + Sticker, Video, VideoNote, Voice) + +import bot as B + +CHAT = Chat(id=1, type="private") +NOW = datetime.datetime.now() + + +def msg(**kw) -> Message: + return Message(message_id=1, date=NOW, chat=CHAT, **kw) + + +def sample_messages(): + photo = PhotoSize(file_id="F_photo", file_unique_id="U_photo", width=1, height=1) + anim = Animation(file_id="F_gif", file_unique_id="U_gif", width=1, height=1, duration=1) + sticker = Sticker(file_id="F_st", file_unique_id="U_st", width=1, height=1, + type="regular", is_animated=False, is_video=False) + return [ + ("photo", msg(photo=[photo])), + # Telegram sets document alongside animation for backward compatibility. + ("gif", msg(animation=anim, document=Document(file_id="F_gif", file_unique_id="U_gif"))), + ("sticker", msg(sticker=sticker)), + ("video", msg(video=Video(file_id="F_v", file_unique_id="U_v", width=1, height=1, duration=1))), + ("video", msg(video_note=VideoNote(file_id="F_vn", file_unique_id="U_vn", length=1, duration=1))), + ("voice", msg(voice=Voice(file_id="F_vo", file_unique_id="U_vo", duration=1))), + ("audio", msg(audio=Audio(file_id="F_a", file_unique_id="U_a", duration=1))), + ("document", msg(document=Document(file_id="F_doc", file_unique_id="U_doc"))), + ] + + +def test_every_type_extracts(): + for expected, message in sample_messages(): + found = B.extract_media(message) + assert found is not None, f"{expected}: nothing extracted" + assert found[0] == expected, f"expected {expected}, got {found[0]}" + print(" every supported media type extracts to the right media_type") + + +def test_animation_beats_document(): + gif = sample_messages()[1][1] + assert gif.document is not None, "test fixture should mimic Telegram's dual field" + assert B.extract_media(gif)[0] == "gif", "a GIF must not be filed as a document" + print(" animation wins over the backward-compat document field") + + +def test_video_note_stored_as_video(): + note = sample_messages()[4][1] + assert B.extract_media(note) == ("video", "F_vn", "U_vn") + print(" video notes are stored as video so inline search can return them") + + +def test_non_media_ignored(): + assert B.extract_media(msg(text="hello")) is None + print(" plain text messages are ignored") + + +def test_every_stored_type_renders_inline(): + stored = {t for t, _ in B.MEDIA_EXTRACTORS} + missing = stored - set(B.INLINE_BUILDERS) + assert not missing, f"stored but not renderable inline: {sorted(missing)}" + print(" every stored media_type has a matching inline builder") + + +def test_builders_construct(): + for media_type, build in B.INLINE_BUILDERS.items(): + build("rid", f"FID_{media_type}", "cat, funny") + # Documents and videos require a non-empty title; keywords can be blank. + build("rid", f"FID_{media_type}", "") + print(" inline builders construct, including with empty keywords") + + +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() + print("\nAll media tests passed.") + + +if __name__ == "__main__": + main()