from __future__ import annotations from dataclasses import dataclass from datetime import date import hashlib import re from typing import Callable, Iterable, Sequence from .models import Candidate @dataclass(frozen=True) class ScanPage: candidates: tuple[Candidate, ...] fingerprint: str @dataclass(frozen=True) class ScanOutcome: status: str candidates: tuple[Candidate, ...] screens_scanned: int unique_candidates: int consecutive_no_new: int def score_candidate(candidate: Candidate, *, query: str, aliases: Sequence[str] = (), institutions: Sequence[str] = (), minimum_pages: int = 1) -> int: title = candidate.title.casefold() terms = [query.casefold(), *(a.casefold() for a in aliases)] score = 0 if any(term and term in title for term in terms): score += 100 if candidate.institution and candidate.institution in institutions: score += 20 if candidate.page_count is not None: score += 10 if candidate.page_count >= minimum_pages else -100 if candidate.report_date: score += 3 return score class FastScanner: def __init__(self, page_provider: Callable[[int], ScanPage]): self.page_provider = page_provider def scan(self, *, quantity: int, min_screens: int = 3, normal_max_screens: int = 5, hard_max_screens: int = 20, hard_max_candidates: int = 100, no_new_stop: int = 2) -> ScanOutcome: selected: dict[str, Candidate] = {} no_new = 0 previous_fingerprint = None for screen in range(1, hard_max_screens + 1): page = self.page_provider(screen) before = len(selected) for candidate in page.candidates: selected.setdefault(candidate.candidate_id, candidate) if len(selected) >= hard_max_candidates: break no_new = no_new + 1 if len(selected) == before or page.fingerprint == previous_fingerprint else 0 previous_fingerprint = page.fingerprint ranked = sorted(selected.values(), key=lambda c: (-c.score, c.candidate_id)) enough = len([c for c in ranked if c.score >= 100]) >= quantity if screen >= min_screens and enough: return ScanOutcome("ENOUGH_CONFIRMED", tuple(ranked), screen, len(selected), no_new) if screen >= min_screens and no_new >= no_new_stop: return ScanOutcome("EXHAUSTED", tuple(ranked), screen, len(selected), no_new) if screen >= normal_max_screens and len(ranked) >= quantity: return ScanOutcome("ENOUGH_CONFIRMED", tuple(ranked), screen, len(selected), no_new) if len(selected) >= hard_max_candidates: return ScanOutcome("EXHAUSTED", tuple(ranked), screen, len(selected), no_new) ranked = sorted(selected.values(), key=lambda c: (-c.score, c.candidate_id)) return ScanOutcome("EXHAUSTED", tuple(ranked), hard_max_screens, len(selected), no_new) def fingerprint_text(lines: Iterable[str]) -> str: normalized = "\n".join(" ".join(line.split()) for line in lines if line.strip()) return hashlib.sha256(normalized.encode("utf-8")).hexdigest()