Spaces:
Sleeping
Sleeping
| """Sentiment API methods for ChainShiftClient. | |
| Mixin class extracted from api_client.py. | |
| Methods access self._get, self._post, self._session, self.base_url, self.headers from the parent class. | |
| """ | |
| from typing import Any | |
| class SentimentApiMixin: | |
| """Gen3 Nudge, Brand Mentions, LLM Verification, Feedback, Keyword APIs.""" | |
| # ======================================================================== | |
| # Gen3 APIs - Nudge Detection | |
| # ======================================================================== | |
| def get_nudge_candidates( | |
| self, | |
| campaign_id: int, | |
| page: int = 1, | |
| page_size: int = 50, | |
| bit_quadrant: str | None = None, | |
| platform: str | None = None, | |
| ) -> dict: | |
| """Get answers flagged for nudge (in-house brand negative mentions). | |
| Returns: | |
| - total_nudge_candidates: int | |
| - by_confidence_tier: {HIGH: n, MEDIUM: n, LOW: n} | |
| - by_platform: {CHATGPT: n, GOOGLE_AI: n, ...} | |
| - by_cej: {AWARENESS_COMPARISON: n, PURCHASE: n, ...} | |
| - by_bit_quadrant: {neutral: n, ...} | |
| - candidates: list of nudge items | |
| """ | |
| params = {"page": page, "page_size": page_size} | |
| if bit_quadrant: | |
| params["bit_quadrant"] = bit_quadrant | |
| if platform: | |
| params["platform"] = platform | |
| return self._get(f"/api/v1/sentiment/campaigns/{campaign_id}/nudge-candidates", params) | |
| def get_brand_mentions( | |
| self, | |
| campaign_id: int, | |
| page: int = 1, | |
| page_size: int = 50, | |
| brand_type: str | None = None, | |
| polarity: str | None = None, | |
| llm_verified: str | None = None, | |
| brand_name: str | None = None, | |
| sort_by: str | None = "analyzed_at", | |
| sort_desc: bool = True, | |
| ) -> dict: | |
| """Get brand mention analysis.""" | |
| params = {"page": page, "page_size": page_size, "sort_desc": sort_desc} | |
| if brand_type: | |
| params["brand_type"] = brand_type | |
| if polarity: | |
| params["polarity"] = polarity | |
| if llm_verified: | |
| params["llm_verified"] = llm_verified | |
| if brand_name: | |
| params["brand_name"] = brand_name | |
| if sort_by: | |
| params["sort_by"] = sort_by | |
| return self._get(f"/api/v1/sentiment/campaigns/{campaign_id}/brand-mentions", params) | |
| def export_brand_mentions( | |
| self, | |
| campaign_id: int, | |
| brand_type: str | None = None, | |
| polarity: str | None = None, | |
| llm_verified: str | None = None, | |
| brand_name: str | None = None, | |
| include_full_answers: bool = False, | |
| ) -> bytes: | |
| """Export brand mentions to Excel.""" | |
| params = {} | |
| if brand_type: | |
| params["brand_type"] = brand_type | |
| if polarity: | |
| params["polarity"] = polarity | |
| if llm_verified: | |
| params["llm_verified"] = llm_verified | |
| if brand_name: | |
| params["brand_name"] = brand_name | |
| if include_full_answers: | |
| params["include_full_answers"] = "true" | |
| url = f"{self.base_url}/api/v1/sentiment/campaigns/{campaign_id}/brand-mentions/export" | |
| response = self._session.get(url, params=params, headers=self.headers, timeout=120) | |
| response.raise_for_status() | |
| return response.content | |
| # ======================================================================== | |
| # LLM Verification & Human Feedback APIs | |
| # ======================================================================== | |
| def verify_answer(self, answer_id: int, force: bool = False) -> dict: | |
| """Request LLM 2차 검증 for an answer.""" | |
| return self._post("/api/v1/sentiment/verify", { | |
| "answer_id": answer_id, | |
| "force": force, | |
| }) | |
| def submit_feedback( | |
| self, | |
| answer_id: int, | |
| campaign_id: int, | |
| feedback_type: str, | |
| corrected_polarity: str | None = None, | |
| wrong_reason: str | None = None, | |
| comment: str | None = None, | |
| ) -> dict: | |
| """Submit human feedback for an answer sentiment.""" | |
| data = { | |
| "answer_id": answer_id, | |
| "campaign_id": campaign_id, | |
| "feedback_type": feedback_type, | |
| } | |
| if corrected_polarity: | |
| data["corrected_polarity"] = corrected_polarity | |
| if wrong_reason: | |
| data["wrong_reason"] = wrong_reason | |
| if comment: | |
| data["comment"] = comment | |
| return self._post("/api/v1/sentiment/feedback", data) | |
| def get_feedback_stats(self, campaign_id: int) -> dict: | |
| """Get feedback statistics for a campaign.""" | |
| return self._get(f"/api/v1/sentiment/campaigns/{campaign_id}/feedback/stats") | |
| def get_full_answers(self, answer_ids: list[int]) -> dict: | |
| """Get full answer text for multiple answers.""" | |
| return self._post("/api/v1/sentiment/answers/full", {"answer_ids": answer_ids}) | |
| def export_nudge_candidates( | |
| self, | |
| campaign_id: int, | |
| include_full_answers: bool = False, | |
| include_evidence: bool = True, | |
| llm_verified_only: bool = False, | |
| platform: str | None = None, | |
| llm_is_negative: bool | None = None, | |
| ) -> bytes: | |
| """Export nudge candidates to Excel.""" | |
| url = f"{self.base_url}/api/v1/sentiment/campaigns/{campaign_id}/nudge-candidates/export" | |
| params = { | |
| "include_full_answers": str(include_full_answers).lower(), | |
| "include_evidence": str(include_evidence).lower(), | |
| "llm_verified_only": str(llm_verified_only).lower(), | |
| } | |
| if platform: | |
| params["platform"] = platform | |
| if llm_is_negative is not None: | |
| params["llm_is_negative"] = str(llm_is_negative).lower() | |
| response = self._session.get(url, headers=self.headers, params=params, timeout=120) | |
| response.raise_for_status() | |
| return response.content | |
| # ======================================================================== | |
| # Keyword Sentiment APIs | |
| # ======================================================================== | |
| def start_keyword_analysis(self, campaign_id: int, keywords: list[str]) -> dict: | |
| """Start a keyword sentiment analysis job.""" | |
| return self._post( | |
| f"/api/v1/sentiment/campaigns/{campaign_id}/keyword-analysis", | |
| {"keywords": keywords}, | |
| ) | |
| def get_keyword_summary( | |
| self, | |
| campaign_id: int, | |
| keyword: str | None = None, | |
| page: int = 1, | |
| page_size: int = 50, | |
| ) -> dict: | |
| """Get keyword sentiment summary with optional filter and pagination.""" | |
| params: dict[str, Any] = {"page": page, "page_size": page_size} | |
| if keyword: | |
| params["keyword"] = keyword | |
| return self._get(f"/api/v1/sentiment/campaigns/{campaign_id}/keyword-summary", params) | |
| def get_keyword_results( | |
| self, | |
| campaign_id: int, | |
| keyword: str | None = None, | |
| sentiment: str | None = None, | |
| brand_name: str | None = None, | |
| competitor: bool = False, | |
| llm_verified: str | None = None, | |
| llm_is_negative: bool | None = None, | |
| platform: str | None = None, | |
| brand_only: bool = False, | |
| no_brand: bool = False, | |
| page: int = 1, | |
| page_size: int = 50, | |
| ) -> dict: | |
| """Get keyword results with filters.""" | |
| params: dict[str, Any] = {"page": page, "page_size": page_size} | |
| if keyword: | |
| params["keyword"] = keyword | |
| if sentiment: | |
| params["sentiment"] = sentiment | |
| if brand_name: | |
| params["brand_name"] = brand_name | |
| if competitor: | |
| params["competitor"] = "true" | |
| if llm_verified: | |
| params["llm_verified"] = llm_verified | |
| if llm_is_negative is not None: | |
| params["llm_is_negative"] = str(llm_is_negative).lower() | |
| if platform: | |
| params["platform"] = platform | |
| if brand_only: | |
| params["brand_only"] = "true" | |
| if no_brand: | |
| params["no_brand"] = "true" | |
| return self._get( | |
| f"/api/v1/sentiment/campaigns/{campaign_id}/keyword-results", params | |
| ) | |
| def get_keyword_brand_analysis( | |
| self, | |
| campaign_id: int, | |
| keyword: str | None = None, | |
| brand_name: str | None = None, | |
| competitor: bool = False, | |
| ) -> dict: | |
| """Get keyword x brand cross analysis.""" | |
| params: dict[str, Any] = {} | |
| if keyword: | |
| params["keyword"] = keyword | |
| if brand_name: | |
| params["brand_name"] = brand_name | |
| if competitor: | |
| params["competitor"] = "true" | |
| return self._get( | |
| f"/api/v1/sentiment/campaigns/{campaign_id}/keyword-brand-analysis", params | |
| ) | |
| def export_keyword_results( | |
| self, | |
| campaign_id: int, | |
| keyword: str | None = None, | |
| sentiment: str | None = None, | |
| brand_name: str | None = None, | |
| competitor: bool = False, | |
| llm_verified: str | None = None, | |
| llm_is_negative: bool | None = None, | |
| platform: str | None = None, | |
| brand_only: bool = False, | |
| no_brand: bool = False, | |
| format: str = "xlsx", | |
| ) -> bytes: | |
| """Export keyword results to Excel or CSV with same filters as list endpoint.""" | |
| params: dict[str, Any] = {"format": format} | |
| if keyword: | |
| params["keyword"] = keyword | |
| if sentiment: | |
| params["sentiment"] = sentiment | |
| if brand_name: | |
| params["brand_name"] = brand_name | |
| if competitor: | |
| params["competitor"] = "true" | |
| if llm_verified: | |
| params["llm_verified"] = llm_verified | |
| if llm_is_negative is not None: | |
| params["llm_is_negative"] = str(llm_is_negative).lower() | |
| if platform: | |
| params["platform"] = platform | |
| if brand_only: | |
| params["brand_only"] = "true" | |
| if no_brand: | |
| params["no_brand"] = "true" | |
| url = f"{self.base_url}/api/v1/sentiment/campaigns/{campaign_id}/keyword-results/export" | |
| response = self._session.get(url, params=params, headers=self.headers, timeout=120) | |
| response.raise_for_status() | |
| return response.content | |