Files
rdgen/rdgenerator/build_backends/gitea.py
T
naeeo c8f41a34c0 修复列表页加载慢并支持 Gitea 1.27 一键停止
1. 状态同步加 20 秒节流、超时缩短到 8 秒;Gitea 改用 runs 列表接口一次批量取回全部任务状态,在途任务并发补查作业,消除逐任务串行 API 请求导致的后台/我的构建/详情页慢。

2. 停止时先区分 404:运行已删除则本地标记取消;Gitea 1.27 无取消 API 时,若后台配置了 Gitea 网页账号密码则模拟网页登录取消,否则回退手动链接提示。

3. 后台 Gitea 连接设置新增网页登录账号/密码两个可选项。
2026-09-30 01:04:16 +08:00

496 lines
22 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
"""Gitea Actions 构建后端(自建 Gitea 1.23+,建议 1.27+)。
API 与 GitHub Actions 几乎同形(已对照 Gitea 官方 Swagger 核实):
- 触发:POST /api/v1/repos/{owner}/{repo}/actions/workflows/{file}/dispatches
?return_run_details=true,body {ref, inputs},200 返回
{workflow_run_id, html_url, run_url},不传该参数为 204;
- 状态:GET /api/v1/repos/{owner}/{repo}/actions/runs/{id},
status/conclusion 字段仿 GitHub;
- 鉴权:Authorization: token <PAT>。
"""
import re
from .. import app_settings, ghnet
from .base import BackendError, DispatchResult, RunInfo, WORKFLOW_FILES, BuildBackend
# Gitea 未结束状态(blocked=等待审批,waiting=等待 runner)
_GITEA_ACTIVE = {'queued', 'in_progress', 'waiting', 'blocked', 'requested', 'pending', 'running'}
# Gitea 原生终态(部分版本 run.status 直接返回结论,不走 completed+conclusion)
_GITEA_TERMINAL = {'success', 'failure', 'cancelled', 'skipped', 'timed_out', 'stopped', 'dead'}
# Gitea 结论 -> rdgen 内部状态
_CONCLUSION_MAP = {
'success': 'success',
'failure': 'failure',
'cancelled': 'cancelled',
'skipped': 'skipped',
'timed_out': 'timed_out',
'stopped': 'cancelled',
'dead': 'failure',
}
# 作业出现这些结论时,整个运行最终必然失败(cleanup 等收尾作业除外)
_JOB_BAD_CONCLUSIONS = {'failure', 'cancelled', 'timed_out', 'dead'}
class GiteaBackend(BuildBackend):
name = 'gitea'
label = 'Gitea'
# ---- 取值辅助 ----
def _server(self):
return app_settings.get_value('GITEA_SERVER_URL').strip().rstrip('/')
def _owner(self):
return app_settings.get_value('GITEA_OWNER').strip()
def _repo(self):
return app_settings.get_value('GITEA_REPO').strip() or 'rdgen'
def _branch(self):
return app_settings.get_value('GITEA_BRANCH').strip() or 'master'
def _ref(self):
branch = self._branch()
return branch if branch.startswith('refs/') else f'refs/heads/{branch}'
def _proxy(self):
return app_settings.get_value('GITEA_PROXY').strip()
def _api(self, path):
return f'{self._server()}/api/v1{path}'
# ---- 配置 ----
def configured(self):
return bool(self._server() and self._owner() and app_settings.get_value('GITEA_TOKEN').strip())
def warnings(self):
warnings = []
if not self._server():
warnings.append('Gitea 服务器地址未配置,无法触发构建。')
if not self._owner():
warnings.append('Gitea 仓库属主未配置,无法触发构建。')
if not app_settings.get_value('GITEA_TOKEN').strip():
warnings.append('Gitea 访问令牌未配置,无法触发构建。')
warnings.extend(self._shared_warnings())
return warnings
# ---- 工作流选择 ----
def workflow_file_for(self, platform, selfhosted=False):
# Gitea 的 runner 全部为自建注册,无需 sh- 变体,统一用标准工作流
return WORKFLOW_FILES.get(platform, 'generator-windows.yml')
# ---- 触发 ----
def dispatch(self, workflow_file, inputs, *, timeout=20):
if not self.configured():
raise BackendError('Gitea 构建服务未完成配置(服务器地址 / 属主 / 令牌缺失)。', kind='config')
url = self._api(
f'/repos/{self._owner()}/{self._repo()}'
f'/actions/workflows/{workflow_file}/dispatches?return_run_details=true'
)
payload = {'ref': self._ref(), 'inputs': inputs}
try:
resp = ghnet.post(
url, json=payload,
token=app_settings.get_value('GITEA_TOKEN'),
timeout=timeout, flavor='gitea', proxy=self._proxy(),
)
except Exception as e:
raise BackendError(f'无法连接 Gitea 构建服务:{e}', kind='network')
if resp.status_code in (200, 204):
data = {}
try:
data = resp.json()
except ValueError:
pass
run_id = data.get('workflow_run_id')
html_url = data.get('html_url') or ''
# 旧版 Gitea 或未回传详情时,按工作流最新一条运行兜底取 run id
if not run_id and resp.status_code == 204:
fallback = self._latest_run_id(workflow_file)
if fallback:
run_id, html_url = fallback
return DispatchResult(run_id=run_id, html_url=html_url or self.log_url(run_id))
detail = ''
try:
detail = (resp.text or '')[:200]
except Exception:
pass
if resp.status_code == 404:
msg = f'Gitea 仓库或工作流 {workflow_file} 不存在(请确认仓库已启用 Repository Actions)。'
elif resp.status_code in (401, 403):
msg = 'Gitea 拒绝访问:请检查令牌是否有效且具备该仓库 Actions 读写权限。'
elif resp.status_code == 422:
msg = f'Gitea 校验失败(分支或工作流参数有误):{detail}'
else:
msg = f'Gitea 构建服务拒绝了启动请求(HTTP {resp.status_code}):{detail}'
raise BackendError(msg, kind='rejected')
def _latest_run_id(self, workflow_file, *, timeout=10):
"""204 兜底:查该工作流最新一条运行(仅在刚触发后调用)。"""
url = self._api(
f'/repos/{self._owner()}/{self._repo()}'
f'/actions/workflows/{workflow_file}/runs?limit=1'
)
try:
resp = ghnet.get(
url, token=app_settings.get_value('GITEA_TOKEN'),
timeout=timeout, flavor='gitea', proxy=self._proxy(),
)
if resp.status_code != 200:
return None
data = resp.json()
runs = data.get('workflow_runs') if isinstance(data, dict) else data
if runs:
run = runs[0]
return run.get('id'), run.get('html_url') or ''
except Exception as e:
print(f'查询 Gitea 最新运行失败:{e}')
return None
# ---- 状态查询 ----
def get_run(self, run_id, *, timeout=12):
url = self._api(f'/repos/{self._owner()}/{self._repo()}/actions/runs/{run_id}')
token = app_settings.get_value('GITEA_TOKEN')
proxy = self._proxy()
try:
resp = ghnet.get(url, token=token, timeout=timeout, flavor='gitea', proxy=proxy)
except Exception as e:
print(f'查询 Gitea 状态出错:{e}')
return None
if resp.status_code != 200:
return None
data = resp.json()
html_url = data.get('html_url') or self.log_url(run_id)
status = (data.get('status') or '').lower()
conclusion = (data.get('conclusion') or '').lower()
if status == 'completed':
return RunInfo(
finished=True,
status=_CONCLUSION_MAP.get(conclusion, 'failure'),
html_url=html_url,
)
# 部分 Gitea 版本直接在 status 返回终态(无 completed 包裹)
if status in _GITEA_TERMINAL:
return RunInfo(
finished=True,
status=_CONCLUSION_MAP.get(status, 'failure'),
html_url=html_url,
)
# 仅在途运行才需要 jobs:识别「前置作业失败但父运行长期挂起」的 Gitea 已知行为
jobs_url = self._api(
f'/repos/{self._owner()}/{self._repo()}/actions/runs/{run_id}/jobs'
)
try:
jobs_resp = ghnet.get(
jobs_url, token=token, timeout=timeout, flavor='gitea', proxy=proxy)
except Exception as e:
print(f'查询 Gitea 作业状态出错:{e}')
jobs_resp = None
if jobs_resp is not None:
bad = self._bad_job_conclusion(jobs_resp)
if bad:
return RunInfo(finished=True, status=bad, html_url=html_url)
# blocked(等待审批)等也算进行中
return RunInfo(finished=False, status=status or 'in_progress', html_url=html_url)
@staticmethod
def _bad_job_conclusion(resp):
"""从 jobs 响应中判定终态:有非 cleanup 作业失败/取消/超时时返回对应状态,否则 None。"""
try:
if resp.status_code != 200:
return None
data = resp.json()
jobs = data.get('jobs', data if isinstance(data, list) else [])
for job in jobs:
name = (job.get('name') or '').lower()
if name.startswith('cleanup'):
continue
conclusion = (job.get('conclusion') or '').lower()
if conclusion in _JOB_BAD_CONCLUSIONS:
return _CONCLUSION_MAP.get(conclusion, 'failure')
except Exception as e:
print(f'解析 Gitea 作业状态出错:{e}')
return None
# ---- 状态批量查询(列表页用,一次请求取回全部运行) ----
def get_runs_batch(self, run_ids, *, timeout=12):
"""一次 runs 列表请求批量同步状态,避免每个在途任务各发两个请求。
返回 {str(run_id): RunInfo};列表拿不到时返回 None 让调用方退化逐个查询。
列表中仍在途的运行,再查 jobs 以识别「前置作业失败但父运行挂起」的僵死;
列表中不存在的 id(已删除或超出最近窗口)不放进返回字典。
"""
want = {str(i) for i in run_ids if i}
if not want:
return {}
url = self._api(
f'/repos/{self._owner()}/{self._repo()}/actions/runs?limit=50'
)
try:
resp = ghnet.get(
url, token=app_settings.get_value('GITEA_TOKEN'),
timeout=timeout, flavor='gitea', proxy=self._proxy(),
)
except Exception as e:
print(f'批量查询 Gitea 状态出错:{e}')
return None
if resp.status_code != 200:
return None
try:
data = resp.json()
items = data.get('workflow_runs') if isinstance(data, dict) else data
except ValueError:
return None
if not items:
return {}
result = {}
active = [] # (id, html_url)
for item in items:
rid = item.get('id')
if rid is None or str(rid) not in want:
continue
html_url = item.get('html_url') or self.log_url(rid)
status = (item.get('status') or '').lower()
conclusion = (item.get('conclusion') or '').lower()
if status == 'completed' or status in _GITEA_TERMINAL:
result[str(rid)] = RunInfo(
finished=True,
status=_CONCLUSION_MAP.get(conclusion or status, 'failure'),
html_url=html_url,
)
else:
active.append((rid, html_url))
# 在途运行查 jobs 识别 Gitea 的作业级失败挂起(通常在途任务很少)
if active:
token = app_settings.get_value('GITEA_TOKEN')
proxy = self._proxy()
owner_repo = f'{self._owner()}/{self._repo()}'
def _probe(rid_html):
rid, html_url = rid_html
jobs_url = self._api(
f'/repos/{owner_repo}/actions/runs/{rid}/jobs'
)
try:
jobs_resp = ghnet.get(
jobs_url, token=token, timeout=timeout,
flavor='gitea', proxy=proxy,
)
except Exception:
return str(rid), RunInfo(False, 'in_progress', html_url)
bad = self._bad_job_conclusion(jobs_resp)
if bad:
return str(rid), RunInfo(True, bad, html_url)
return str(rid), RunInfo(False, 'in_progress', html_url)
from concurrent.futures import ThreadPoolExecutor
with ThreadPoolExecutor(max_workers=min(8, len(active))) as pool:
for key, info in pool.map(_probe, active):
result[key] = info
return result
# ---- 取消运行 ----
def cancel_run(self, run_id, *, timeout=15):
"""请求平台取消运行。
- 2xx:已接受取消;
- 409:运行已结束(调用方应改为拉取真实状态);
- 运行在平台上已不存在(如已被手工删除):kind='missing';
- 旧版 Gitea(≤1.27)无取消 API:配置了 Web 账密则模拟网页登录取消,
否则 kind='unsupported',引导用户去运行页手动 Cancel。
"""
url = self._api(f'/repos/{self._owner()}/{self._repo()}/actions/runs/{run_id}/cancel')
try:
resp = ghnet.post(
url, token=app_settings.get_value('GITEA_TOKEN'),
timeout=timeout, flavor='gitea', proxy=self._proxy(),
)
except Exception as e:
raise BackendError(f'无法连接 Gitea 构建服务:{e}', kind='network')
if 200 <= resp.status_code < 300:
return
if resp.status_code == 409:
raise BackendError('该构建运行已经结束,无法取消。', kind='finished')
if resp.status_code == 404:
exists = self._run_exists(run_id, timeout=timeout)
if exists is False:
raise BackendError(
'该运行在 Gitea 上已不存在(可能已被删除)。', kind='missing')
# 运行存在但 cancel 端点 404:Gitea ≤ 1.27 没有取消 API
if self._cancel_via_web(run_id, timeout=timeout):
return
raise BackendError(
'当前 Gitea 版本(1.27 及更早)未开放取消运行的 API。'
'可在后台「Gitea 连接」填写 Gitea 网页登录账号以启用一键停止,'
'或在 Gitea 的运行页面手动点击 Cancel 停止;升级到 Gitea 1.28+ 后可直接一键停止。',
kind='unsupported',
)
detail = ''
try:
detail = (resp.text or '')[:200]
except Exception:
pass
raise BackendError(f'Gitea 拒绝了取消请求(HTTP {resp.status_code}):{detail}')
def _run_exists(self, run_id, *, timeout=10):
"""GET run:存在 True / 404 False / 无法确定 None。"""
url = self._api(f'/repos/{self._owner()}/{self._repo()}/actions/runs/{run_id}')
try:
resp = ghnet.get(
url, token=app_settings.get_value('GITEA_TOKEN'),
timeout=timeout, flavor='gitea', proxy=self._proxy(),
)
except Exception:
return None
if resp.status_code == 200:
return True
if resp.status_code == 404:
return False
return None
def _cancel_via_web(self, run_id, *, timeout=20):
"""Gitea ≤1.27 无取消 API 时,用网页会话方式取消(需后台配置 Web 账密)。
模拟浏览器:登录拿会话 → 打开运行页拿 CSRF → POST 网页取消路由。
注意:部分 1.27 实例的登录页本身不输出 CSRF(登录 POST 豁免),
但取消操作需要登录后页面里的令牌;未配置账密、登录失败
(含开启两步验证)时返回 False。
"""
import requests
username = app_settings.get_value('GITEA_WEB_USERNAME').strip()
password = app_settings.get_value('GITEA_WEB_PASSWORD')
if not username or not password:
return False
server = self._server()
owner, repo = self._owner(), self._repo()
proxies = ghnet.proxies(self._proxy())
sess = requests.Session()
sess.proxies.update(proxies)
ua = 'Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 rdgen'
try:
# 1. 登录页(建立会话;CSRF 若页面有则带上,没有也无妨——该实例登录豁免)
r = sess.get(f'{server}/user/login', timeout=timeout,
headers={'User-Agent': ua}, allow_redirects=True)
csrf = self._extract_csrf(r.text)
login_data = {'user_name': username, 'password': password}
login_headers = {'User-Agent': ua, 'Referer': f'{server}/user/login'}
if csrf:
login_data['_csrf'] = csrf
login_headers['X-Csrf-Token'] = csrf
# 2. 提交登录(Gitea 用 user_name 字段;成功后 302 离开 /user/login)
r = sess.post(
f'{server}/user/login', data=login_data,
headers=login_headers, timeout=timeout, allow_redirects=True,
)
if '/user/login' in r.url or 'name="password"' in r.text:
print('Gitea 网页取消:登录失败(账号密码错误或启用了两步验证)')
return False
# 3. 打开运行页取 CSRF(登录后页面才输出令牌)
run_page = f'{server}/{owner}/{repo}/actions/runs/{run_id}'
r = sess.get(run_page, timeout=timeout, headers={'User-Agent': ua})
if r.status_code != 200:
print(f'Gitea 网页取消:运行页 HTTP {r.status_code}(账号可能无该仓库访问权限)')
return False
csrf = self._extract_csrf(r.text)
if not csrf:
print('Gitea 网页取消:登录后运行页仍未找到 CSRF 令牌(账号可能无该仓库写权限)')
return False
# 4. POST 网页取消路由(302/303 回运行页即成功)
r = sess.post(
f'{run_page}/cancel',
data={'_csrf': csrf},
headers={'User-Agent': ua, 'Referer': run_page,
'X-Csrf-Token': csrf},
timeout=timeout, allow_redirects=False,
)
if r.status_code in (200, 302, 303):
return True
print(f'Gitea 网页取消:取消请求被拒绝 HTTP {r.status_code}')
return False
except Exception as e:
print(f'Gitea 网页取消出错:{e}')
return False
@staticmethod
def _extract_csrf(html):
"""从 Gitea 页面提取 CSRF:优先 meta,其次隐藏域,最后前端 JSON 配置。"""
if not html:
return None
for pattern in (
r'name="_csrf"\s+content="([^"]+)"',
r'content="([^"]+)"\s+name="_csrf"',
r'name="_csrf"\s+value="([^"]+)"',
r'"(?:csrfToken|_csrf)"\s*:\s*"([^"]+)"',
):
m = re.search(pattern, html)
if m:
return m.group(1)
return None
# ---- 日志链接 ----
def log_url(self, run_id):
if not run_id:
return ''
return f'{self._server()}/{self._owner()}/{self._repo()}/actions/runs/{run_id}'
# ---- 连接测试 ----
def test_connection(self):
server = self._server()
owner = self._owner()
token = app_settings.get_value('GITEA_TOKEN').strip()
repo = self._repo()
if not server or not owner or not token:
return {'ok': False, 'message': 'Gitea 服务器地址、仓库属主或访问令牌为空,请先保存配置。'}
if not (server.startswith('http://') or server.startswith('https://')):
return {'ok': False, 'message': 'Gitea 服务器地址需以 http:// 或 https:// 开头。'}
try:
resp = ghnet.get(
self._api(f'/repos/{owner}/{repo}'),
token=token, timeout=10, flavor='gitea', proxy=self._proxy(),
)
if resp.status_code == 401:
return {'ok': False, 'message': '令牌无效或已过期(Gitea 返回 401)。'}
if resp.status_code == 404:
return {'ok': False,
'message': f'找不到 Gitea 仓库 {owner}/{repo},或令牌无权访问该仓库。'}
if resp.status_code != 200:
return {'ok': False, 'message': f'Gitea 返回异常状态码:{resp.status_code}。'}
act = ghnet.get(
self._api(f'/repos/{owner}/{repo}/actions/workflows'),
token=token, timeout=10, flavor='gitea', proxy=self._proxy(),
)
if act.status_code != 200:
return {'ok': False,
'message': '可以访问仓库,但无法读取 Actions 工作流;请在仓库设置中启用 Repository Actions。'}
# 版本检查:1.23 起支持 workflow_dispatch;建议 1.27+(return_run_details 与缓存兼容性更好)
version_msg = ''
ver = ghnet.get(
self._api('/version'),
token=token, timeout=10, flavor='gitea', proxy=self._proxy(),
)
if ver.status_code == 200:
raw = (ver.json().get('version') or '').strip()
parts = raw.replace('+dev', '').split('.')
try:
major, minor = int(parts[0]), int(parts[1])
if (major, minor) < (1, 23):
return {'ok': False,
'message': f'Gitea 版本为 {raw},低于 1.23,不支持 workflow_dispatch 远程触发,请先升级。'}
if (major, minor) < (1, 27):
version_msg = f'当前 Gitea 版本为 {raw},可触发构建,但建议升级到 1.27+(运行详情回传与缓存兼容性更好)。'
except (ValueError, IndexError):
pass
msg = f'连接正常:令牌可访问仓库 {owner}/{repo},Actions 已启用。'
if version_msg:
msg += version_msg
return {'ok': True, 'message': msg}
except Exception as e:
return {'ok': False, 'message': f'无法连接 Gitea API:{e}(请检查服务器地址与网络/代理配置)'}