Spaces:
Running on Zero
Running on Zero
| import re | |
| import time | |
| from typing import Optional | |
| from aiogram import Bot | |
| from sqlalchemy.ext.asyncio import AsyncSession | |
| from app.config import get_settings | |
| from app.db.models import Course, Material, User | |
| from app.db.repos import library as repo | |
| from app.services import fuzzy | |
| from app.services.classifier import display_title as fuzzy_display | |
| PAGE_SIZE = 8 | |
| _STRIP = re.compile(r"[%_\\]") | |
| def tokenize(query: str, limit: int = 5) -> list[str]: | |
| return [t for t in _STRIP.sub(" ", query.lower()).split() if t][:limit] | |
| async def courses_page(session: AsyncSession, user: User, page: int, scope: str = "c") -> tuple[list[Course], int, bool]: | |
| """(courses on this page, total, are there more courses in other streams?). | |
| scope "c" = the student's own stream and year, "a" = every course that has files.""" | |
| only = get_settings().hide_empty_courses | |
| if scope == "a": | |
| return ( | |
| await repo.list_all_courses(session, page * PAGE_SIZE, PAGE_SIZE, only), | |
| await repo.count_all_courses(session, only), | |
| False, | |
| ) | |
| total = await repo.count_courses(session, user.stream, user.year, only) | |
| courses = await repo.list_courses(session, user.stream, user.year, page * PAGE_SIZE, PAGE_SIZE, only) | |
| return courses, total, await repo.count_all_courses(session, only) > total | |
| async def kinds_of(session: AsyncSession, course_id: int) -> tuple[Optional[Course], dict[str, int]]: | |
| return await repo.get_course(session, course_id), await repo.kind_counts(session, course_id) | |
| async def materials_page( | |
| session: AsyncSession, course_id: int, kind: str, page: int | |
| ) -> tuple[Optional[Course], list[Material], int]: | |
| course = await repo.get_course(session, course_id) | |
| total = await repo.count_materials(session, course_id, kind) | |
| items = await repo.list_materials(session, course_id, kind, page * PAGE_SIZE, PAGE_SIZE) | |
| return course, items, total | |
| _courses_cache: dict = {"at": 0.0, "items": []} | |
| async def all_courses_cached(session: AsyncSession) -> list[Course]: | |
| """The course list is tiny and rarely changes: keep it in memory for a minute.""" | |
| if time.monotonic() - _courses_cache["at"] > 60: | |
| _courses_cache["items"] = await repo.all_courses(session) | |
| _courses_cache["at"] = time.monotonic() | |
| return _courses_cache["items"] | |
| def invalidate_courses() -> None: | |
| _courses_cache["at"] = 0.0 | |
| async def search( | |
| session: AsyncSession, query: str, course_limit: int = 5, material_limit: int = 8 | |
| ) -> tuple[list[Course], list[Material]]: | |
| tokens = tokenize(query) | |
| if not tokens: | |
| return [], [] | |
| scored = [] | |
| for course in await all_courses_cached(session): | |
| score = fuzzy.match_score(tokens, [course.name, course.code or ""]) | |
| if score is not None: | |
| scored.append((score, course)) | |
| if get_settings().hide_empty_courses: # an empty course is no use to a student | |
| with_files = await repo.file_counts(session) | |
| scored = [(sc, c) for sc, c in scored if with_files.get(c.id)] | |
| scored.sort(key=lambda item: (-item[0], item[1].id)) | |
| courses = [c for _, c in scored[:course_limit]] | |
| materials = await repo.search_materials(session, tokens, material_limit) | |
| if scored and len(materials) < material_limit: # also include files of the matching courses | |
| seen = {m.id for m in materials} | |
| extra = await repo.materials_for_courses(session, [c.id for _, c in scored[:5]], material_limit) | |
| materials += [m for m in extra if m.id not in seen] | |
| return courses, materials[:material_limit] | |
| async def send_material(bot: Bot, chat_id: int, material: Material, course: Optional[Course]) -> None: | |
| caption = f"📄 {fuzzy_display(material.title)}" | |
| if course: | |
| caption += f"\n📚 {course.name}" | |
| if material.credit: | |
| caption += f"\n✍️ {material.credit}" | |
| caption += f"\n#{material.kind}" | |
| caption += f"\n🤖 @{(await bot.me()).username}" | |
| kwargs = {material.file_type: material.file_id, "caption": caption} | |
| sender = { | |
| "document": bot.send_document, | |
| "video": bot.send_video, | |
| "audio": bot.send_audio, | |
| "photo": bot.send_photo, | |
| }[material.file_type] | |
| await sender(chat_id, **kwargs) | |