main.py

import datetime as dt
import math
import os
import re
import secrets
import sqlite3
import time
from typing import Any
from urllib.parse import urlencode

from fastapi import FastAPI, Request
from fastapi.responses import HTMLResponse, RedirectResponse
from fastapi.staticfiles import StaticFiles
from fastapi.templating import Jinja2Templates
import httpx
from dotenv import load_dotenv

load_dotenv()

app = FastAPI(title="Centauri")
templates = Jinja2Templates(directory="templates")

app.mount("/static", StaticFiles(directory="static"), name="static")

DB_PATH = os.getenv("CENTAURI_DB_PATH", "centauri.db")
OAUTH_AUTHORIZE_URL = "https://github.com/login/oauth/authorize"
OAUTH_TOKEN_URL = "https://github.com/login/oauth/access_token"
GITHUB_API_BASE = "https://api.github.com"
DEFAULT_CALLBACK_URL = "http://localhost:1580/oauth/callback"
STARRED_CACHE_TTL_SECONDS = int(os.getenv("STARRED_CACHE_TTL_SECONDS", "300"))

_STARRED_CACHE: dict[str, tuple[float, list[dict[str, str | int | None]]]] = {}


def _get_connection() -> sqlite3.Connection:
    conn = sqlite3.connect(DB_PATH)
    conn.row_factory = sqlite3.Row
    return conn


def _init_db() -> None:
    with _get_connection() as conn:
        conn.execute(
            """
            CREATE TABLE IF NOT EXISTS github_tokens (
                id INTEGER PRIMARY KEY,
                user_id INTEGER,
                login TEXT,
                access_token TEXT NOT NULL,
                created_at TEXT NOT NULL,
                updated_at TEXT NOT NULL
            )
            """
        )
        conn.execute(
            """
            CREATE TABLE IF NOT EXISTS oauth_state (
                state TEXT PRIMARY KEY,
                created_at TEXT NOT NULL
            )
            """
        )


@app.on_event("startup")
def _on_startup() -> None:
    _init_db()


@app.get("/", response_class=HTMLResponse)
async def index(request: Request) -> HTMLResponse:
    token_present = bool(_get_active_token())
    return templates.TemplateResponse(
        "index.html",
        {
            "request": request,
            "page_title": "Centauri",
            "github_token_present": token_present,
            "oauth_ready": _oauth_ready(),
        },
    )


def _parse_last_page(link_header: str | None) -> int | None:
    if not link_header:
        return None
    for part in link_header.split(","):
        if 'rel="last"' not in part:
            continue
        match = re.search(r"[?&]page=(\d+)", part)
        if match:
            return int(match.group(1))
    return None


def _oauth_ready() -> bool:
    return bool(os.getenv("GITHUB_CLIENT_ID") and os.getenv("GITHUB_CLIENT_SECRET"))


def _get_callback_url() -> str:
    return os.getenv("GITHUB_CALLBACK_URL", DEFAULT_CALLBACK_URL)


def _store_state(state: str) -> None:
    with _get_connection() as conn:
        conn.execute(
            "INSERT OR REPLACE INTO oauth_state (state, created_at) VALUES (?, ?)",
            (state, dt.datetime.utcnow().isoformat()),
        )


def _consume_state(state: str) -> bool:
    with _get_connection() as conn:
        row = conn.execute(
            "SELECT state FROM oauth_state WHERE state = ?",
            (state,),
        ).fetchone()
        if not row:
            return False
        conn.execute("DELETE FROM oauth_state WHERE state = ?", (state,))
        return True


def _store_token(access_token: str, user_id: int | None, login: str | None) -> None:
    now = dt.datetime.utcnow().isoformat()
    with _get_connection() as conn:
        conn.execute("DELETE FROM github_tokens")
        conn.execute(
            """
            INSERT INTO github_tokens (user_id, login, access_token, created_at, updated_at)
            VALUES (?, ?, ?, ?, ?)
            """,
            (user_id, login, access_token, now, now),
        )


def _load_db_token() -> str | None:
    with _get_connection() as conn:
        row = conn.execute(
            "SELECT access_token FROM github_tokens ORDER BY updated_at DESC LIMIT 1"
        ).fetchone()
    if row:
        return row["access_token"]
    return None


def _get_active_token() -> str | None:
    return _load_db_token() or os.getenv("GITHUB_TOKEN")


def _normalize_repo(item: dict[str, Any]) -> dict[str, str | int | None]:
    return {
        "name": item.get("full_name") or item.get("name") or "unknown",
        "url": item.get("html_url") or "#",
        "description": item.get("description"),
        "language": item.get("language"),
        "stars": item.get("stargazers_count") or 0,
        "forks": item.get("forks_count") or 0,
    }


def _get_cached_starred(token: str) -> list[dict[str, str | int | None]] | None:
    cached = _STARRED_CACHE.get(token)
    if not cached:
        return None
    fetched_at, repos = cached
    if time.monotonic() - fetched_at > STARRED_CACHE_TTL_SECONDS:
        _STARRED_CACHE.pop(token, None)
        return None
    return repos


def _set_cached_starred(
    token: str, repos: list[dict[str, str | int | None]]
) -> None:
    _STARRED_CACHE[token] = (time.monotonic(), repos)


@app.get("/stars", response_class=HTMLResponse)
async def stars(
    request: Request,
    page: int = 1,
    per_page: int = 20,
    sort: str = "default",
    language: str = "all",
) -> HTMLResponse:
    safe_page = max(1, page)
    safe_per_page = min(max(5, per_page), 50)
    token = _get_active_token()
    oauth_ready = _oauth_ready()

    if not token:
        return templates.TemplateResponse(
            "partials/stars_list.html",
            {
                "request": request,
                "repos": [],
                "page": safe_page,
                "per_page": safe_per_page,
                "total_pages": None,
                "error": "Connect GitHub to enable API calls.",
                "login_url": "/oauth/login" if oauth_ready else None,
                "sort": sort,
                "language": language,
                "languages": [],
            },
        )

    headers = {
        "Authorization": f"Bearer {token}",
        "Accept": "application/vnd.github+json",
        "X-GitHub-Api-Version": "2022-11-28",
    }
    cached_repos = _get_cached_starred(token)
    if cached_repos is None:
        all_items: list[dict[str, Any]] = []
        try:
            async with httpx.AsyncClient(timeout=10.0) as client:
                first_response = await client.get(
                    f"{GITHUB_API_BASE}/user/starred",
                    headers=headers,
                    params={"page": 1, "per_page": 100},
                )
                responses = [first_response]
                total_pages = _parse_last_page(first_response.headers.get("Link")) or 1
                for page_number in range(2, total_pages + 1):
                    page_response = await client.get(
                        f"{GITHUB_API_BASE}/user/starred",
                        headers=headers,
                        params={"page": page_number, "per_page": 100},
                    )
                    responses.append(page_response)
        except httpx.HTTPError as exc:
            return templates.TemplateResponse(
                "partials/stars_list.html",
                {
                    "request": request,
                    "repos": [],
                    "page": safe_page,
                    "per_page": safe_per_page,
                    "total_pages": None,
                    "error": f"GitHub request failed: {exc}",
                    "login_url": None,
                    "sort": sort,
                    "language": language,
                    "languages": [],
                },
            )

        for response in responses:
            if response.status_code != 200:
                try:
                    message = response.json().get("message", "Unknown error")
                except ValueError:
                    message = response.text[:200] or "Unknown error"
                return templates.TemplateResponse(
                    "partials/stars_list.html",
                    {
                        "request": request,
                        "repos": [],
                        "page": safe_page,
                        "per_page": safe_per_page,
                        "total_pages": None,
                        "error": f"GitHub API error {response.status_code}: {message}",
                        "login_url": None,
                        "sort": sort,
                        "language": language,
                        "languages": [],
                    },
                )

        for response in responses:
            all_items.extend(response.json())

        cached_repos = [_normalize_repo(item) for item in all_items]
        _set_cached_starred(token, cached_repos)

    languages_set = {repo.get("language") for repo in cached_repos}
    languages = sorted(lang for lang in languages_set if lang)
    if any(lang is None for lang in languages_set):
        languages.append("Unknown")

    if language == "all":
        filtered_repos = list(cached_repos)
    elif language == "unknown":
        filtered_repos = [repo for repo in cached_repos if not repo.get("language")]
    else:
        filtered_repos = [
            repo for repo in cached_repos if repo.get("language") == language
        ]

    if sort == "stars_desc":
        filtered_repos.sort(key=lambda repo: int(repo.get("stars") or 0), reverse=True)
    elif sort == "stars_asc":
        filtered_repos.sort(key=lambda repo: int(repo.get("stars") or 0))

    total_pages = max(1, math.ceil(len(filtered_repos) / safe_per_page))
    if safe_page > total_pages:
        safe_page = total_pages
    start = (safe_page - 1) * safe_per_page
    end = start + safe_per_page
    repos = filtered_repos[start:end]

    return templates.TemplateResponse(
        "partials/stars_list.html",
        {
            "request": request,
            "repos": repos,
            "page": safe_page,
            "per_page": safe_per_page,
            "total_pages": total_pages,
            "error": None,
            "login_url": None,
            "sort": sort,
            "language": language,
            "languages": languages,
        },
    )


@app.get("/oauth/login", response_model=None)
async def oauth_login() -> RedirectResponse | HTMLResponse:
    if not _oauth_ready():
        return HTMLResponse(
            "Missing GITHUB_CLIENT_ID or GITHUB_CLIENT_SECRET.", status_code=500
        )

    state = secrets.token_urlsafe(24)
    _store_state(state)

    params = {
        "client_id": os.getenv("GITHUB_CLIENT_ID"),
        "redirect_uri": _get_callback_url(),
        "scope": "read:user",
        "state": state,
    }
    url = f"{OAUTH_AUTHORIZE_URL}?{urlencode(params)}"
    return RedirectResponse(url)


@app.get("/oauth/callback", response_model=None)
async def oauth_callback(
    request: Request,
    code: str | None = None,
    state: str | None = None,
    error: str | None = None,
    error_description: str | None = None,
) -> RedirectResponse | HTMLResponse:
    if error:
        detail = error_description or error
        return HTMLResponse(f"OAuth failed: {detail}", status_code=400)

    if not code or not state:
        return HTMLResponse("Missing OAuth parameters.", status_code=400)

    if not _consume_state(state):
        return HTMLResponse("Invalid OAuth state.", status_code=400)

    token_payload = {
        "client_id": os.getenv("GITHUB_CLIENT_ID"),
        "client_secret": os.getenv("GITHUB_CLIENT_SECRET"),
        "code": code,
        "redirect_uri": _get_callback_url(),
    }

    try:
        async with httpx.AsyncClient(timeout=10.0) as client:
            token_response = await client.post(
                OAUTH_TOKEN_URL,
                data=token_payload,
                headers={"Accept": "application/json"},
            )
            token_response.raise_for_status()
    except httpx.HTTPError as exc:
        return HTMLResponse(f"Token exchange failed: {exc}", status_code=400)

    token_data = token_response.json()
    access_token = token_data.get("access_token")
    if not access_token:
        return HTMLResponse("Missing access token in response.", status_code=400)

    try:
        async with httpx.AsyncClient(timeout=10.0) as client:
            user_response = await client.get(
                f"{GITHUB_API_BASE}/user",
                headers={
                    "Authorization": f"Bearer {access_token}",
                    "Accept": "application/vnd.github+json",
                    "X-GitHub-Api-Version": "2022-11-28",
                },
            )
            user_response.raise_for_status()
    except httpx.HTTPError as exc:
        return HTMLResponse(f"Failed to fetch user profile: {exc}", status_code=400)

    user_data = user_response.json()
    _store_token(access_token, user_data.get("id"), user_data.get("login"))

    return RedirectResponse("/")