110 lines
3.2 KiB
Python
110 lines
3.2 KiB
Python
from __future__ import annotations
|
||
|
||
import os
|
||
from typing import Any
|
||
from urllib.parse import quote
|
||
|
||
import requests
|
||
|
||
|
||
def _env(name: str, default: str) -> str:
|
||
value = os.getenv(name)
|
||
return value if value else default
|
||
|
||
|
||
def _normalize_base(url: str) -> str:
|
||
u = (url or "").strip().rstrip("/")
|
||
if not u:
|
||
return ""
|
||
if not u.startswith(("http://", "https://")):
|
||
u = f"https://{u}"
|
||
return u
|
||
|
||
|
||
def _job_url(base: str, job_path: str) -> str:
|
||
"""https://jenkins/job/a/job/b/job/c"""
|
||
parts = [p for p in (job_path or "").strip("/").split("/") if p]
|
||
if not parts:
|
||
raise ValueError("job path is empty")
|
||
segs = "/".join(f"job/{quote(p, safe='')}" for p in parts)
|
||
return f"{_normalize_base(base)}/{segs}"
|
||
|
||
|
||
def trigger_jenkins_job(
|
||
*,
|
||
server_url: str,
|
||
job_path: str,
|
||
parameters: dict[str, str],
|
||
username: str | None,
|
||
token: str | None,
|
||
verify_ssl: bool = True,
|
||
) -> tuple[bool, int, str, dict[str, Any]]:
|
||
"""
|
||
Запускает сборку через Jenkins REST API: /build или /buildWithParameters.
|
||
При включённом CSRF получает crumb с /crumbIssuer/api/json.
|
||
"""
|
||
base = _normalize_base(server_url)
|
||
if not base:
|
||
return False, 0, "server_url is empty", {}
|
||
|
||
job_url = _job_url(base, job_path)
|
||
auth = (username, token) if username and token else None
|
||
|
||
headers: dict[str, str] = {}
|
||
if auth:
|
||
try:
|
||
cr = requests.get(
|
||
f"{base}/crumbIssuer/api/json",
|
||
auth=auth,
|
||
timeout=30,
|
||
verify=verify_ssl,
|
||
)
|
||
if cr.ok:
|
||
cj = cr.json()
|
||
field = cj.get("crumbRequestField", "Jenkins-Crumb")
|
||
headers[field] = cj["crumb"]
|
||
except Exception:
|
||
pass
|
||
|
||
if parameters:
|
||
target = f"{job_url}/buildWithParameters"
|
||
resp = requests.post(
|
||
target,
|
||
auth=auth,
|
||
headers=headers,
|
||
data=parameters,
|
||
timeout=120,
|
||
verify=verify_ssl,
|
||
)
|
||
else:
|
||
target = f"{job_url}/build"
|
||
resp = requests.post(
|
||
target,
|
||
auth=auth,
|
||
headers=headers,
|
||
timeout=120,
|
||
verify=verify_ssl,
|
||
)
|
||
|
||
detail: dict[str, Any] = {
|
||
"http_status": resp.status_code,
|
||
"location": resp.headers.get("Location"),
|
||
"jenkins_body_preview": (resp.text or "")[:500],
|
||
}
|
||
ok = 200 <= resp.status_code < 300
|
||
if resp.status_code == 403 and "crumb" in (resp.text or "").lower():
|
||
return False, resp.status_code, "Jenkins вернул 403 — проверьте crumb/права или отключите CSRF для API.", detail
|
||
if not ok:
|
||
return False, resp.status_code, f"Jenkins HTTP {resp.status_code}", detail
|
||
return True, resp.status_code, "queued", detail
|
||
|
||
|
||
def default_jenkins_credentials() -> tuple[str, str, str, bool]:
|
||
"""default_url, user, token, verify_ssl"""
|
||
return (
|
||
_env("JENKINS_DEFAULT_URL", ""),
|
||
_env("JENKINS_USER", ""),
|
||
_env("JENKINS_TOKEN", ""),
|
||
_env("JENKINS_VERIFY_SSL", "true").lower() in ("1", "true", "yes", "on"),
|
||
)
|