103 lines
3.6 KiB
Python
103 lines
3.6 KiB
Python
import asyncio
|
|
import json
|
|
import logging
|
|
from typing import Any
|
|
|
|
import httpx
|
|
|
|
from app import config
|
|
from app.models import Settings, Review
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class GPTClient:
|
|
def __init__(self):
|
|
self.api_url = config.GPT_API_URL
|
|
self.headers = {
|
|
"Authorization": f"Bearer {config.GPTUNNEL_API_KEY}",
|
|
"Content-Type": "application/json",
|
|
}
|
|
|
|
async def _make_request(
|
|
self, messages: list, model: str = "gpt-4o", max_tokens: int = 100, retry: bool = False
|
|
) -> str:
|
|
data = {
|
|
"model": model,
|
|
"max_tokens": max_tokens,
|
|
"messages": messages,
|
|
}
|
|
|
|
try:
|
|
async with httpx.AsyncClient() as client:
|
|
response = await client.post(
|
|
self.api_url, headers=self.headers, json=data
|
|
)
|
|
response.raise_for_status()
|
|
result = response.json()
|
|
logger.info("GPT response: %s", result)
|
|
answer = (
|
|
result.get("choices", [{}])[0].get("message", {}).get("content", "")
|
|
)
|
|
return answer
|
|
except httpx.HTTPError as e:
|
|
logger.error(f"Error making GPT request: {e.__class__.__name__}: {e}")
|
|
if retry:
|
|
return ""
|
|
logger.info("Waiting 15 seconds and retrying")
|
|
await asyncio.sleep(15)
|
|
return await self._make_request(messages, model, max_tokens, retry=True)
|
|
|
|
async def analyze_review(self, settings: Settings, review: Review) -> str:
|
|
review_text = (
|
|
f"Комментарий: {review.text}\n"
|
|
f"Название товара: {review.product_name or '-'}\n"
|
|
f"Оценка пользователя: {review.rating}"
|
|
)
|
|
messages = [
|
|
{"role": "system", "content": settings.analysis_prompt},
|
|
{
|
|
"role": "user",
|
|
"content": review_text,
|
|
},
|
|
]
|
|
|
|
return await self._make_request(messages, model=settings.gpt_model, max_tokens=settings.max_tokens)
|
|
|
|
async def generate_answer(self, settings: Settings, review: Review) -> str:
|
|
review_text = (
|
|
f"Комментарий: {review.text}\n"
|
|
f"Название товара: {review.product_name or '-'}\n"
|
|
f"Оценка пользователя: {review.rating}"
|
|
)
|
|
messages = [
|
|
{"role": "system", "content": settings.answer_prompt},
|
|
{
|
|
"role": "user",
|
|
"content": f"{review_text}\nОтвет представителя бренда:",
|
|
},
|
|
]
|
|
|
|
return await self._make_request(
|
|
messages, model=settings.gpt_model, max_tokens=settings.max_tokens
|
|
)
|
|
|
|
@staticmethod
|
|
def extract_analysis_data(analysis_text: str) -> dict[str, Any]:
|
|
try:
|
|
analysis_text = analysis_text.replace("```json", "").replace("```", "")
|
|
data = json.loads(analysis_text)
|
|
|
|
tone = (data.get("tone") or "").lower()
|
|
tone_reason = data.get("reason") or ""
|
|
is_critical = data.get("is_critical") or None
|
|
except (json.JSONDecodeError, ValueError, AttributeError) as e:
|
|
logger.error(f"Error parsing analysis JSON: {e}. Raw text: {analysis_text}")
|
|
raise e
|
|
|
|
return {
|
|
"tone": tone,
|
|
"tone_reason": tone_reason,
|
|
"is_critical": is_critical,
|
|
}
|