john
05e173b293
Replace the WebView shell with SwiftUI screens, add account-scoped Wiki and TTS APIs with adaptive review and photo scan support, and keep web/iOS pages usable while data loads asynchronously. Co-authored-by: Cursor <cursoragent@cursor.com>
144 lines
5.4 KiB
Python
144 lines
5.4 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
import urllib.error
|
|
import urllib.request
|
|
|
|
from config import get_settings
|
|
|
|
|
|
class LlmServiceError(Exception):
|
|
pass
|
|
|
|
|
|
class LlmService:
|
|
def is_configured(self) -> bool:
|
|
return bool(get_settings().deepseek_api_key.strip())
|
|
|
|
def model_name(self) -> str:
|
|
return get_settings().deepseek_model
|
|
|
|
def _chat_completion(
|
|
self,
|
|
*,
|
|
messages: list[dict],
|
|
temperature: float = 0.2,
|
|
max_tokens: int = 1800,
|
|
response_format: dict | None = None,
|
|
timeout: int = 15,
|
|
) -> str:
|
|
settings = get_settings()
|
|
api_key = settings.deepseek_api_key.strip()
|
|
if not api_key:
|
|
raise LlmServiceError("DeepSeek API key is not configured")
|
|
|
|
payload: dict = {
|
|
"model": settings.deepseek_model,
|
|
"messages": messages,
|
|
"temperature": temperature,
|
|
"max_tokens": max_tokens,
|
|
}
|
|
if response_format:
|
|
payload["response_format"] = response_format
|
|
|
|
request = urllib.request.Request(
|
|
f"{settings.deepseek_base_url.rstrip('/')}/v1/chat/completions",
|
|
data=json.dumps(payload).encode("utf-8"),
|
|
headers={
|
|
"Authorization": f"Bearer {api_key}",
|
|
"Content-Type": "application/json",
|
|
},
|
|
method="POST",
|
|
)
|
|
try:
|
|
with urllib.request.urlopen(request, timeout=timeout) as response:
|
|
body = json.loads(response.read().decode("utf-8"))
|
|
except urllib.error.HTTPError as exc:
|
|
detail = exc.read().decode("utf-8", errors="replace")
|
|
raise LlmServiceError(f"DeepSeek HTTP {exc.code}: {detail}") from exc
|
|
except urllib.error.URLError as exc:
|
|
raise LlmServiceError(f"DeepSeek request failed: {exc.reason}") from exc
|
|
|
|
try:
|
|
content = body["choices"][0]["message"]["content"]
|
|
except (KeyError, IndexError, TypeError) as exc:
|
|
raise LlmServiceError("DeepSeek response format is invalid") from exc
|
|
return str(content).strip()
|
|
|
|
@staticmethod
|
|
def _strip_code_fence(text: str) -> str:
|
|
cleaned = text.strip()
|
|
if not cleaned.startswith("```"):
|
|
return cleaned
|
|
cleaned = cleaned.strip("`").strip()
|
|
if cleaned.lower().startswith("json"):
|
|
cleaned = cleaned[4:].lstrip()
|
|
elif cleaned.lower().startswith("markdown"):
|
|
cleaned = cleaned[8:].lstrip()
|
|
return cleaned.strip()
|
|
|
|
def enhance_wiki_page(self, *, title: str, draft_markdown: str) -> str:
|
|
prompt = (
|
|
"你是 WordLoop 个人学习 Wiki 的编译器，遵循 Karpathy Wiki 思路："
|
|
"把原始学习事件整理成简洁、可检索的 Markdown 知识页。\n"
|
|
"要求：\n"
|
|
"1. 保留草稿中的事实数据（释义、掌握度、练习记录等），不要编造。\n"
|
|
"2. 优化结构、摘要与关联说明，便于日后复习。\n"
|
|
"3. 只输出 Markdown 正文，不要代码块包裹，不要额外解释。\n"
|
|
f"\n页面标题：{title}\n\n草稿：\n{draft_markdown}"
|
|
)
|
|
content = self._chat_completion(
|
|
messages=[
|
|
{"role": "system", "content": "你是严谨的学习笔记整理助手。"},
|
|
{"role": "user", "content": prompt},
|
|
],
|
|
temperature=0.2,
|
|
max_tokens=1800,
|
|
)
|
|
cleaned = self._strip_code_fence(content)
|
|
return cleaned or draft_markdown
|
|
|
|
def parse_word_pairs_from_text(self, text: str) -> list[dict[str, str]]:
|
|
prompt = (
|
|
"你是英语学习词表整理助手。从 OCR 识别文本中提取英汉单词对。\n"
|
|
"要求：\n"
|
|
"1. 只提取明确的单词/短语与其中文释义，不要编造。\n"
|
|
"2. 忽略页码、标题、序号、噪声行。\n"
|
|
"3. 输出 JSON 对象，格式为 {\"items\":[{\"en\":\"...\",\"zh\":\"...\"}]}。\n"
|
|
"4. en 为英文，zh 为中文；若原文是中文在前英文在后，也要正确归位。\n"
|
|
"5. 只输出 JSON，不要解释。\n"
|
|
f"\nOCR 文本：\n{text.strip()}"
|
|
)
|
|
content = self._chat_completion(
|
|
messages=[
|
|
{"role": "system", "content": "你只输出合法 JSON。"},
|
|
{"role": "user", "content": prompt},
|
|
],
|
|
temperature=0.1,
|
|
max_tokens=4000,
|
|
response_format={"type": "json_object"},
|
|
timeout=30,
|
|
)
|
|
cleaned = self._strip_code_fence(content)
|
|
try:
|
|
payload = json.loads(cleaned)
|
|
except json.JSONDecodeError as exc:
|
|
raise LlmServiceError("词表解析结果不是合法 JSON") from exc
|
|
|
|
raw_items = payload.get("items") if isinstance(payload, dict) else payload
|
|
if not isinstance(raw_items, list):
|
|
raise LlmServiceError("词表解析结果缺少 items 数组")
|
|
|
|
pairs: list[dict[str, str]] = []
|
|
for item in raw_items:
|
|
if not isinstance(item, dict):
|
|
continue
|
|
en = str(item.get("en") or item.get("source_text") or "").strip()
|
|
zh = str(item.get("zh") or item.get("target_text") or "").strip()
|
|
if en and zh:
|
|
pairs.append({"en": en, "zh": zh})
|
|
return pairs
|
|
|
|
|
|
llm_service = LlmService()
|