Spaces:
Running on Zero
Running on Zero
File size: 4,249 Bytes
120bd9a d896946 120bd9a a0261c0 120bd9a d896946 9318723 120bd9a d925e38 a0261c0 d925e38 a0261c0 d925e38 120bd9a d896946 120bd9a d896946 a0261c0 d896946 120bd9a d896946 120bd9a 9318723 120bd9a 4e64e70 120bd9a c3b9247 120bd9a | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 | 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)
|