Blame for app/services/classifier.py

"""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