"""Gitea API client с простым in-memory кэшем (5 минут).""" import re import time import requests from . import config _cache = {} CACHE_TTL = 300 # 5 минут def _headers(): if config.GITEA_API_TOKEN: return {"Authorization": f"token {config.GITEA_API_TOKEN}"} return {} def _cache_get(key): entry = _cache.get(key) if not entry: return None ts, data = entry if time.time() - ts > CACHE_TTL: return None return data def _cache_set(key, data): _cache[key] = (time.time(), data) def _api_get(path, params=None): """Возвращает (data, total) или (None, None) при ошибке.""" url = f"{config.GITEA_URL}/api/v1{path}" try: r = requests.get(url, headers=_headers(), params=params, timeout=5) if r.status_code == 200: total_header = r.headers.get("X-Total-Count") total = int(total_header) if total_header and total_header.isdigit() else None try: data = r.json() except Exception: data = None return data, total return None, None except Exception as e: print(f"[gitea_api] {path}: {e}", flush=True) return None, None def extract_repo_from_url(url): """https://git.morozovka.space/owner/repo(.git)? → owner/repo""" if not url: return None m = re.search(r"[:/]([^/\s]+)/([^/\s]+?)(?:\.git)?/?$", url.strip()) if m: return f"{m.group(1)}/{m.group(2)}" return None def get_repo_info(owner, repo): key = f"repo:{owner}/{repo}" cached = _cache_get(key) if cached is not None: return cached data, _ = _api_get(f"/repos/{owner}/{repo}") _cache_set(key, data) return data def get_latest_commits(owner, repo, limit=5): key = f"commits:{owner}/{repo}:{limit}" cached = _cache_get(key) if cached is not None: return cached data, _ = _api_get( f"/repos/{owner}/{repo}/commits", params={"limit": limit, "page": 1, "stat": "false"}, ) result = [] if isinstance(data, list): for c in data[:limit]: commit_obj = c.get("commit") or {} author_obj = commit_obj.get("author") or {} result.append({ "sha_short": (c.get("sha") or "")[:7], "message": (commit_obj.get("message") or "").split("\n")[0][:120], "author": author_obj.get("name") or "", "date": author_obj.get("date") or "", "html_url": c.get("html_url") or "", }) _cache_set(key, result) return result def get_open_issues_count(owner, repo): key = f"issues:{owner}/{repo}" cached = _cache_get(key) if cached is not None: return cached _, total = _api_get( f"/repos/{owner}/{repo}/issues", params={"state": "open", "type": "issues", "limit": 1}, ) _cache_set(key, total or 0) return total or 0 def get_open_prs_count(owner, repo): key = f"prs:{owner}/{repo}" cached = _cache_get(key) if cached is not None: return cached _, total = _api_get( f"/repos/{owner}/{repo}/pulls", params={"state": "open", "limit": 1}, ) _cache_set(key, total or 0) return total or 0 def get_project_stats(gitea_repo): """gitea_repo: 'owner/repo'. Возвращает dict для JSON-ответа.""" if not gitea_repo or "/" not in gitea_repo: return {"ok": False, "error": "no_repo", "message": "Не указан репозиторий проекта."} owner, repo = gitea_repo.split("/", 1) info = get_repo_info(owner, repo) if info is None: # 404 или 401 — не смогли прочитать if not config.GITEA_API_TOKEN: return {"ok": False, "error": "auth_required", "message": "Репозиторий приватный или не найден. " "Настройте GITEA_API_TOKEN в config.py."} return {"ok": False, "error": "api_unreachable", "message": "Не удалось получить данные из Gitea."} commits = get_latest_commits(owner, repo, limit=5) open_issues = get_open_issues_count(owner, repo) open_prs = get_open_prs_count(owner, repo) return { "ok": True, "repo": { "full_name": info.get("full_name", gitea_repo), "description": info.get("description") or "", "html_url": info.get("html_url", ""), "default_branch": info.get("default_branch") or "main", "stars": info.get("stars_count", 0), "forks": info.get("forks_count", 0), "is_private": bool(info.get("private")), }, "commits": commits, "open_issues": open_issues, "open_prs": open_prs, }