157 lines
4.8 KiB
Python
157 lines
4.8 KiB
Python
"""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,
|
||
}
|