gabriel / musehub public
_ui_helpers.py python
272 lines 9.6 KB
Raw
sha256:e17dff4303a5885a1a61af6b39594fe316aeb62821bad17ccb66c52128a315f5 fix: enforce repo visibility gate on all SSR route handlers… Sonnet 4.6 minor ⚠ breaking 34 days ago
1 """Shared helpers for MuseHub UI route handlers.
2
3 All route files under musehub/api/routes/musehub/ui_*.py import from here.
4 Nothing in this module registers routes or has side effects on import.
5 """
6
7 import asyncio
8 import json as _json
9 import logging
10 from pathlib import Path
11 from fastapi import HTTPException
12 from fastapi import status as http_status
13 from sqlalchemy import func, select as sa_select
14 from sqlalchemy.ext.asyncio import AsyncSession
15
16 from musehub.auth.request_signing import MSignContext
17 from musehub.db.musehub_intel_models import MusehubIntelResult
18 from musehub.db.musehub_social_models import MusehubIssue, MusehubProposal
19 from musehub.models.musehub import RepoResponse, TreeEntryResponse
20 from musehub.services import musehub_repository
21 from musehub.storage.backends import get_backend as _get_storage_backend, read_object_bytes as _read_object_bytes
22 from musehub.types.json_types import Breadcrumb, JSONObject, StrDict
23
24 logger = logging.getLogger(__name__)
25
26 # ---------------------------------------------------------------------------
27 # URL helpers
28 # ---------------------------------------------------------------------------
29
30 def _base_url(owner: str, repo_slug: str) -> str:
31 """Return the canonical UI base URL for a repo."""
32 return f"/{owner}/{repo_slug}"
33
34 def _breadcrumbs(*segments: tuple[str, str]) -> list[Breadcrumb]:
35 """Build breadcrumb_data list from (label, url) pairs."""
36 return [{"label": label, "url": url} for label, url in segments]
37
38 def _og_tags(
39 *,
40 title: str,
41 description: str = "",
42 image: str = "",
43 og_type: str = "website",
44 twitter_card: str = "summary",
45 ) -> StrDict:
46 """Build Open Graph and Twitter Card meta tag dict for a page template."""
47 tags: StrDict = {
48 "og:title": title,
49 "og:type": og_type,
50 "twitter:card": twitter_card,
51 "twitter:title": title,
52 }
53 if description:
54 tags["og:description"] = description
55 tags["twitter:description"] = description
56 if image:
57 tags["og:image"] = image
58 tags["twitter:image"] = image
59 return tags
60
61 # ---------------------------------------------------------------------------
62 # Language / blob detection
63 # ---------------------------------------------------------------------------
64
65 _LANG_MAP: StrDict = {
66 ".py": "python",
67 ".js": "javascript",
68 ".ts": "typescript",
69 ".json": "json",
70 ".yaml": "yaml",
71 ".yml": "yaml",
72 ".md": "markdown",
73 ".txt": "text",
74 ".xml": "xml",
75 ".html": "html",
76 ".css": "css",
77 ".sh": "bash",
78 ".toml": "toml",
79 ".bats": "bash",
80 }
81
82 # Dotfiles have no extension — map by exact filename.
83 _FILENAME_LANG_MAP: StrDict = {
84 ".museattributes": "toml",
85 ".museignore": "toml",
86 }
87
88 _BLOB_BINARY_TYPES: frozenset[str] = frozenset(
89 [".webp", ".png", ".jpg", ".jpeg", ".gif", ".svg"]
90 )
91
92 def _detect_language(path: str) -> str:
93 """Return a display-friendly language name for a file path based on its extension."""
94 filename = Path(path).name
95 if filename in _FILENAME_LANG_MAP:
96 return _FILENAME_LANG_MAP[filename]
97 if filename.startswith("_") and "." not in filename:
98 return "bash"
99 return _LANG_MAP.get(Path(path).suffix.lower(), "")
100
101 # ---------------------------------------------------------------------------
102 # Repo resolution
103 # ---------------------------------------------------------------------------
104
105 async def _resolve_repo(
106 owner: str,
107 repo_slug: str,
108 db: AsyncSession,
109 claims: MSignContext | None = None,
110 ) -> tuple[str, str, JSONObject]:
111 """Resolve owner+slug to repo_id; raise 404 if not found or not accessible.
112
113 Returns (repo_id, base_url, nav_ctx) where nav_ctx contains SSR-ready
114 nav-header fields including open proposal/issue counts for the tab strip.
115
116 Private repos are only accessible when a valid MSign token is present.
117 Anonymous browser requests (no Authorization header) receive 404 — not 403,
118 so the repo's existence is not confirmed.
119 """
120 row = await musehub_repository.get_repo_orm_by_owner_slug(db, owner, repo_slug)
121 if row is None:
122 raise HTTPException(
123 status_code=http_status.HTTP_404_NOT_FOUND,
124 detail=f"Repo '{owner}/{repo_slug}' not found",
125 )
126 if row.visibility != "public" and claims is None:
127 raise HTTPException(
128 status_code=http_status.HTTP_404_NOT_FOUND,
129 detail=f"Repo '{owner}/{repo_slug}' not found",
130 )
131 repo_id = str(row.repo_id)
132
133 proposal_count_result, issue_count_result = await asyncio.gather(
134 db.execute(
135 sa_select(func.count()).select_from(MusehubProposal).where(
136 MusehubProposal.repo_id == repo_id,
137 MusehubProposal.state.in_(("open", "in_review", "approved", "drafting", "settling")),
138 )
139 ),
140 db.execute(
141 sa_select(func.count()).select_from(MusehubIssue).where(
142 MusehubIssue.repo_id == repo_id,
143 MusehubIssue.state == "open",
144 )
145 ),
146 )
147 open_proposal_count: int = proposal_count_result.scalar_one_or_none() or 0
148 open_issue_count: int = issue_count_result.scalar_one_or_none() or 0
149
150 nav_ctx: JSONObject = {
151 "repo_tags": row.tags or [],
152 "repo_visibility": row.visibility or "private",
153 "nav_open_proposal_count": open_proposal_count,
154 "nav_open_issue_count": open_issue_count,
155 }
156 return repo_id, _base_url(owner, repo_slug), nav_ctx
157
158 async def _resolve_repo_full(
159 owner: str,
160 repo_slug: str,
161 db: AsyncSession,
162 claims: MSignContext | None = None,
163 ) -> tuple[RepoResponse, str]:
164 """Resolve owner+slug to a full RepoResponse; raise 404 if not found or not accessible.
165
166 Returns (repo_response, base_url). Use when the handler needs structured
167 repo data (e.g. to return JSON via negotiate_response).
168
169 Private repos are only accessible when a valid MSign token is present.
170 """
171 repo = await musehub_repository.get_repo_by_owner_slug(db, owner, repo_slug)
172 if repo is None:
173 raise HTTPException(
174 status_code=http_status.HTTP_404_NOT_FOUND,
175 detail=f"Repo '{owner}/{repo_slug}' not found",
176 )
177 if repo.visibility != "public" and claims is None:
178 raise HTTPException(
179 status_code=http_status.HTTP_404_NOT_FOUND,
180 detail=f"Repo '{owner}/{repo_slug}' not found",
181 )
182 return repo, _base_url(owner, repo_slug)
183
184 # ---------------------------------------------------------------------------
185 # Ref resolution — handles branch names containing slashes (e.g. feat/my-thing)
186 # ---------------------------------------------------------------------------
187
188 async def _resolve_ref_and_path(
189 db: AsyncSession,
190 repo_id: str,
191 refpath: str,
192 ) -> tuple[str, str]:
193 """Split a combined ``{ref}/{path}`` string into (branch_name, file_path).
194
195 FastAPI routes like ``/blob/{ref}/{path:path}`` only capture one segment
196 for ``ref``, so ``feat/my-branch/file.py`` is parsed as ref=``feat`` and
197 path=``my-branch/file.py``. This helper tries progressively longer
198 prefixes of *refpath* until one matches an existing branch, returning
199 ``(branch, remainder)`` for the longest match found.
200
201 Falls back to the first segment as the ref when no branch matches.
202 """
203 from musehub.services import musehub_repository as _repo_svc
204 branches = await _repo_svc.list_branches(db, repo_id)
205 branch_names = {b.name for b in branches}
206
207 # Try longest-prefix match: split refpath at each '/' from left to right.
208 segments = refpath.split("/")
209 best_ref, best_path = segments[0], "/".join(segments[1:])
210 for i in range(1, len(segments)):
211 candidate_ref = "/".join(segments[:i])
212 if candidate_ref in branch_names:
213 best_ref = candidate_ref
214 best_path = "/".join(segments[i:])
215 return best_ref, best_path
216
217
218 # ---------------------------------------------------------------------------
219 # README / intel helpers
220 # ---------------------------------------------------------------------------
221
222 _README_NAMES: frozenset[str] = frozenset(
223 {"readme.md", "readme", "readme.txt", "readme.rst"}
224 )
225
226 async def _load_intel_strip(db: AsyncSession, repo_id: str) -> JSONObject | None:
227 """Return intel strip data for the repo home page, or None if no index exists."""
228 try:
229 result = await db.execute(
230 sa_select(MusehubIntelResult.data_json)
231 .where(
232 MusehubIntelResult.repo_id == repo_id,
233 MusehubIntelResult.intel_type == "code.intel_summary",
234 )
235 )
236 row = result.first()
237 if row is None or not row.data_json:
238 return None
239 return _json.loads(row.data_json)
240 except Exception:
241 return None
242
243 async def _fetch_readme(
244 db: AsyncSession,
245 repo_id: str,
246 ref: str,
247 entries: list[TreeEntryResponse],
248 owner: str = "",
249 slug: str = "",
250 ) -> str:
251 """Return the raw text of the first README found in the tree root, or ''."""
252 readme_entry = next(
253 (e for e in entries if getattr(e, "name", "").lower() in _README_NAMES),
254 None,
255 )
256 if readme_entry is None:
257 return ""
258 object_id: str = getattr(readme_entry, "object_id", "") or ""
259 if not object_id:
260 return ""
261 try:
262 from musehub.db.musehub_repo_models import MusehubObject
263 obj = await db.get(MusehubObject, object_id)
264 if obj is None:
265 return ""
266 raw: bytes | None = await _read_object_bytes(obj, session=db)
267 if not raw:
268 return ""
269 from musehub.types.compression import decompress_if_needed
270 return decompress_if_needed(raw).decode("utf-8", errors="replace")
271 except Exception:
272 return ""
File History 3 commits
sha256:e17dff4303a5885a1a61af6b39594fe316aeb62821bad17ccb66c52128a315f5 fix: enforce repo visibility gate on all SSR route handlers… Sonnet 4.6 minor ⚠ 34 days ago
sha256:65f2fd8d910e1eeb00b7bc8740d3cbf1b2e14dad83b2eb999fbbbc44e97cd936 getting intel jobs to run properly Human minor ⚠ 43 days ago
sha256:92528ae07d0e1239d87fd5fd1f439e8fbb49c9778a9a400bc4a736073fb28316 feat: byte-range blob reads, file attribution DAG walk, bra… Sonnet 4.6 minor ⚠ 43 days ago