build-small-hackathon/hackathon-advisor

1from __future__ import annotations2 3import ast4from collections.abc import Callable, Sequence5from dataclasses import dataclass6from hashlib import sha2567import json8import math9from pathlib import Path10from pathlib import PurePosixPath11import re12from typing import Any13 14from hackathon_advisor._text import utc_now15 16 17TOKEN_RE = re.compile(r"[a-z0-9][a-z0-9.+_-]*", re.IGNORECASE)18HTML_TAG_RE = re.compile(r"<[^>]+>")19GENERIC_PUBLIC_TITLE_RE = re.compile(20    r"^(?:my\s+)?build\s+small\s+hackathon$",21    re.IGNORECASE,22)23GENERIC_PUBLIC_SUMMARY_RE = re.compile(24    r"(?:\bthis\s+(?:is\s+)?(?:space\s+is\s+for|my\s+submission)\b.*\b(?:build[-\s]*small|hackathon)\b)"25    r"|(?:\bhacka?ton\s+project\b)"26    r"|(?:^\s*todo\s*$)",27    re.IGNORECASE,28)29 30INDEX_SCHEMA_VERSION = 331INDEX_ALGORITHM = "llama-cpp-embedding-v1"32DEFAULT_EMBEDDING_MODEL_REPO = "ggml-org/embeddinggemma-300m-qat-q8_0-GGUF"33DEFAULT_EMBEDDING_MODEL_FILE = "embeddinggemma-300m-qat-Q8_0.gguf"34DEFAULT_EMBEDDING_RUNTIME = "llama.cpp via llama-cpp-python"35APP_FILE_EMBEDDING_CHAR_LIMIT = 200036HOSTING_METADATA_TAG_PREFIXES = ("region:",)37 38 39EmbeddingFunction = Callable[[str], Sequence[float]]40 41 42@dataclass(frozen=True)43class Project:44    id: str45    title: str46    summary: str47    tags: tuple[str, ...]48    models: tuple[str, ...]49    datasets: tuple[str, ...]50    likes: int51    sdk: str52    license: str53    created_at: str54    last_modified: str55    host: str56    url: str57    app_file: str = ""58    app_file_embedding_text: str = ""59    readme_body: str = ""60    app_file_source: str = ""61 62    @classmethod63    def from_dict(cls, data: dict) -> "Project":64        return cls(65            id=str(data["id"]),66            title=str(data.get("title") or data["id"].rsplit("/", 1)[-1]),67            summary=str(data.get("summary") or ""),68            tags=tuple(data.get("tags") or ()),69            models=tuple(data.get("models") or ()),70            datasets=tuple(data.get("datasets") or ()),71            likes=int(data.get("likes") or 0),72            sdk=str(data.get("sdk") or ""),73            license=str(data.get("license") or ""),74            created_at=str(data.get("created_at") or ""),75            last_modified=str(data.get("last_modified") or ""),76            host=str(data.get("host") or ""),77            url=str(data.get("url") or f"https://huggingface.co/spaces/{data['id']}"),78            app_file=str(data.get("app_file") or ""),79            app_file_embedding_text=str(data.get("app_file_embedding_text") or ""),80            readme_body=str(data.get("readme_body") or ""),81            app_file_source=str(data.get("app_file_source") or data.get("app_source") or ""),82        )83 84    @property85    def slug(self) -> str:86        return self.id.rsplit("/", 1)[-1]87 88    @property89    def searchable_text(self) -> str:90        return "\n".join(91            part92            for part in [93                f"title: {self.title}",94                f"slug: {self.slug.replace('-', ' ').replace('_', ' ')}",95                f"summary: {self.summary}",96                f"tags: {' '.join(self.tags)}",97                f"models: {' '.join(self.models)}",98                f"datasets: {' '.join(self.datasets)}",99                f"main app file: {self.app_file}" if self.app_file else "",100                "main app file content:\n"101                f"{bounded_embedding_text(self.app_file_embedding_text, APP_FILE_EMBEDDING_CHAR_LIMIT)}"102                if self.app_file_embedding_text103                else "",104            ]105            if part.strip()106        )107 108    def to_public_dict(self) -> dict:109        return {110            "id": self.id,111            "title": public_project_title(self.title),112            "summary": public_project_summary(self.summary),113            "tags": list(normalize_project_tags(self.tags)),114            "models": list(self.models),115            "datasets": list(self.datasets),116            "likes": self.likes,117            "sdk": self.sdk,118            "license": self.license,119            "created_at": self.created_at,120            "last_modified": self.last_modified,121            "host": self.host,122            "url": self.url,123            "app_file": self.app_file,124        }125 126    def to_snapshot_dict(self) -> dict:127        return {128            "id": self.id,129            "title": self.title,130            "summary": self.summary,131            "tags": list(self.tags),132            "models": list(self.models),133            "datasets": list(self.datasets),134            "likes": self.likes,135            "sdk": self.sdk,136            "license": self.license,137            "created_at": self.created_at,138            "last_modified": self.last_modified,139            "host": self.host,140            "url": self.url,141            "app_file": self.app_file,142            "app_file_embedding_text": self.app_file_embedding_text,143        }144 145    def to_refresh_snapshot_dict(self) -> dict:146        payload = self.to_snapshot_dict()147        payload.update(148            {149                "readme_body": self.readme_body,150                "app_file_source": self.app_file_source,151            }152        )153        return payload154 155 156@dataclass(frozen=True)157class SearchHit:158    project: Project159    score: float160    matched_terms: tuple[str, ...]161    page_number: int162 163 164@dataclass(frozen=True)165class WhitespaceItem:166    label: str167    pitch: str168    evidence: str169    score: float170    nearby_projects: tuple[Project, ...]171 172    def to_dict(self) -> dict:173        return {174            "label": self.label,175            "pitch": self.pitch,176            "evidence": self.evidence,177            "score": round(self.score, 3),178            "nearby_projects": [project.to_public_dict() for project in self.nearby_projects],179        }180 181 182def public_project_title(title: str) -> str:183    cleaned = " ".join(str(title).split())184    if not cleaned:185        return "Untitled project"186    if GENERIC_PUBLIC_TITLE_RE.search(cleaned):187        return "Untitled project"188    return cleaned189 190 191def normalize_project_tags(tags: Sequence[Any]) -> tuple[str, ...]:192    cleaned: list[str] = []193    seen: set[str] = set()194    for raw_tag in tags or ():195        tag = " ".join(str(raw_tag or "").split())196        if not tag or is_hosting_metadata_tag(tag):197            continue198        if tag in seen:199            continue200        seen.add(tag)201        cleaned.append(tag)202    return tuple(cleaned)203 204 205def is_hosting_metadata_tag(tag: str) -> bool:206    folded = str(tag or "").strip().casefold()207    return any(folded.startswith(prefix) for prefix in HOSTING_METADATA_TAG_PREFIXES)208 209 210def public_project_summary(summary: str) -> str:211    cleaned = " ".join(str(summary).split())212    if not cleaned:213        return ""214    if GENERIC_PUBLIC_SUMMARY_RE.search(cleaned):215        return ""216    return cleaned217 218 219def extract_app_file_embedding_text(app_file: str, text: str) -> str:220    cleaned_file = str(app_file).strip()221    cleaned_text = str(text or "")222    if not cleaned_file or not cleaned_text.strip():223        return ""224 225    suffix = PurePosixPath(cleaned_file).suffix.lower()226    if suffix == ".py":227        body = python_app_signals(cleaned_text)228    else:229        body = cleaned_text230    return bounded_embedding_text(body, APP_FILE_EMBEDDING_CHAR_LIMIT)231 232 233def python_app_signals(source: str) -> str:234    try:235        tree = ast.parse(source)236    except SyntaxError:237        return source238 239    signals: list[str] = []240    for node in ast.walk(tree):241        if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)):242            signals.append(node.name)243            for arg in node.args.args:244                signals.append(arg.arg)245        elif isinstance(node, ast.ClassDef):246            signals.append(node.name)247        elif isinstance(node, ast.Call):248            name = call_name(node.func)249            if name:250                signals.append(name)251            signals.extend(keyword.arg for keyword in node.keywords if keyword.arg)252        elif isinstance(node, ast.Constant) and isinstance(node.value, str):253            signals.append(node.value)254 255    return ordered_normalized_text(signals)256 257 258def call_name(node: ast.AST) -> str:259    if isinstance(node, ast.Name):260        return node.id261    if isinstance(node, ast.Attribute):262        parent = call_name(node.value)263        return f"{parent}.{node.attr}" if parent else node.attr264    return ""265 266 267def ordered_normalized_text(values: Sequence[str]) -> str:268    seen: set[str] = set()269    ordered: list[str] = []270    for value in values:271        cleaned = clean_embedding_signal(value)272        if not cleaned:273            continue274        if cleaned in seen:275            continue276        seen.add(cleaned)277        ordered.append(cleaned)278    return "\n".join(ordered)279 280 281def clean_embedding_signal(value: str) -> str:282    cleaned = HTML_TAG_RE.sub(" ", str(value))283    cleaned = " ".join(cleaned.split())284    if looks_like_style_blob(cleaned):285        return ""286    return cleaned287 288 289def looks_like_style_blob(text: str) -> bool:290    if len(text) < 80:291        return False292    style_markers = (293        text.count("{")294        + text.count("}")295        + text.count(";")296        + text.count("!important")297        + text.count("rgba(")298        + text.count("linear-gradient")299    )300    return style_markers >= 8 and style_markers / len(text) > 0.015301 302 303def bounded_embedding_text(text: str, limit: int) -> str:304    cleaned = " ".join(str(text).split())305    if len(cleaned) <= limit:306        return cleaned307    marker = " ... "308    edge = max(1, (limit - len(marker)) // 2)309    return f"{cleaned[:edge].rstrip()}{marker}{cleaned[-edge:].lstrip()}"310 311 312@dataclass(frozen=True)313class WhitespaceSeed:314    label: str315    query: str316    pitch: str317 318 319WHITESPACE_SEEDS: tuple[WhitespaceSeed, ...] = (320    WhitespaceSeed(321        "Tiny civic repair desk",322        "local government forms benefits tenant aid accessibility paperwork",323        "A small agent that turns intimidating public-service forms into one-page action plans.",324    ),325    WhitespaceSeed(326        "Hands-on science coach",327        "kitchen science experiment kids sensor notebook classroom",328        "A lab-notebook companion that designs safe experiments from household materials.",329    ),330    WhitespaceSeed(331        "Offline field translator",332        "offline translation field guide travel emergency low connectivity",333        "A local-first phrase and intent helper for stressful travel or field-work moments.",334    ),335    WhitespaceSeed(336        "Personal archive cartographer",337        "photos notes memories archive timeline family history scrapbook",338        "A tiny model that maps a private archive into stories without sending it to cloud APIs.",339    ),340    WhitespaceSeed(341        "Small-team incident scribe",342        "incident retrospective logs on call debugging timeline root cause",343        "A local incident historian that turns messy notes into a calm timeline and next actions.",344    ),345    WhitespaceSeed(346        "Accessibility rehearsal room",347        "accessibility captions alt text screen reader rehearsal inclusive design",348        "A practice space that lets makers rehearse their demo for captions, contrast, and clarity.",349    ),350    WhitespaceSeed(351        "Neighborhood seed library",352        "garden plants seed library neighborhood seasons climate local exchange",353        "An advisor for hyperlocal seed swaps, planting plans, and community garden knowledge.",354    ),355)356 357 358class ProjectIndex:359    def __init__(360        self,361        projects: list[Project],362        generated_at: str,363        source: str,364        index_payload: dict,365        query_embedder: EmbeddingFunction | None = None,366    ) -> None:367        if not projects:368            raise ValueError("project index requires at least one project")369        validate_index_payload(index_payload, projects, generated_at, source)370        self.projects = projects371        self.generated_at = generated_at372        self.source = source373        self.index_generated_at = str(index_payload["generated_at"])374        self.index_algorithm = str(index_payload["algorithm"])375        self.snapshot_digest = str(index_payload["snapshot_digest"])376        self.index_payload = index_payload377        self.embedding_metadata = dict(index_payload["embedding"])378        self.embedding_dimensions = int(self.embedding_metadata["dimensions"])379        self._query_embedder = query_embedder380        self._vectors = [381            tuple(float(value) for value in document["vector"])382            for document in index_payload["documents"]383        ]384        self._vector_by_id = {385            project.id: vector for project, vector in zip(self.projects, self._vectors)386        }387 388    def vector_for(self, project_id: str) -> tuple[float, ...] | None:389        return self._vector_by_id.get(project_id)390 391    def project_vectors(self) -> tuple[tuple[float, ...], ...]:392        return tuple(self._vectors)393 394    def embed_query(self, text: str) -> tuple[float, ...]:395        return tuple(normalize_vector(self._embed_query(text)))396 397    @classmethod398    def from_file(cls, path: Path, query_embedder: EmbeddingFunction | None = None) -> "ProjectIndex":399        json.loads(path.read_text(encoding="utf-8"))400        raise ValueError("ProjectIndex.from_file requires a separate embedding index payload")401 402    @classmethod403    def from_files(404        cls,405        project_path: Path,406        index_path: Path,407        query_embedder: EmbeddingFunction | None = None,408    ) -> "ProjectIndex":409        data = json.loads(project_path.read_text(encoding="utf-8"))410        index_payload = json.loads(index_path.read_text(encoding="utf-8"))411        projects = [Project.from_dict(item) for item in data["projects"]]412        return cls(413            projects=projects,414            generated_at=str(data.get("generated_at") or ""),415            source=str(data.get("source") or ""),416            index_payload=index_payload,417            query_embedder=query_embedder,418        )419 420    def set_query_embedder(self, embedder: EmbeddingFunction) -> None:421        self._query_embedder = embedder422 423    def top_projects(self, limit: int = 8) -> list[Project]:424        return sorted(425            self.projects,426            key=lambda project: (project.likes, project.last_modified, project.title.lower()),427            reverse=True,428        )[:limit]429 430    def search(self, query: str, limit: int = 5) -> list[SearchHit]:431        query_terms = set(tokenize(query))432        if not query_terms:433            return []434        query_vector = normalize_vector(self._embed_query(query))435        hits: list[SearchHit] = []436        for page_number, (project, vector) in enumerate(437            zip(self.projects, self._vectors, strict=True),438            start=1,439        ):440            score = max(0.0, min(1.0, (dot_product(query_vector, vector) + 1.0) / 2.0))441            hits.append(442                SearchHit(443                    project=project,444                    score=score,445                    matched_terms=matched_terms(query_terms, project),446                    page_number=page_number,447                )448            )449        hits.sort(key=lambda hit: (hit.score, hit.project.likes), reverse=True)450        return hits[:limit]451 452    def get(self, project_id: str) -> Project | None:453        for project in self.projects:454            if project.id == project_id or project.slug == project_id:455                return project456        return None457 458    def find_whitespace(self, limit: int = 5) -> list[WhitespaceItem]:459        items: list[WhitespaceItem] = []460        for seed in WHITESPACE_SEEDS:461            hits = self.search(seed.query, limit=3)462            saturation = sum(hit.score for hit in hits) / max(len(hits), 1)463            score = max(0.0, min(1.0, 1.0 - max(0.0, saturation - 0.35) / 0.60))464            if hits:465                evidence = f"Nearest echoes are weak: {', '.join(hit.project.title for hit in hits[:2])}."466            else:467                evidence = "No close project echoes in the current snapshot."468            items.append(469                WhitespaceItem(470                    label=seed.label,471                    pitch=seed.pitch,472                    evidence=evidence,473                    score=score,474                    nearby_projects=tuple(hit.project for hit in hits),475                )476            )477        items.sort(key=lambda item: item.score, reverse=True)478        return items[:limit]479 480    def starter_directions(self, limit: int = 5) -> list[WhitespaceItem]:481        return [482            WhitespaceItem(483                label=seed.label,484                pitch=seed.pitch,485                evidence="Press this direction to test it against the current project map.",486                score=0.0,487                nearby_projects=(),488            )489            for seed in WHITESPACE_SEEDS[:limit]490        ]491 492    def _embed_query(self, query: str) -> Sequence[float]:493        if self._query_embedder is None:494            from hackathon_advisor.llama_embedding import create_llama_cpp_embedder495 496            self._query_embedder = create_llama_cpp_embedder(self.embedding_metadata)497        return self._query_embedder(query)498 499 500def tokenize(text: str) -> list[str]:501    return [token.lower().strip("._-+") for token in TOKEN_RE.findall(text) if len(token.strip("._-+")) > 1]502 503 504def matched_terms(query_terms: set[str], project: Project) -> tuple[str, ...]:505    project_terms = set(tokenize(project.searchable_text))506    return tuple(sorted(query_terms & project_terms)[:8])507 508 509def build_index_payload(510    projects: list[Project],511    snapshot_generated_at: str,512    source: str,513    embeddings: Sequence[Sequence[float]],514    *,515    embedding_metadata: dict[str, Any] | None = None,516) -> dict:517    if len(embeddings) != len(projects):518        raise ValueError("embedding count must match project count")519    normalized = [normalize_vector(vector) for vector in embeddings]520    dimensions = len(normalized[0]) if normalized else 0521    if dimensions <= 0:522        raise ValueError("embedding vectors must not be empty")523    if any(len(vector) != dimensions for vector in normalized):524        raise ValueError("embedding vectors must have one shared dimension")525 526    metadata = {527        "model_repo": DEFAULT_EMBEDDING_MODEL_REPO,528        "model_file": DEFAULT_EMBEDDING_MODEL_FILE,529        "runtime": DEFAULT_EMBEDDING_RUNTIME,530        "dimensions": dimensions,531        "normalized": True,532        **(embedding_metadata or {}),533    }534    indexed_documents = []535    for project, vector in zip(projects, normalized, strict=True):536        indexed_documents.append(537            {538                "project_id": project.id,539                "text_digest": sha256(project.searchable_text.encode("utf-8")).hexdigest(),540                "norm": round(vector_norm(vector), 8),541                "vector": [round(value, 8) for value in vector],542            }543        )544    return {545        "schema_version": INDEX_SCHEMA_VERSION,546        "algorithm": INDEX_ALGORITHM,547        "generated_at": utc_now(),548        "snapshot_generated_at": snapshot_generated_at,549        "snapshot_source": source,550        "snapshot_digest": project_snapshot_digest(projects, snapshot_generated_at, source),551        "document_count": len(projects),552        "embedding": metadata,553        "documents": indexed_documents,554    }555 556 557def validate_index_payload(558    payload: dict,559    projects: list[Project],560    snapshot_generated_at: str,561    snapshot_source: str,562) -> None:563    if payload.get("schema_version") != INDEX_SCHEMA_VERSION:564        raise ValueError("unsupported project index schema version")565    if payload.get("algorithm") != INDEX_ALGORITHM:566        raise ValueError(f"unsupported project index algorithm: {payload.get('algorithm')}")567    if payload.get("snapshot_generated_at") != snapshot_generated_at:568        raise ValueError("project index was built from a different snapshot timestamp")569    if payload.get("snapshot_source") != snapshot_source:570        raise ValueError("project index was built from a different snapshot source")571    if payload.get("snapshot_digest") != project_snapshot_digest(572        projects,573        snapshot_generated_at,574        snapshot_source,575    ):576        raise ValueError("project index digest does not match projects snapshot")577 578    embedding = payload.get("embedding")579    if not isinstance(embedding, dict):580        raise ValueError("project index embedding metadata is missing")581    dimensions = int(embedding.get("dimensions") or 0)582    if dimensions <= 0:583        raise ValueError("project index embedding dimensions must be positive")584    if embedding.get("runtime") != DEFAULT_EMBEDDING_RUNTIME:585        raise ValueError("project index embedding runtime must be llama.cpp")586 587    documents = payload.get("documents")588    if not isinstance(documents, list) or len(documents) != len(projects):589        raise ValueError("project index document count does not match projects snapshot")590    project_ids = [project.id for project in projects]591    indexed_ids = [document.get("project_id") for document in documents]592    if indexed_ids != project_ids:593        raise ValueError("project index project order does not match projects snapshot")594    for project, document in zip(projects, documents, strict=True):595        if document.get("text_digest") != sha256(project.searchable_text.encode("utf-8")).hexdigest():596            raise ValueError("project index text digest does not match searchable project text")597        vector = document.get("vector")598        if not isinstance(vector, list) or len(vector) != dimensions:599            raise ValueError("project index vector dimensions do not match embedding metadata")600        norm = vector_norm(float(value) for value in vector)601        if not 0.99 <= norm <= 1.01:602            raise ValueError("project index vectors must be normalized")603 604 605def normalize_vector(vector: Sequence[float]) -> tuple[float, ...]:606    values = tuple(float(value) for value in vector)607    norm = vector_norm(values)608    if norm == 0.0:609        raise ValueError("embedding vector norm must be non-zero")610    return tuple(value / norm for value in values)611 612 613def vector_norm(vector: Sequence[float]) -> float:614    return math.sqrt(sum(float(value) * float(value) for value in vector))615 616 617def dot_product(left: Sequence[float], right: Sequence[float]) -> float:618    if len(left) != len(right):619        raise ValueError("embedding vectors must have equal dimensions")620    return sum(float(a) * float(b) for a, b in zip(left, right, strict=True))621 622 623def project_snapshot_digest(projects: list[Project], generated_at: str, source: str) -> str:624    payload = {625        "generated_at": generated_at,626        "source": source,627        "projects": [project.to_snapshot_dict() for project in projects],628    }629    encoded = json.dumps(payload, sort_keys=True, separators=(",", ":"), ensure_ascii=False).encode("utf-8")630    return sha256(encoded).hexdigest()631