Spaces:
Running on Zero
Running on Zero
| """Inline mode: type @YourBot physics in ANY chat to share a file (or a course link) instantly. | |
| Every use shows the bot to the whole chat: the main organic-growth channel.""" | |
| import html | |
| from aiogram import Router | |
| from aiogram.types import ( | |
| InlineKeyboardButton, | |
| InlineKeyboardMarkup, | |
| InlineQuery, | |
| InlineQueryResultArticle, | |
| InlineQueryResultCachedAudio, | |
| InlineQueryResultCachedDocument, | |
| InlineQueryResultCachedPhoto, | |
| InlineQueryResultCachedVideo, | |
| InlineQueryResultsButton, | |
| InputTextMessageContent, | |
| ) | |
| from app.config import get_settings | |
| from app.core.i18n import t | |
| from app.db.models import ActivityEvent, Course, Material | |
| from app.db.repos import library as repo | |
| from app.db.session import SessionFactory | |
| from app.services import library as lib | |
| from app.services.classifier import display_title | |
| router = Router() | |
| def _file_result(m: Material, course: Course | None): | |
| m_title = display_title(m.title) | |
| desc = f"{course.name} · #{m.kind}" if course else f"#{m.kind}" | |
| caption = f"📄 {m_title}\n{desc}" | |
| rid = f"m{m.id}" | |
| if m.file_type == "video": | |
| return InlineQueryResultCachedVideo(id=rid, video_file_id=m.file_id, title=m_title, description=desc, caption=caption) | |
| if m.file_type == "audio": | |
| return InlineQueryResultCachedAudio(id=rid, audio_file_id=m.file_id, caption=caption) | |
| if m.file_type == "photo": | |
| return InlineQueryResultCachedPhoto(id=rid, photo_file_id=m.file_id, title=m_title, description=desc, caption=caption) | |
| return InlineQueryResultCachedDocument(id=rid, document_file_id=m.file_id, title=m_title, description=desc, caption=caption) | |
| def _course_result(c: Course, n_files: int, bot_username: str) -> InlineQueryResultArticle: | |
| """A course card that works even before any file exists; it links into the bot.""" | |
| link = f"https://t.me/{bot_username}?start=c{c.id}" | |
| title = f"📘 {c.code} · {c.name}" if c.code else f"📘 {c.name}" | |
| label = f"📁 {n_files} files" if n_files else "📁 No files yet" | |
| code = f" ({html.escape(c.code)})" if c.code else "" | |
| return InlineQueryResultArticle( | |
| id=f"c{c.id}", | |
| title=title, | |
| description=label, | |
| input_message_content=InputTextMessageContent( | |
| message_text=f"📘 <b>{html.escape(c.name)}</b>{code}\n{label}\n👉 {link}", parse_mode="HTML" | |
| ), | |
| reply_markup=InlineKeyboardMarkup(inline_keyboard=[[InlineKeyboardButton(text="📚 Open in bot", url=link)]]), | |
| ) | |
| async def inline_search(query: InlineQuery) -> None: | |
| text = (query.query or "").strip() | |
| me = await query.bot.me() | |
| async with SessionFactory() as session: | |
| if len(text) >= 2: | |
| courses, materials = await lib.search(session, text, course_limit=8, material_limit=20) | |
| else: # nothing typed yet: popular files, else the first courses | |
| materials = await repo.top_materials(session, 20) | |
| courses = (await lib.all_courses_cached(session))[:10] | |
| by_id = await repo.get_courses_map(session, {m.course_id for m in materials if m.course_id}) | |
| counts = await repo.file_counts(session) | |
| if get_settings().hide_empty_courses: | |
| courses = [c for c in courses if counts.get(c.id)] | |
| results = [_file_result(m, by_id.get(m.course_id)) for m in materials if m.file_id] | |
| results += [_course_result(c, counts.get(c.id, 0), me.username) for c in courses] | |
| await query.answer( | |
| results=results[:50], | |
| cache_time=10, | |
| is_personal=False, | |
| button=InlineQueryResultsButton(text=t("en", "inline_open_bot"), start_parameter="inline"), | |
| ) | |
| if len(text) >= 2: # remember what was searched (after answering, so the student never waits for it) | |
| try: | |
| async with SessionFactory() as session: | |
| session.add(ActivityEvent(user_id=query.from_user.id, kind="inline", ref=text[:100].lower(), n=len(results))) | |
| await session.commit() | |
| except Exception: | |
| pass | |