mirror of
https://github.com/hansjone/netx.git
synced 2026-10-09 04:20:45 +08:00
feat: initialize netx ops tool repository
Set up netx as a standalone git-managed project with isolated Python dependencies, alarm ingestion/query APIs, oclaw bridge integration, and ops-focused UI enhancements including history and raw field access. Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
commit
f49c0b5bad
56 changed files with 6974 additions and 0 deletions
50
netx_api/ap_client.py
Normal file
50
netx_api/ap_client.py
Normal file
|
|
@ -0,0 +1,50 @@
|
|||
from __future__ import annotations
|
||||
|
||||
from typing import Any
|
||||
|
||||
import httpx
|
||||
|
||||
from .config import settings
|
||||
|
||||
|
||||
def _analyze_httpx_timeout() -> httpx.Timeout:
|
||||
read_s = max(30.0, float(settings.oclaw_analyze_read_timeout_sec))
|
||||
connect_s = max(3.0, float(settings.oclaw_connect_timeout_sec))
|
||||
return httpx.Timeout(
|
||||
connect=connect_s,
|
||||
read=read_s,
|
||||
write=min(120.0, read_s),
|
||||
pool=connect_s,
|
||||
)
|
||||
|
||||
|
||||
def analyze_with_oclaw(payload: dict[str, Any]) -> dict[str, Any]:
|
||||
headers = {"content-type": "application/json", "accept": "application/json"}
|
||||
token = str(settings.oclaw_analyze_token or "").strip()
|
||||
if token:
|
||||
headers["authorization"] = f"Bearer {token}"
|
||||
with httpx.Client(timeout=_analyze_httpx_timeout()) as client:
|
||||
resp = client.post(settings.oclaw_analyze_url, json=payload, headers=headers)
|
||||
text = resp.text
|
||||
if not resp.is_success:
|
||||
raise RuntimeError(f"oclaw analyze failed: {resp.status_code} {text[:300]}")
|
||||
if not text:
|
||||
return {}
|
||||
return resp.json()
|
||||
|
||||
|
||||
def health_with_oclaw() -> dict[str, Any]:
|
||||
headers = {"accept": "application/json"}
|
||||
token = str(settings.oclaw_analyze_token or "").strip()
|
||||
if token:
|
||||
headers["authorization"] = f"Bearer {token}"
|
||||
health_s = max(3.0, float(settings.oclaw_health_timeout_sec))
|
||||
with httpx.Client(timeout=health_s) as client:
|
||||
resp = client.get(settings.oclaw_health_url, headers=headers)
|
||||
status = int(resp.status_code)
|
||||
text = resp.text
|
||||
if not resp.is_success:
|
||||
raise RuntimeError(f"oclaw health failed: {status} {text[:200]}")
|
||||
if not text:
|
||||
return {"status_code": status, "data": {}}
|
||||
return {"status_code": status, "data": resp.json()}
|
||||
Loading…
Add table
Add a link
Reference in a new issue