build-small-hackathon/hackathon-advisor

1"""Read-only query layer over one immutable dashboard snapshot.2 3The atlas chat feature (and any future consumer) asks questions like "what is4everyone building", "which projects completed the most quests", or "what does the5Voice cluster contain". Those queries belong in one place — not in tool prompts,6not in route handlers — so this module wraps a single dashboard snapshot7(``dashboard_payload`` + its ``DashboardSearchIndex``) behind typed query methods8that return plain JSON-ready dicts.9 10A repository instance never touches module globals, locks, or models: the caller11captures a consistent snapshot (app.py does so under ``_runtime_lock``, the same12pattern as ``/api/dashboard/search``) and constructs the repository outside the13lock. Quest data may be absent (``quest_report.status == "not_analyzed"``); every14method degrades to empty-but-well-formed results in that case.15"""16 17from __future__ import annotations18 19from collections.abc import Mapping20from difflib import SequenceMatcher21from typing import Any22 23from hackathon_advisor._text import clean, list_of_dicts24from hackathon_advisor.dashboard_search import DashboardSearchIndex25from hackathon_advisor.data import (26    normalize_project_tags,27    public_project_summary,28    public_project_title,29)30from hackathon_advisor.quest_taxonomy import (31    build_app_segment,32    build_readme_segment,33    canonical_quest_id,34    quest_label,35    quest_profiles,36)37 38CLUSTER_LABEL_MATCH_THRESHOLD = 0.639DEFAULT_LEADERBOARD_LIMIT = 840DEFAULT_SEARCH_LIMIT = 841DEFAULT_RECENT_LIMIT = 642DEFAULT_EXAMPLE_LIMIT = 643README_EXCERPT_CHARS = 150044APP_EXCERPT_CHARS = 190045 46 47class DashboardRepository:48    """Pure queries over one dashboard snapshot; safe to use without locks."""49 50    def __init__(51        self, dashboard_payload: Mapping[str, Any], search_index: DashboardSearchIndex52    ) -> None:53        self._payload = dashboard_payload54        self._search_index = search_index55        self._points = list_of_dicts(dashboard_payload.get("points"))56        self._clusters = list_of_dicts(dashboard_payload.get("clusters"))57        quest_report = dashboard_payload.get("quest_report")58        self._quest_report = quest_report if isinstance(quest_report, Mapping) else {}59        self._cluster_label_by_id = {60            str(cluster.get("id") or ""): clean(cluster.get("label")) for cluster in self._clusters61        }62        # Full Project objects (incl. readme_body / app_file_source, which the public63        # dashboard points strip) ride along inside the search index's documents.64        self._project_by_id = {65            document.project.id: document.project for document in search_index.documents66        }67        self._point_by_id = {str(point.get("id") or ""): point for point in self._points}68 69    def quests_analyzed(self) -> bool:70        return str(self._quest_report.get("status") or "") == "analyzed"71 72    def overview(self) -> dict[str, Any]:73        """Field-wide counts plus the brightest clusters, quests, and projects."""74        top_quests = [75            {"id": quest["id"], "label": quest["label"], "project_count": quest["project_count"]}76            for quest in self._quests_by_coverage()[:3]77            if quest["project_count"] > 078        ]79        most_liked = sorted(80            self._points,81            key=lambda point: (int(point.get("likes") or 0), clean(point.get("title")).casefold()),82            reverse=True,83        )[:3]84        return {85            "project_count": int(self._payload.get("project_count") or len(self._points)),86            "cluster_count": len(self._clusters),87            "generated_at": str(self._payload.get("generated_at") or ""),88            "quest_status": str(self._quest_report.get("status") or "not_analyzed"),89            "top_clusters": [90                {91                    "label": clean(cluster.get("label")),92                    "project_count": int(cluster.get("project_count") or 0),93                }94                for cluster in self._clusters[:3]95            ],96            "top_quests": top_quests,97            "most_liked": [self._project_row(point) for point in most_liked],98        }99 100    def list_clusters(self) -> dict[str, Any]:101        return {102            "cluster_count": len(self._clusters),103            "clusters": [104                {105                    "label": clean(cluster.get("label")),106                    "project_count": int(cluster.get("project_count") or 0),107                    "keywords": [clean(keyword) for keyword in (cluster.get("keywords") or [])[:4]],108                }109                for cluster in self._clusters110            ],111        }112 113    def cluster_detail(self, label: str) -> dict[str, Any] | None:114        """Resolve a cluster by (fuzzy) label or id; cluster ids are unstable across refreshes."""115        cluster = self._resolve_cluster(label)116        if cluster is None:117            return None118        examples = list_of_dicts(cluster.get("representative_projects"))[:DEFAULT_EXAMPLE_LIMIT]119        return {120            "label": clean(cluster.get("label")),121            "project_count": int(cluster.get("project_count") or 0),122            "keywords": [clean(keyword) for keyword in (cluster.get("keywords") or [])[:5]],123            "examples": [self._project_row(example) for example in examples],124        }125 126    def list_quests(self) -> dict[str, Any]:127        return {128            "status": str(self._quest_report.get("status") or "not_analyzed"),129            "quests": self._quests_by_coverage(),130        }131 132    def quest_detail(self, quest: str) -> dict[str, Any] | None:133        try:134            quest_id = canonical_quest_id(quest)135        except ValueError:136            quest_id = self._find_quest_in_text(quest)137            if quest_id is None:138                return None139        report_entry = next(140            (141                entry142                for entry in list_of_dicts(self._quest_report.get("quests"))143                if str(entry.get("id") or "") == quest_id144            ),145            {},146        )147        profile = next(148            (profile for profile in quest_profiles() if profile["id"] == quest_id),149            {"id": quest_id, "label": quest_id, "description": ""},150        )151        matched = [point for point in self._points if quest_id in (point.get("quest_ids") or [])]152        examples = list_of_dicts(report_entry.get("examples"))[:DEFAULT_EXAMPLE_LIMIT] or [153            self._project_row(point) for point in matched[:DEFAULT_EXAMPLE_LIMIT]154        ]155        return {156            "id": quest_id,157            "label": profile["label"],158            "description": profile["description"],159            "status": str(self._quest_report.get("status") or "not_analyzed"),160            "project_count": int(report_entry.get("project_count") or len(matched)),161            "examples": [self._project_row(example) for example in examples],162        }163 164    def top_by_quests(self, limit: int = DEFAULT_LEADERBOARD_LIMIT) -> dict[str, Any]:165        """Per-project quest leaderboard (projects ARE the teams: no author field exists)."""166        rows = [167            {168                **self._project_row(point),169                "quest_count": len(point.get("quest_ids") or []),170                "quest_ids": [str(quest) for quest in point.get("quest_ids") or []],171            }172            for point in self._points173            if point.get("quest_ids")174        ]175        rows.sort(176            key=lambda row: (row["quest_count"], row["likes"], row["title"].casefold()),177            reverse=True,178        )179        return {180            "status": str(self._quest_report.get("status") or "not_analyzed"),181            "rows": rows[: max(1, int(limit))],182            "projects_with_quests": len(rows),183        }184 185    def search(self, query: str, limit: int = DEFAULT_SEARCH_LIMIT) -> dict[str, Any]:186        payload = self._search_index.search(clean(query), limit=max(1, int(limit)))187        return {188            "query": payload["query"],189            "total": int(payload["total"]),190            "results": [191                {192                    "id": str(result.get("project_id") or ""),193                    "title": clean(result.get("title")),194                    "summary": clean(result.get("summary")),195                    "url": str(result.get("url") or ""),196                    "score": float(result.get("score") or 0.0),197                }198                for result in payload["results"]199            ],200        }201 202    def recent_activity(self, limit: int = DEFAULT_RECENT_LIMIT) -> dict[str, Any]:203        ordered = sorted(204            self._points,205            key=lambda point: str(point.get("last_modified") or ""),206            reverse=True,207        )[: max(1, int(limit))]208        return {209            "projects": [210                {211                    **self._project_row(point),212                    "last_modified": str(point.get("last_modified") or ""),213                    "cluster_label": self._cluster_label_by_id.get(214                        str(point.get("cluster_id") or ""), ""215                    ),216                }217                for point in ordered218            ],219        }220 221    def _quests_by_coverage(self) -> list[dict[str, Any]]:222        entries = [223            {224                "id": str(entry.get("id") or ""),225                "label": clean(entry.get("label")) or str(entry.get("id") or ""),226                "description": clean(entry.get("description")),227                "project_count": int(entry.get("project_count") or 0),228            }229            for entry in list_of_dicts(self._quest_report.get("quests"))230        ]231        return sorted(232            entries, key=lambda entry: (-entry["project_count"], entry["label"].casefold())233        )234 235    def project_detail(self, name: str) -> dict[str, Any] | None:236        """One project's card plus its README and main-app-file excerpts.237 238        The excerpts reuse the quest classifier's prompt view (build_readme_segment /239        build_app_segment) — the same budgeted slices MiniCPM already reads well."""240        project = self._resolve_project(name)241        if project is None:242            return None243        point = self._point_by_id.get(project.id, {})244        app_excerpt = _clip_excerpt(245            build_app_segment(project.app_file_source, project.app_file_embedding_text),246            APP_EXCERPT_CHARS,247        )248        return {249            "id": project.id,250            "title": public_project_title(project.title),251            "summary": public_project_summary(project.summary),252            "url": project.url,253            "likes": project.likes,254            "sdk": project.sdk,255            "models": list(project.models)[:4],256            "tags": list(normalize_project_tags(project.tags))[:6],257            "last_modified": project.last_modified,258            "cluster_label": self._cluster_label_by_id.get(str(point.get("cluster_id") or ""), ""),259            "quests": [quest_label(str(quest)) for quest in point.get("quest_ids") or []],260            "readme_excerpt": _clip_excerpt(261                build_readme_segment(project.readme_body), README_EXCERPT_CHARS262            ),263            "app_file": project.app_file,264            "app_excerpt": app_excerpt,265        }266 267    def _resolve_project(self, name: str) -> Any | None:268        """Match a project by id, slug, or title — exact first, then embedded in a269        longer question ("tell me about Jawbreaker"), longest title winning."""270        wanted = clean(name).casefold()271        if not wanted:272            return None273        for project in self._project_by_id.values():274            slug = project.id.rsplit("/", 1)[-1]275            if wanted in (project.id.casefold(), slug.casefold()):276                return project277            if public_project_title(project.title).casefold() == wanted:278                return project279        best, best_length = None, 0280        for project in self._project_by_id.values():281            title = public_project_title(project.title).casefold()282            slug = project.id.rsplit("/", 1)[-1].casefold()283            for candidate in (title, slug):284                if len(candidate) > 3 and candidate in wanted and len(candidate) > best_length:285                    best, best_length = project, len(candidate)286        return best287 288    def _find_quest_in_text(self, text: str) -> str | None:289        """Spot a quest id or label embedded in a longer question."""290        wanted = clean(text).casefold()291        if not wanted:292            return None293        for profile in quest_profiles():294            if profile["id"].casefold() in wanted or profile["label"].casefold() in wanted:295                return profile["id"]296        return None297 298    def _resolve_cluster(self, label: str) -> Mapping[str, Any] | None:299        wanted = clean(label).casefold()300        if not wanted:301            return None302        for cluster in self._clusters:303            if str(cluster.get("id") or "").casefold() == wanted:304                return cluster305        for cluster in self._clusters:306            if clean(cluster.get("label")).casefold() == wanted:307                return cluster308        for cluster in self._clusters:309            cluster_label = clean(cluster.get("label")).casefold()310            if wanted in cluster_label or cluster_label in wanted:311                return cluster312        for cluster in self._clusters:313            keywords = {clean(keyword).casefold() for keyword in cluster.get("keywords") or []}314            if any(token in keywords for token in wanted.split()):315                return cluster316        best, best_score = None, 0.0317        for cluster in self._clusters:318            score = SequenceMatcher(None, wanted, clean(cluster.get("label")).casefold()).ratio()319            if score > best_score:320                best, best_score = cluster, score321        return best if best_score >= CLUSTER_LABEL_MATCH_THRESHOLD else None322 323    def _project_row(self, point: Mapping[str, Any]) -> dict[str, Any]:324        return {325            "id": str(point.get("id") or ""),326            "title": clean(point.get("title")) or str(point.get("id") or ""),327            "url": str(point.get("url") or ""),328            "likes": int(point.get("likes") or 0),329        }330 331 332def _clip_excerpt(text: str, limit: int) -> str:333    # Newlines stay (app files read as code); only the length is bounded.334    cleaned = str(text or "").strip()335    if len(cleaned) <= limit:336        return cleaned337    return cleaned[:limit].rstrip() + " ..."338