239 lines
7.9 KiB
Python
239 lines
7.9 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
import re
|
|
from typing import Any
|
|
|
|
import httpx
|
|
|
|
from .config import Settings
|
|
from .models import RawItem
|
|
|
|
|
|
DEFAULT_ANALYSIS = {
|
|
"sentiment": "neutral",
|
|
"is_positive": False,
|
|
"is_negative": False,
|
|
"has_actionable_feedback": False,
|
|
"feedback_types": [],
|
|
"reply_recommended": False,
|
|
"reply_priority": "none",
|
|
"reply_suggestion": "",
|
|
"summary": "",
|
|
"priority": "low",
|
|
"confidence": 0.0,
|
|
"reason": "",
|
|
}
|
|
|
|
|
|
TRANSLATION_SCHEMA = {
|
|
"type": "object",
|
|
"properties": {
|
|
"translated_content": {"type": "string"},
|
|
},
|
|
"required": ["translated_content"],
|
|
"additionalProperties": False,
|
|
}
|
|
|
|
|
|
SCHEMA = {
|
|
"type": "object",
|
|
"properties": {
|
|
"sentiment": {"type": "string", "enum": ["positive", "negative", "mixed", "neutral"]},
|
|
"is_positive": {"type": "boolean"},
|
|
"is_negative": {"type": "boolean"},
|
|
"has_actionable_feedback": {"type": "boolean"},
|
|
"feedback_types": {
|
|
"type": "array",
|
|
"items": {
|
|
"type": "string",
|
|
"enum": [
|
|
"bug",
|
|
"suggestion",
|
|
"balance",
|
|
"ui",
|
|
"localization",
|
|
"performance",
|
|
"pricing",
|
|
"content",
|
|
"question",
|
|
"other",
|
|
],
|
|
},
|
|
},
|
|
"reply_recommended": {"type": "boolean"},
|
|
"reply_priority": {"type": "string", "enum": ["none", "low", "medium", "high"]},
|
|
"reply_suggestion": {"type": "string"},
|
|
"summary": {"type": "string"},
|
|
"priority": {"type": "string", "enum": ["low", "medium", "high"]},
|
|
"confidence": {"type": "number", "minimum": 0, "maximum": 1},
|
|
"reason": {"type": "string"},
|
|
},
|
|
"required": [
|
|
"sentiment",
|
|
"is_positive",
|
|
"is_negative",
|
|
"has_actionable_feedback",
|
|
"feedback_types",
|
|
"reply_recommended",
|
|
"reply_priority",
|
|
"reply_suggestion",
|
|
"summary",
|
|
"priority",
|
|
"confidence",
|
|
"reason",
|
|
],
|
|
"additionalProperties": False,
|
|
}
|
|
|
|
|
|
class OpenRouterClient:
|
|
def __init__(self, settings: Settings) -> None:
|
|
self.settings = settings
|
|
self.enabled = bool(settings.openrouter_api_key)
|
|
self.client = httpx.Client(timeout=60)
|
|
|
|
def close(self) -> None:
|
|
self.client.close()
|
|
|
|
def analyze(self, item: RawItem) -> dict[str, Any]:
|
|
if not self.enabled:
|
|
raise MissingOpenRouterKey("OPENROUTER_API_KEY is not configured")
|
|
|
|
payload = {
|
|
"model": self.settings.openrouter_model,
|
|
"messages": [
|
|
{
|
|
"role": "system",
|
|
"content": (
|
|
"你是独立游戏《帝国幻想乡~TOHOTOPIA》的社区运营助手。"
|
|
"请判断 Steam、Twitter/X 等社区内容的情绪、是否包含具体可处理反馈、"
|
|
"以及是否建议制作人回复。summary、reason、reply_suggestion 必须使用中文。"
|
|
"只输出符合 JSON Schema 的 JSON。"
|
|
),
|
|
},
|
|
{
|
|
"role": "user",
|
|
"content": self._prompt(item),
|
|
},
|
|
],
|
|
"temperature": 0.1,
|
|
"response_format": {
|
|
"type": "json_schema",
|
|
"json_schema": {
|
|
"name": "community_item_analysis",
|
|
"strict": True,
|
|
"schema": SCHEMA,
|
|
},
|
|
},
|
|
}
|
|
headers = {
|
|
"Authorization": f"Bearer {self.settings.openrouter_api_key}",
|
|
"HTTP-Referer": self.settings.openrouter_referer,
|
|
"X-Title": self.settings.openrouter_title,
|
|
}
|
|
response = self.client.post(
|
|
"https://openrouter.ai/api/v1/chat/completions",
|
|
headers=headers,
|
|
json=payload,
|
|
)
|
|
response.raise_for_status()
|
|
data = response.json()
|
|
content = data["choices"][0]["message"]["content"]
|
|
parsed = self._parse_json(content)
|
|
return self._normalize(parsed)
|
|
|
|
def translate_to_chinese(self, content: str) -> str:
|
|
if not self.enabled:
|
|
raise MissingOpenRouterKey("OPENROUTER_API_KEY is not configured")
|
|
|
|
payload = {
|
|
"model": self.settings.openrouter_model,
|
|
"messages": [
|
|
{
|
|
"role": "system",
|
|
"content": (
|
|
"你是独立游戏社区运营翻译助手。"
|
|
"把用户提供的社区内容准确翻译成简体中文，保留原意、语气、问题细节、游戏术语、链接和编号。"
|
|
"不要添加解释。只输出符合 JSON Schema 的 JSON。"
|
|
),
|
|
},
|
|
{
|
|
"role": "user",
|
|
"content": content[:6000],
|
|
},
|
|
],
|
|
"temperature": 0,
|
|
"response_format": {
|
|
"type": "json_schema",
|
|
"json_schema": {
|
|
"name": "manual_item_translation",
|
|
"strict": True,
|
|
"schema": TRANSLATION_SCHEMA,
|
|
},
|
|
},
|
|
}
|
|
headers = {
|
|
"Authorization": f"Bearer {self.settings.openrouter_api_key}",
|
|
"HTTP-Referer": self.settings.openrouter_referer,
|
|
"X-Title": self.settings.openrouter_title,
|
|
}
|
|
response = self.client.post(
|
|
"https://openrouter.ai/api/v1/chat/completions",
|
|
headers=headers,
|
|
json=payload,
|
|
)
|
|
response.raise_for_status()
|
|
data = response.json()
|
|
parsed = self._parse_json(data["choices"][0]["message"]["content"])
|
|
translated = str(parsed.get("translated_content") or "").strip()
|
|
return translated or content
|
|
|
|
def _prompt(self, item: RawItem) -> str:
|
|
metadata = {
|
|
"source": item.source,
|
|
"content_type": item.content_type,
|
|
"source_url": item.source_url,
|
|
"author": item.author_name,
|
|
"title": item.title,
|
|
"steam_review_voted_up": item.raw.get("voted_up"),
|
|
"language": item.raw.get("language"),
|
|
"in_reply_to": item.raw.get("parent_url") or item.raw.get("in_reply_to"),
|
|
"likes": item.raw.get("likes"),
|
|
"replies": item.raw.get("replies"),
|
|
"retweets": item.raw.get("retweets"),
|
|
"views": item.raw.get("views"),
|
|
}
|
|
return (
|
|
"请分析以下社区内容。\n\n"
|
|
f"元数据：{json.dumps(metadata, ensure_ascii=False)}\n\n"
|
|
f"正文：\n{item.content[:6000]}"
|
|
)
|
|
|
|
def _parse_json(self, content: str) -> dict[str, Any]:
|
|
try:
|
|
return json.loads(content)
|
|
except json.JSONDecodeError:
|
|
match = re.search(r"\{.*\}", content, re.S)
|
|
if not match:
|
|
raise
|
|
return json.loads(match.group(0))
|
|
|
|
def _normalize(self, value: dict[str, Any]) -> dict[str, Any]:
|
|
result = dict(DEFAULT_ANALYSIS)
|
|
result.update(value)
|
|
result["feedback_types"] = list(result.get("feedback_types") or [])
|
|
result["is_positive"] = bool(result.get("is_positive"))
|
|
result["is_negative"] = bool(result.get("is_negative"))
|
|
result["has_actionable_feedback"] = bool(result.get("has_actionable_feedback"))
|
|
result["reply_recommended"] = bool(result.get("reply_recommended"))
|
|
try:
|
|
result["confidence"] = float(result.get("confidence", 0.0))
|
|
except (TypeError, ValueError):
|
|
result["confidence"] = 0.0
|
|
return result
|
|
|
|
|
|
class MissingOpenRouterKey(RuntimeError):
|
|
pass
|