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()
|
