update low and add jenkins api

backend/jenkins_client.py

@@ -0,0 +1,109 @@
+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"),
+    )