Spaces:
Running on Zero
Running on Zero
| """Smart file understanding: which course, which kind, a clean title, duplicate fingerprints. | |
| Pure functions (no I/O) so everything here is unit-tested.""" | |
| import hashlib | |
| import re | |
| import urllib.parse | |
| from typing import Optional, Sequence | |
| from app.services.fuzzy import words | |
| # ---------------------------------------------------------------- helpers | |
| _STOP = {"for", "of", "the", "and", "to", "in", "on", "a", "an"} | |
| _GENERIC = {"general", "introduction", "skills", "language", "sciences", "science", "education", "technologies", "communicative"} | |
| def collapse(text: str) -> str: | |
| return re.sub(r"[^a-z0-9\u1200-\u137f]+", "", (text or "").lower()) | |
| def norm_title_key(title: str) -> str: | |
| return collapse(title) | |
| # Short, hand-written synonyms (collapsed course name -> aliases). Cheap to extend. | |
| ALIASES: dict[str, list[str]] = { | |
| "mathematicsfornaturalsciences": ["ሒሳብ", "ሂሳብ", "calculus", "natural math", "math natural"], | |
| "appliedmathematicsi": ["applied math", "applied maths", "applied mathematics"], | |
| "mathematicsforsocialsciences": ["social math", "math social", "maths social"], | |
| "generalphysics": ["physics", "phys", "mechanics", "ፊዚክስ"], | |
| "generalchemistry": ["chemistry", "chem", "ኬሚስትሪ"], | |
| "generalbiology": ["biology", "bio", "ባዮሎጂ"], | |
| "communicativeenglishlanguageskillsi": ["english", "eng", "flen", "comm english", "እንግሊዝኛ"], | |
| "communicativeenglishlanguageskillsii": ["english", "eng", "flen", "comm english", "እንግሊዝኛ"], | |
| "generalpsychology": ["psychology", "psych", "psys", "ሳይኮሎጂ"], | |
| "criticalthinking": ["አመክንዮ", "logic", "loct", "logic and critical thinking"], | |
| "geographyofethiopiaandthehorn": ["geography", "geo", "gees", "ጂኦግራፊ"], | |
| "historyofethiopiaandthehorn": ["history", "hist", "ታሪክ"], | |
| "socialanthropology": ["አንትሮፖሎጂ", "anthropology", "anth"], | |
| "entrepreneurship": ["ስራ ፈጠራ", "ሥራ ፈጠራ", "entrepreneur", "entre", "mgmt"], | |
| "introductiontoemergingtechnologies": ["emerging", "emte", "emerging technology"], | |
| "computerprogramming": ["programming", "coding", "eceg", "python", "c++"], | |
| "moralandciviceducation": ["ሲቪክስ", "ሥነ ምግባር", "civic", "civics", "moral", "mcie", "ethics"], | |
| "physicalfitness": ["ስፖርት", "fitness", "sports", "spsc"], | |
| "introductiontoeconomics": ["ኢኮኖሚክስ", "ምጣኔ ሀብት", "economics", "econ"], | |
| "globalaffairs": ["global", "global trends"], | |
| "introductiontolawandtheethiopianlegalsystem": ["ህግ", "ሕግ", "law", "legal"], | |
| "freshmanguideandgeneralresources": [ | |
| "cut point", "cutt point", "cutpoint", "cutoff", "cut off", "placement", "department selection", | |
| "quantitative reasoning", "verbal reasoning", "logical reasoning", "aptitude", "reasoning", "verbal", | |
| "quantitative", "guide", "handbook", "calendar", "schedule", "scholarship", "orientation", "exit exam", | |
| ], | |
| } | |
| def _hit(word: str, target: str) -> bool: | |
| """word (from a file name) refers to target (a course-name word).""" | |
| if word == target: | |
| return True | |
| if len(word) >= 4 and (target.startswith(word) or word.startswith(target)): | |
| return True | |
| return len(word) >= 5 and len(target) >= 5 and _ratio(word, target) >= 0.85 | |
| def _ratio(a: str, b: str) -> float: | |
| import difflib | |
| return difflib.SequenceMatcher(None, a, b).ratio() | |
| def _alias_hit(alias: str, ws: Sequence[str], fuzzy_ok: bool = True) -> int: | |
| """0 = no match, 1 = prefix match, 2 = exact match, 3 = close typo of a whole word.""" | |
| parts = words(alias) | |
| if not parts: | |
| return 0 | |
| if len(parts) > 1: | |
| return 2 if all(any(w == p for w in ws) for p in parts) else 0 | |
| p = parts[0] | |
| best = 0 | |
| for w in ws: | |
| base = re.sub(r"\d+$", "", w) | |
| if w == p or base == p: | |
| return 2 | |
| if len(p) >= 4 and w.startswith(p): | |
| best = max(best, 1) | |
| elif fuzzy_ok and len(p) >= 6 and len(w) >= 6 and w[0] == p[0] and _ratio(w, p) >= 0.7: # "psyscho" -> psychology | |
| best = max(best, 3) | |
| return best | |
| def _level(ws: Sequence[str]) -> Optional[int]: | |
| if any(w in ("ii", "2") for w in ws) or any(re.fullmatch(r"[a-z]*(1002|ii)", w) for w in ws): | |
| return 2 | |
| if any(w in ("i", "1") for w in ws): | |
| return 1 | |
| return None | |
| # ---------------------------------------------------------------- course matching | |
| def rank_courses(courses: Sequence, sources: Sequence[tuple[str, str]]) -> list[tuple[float, object]]: | |
| """sources: [(text, 'name'|'text')]. 'name' = file name/caption (strong), 'text' = content (weak). | |
| Returns [(score, course)] best first, only courses with score > 0.""" | |
| prepared = [] | |
| for text, kind in sources: | |
| if text and text.strip(): | |
| prepared.append((collapse(text), words(text), 1.0 if kind == "name" else 0.5)) | |
| if not prepared: | |
| return [] | |
| out = [] | |
| for c in courses: | |
| cname_key = collapse(c.name) | |
| code_key = collapse(c.code) if c.code else "" | |
| sig_words = [w for w in words(c.name) if len(w) >= 4 and w not in _STOP and w not in _GENERIC] | |
| score = 0.0 | |
| for compact, ws, weight in prepared: | |
| if code_key and code_key in compact: | |
| score += 10 * weight | |
| if cname_key and cname_key in compact: | |
| score += 8 * weight | |
| best_alias = 0 | |
| for alias in ALIASES.get(cname_key, ()): | |
| # the catch-all bucket only matches exact words ('reading' must never look like 'reasoning') | |
| strength = _alias_hit(alias, ws, fuzzy_ok=cname_key != "freshmanguideandgeneralresources") | |
| if strength: | |
| multi = len(alias.split()) > 1 | |
| best_alias = max(best_alias, 8 if multi else (5 if strength == 1 else 6)) | |
| score += best_alias * weight | |
| hits = sum(1 for nw in sig_words if any(_hit(w, nw) for w in ws)) | |
| score += min(hits * 2.0 * weight, 4.0 if weight < 1 else 8.0) | |
| # "English I" vs "English II": use the level written in the file name | |
| name_words = words(c.name) | |
| if name_words and name_words[-1] in ("i", "ii") and score > 0: | |
| lvl = _level([w for _, ws, wt in prepared if wt == 1.0 for w in ws]) | |
| if lvl: | |
| is_two = name_words[-1] == "ii" | |
| score += 3 if (lvl == 2) == is_two else -3 | |
| if score > 0: | |
| out.append((score, c)) | |
| out.sort(key=lambda item: (-item[0], item[1].id)) | |
| return out | |
| def decide(ranked: Sequence[tuple[float, object]]) -> tuple[Optional[object], bool, list]: | |
| """-> (best course or None, confident?, top-3 suggestions).""" | |
| if not ranked: | |
| return None, False, [] | |
| top = ranked[0][0] | |
| second = ranked[1][0] if len(ranked) > 1 else 0.0 | |
| confident = top >= 6 and (top - second >= 3 or second <= 0) | |
| return ranked[0][1], confident, [c for _, c in ranked[:3]] | |
| # ---------------------------------------------------------------- kind | |
| _EXAM_STRONG = {"exam", "exams", "examination", "midterm", "midexam", "quiz", "quizzes", "ፈተና", "ፈተናዎች", "ፈተናው"} | |
| _EXAM_PHRASES = ("past paper", "model exam", "question paper", "mid exam", "final exam") | |
| _EXAM_WEAK = {"final", "test", "tests", "question", "questions", "previous", "mid", "ጥያቄ", "ጥያቄዎች"} # "Module Final" is NOT an exam | |
| _SLIDE = {"slide", "slides", "ppt", "pptx", "presentation", "lecture", "lectures", "ስላይድ", "ስላይዶች", "ፕሬዘንቴሽን"} | |
| _NOTE = {"note", "notes", "handout", "handouts", "summary", "summaries", "worksheet", "assignment", "tutorial", "module", | |
| "modules", "lab", "ማስታወሻ", "ሞጁል", "ሞዱል"} | |
| _BOOK = {"textbook", "book", "books", "edition", "isbn", "volume", "መጽሐፍ", "መፅሐፍ", "መጽሀፍ", "መፅሀፍ"} | |
| _EXAM_TEXT = ("time allowed", "answer all questions", "choose the best answer", "instruction:", "id no", "name:") | |
| def _kind_hit(tokens: Sequence[str], vocab: set) -> bool: | |
| """A whole word of the file name is a keyword, or a close typo of a longer one ('exm', 'quize', 'slid').""" | |
| for t in tokens: | |
| if t in vocab: | |
| return True | |
| if len(t) >= 3 and t.isascii(): | |
| for k in vocab: | |
| if k.isascii() and len(k) >= 4 and t[0] == k[0] and abs(len(t) - len(k)) <= 2 and _ratio(t, k) >= 0.8: | |
| return True | |
| return False | |
| def detect_kind(name: str, caption: str, text: str, pages: Optional[int], size: Optional[int], ext: str) -> tuple[str, str]: | |
| label = f"{name or ''} {caption or ''}".lower() | |
| tokens = words(label) | |
| spaced = " " + " ".join(tokens) + " " | |
| if ext in (".ppt", ".pptx"): | |
| return "slide", "presentation file" | |
| if _kind_hit(tokens, _EXAM_STRONG) or any(f" {p} " in spaced for p in _EXAM_PHRASES): | |
| return "exam", "name says exam" | |
| if _kind_hit(tokens, _BOOK): | |
| return "book", "name says book" | |
| if _kind_hit(tokens, _NOTE): | |
| return "note", "name says module/notes" # beats the ambiguous words "final" and "chapter" | |
| if _kind_hit(tokens, _SLIDE): | |
| return "slide", "name says slides/lecture" | |
| if _kind_hit(tokens, _EXAM_WEAK): | |
| return "exam", "name hints at an exam" | |
| if sum(1 for k in _EXAM_TEXT if k in (text or "").lower()) >= 2: | |
| return "exam", "content looks like an exam" | |
| if pages and pages >= 350: # national modules are 150-300 pages: those stay "note" | |
| return "book", f"{pages} pages" | |
| if size and size > 40 * 1024 * 1024: | |
| return "book", "very large file" | |
| return "note", "default" | |
| # ---------------------------------------------------------------- title | |
| _JUNK = re.compile(r"(https?://\S+|t\.me/\S+|@\w+|www\.\S+|\(\d+\)|\bcopy\b|\bscanned\b)", re.I) | |
| def detect_year(*texts: str) -> Optional[str]: | |
| for t in texts: | |
| m = re.search(r"\b(20[0-3]\d)\b", t or "") | |
| if m: | |
| return m.group(1) | |
| return None | |
| def exam_stage(*texts: str) -> Optional[str]: | |
| blob = " ".join(texts).lower() | |
| if "mid" in blob: | |
| return "Mid" | |
| if "final" in blob: | |
| return "Final" | |
| if "quiz" in blob: | |
| return "Quiz" | |
| return None | |
| def clean_title(file_name: Optional[str], caption: Optional[str], course_name: str, kind: str, year: Optional[str], stage: Optional[str]) -> str: | |
| base = re.sub(r"\.[A-Za-z0-9]{2,5}$", "", urllib.parse.unquote(file_name or "")) | |
| base = _JUNK.sub(" ", base) # junk first: "@some_channel" must go before underscores become spaces | |
| base = base.replace("_", " ").replace("-", " ").replace(".", " ") | |
| parts = [] | |
| for tok in base.split(): | |
| if re.fullmatch(r"\d{8,}", tok) or (len(tok) >= 12 and re.search(r"\d", tok) and re.search(r"[a-z]", tok, re.I)): | |
| continue # phone-number / hash-like junk | |
| parts.append(tok) | |
| deduped: list[str] = [] | |
| for tok in parts: | |
| if deduped and deduped[-1].lower() == tok.lower(): | |
| continue # "final FINAL" -> "final" | |
| deduped.append(tok.capitalize() if tok.islower() and len(tok) > 3 else tok) | |
| title = " ".join(deduped).strip() | |
| if not title and caption: | |
| title = _JUNK.sub(" ", caption.strip().split("\n")[0]).strip() | |
| if not title: | |
| title = f"{course_name} {kind}" | |
| if title.isupper() or title.islower(): | |
| title = title.title() | |
| if kind == "exam": | |
| if stage and stage.lower() not in title.lower(): | |
| title += f" {stage}" | |
| if year and year not in title: | |
| title += f" {year}" | |
| return re.sub(r"\s+", " ", title)[:80] | |
| def display_title(title: str) -> str: | |
| """What students see. Fixes titles saved by older versions (%28 codes, underscores, 'final FINAL') | |
| without touching anything that was renamed by hand.""" | |
| out: list[str] = [] | |
| for tok in urllib.parse.unquote(title or "").replace("_", " ").split(): | |
| if out and out[-1].lower() == tok.lower(): | |
| continue | |
| out.append(tok) | |
| return " ".join(out) | |
| def disambiguate(title: str, taken: set, *, year: Optional[str], pages: Optional[int], size: Optional[int]) -> str: | |
| """Same title already used for a DIFFERENT file in this course/kind -> add what tells them apart.""" | |
| options = [] | |
| if year and year not in title: | |
| options.append(f"{title} ({year})") | |
| if pages: | |
| options.append(f"{title} ({pages} p)") | |
| if size: | |
| options.append(f"{title} ({size / 1048576:.1f} MB)") | |
| for candidate in options: | |
| if norm_title_key(candidate) not in taken: | |
| return candidate | |
| n = 2 | |
| while norm_title_key(f"{title} ({n})") in taken: | |
| n += 1 | |
| return f"{title} ({n})" | |
| # ---------------------------------------------------------------- near-duplicate fingerprint | |
| def simhash64(text: str) -> Optional[int]: | |
| """64-bit SimHash of word 3-grams (signed, fits a BIGINT). None if there is too little text.""" | |
| ws = words(text) | |
| if len(ws) < 40: | |
| return None | |
| vec = [0] * 64 | |
| for i in range(len(ws) - 2): | |
| h = int.from_bytes(hashlib.blake2b(" ".join(ws[i : i + 3]).encode(), digest_size=8).digest(), "big") | |
| for b in range(64): | |
| vec[b] += 1 if (h >> b) & 1 else -1 | |
| out = 0 | |
| for b in range(64): | |
| if vec[b] > 0: | |
| out |= 1 << b | |
| return out - (1 << 64) if out >= (1 << 63) else out | |
| def hamming(a: int, b: int) -> int: | |
| return ((a ^ b) & ((1 << 64) - 1)).bit_count() | |
| NEAR_DUPLICATE_BITS = 3 | |