Spaces:
Sleeping
Sleeping
| """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") | |