90966663bf
- Обновлены скрипты для поддержки групп сертификатов (SAN) в конфигурации. - Добавлен новый скрипт для запуска GitHub Actions workflows. - Обновлены Makefile и документация для отражения новых возможностей. - Улучшена обработка доменов в коде и конфигурации.
138 lines
4.5 KiB
Python
138 lines
4.5 KiB
Python
#!/usr/bin/env python3
|
||
import argparse
|
||
import json
|
||
import os
|
||
import re
|
||
import sys
|
||
import urllib.error
|
||
import urllib.request
|
||
from subprocess import PIPE, run
|
||
|
||
|
||
def get_env(name: str, default: str = "") -> str:
|
||
value = os.getenv(name, default)
|
||
return value.strip() if isinstance(value, str) else default
|
||
|
||
|
||
def fail(message: str, code: int = 1) -> None:
|
||
print(f"✗ {message}")
|
||
sys.exit(code)
|
||
|
||
|
||
def parse_args() -> argparse.Namespace:
|
||
parser = argparse.ArgumentParser(description="Dispatch Gitea Actions workflow")
|
||
parser.add_argument("--workflow", default="wiki-sync.yml", help="Workflow file name (e.g. wiki-sync.yml)")
|
||
parser.add_argument("--ref", default="master", help="Git ref for workflow dispatch")
|
||
parser.add_argument(
|
||
"--input",
|
||
action="append",
|
||
default=[],
|
||
help="Workflow input in key=value format. Can be provided multiple times.",
|
||
)
|
||
parser.add_argument(
|
||
"--require-input",
|
||
action="append",
|
||
default=[],
|
||
help="Required input key(s). Fails if key is missing or empty.",
|
||
)
|
||
return parser.parse_args()
|
||
|
||
|
||
def parse_inputs(items: list[str]) -> dict[str, str]:
|
||
result: dict[str, str] = {}
|
||
for item in items:
|
||
if "=" not in item:
|
||
fail(f"Неверный формат --input '{item}'. Используйте key=value")
|
||
key, value = item.split("=", 1)
|
||
key = key.strip()
|
||
value = value.strip()
|
||
if not key:
|
||
fail(f"Пустой ключ в --input '{item}'")
|
||
result[key] = value
|
||
return result
|
||
|
||
|
||
def get_repo_path() -> str:
|
||
result = run(["git", "config", "--get", "remote.origin.url"], stdout=PIPE, stderr=PIPE, text=True)
|
||
origin_url = (result.stdout or "").strip()
|
||
if result.returncode != 0 or not origin_url:
|
||
fail("remote.origin.url не найден. Проверьте, что команда запущена внутри git-репозитория.")
|
||
|
||
repo_path = re.sub(r"^(https?://[^/]+/|git@[^:]+:)", "", origin_url)
|
||
repo_path = re.sub(r"\.git$", "", repo_path)
|
||
if not repo_path:
|
||
fail(f"Не удалось вычислить owner/repo из remote.origin.url: {origin_url}")
|
||
return repo_path
|
||
|
||
|
||
def main() -> None:
|
||
args = parse_args()
|
||
|
||
token = get_env("DFGIT_TOKEN") or get_env("GIT_TOKEN")
|
||
if not token:
|
||
fail("не задан DFGIT_TOKEN (или GIT_TOKEN).")
|
||
|
||
dfgit_url = get_env("DFGIT_URL", "http://dfgit").rstrip("/")
|
||
workflow = args.workflow.strip() or "wiki-sync.yml"
|
||
ref = args.ref.strip() or "master"
|
||
inputs = parse_inputs(args.input)
|
||
|
||
for required_key in args.require_input:
|
||
required_key = required_key.strip()
|
||
if required_key and not inputs.get(required_key, "").strip():
|
||
fail(
|
||
f"Обязательный input '{required_key}' не задан. "
|
||
f"Передайте --input {required_key}=..."
|
||
)
|
||
|
||
repo_path = get_repo_path()
|
||
dispatch_url = f"{dfgit_url}/api/v1/repos/{repo_path}/actions/workflows/{workflow}/dispatches"
|
||
|
||
payload_data: dict[str, object] = {"ref": ref}
|
||
if inputs:
|
||
payload_data["inputs"] = inputs
|
||
|
||
payload = json.dumps(payload_data).encode("utf-8")
|
||
|
||
print(f"→ Репозиторий: {repo_path}")
|
||
print(f"→ Workflow: {workflow}, ref: {ref}")
|
||
if inputs:
|
||
print(f"→ Inputs: {json.dumps(inputs, ensure_ascii=False)}")
|
||
print(f"→ Запрос: {dispatch_url}")
|
||
|
||
request = urllib.request.Request(
|
||
dispatch_url,
|
||
data=payload,
|
||
method="POST",
|
||
headers={
|
||
"Authorization": f"token {token}",
|
||
"Content-Type": "application/json",
|
||
"Accept": "application/json",
|
||
},
|
||
)
|
||
|
||
try:
|
||
with urllib.request.urlopen(request, timeout=30) as response:
|
||
status = response.getcode()
|
||
body = response.read().decode("utf-8", errors="replace").strip()
|
||
except urllib.error.HTTPError as error:
|
||
status = error.code
|
||
body = error.read().decode("utf-8", errors="replace").strip()
|
||
except urllib.error.URLError as error:
|
||
fail(f"Ошибка сети при обращении к dfgit: {error}")
|
||
|
||
if status in (200, 201, 204):
|
||
print("✓ Workflow успешно запущен")
|
||
print("Проверьте статус в Actions на сервере dfgit.")
|
||
return
|
||
|
||
print(f"✗ Ошибка запуска workflow (HTTP {status})")
|
||
if body:
|
||
print("Ответ API:")
|
||
print(body)
|
||
sys.exit(1)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|