Spaces:
Sleeping
Sleeping
File size: 3,922 Bytes
ef78361 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 | """Hierarchy & Analysis Job API methods for ChainShiftClient.
Mixin class extracted from api_client.py.
Methods access self._get, self._post from the parent class.
"""
from typing import Any
class HierarchyApiMixin:
"""Hierarchy Analysis and Analysis Job APIs."""
# ========================================================================
# Analysis Job APIs
# ========================================================================
def start_analysis_job(self, campaign_id: int, options: dict | None = None) -> dict:
"""Start a sentiment analysis job. POST /api/v1/sentiment/campaigns/{campaign_id}/run"""
body: dict = {}
if options:
body["options"] = options
return self._post(
f"/api/v1/sentiment/campaigns/{campaign_id}/run",
body,
)
def list_analysis_jobs(
self,
campaign_id: int | None = None,
status: str | None = None,
page: int = 1,
page_size: int = 20,
) -> dict:
"""List analysis jobs. GET /api/v1/sentiment/jobs"""
params: dict[str, Any] = {"page": page, "page_size": page_size}
if campaign_id is not None:
params["campaign_id"] = campaign_id
if status:
params["status"] = status
return self._get("/api/v1/sentiment/jobs", params)
def get_analysis_job(self, job_id: str) -> dict:
"""Get analysis job details. GET /api/v1/sentiment/jobs/{job_id}"""
return self._get(f"/api/v1/sentiment/jobs/{job_id}")
def cancel_analysis_job(self, job_id: str) -> dict:
"""Cancel a queued or running analysis job. POST /api/v1/sentiment/jobs/{job_id}/cancel"""
return self._post(f"/api/v1/sentiment/jobs/{job_id}/cancel")
# ========================================================================
# Hierarchy Analysis APIs
# ========================================================================
def create_hierarchy_job(
self,
prompt: str,
title: str | None = None,
processor_config: dict | None = None,
settings: dict | None = None,
) -> dict:
"""Create a hierarchy analysis job. POST /api/v1/hierarchy/jobs"""
body: dict = {"prompt": prompt, "processor_config": processor_config or {}}
if title:
body["title"] = title
if settings:
body["settings"] = settings
return self._post("/api/v1/hierarchy/jobs", body)
def list_hierarchy_jobs(
self,
page: int = 1,
page_size: int = 20,
status: str | None = None,
) -> dict:
"""List hierarchy analysis jobs. GET /api/v1/hierarchy/jobs"""
params: dict[str, Any] = {"page": page, "page_size": page_size}
if status:
params["status"] = status
return self._get("/api/v1/hierarchy/jobs", params)
def get_hierarchy_job(self, job_id: str) -> dict:
"""Get hierarchy job details. GET /api/v1/hierarchy/jobs/{job_id}"""
return self._get(f"/api/v1/hierarchy/jobs/{job_id}")
def get_hierarchy_questions(
self,
job_id: str,
page: int = 1,
page_size: int = 50,
journey_depth1: str | None = None,
export: str | None = None,
) -> dict:
"""Get questions for a hierarchy job. GET /api/v1/hierarchy/jobs/{job_id}/questions"""
params: dict[str, Any] = {"page": page, "page_size": page_size}
if journey_depth1:
params["journey_depth1"] = journey_depth1
if export:
params["export"] = export
return self._get(f"/api/v1/hierarchy/jobs/{job_id}/questions", params)
def get_hierarchy_question_stats(self, job_id: str) -> dict:
"""Get question statistics. GET /api/v1/hierarchy/jobs/{job_id}/questions/stats"""
return self._get(f"/api/v1/hierarchy/jobs/{job_id}/questions/stats")
|