From cc677eecb141535d5454dee96bc3641cee50049f Mon Sep 17 00:00:00 2001 From: timmy Date: Tue, 7 Jul 2026 00:58:58 +0000 Subject: [PATCH] fix: restore backend code after stale checkout --- src/gitea_proxy.py | 37 ++++++++++++++++++++++ src/main.py | 56 +++++++++++++++++++++++++++++++++ src/models.py | 52 ++++++++++++++++++++++++++++++ src/suggestion_engine.py | 36 +++++++++++++++++++++ src/views.py | 9 ++++++ tests/test_suggestion_engine.py | 16 ++++++++++ 6 files changed, 206 insertions(+) diff --git a/src/gitea_proxy.py b/src/gitea_proxy.py index e69de29..c45bdc7 100644 --- a/src/gitea_proxy.py +++ b/src/gitea_proxy.py @@ -0,0 +1,37 @@ +import os +from typing import Any + +import httpx + +GITEA_URL = os.getenv("GITEA_URL", "http://127.0.0.1:3000").rstrip("/") +GITEA_TOKEN = os.getenv("GITEA_TOKEN", "") + + +def _auth() -> dict[str, str]: + headers: dict[str, str] = {"Accept": "application/json"} + if GITEA_TOKEN: + headers["Authorization"] = f"token {GITEA_TOKEN}" + return headers + + +async def fetch(path: str) -> Any: + async with httpx.AsyncClient(base_url=GITEA_URL, timeout=10) as client: + r = await client.get(f"/api/v1/{path}", headers=_auth()) + r.raise_for_status() + return r.json() + + +async def current_user() -> dict: + return await fetch("user") + + +async def repos() -> list[dict]: + return await fetch("user/repos?limit=50") + + +async def issues() -> list[dict]: + return await fetch("user/issues?limit=50&type=all") + + +async def pull_requests() -> list[dict]: + return await fetch("user/pulls?limit=50") diff --git a/src/main.py b/src/main.py index e69de29..2e9eec4 100644 --- a/src/main.py +++ b/src/main.py @@ -0,0 +1,56 @@ +from fastapi import FastAPI +from fastapi.middleware.cors import CORSMiddleware +from fastapi.responses import JSONResponse +from fastapi.staticfiles import StaticFiles + +from src.gitea_proxy import current_user, repos, issues, pull_requests +from src.models import User, Repo, Issue, PullRequest +from src.suggestion_engine import compute +from src.views import router as frontend_router + +app = FastAPI(title="Stackchain Dashboard") + +app.add_middleware( + CORSMiddleware, + allow_origins=["*"], + allow_credentials=False, + allow_methods=["*"], + allow_headers=["*"], +) + +app.mount("/static", StaticFiles(directory="frontend/static"), name="static") +app.include_router(frontend_router) + + +@app.get("/api/v1/context") +async def context() -> JSONResponse: + user_data = await current_user() + user_model = User(id=user_data["id"], login=user_data["login"], full_name=user_data.get("full_name", ""), email=user_data.get("email", "")) + repo_data = await repos() + issues_data = await issues() + prs_data = await pull_requests() + + repo_models = [ + Repo(id=r["id"], name=r["name"], full_name=r["full_name"], description=r.get("description", ""), url=r["html_url"], updated_at=r.get("updated_at", "")) + for r in repo_data[:50] + if isinstance(r, dict) + ] + issue_models = [ + Issue(id=i["id"], number=i["number"], title=i["title"], state=i["state"], labels=[l.get("name", "") for l in i.get("labels", [])], assignees=[a.get("login", "") for a in i.get("assignees", [])], url=i["html_url"]) + for i in issues_data[:50] + if isinstance(i, dict) + ] + pr_models = [ + PullRequest(id=p["id"], number=p["number"], title=p["title"], state=p["state"], user=p.get("user", {}).get("login", ""), url=p["html_url"]) + for p in prs_data[:50] + if isinstance(p, dict) + ] + ctx = compute(user_model, repo_models, issue_models, pr_models) + return JSONResponse({ + "user": ctx.user.dict(), + "repos": [r.dict() for r in ctx.repos], + "issues": [i.dict() for i in ctx.issues], + "pull_requests": [p.dict() for p in ctx.pull_requests], + "view": ctx.view, + "deltas": [d.dict() for d in ctx.deltas], + }) diff --git a/src/models.py b/src/models.py index e69de29..357c9b8 100644 --- a/src/models.py +++ b/src/models.py @@ -0,0 +1,52 @@ +from pydantic import BaseModel + + +class User(BaseModel): + id: int + login: str + full_name: str = "" + email: str = "" + + +class Repo(BaseModel): + id: int + name: str + full_name: str + description: str = "" + url: str + updated_at: str = "" + + +class Issue(BaseModel): + id: int + number: int + title: str + state: str + labels: list[str] = [] + assignees: list[str] = [] + url: str + + +class PullRequest(BaseModel): + id: int + number: int + title: str + state: str + user: str + url: str + + +class ViewContext(BaseModel): + user: User + repos: list[Repo] = [] + issues: list[Issue] = [] + pull_requests: list[PullRequest] = [] + view: str = "dashboard" + deltas: list[dict] = [] + + +class SuggestionDelta(BaseModel): + panel: str + action: str + target: str = "" + priority: str = "medium" diff --git a/src/suggestion_engine.py b/src/suggestion_engine.py index e69de29..0774a5f 100644 --- a/src/suggestion_engine.py +++ b/src/suggestion_engine.py @@ -0,0 +1,36 @@ +from src.models import Issue, PullRequest, Repo, SuggestionDelta, User, ViewContext + + +def _load(user: User, repos: list[Repo], issues: list[Issue], prs: list[PullRequest], view: str) -> ViewContext: + return ViewContext(user=user, repos=repos, issues=issues, pull_requests=prs, view=view) + + +def _suggest_review(prs: list[PullRequest]) -> list[SuggestionDelta]: + deltas: list[SuggestionDelta] = [] + for pr in prs: + deltas.append(SuggestionDelta(panel="review", action="review", target=str(pr.number), priority="high")) + return deltas[:5] + + +def _suggest_triage(issues: list[Issue]) -> list[SuggestionDelta]: + deltas: list[SuggestionDelta] = [] + for issue in issues: + if issue.state != "open": + continue + priority = "high" if any(label.lower() in {"p0", "priority-high"} for label in issue.labels) else "medium" + deltas.append(SuggestionDelta(panel="triage", action="open", target=str(issue.number), priority=priority)) + return deltas[:8] + + +def _suggest_focus(repos: list[Repo], prs: list[PullRequest], issues: list[Issue]) -> list[SuggestionDelta]: + if len(prs) > len(issues): + return [SuggestionDelta(panel="layout", action="highlight", target="review", priority="medium")] + if len(issues) > len(prs): + return [SuggestionDelta(panel="layout", action="highlight", target="issues", priority="medium")] + return [SuggestionDelta(panel="layout", action="highlight", target="dashboard", priority="low")] + + +def compute(user: User, repos: list[Repo], issues: list[Issue], prs: list[PullRequest], view: str = "dashboard") -> ViewContext: + ctx = _load(user, repos, issues, prs, view) + ctx.deltas = _suggest_review(prs) + _suggest_triage(issues) + _suggest_focus(repos, prs, issues) + return ctx diff --git a/src/views.py b/src/views.py index e69de29..a891d9f 100644 --- a/src/views.py +++ b/src/views.py @@ -0,0 +1,9 @@ +from fastapi import APIRouter +from fastapi.responses import HTMLResponse + +router = APIRouter() + + +@router.get("/", response_class=HTMLResponse) +async def dashboard() -> str: + return open("frontend/index.html").read() diff --git a/tests/test_suggestion_engine.py b/tests/test_suggestion_engine.py index e69de29..8549a6c 100644 --- a/tests/test_suggestion_engine.py +++ b/tests/test_suggestion_engine.py @@ -0,0 +1,16 @@ +from src.models import Issue, PullRequest, Repo, User +from src.suggestion_engine import compute + + +def test_suggest_review_when_more_prs_than_issues(): + user = User(id=1, login="timmy", full_name="Timmy", email="timmy@example.com") + prs = [PullRequest(id=1, number=1, title="p", state="open", user="timmy", url="http://example.com")] + ctx = compute(user, [], [], prs) + assert [d for d in getattr(ctx, "deltas", []) if d.panel == "review"] + + +def test_suggest_triage_when_open_issues(): + user = User(id=1, login="timmy", full_name="Timmy", email="timmy@example.com") + issues = [Issue(id=1, number=1, title="t", state="open", labels=["P0"], assignees=[], url="http://example.com")] + ctx = compute(user, [], issues, []) + assert [d for d in getattr(ctx, "deltas", []) if d.panel == "triage"]