etc.\n"
"- Clean up excessive whitespace and blank lines\n"
@@ -89,14 +89,13 @@ class TranslationAgent:
"Respond with ONLY the translated text in clean Markdown format."
)
- def __init__(self, provider: LLMProvider | None = None):
- """
- Initialize the translation agent.
-
- Args:
- provider: LLMProvider instance. If None, uses default provider from registry.
- """
- self.provider = provider or get_provider()
+ def __init__(self, provider: LLMProvider | None = None, use_mock: bool = False):
+ if provider is not None:
+ self.provider = provider
+ elif use_mock:
+ self.provider = MockTranslationProvider()
+ else:
+ self.provider = get_provider()
async def translate_chunk(
self,
@@ -105,20 +104,7 @@ async def translate_chunk(
temperature: float = 0.3,
max_retries: int = 3,
) -> dict:
- """翻译单个片段，失败时按指数退避重试。
-
- Args:
- content: 待翻译片段
- target_lang: 目标语言
- temperature: LLM temperature
- max_retries: 最大尝试次数（含首次）
-
- Returns:
- dict with text + token usage.
-
- Raises:
- LLMProviderError: 重试耗尽后仍失败时抛出，由调用方决定如何降级。
- """
+ """Translate a single chunk with retry-on-provider-error behavior."""
messages = [
ChatMessage(role="system", content=self.SYSTEM_PROMPT),
ChatMessage(
@@ -142,53 +128,58 @@ async def translate_chunk(
last_error = exc
if attempt == max_retries - 1:
break
- # 指数退避：0.5s, 1s, 2s ...，缓解限流/瞬时网络错误
await asyncio.sleep(0.5 * (2**attempt))
raise last_error if last_error else LLMProviderError("Translation failed")
- async def translate_bilingual(
+ async def translate_whole_article(
self,
content: str,
target_lang: str,
temperature: float = 0.3,
) -> dict:
- """
- 双语对照翻译：英文一段，中文一段，交替显示。
+ """Translate the whole article in one request, summary-style."""
+ messages = [
+ ChatMessage(role="system", content=self.SYSTEM_PROMPT),
+ ChatMessage(
+ role="user",
+ content=f"Please translate to {target_lang}:\n\n{content}",
+ ),
+ ]
+ options = ChatOptions(temperature=temperature)
- Args:
- content: Article text to translate
- target_lang: Target language (e.g., "Chinese")
- temperature: LLM temperature
+ completion = await self.provider.chat(messages, options=options)
+ return {
+ "translated_text": completion.content,
+ "provider": self.provider.name,
+ "model": self.provider.model,
+ "usage": {
+ "prompt_tokens": completion.usage.prompt_tokens,
+ "completion_tokens": completion.usage.completion_tokens,
+ },
+ }
- Returns:
- Dictionary with translated_text (bilingual format) and usage
+ async def translate_bilingual(
+ self,
+ content: str,
+ target_lang: str,
+ temperature: float = 0.3,
+ ) -> dict:
"""
- # 按段落分段
- paragraphs = [p.strip() for p in content.split('\n\n') if p.strip()]
+ Translate into alternating original/translation blocks.
- # 合并短段落，避免太碎片化
- merged_paragraphs = []
- current = ""
- for para in paragraphs:
- if len(current) + len(para) < 500:
- current = current + "\n\n" + para if current else para
- else:
- if current:
- merged_paragraphs.append(current)
- current = para
- if current:
- merged_paragraphs.append(current)
+ Each markdown paragraph is translated independently so the rendered
+ bilingual view stays aligned block by block.
+ """
+ paragraphs = [p.strip() for p in re.split(r"\n\s*\n", content) if p.strip()]
total_prompt_tokens = 0
total_completion_tokens = 0
- bilingual_parts = []
+ bilingual_parts: list[str] = []
failed_chunks = 0
- for para in merged_paragraphs:
- # 检查是否是标题
- if para.startswith('#'):
- # 标题直接翻译，不保留原文；失败则退化为原标题
+ for para in paragraphs:
+ if para.startswith("#"):
try:
result = await self.translate_chunk(para, target_lang, temperature)
bilingual_parts.append(result["text"])
@@ -197,32 +188,29 @@ async def translate_bilingual(
except LLMProviderError:
failed_chunks += 1
bilingual_parts.append(para)
- else:
- # 普通段落：先原文，再翻译。单段失败不影响整篇，保留原文占位。
- original = f'
{para}
'
- try:
- result = await self.translate_chunk(para, target_lang, temperature)
- translated = result["text"]
- total_prompt_tokens += result["prompt_tokens"]
- total_completion_tokens += result["completion_tokens"]
- except LLMProviderError:
- failed_chunks += 1
- translated = para
- translation = f'
{translated}
'
- bilingual_parts.append(original)
- bilingual_parts.append("")
- bilingual_parts.append(translation)
- bilingual_parts.append("")
-
- # 全部片段都失败时，视为整体失败，交由 service 标记 failure
- if merged_paragraphs and failed_chunks == len(merged_paragraphs):
- raise LLMProviderError("All translation chunks failed")
+ continue
- # 合并结果
- translated_text = "\n\n".join(bilingual_parts)
+ original = f'
{para}
'
+ try:
+ result = await self.translate_chunk(para, target_lang, temperature)
+ translated = result["text"]
+ total_prompt_tokens += result["prompt_tokens"]
+ total_completion_tokens += result["completion_tokens"]
+ except LLMProviderError:
+ failed_chunks += 1
+ translated = para
+
+ translation = f'
{translated}
'
+ bilingual_parts.append(original)
+ bilingual_parts.append("")
+ bilingual_parts.append(translation)
+ bilingual_parts.append("")
+
+ if paragraphs and failed_chunks == len(paragraphs):
+ raise LLMProviderError("All translation chunks failed")
return {
- "translated_text": translated_text,
+ "translated_text": "\n\n".join(bilingual_parts),
"provider": self.provider.name,
"model": self.provider.model,
"usage": {
@@ -241,34 +229,18 @@ async def translate(
"""
Translate content to target language.
Automatically chunks long articles by headings.
-
- Args:
- content: Article text to translate (cleaned markdown/plain text)
- target_lang: Target language (e.g., "English", "Chinese", "Spanish")
- temperature: LLM temperature (0.0-1.0), lower = more consistent
- bilingual: If True, return bilingual format (original + translation)
-
- Returns:
- Dictionary with:
- - translated_text: The translated content
- - provider: Provider name
- - model: Model name
- - usage: Token usage (prompt_tokens, completion_tokens)
"""
- # 如果是双语对照模式
if bilingual:
return await self.translate_bilingual(content, target_lang, temperature)
- # 分段
chunks = chunk_by_headings(content, max_chars=4000)
total_prompt_tokens = 0
total_completion_tokens = 0
- translated_chunks = []
+ translated_chunks: list[str] = []
failed_chunks = 0
for chunk in chunks:
- # 单段失败不影响整篇：保留原文占位，继续翻译其余片段
try:
result = await self.translate_chunk(chunk, target_lang, temperature)
translated_chunks.append(result["text"])
@@ -278,15 +250,11 @@ async def translate(
failed_chunks += 1
translated_chunks.append(chunk)
- # 全部片段都失败时，视为整体失败，交由 service 标记 failure
if chunks and failed_chunks == len(chunks):
raise LLMProviderError("All translation chunks failed")
- # 合并翻译结果
- translated_text = "\n\n".join(translated_chunks)
-
return {
- "translated_text": translated_text,
+ "translated_text": "\n\n".join(translated_chunks),
"provider": self.provider.name,
"model": self.provider.model,
"usage": {
@@ -294,3 +262,70 @@ async def translate(
"completion_tokens": total_completion_tokens,
},
}
+
+
+class MockTranslationProvider:
+ """Local mock provider used when no real translation provider is configured."""
+
+ @property
+ def name(self) -> str:
+ return "mock"
+
+ @property
+ def model(self) -> str:
+ return "mock-translation"
+
+ async def chat(
+ self,
+ messages: list[ChatMessage],
+ *,
+ stream: bool = False,
+ options: ChatOptions | None = None,
+ ) -> ChatCompletion:
+ user_prompt = next(
+ (
+ message.content
+ for message in reversed(messages)
+ if message.role == "user"
+ ),
+ "",
+ )
+ match = re.match(r"Please translate to\s+(.+?):\s*\n\n([\s\S]*)", user_prompt)
+ if match:
+ target_lang = match.group(1).strip()
+ source_text = match.group(2).strip()
+ else:
+ target_lang = "translated"
+ source_text = user_prompt.strip()
+
+ translated = _mock_translate_text(source_text, target_lang)
+ return ChatCompletion(
+ content=translated,
+ model=self.model,
+ usage=TokenUsage(
+ prompt_tokens=max(len(source_text) // 4, 1),
+ completion_tokens=max(len(translated) // 4, 1),
+ ),
+ )
+
+
+def _mock_translate_text(source_text: str, target_lang: str) -> str:
+ cleaned = source_text.strip()
+ if not cleaned:
+ return ""
+
+ if cleaned.startswith("#"):
+ heading_marks, _, heading_text = cleaned.partition(" ")
+ translated_heading = _mock_translate_sentence(heading_text or cleaned, target_lang)
+ return f"{heading_marks} {translated_heading}".strip()
+
+ return _mock_translate_sentence(cleaned, target_lang)
+
+
+def _mock_translate_sentence(text: str, target_lang: str) -> str:
+ language = target_lang.strip().lower()
+ if language == "english":
+ return f"[Mock English] {text}"
+ if language == "chinese":
+ return f"[Mock Chinese] {text}"
+ return f"[Mock {target_lang}] {text}"
diff --git a/backend/agent_translation/http/__init__.py b/backend/agent_translation/http/__init__.py
index 8ad8b69..7445a5c 100644
--- a/backend/agent_translation/http/__init__.py
+++ b/backend/agent_translation/http/__init__.py
@@ -1,5 +1 @@
-"""HTTP layer for translation agent."""
-
-from .router import router
-
-__all__ = ["router"]
+"""HTTP package for translation agent routes."""
diff --git a/backend/agent_translation/service.py b/backend/agent_translation/service.py
index 712112f..3d882f4 100644
--- a/backend/agent_translation/service.py
+++ b/backend/agent_translation/service.py
@@ -1,22 +1,32 @@
-"""Application service for article translation.
-
-Responsible for:
-1. Loading article content from database
-2. Invoking translation agent with LLM provider
-3. Handling errors and recording results
-4. Persisting translations to database
-"""
+from __future__ import annotations
+import logging
+import os
import re
from dataclasses import dataclass
from datetime import UTC
+from pathlib import Path
+
+from pydantic import SecretStr
+from agent_summary.llm_client import LLMClient, LLMClientError
from app.schemas.agent import TranslationRequest, TranslationResult
from db import get_article, get_article_content, record_usage, save_agent_result
from llm_providers import LLMProviderError, ProviderNotFoundError, get_provider
+from llm_providers.base import (
+ ChatCompletion,
+ ChatMessage,
+ ChatOptions,
+ LLMProvider,
+ TokenUsage,
+)
+from llm_providers.config import ProviderConfig, ProviderKind
+from llm_providers.openai_compatible import OpenAICompatibleProvider
from .agent import TranslationAgent
+logger = logging.getLogger(__name__)
+
class TranslationServiceError(Exception):
"""Base class for translation request failures."""
@@ -32,176 +42,116 @@ class ArticleContentUnavailableError(TranslationServiceError):
@dataclass(slots=True)
class TranslationService:
- """
- Application service for article translation.
+ """Application service for loading article content and persisting translation results."""
- Orchestrates:
- - Data loading from database
- - LLM provider selection
- - Translation execution
- - Result persistence
- - Usage tracking
- """
+ agent_factory: type[TranslationAgent] = TranslationAgent
async def generate(self, request: TranslationRequest) -> TranslationResult:
- """
- Generate translation for an article.
-
- Workflow:
- 1. Load article metadata
- 2. Resolve content (prefer cleaned over raw)
- 3. Get LLM provider
- 4. Execute translation
- 5. Handle errors gracefully
- 6. Save result and usage to database
-
- Args:
- request: TranslationRequest with entry_id, target_lang, provider, model
-
- Returns:
- TranslationResult with translated_html and status.
- Always succeeds - status indicates success/failure
-
- Raises:
- ArticleNotFoundError: If article doesn't exist
- ArticleContentUnavailableError: If article has no content
- """
- # 1. Load article from database
article = get_article(request.entry_id)
if article is None:
raise ArticleNotFoundError(request.entry_id)
- # 2. Resolve content - prefer cleaned markdown/text over raw HTML
content = self._resolve_content(request.entry_id, article.reader_html)
if not content:
raise ArticleContentUnavailableError(request.entry_id)
- # 3. Execute translation (with error handling)
- result = await self._execute_translation(
- entry_id=request.entry_id,
- target_lang=request.target_lang,
- content=content,
- provider_name=request.provider,
- model_name=request.model,
- )
-
- # 4. Save result to database
- save_agent_result(result)
-
- return result
-
- async def _execute_translation(
- self,
- entry_id: str,
- target_lang: str,
- content: str,
- provider_name: str | None,
- model_name: str | None,
- ) -> TranslationResult:
- """
- Execute translation with error handling.
-
- Returns SUCCESS or FAILURE status (never raises exceptions).
- Failures are recorded in the result for frontend handling.
-
- Args:
- entry_id: Article ID
- target_lang: Target language
- content: Article content to translate
- provider_name: Optional provider name (uses default if None)
- model_name: Optional model name (uses provider's default if None)
-
- Returns:
- TranslationResult with status SUCCESS or FAILURE
- """
try:
- # Get LLM provider (uses default if name not specified)
- provider = get_provider(name=provider_name)
-
- # Build agent with selected provider
- agent = TranslationAgent(provider=provider)
-
- # Execute translation (bilingual mode: original + translation)
- agent_result = await agent.translate(
- content=content,
- target_lang=target_lang,
- temperature=0.3, # Lower temperature for consistency
- bilingual=True, # 双语对照模式
- )
-
- # Record token usage
- from datetime import datetime
- today = datetime.now(UTC).strftime("%Y-%m-%d")
- record_usage(
- day=today,
- provider=agent_result["provider"],
- model=agent_result["model"],
- agent="translation",
- prompt_tokens=agent_result["usage"]["prompt_tokens"],
- completion_tokens=agent_result["usage"]["completion_tokens"],
- )
-
- # Return success result
- return TranslationResult(
- entry_id=entry_id,
- target_lang=target_lang,
- translation_html=agent_result["translated_text"],
+ if (
+ self.agent_factory is TranslationAgent
+ and not request.provider
+ and not request.model
+ and not _use_mock_llm()
+ ):
+ result = await self._translate_via_legacy_client(content, request.target_lang)
+ else:
+ agent = self._build_agent(request)
+ result = await agent.translate(
+ content=content,
+ target_lang=request.target_lang,
+ temperature=0.3,
+ bilingual=True,
+ )
+
+ translation = TranslationResult(
+ entry_id=request.entry_id,
+ target_lang=request.target_lang,
+ translation_html=result["translated_text"],
status="success",
- provider=agent_result["provider"],
- model=agent_result["model"],
+ provider=result["provider"],
+ model=result["model"],
)
-
- except ProviderNotFoundError:
- # No provider configured
- return TranslationResult(
- entry_id=entry_id,
- target_lang=target_lang,
+ save_agent_result(translation)
+
+ usage = result.get("usage", {})
+ if usage:
+ from datetime import datetime
+
+ today = datetime.now(UTC).strftime("%Y-%m-%d")
+ record_usage(
+ day=today,
+ provider=result["provider"],
+ model=result["model"],
+ agent="translation",
+ prompt_tokens=usage.get("prompt_tokens", 0),
+ completion_tokens=usage.get("completion_tokens", 0),
+ )
+
+ return translation
+ except ProviderNotFoundError as exc:
+ translation = TranslationResult(
+ entry_id=request.entry_id,
+ target_lang=request.target_lang,
translation_html="",
status="failure",
- provider=provider_name or "unknown",
- model=model_name or "unknown",
+ provider=request.provider or "unknown",
+ model=request.model or "unknown",
)
-
- except LLMProviderError:
- # LLM API error (auth, network, rate limit, etc.)
- # Don't propagate - record as failure
- return TranslationResult(
- entry_id=entry_id,
- target_lang=target_lang,
+ save_agent_result(translation)
+ logger.warning("Translation provider not found for %s: %s", request.entry_id, exc)
+ return translation
+ except LLMProviderError as exc:
+ translation = TranslationResult(
+ entry_id=request.entry_id,
+ target_lang=request.target_lang,
translation_html="",
status="failure",
- provider=provider_name or "unknown",
- model=model_name or "unknown",
+ provider=request.provider or "unknown",
+ model=request.model or "unknown",
)
-
- except Exception:
- # Unexpected error - record as failure
- return TranslationResult(
- entry_id=entry_id,
- target_lang=target_lang,
+ save_agent_result(translation)
+ logger.warning("Translation provider error for %s: %s", request.entry_id, exc)
+ return translation
+ except Exception as exc:
+ translation = TranslationResult(
+ entry_id=request.entry_id,
+ target_lang=request.target_lang,
translation_html="",
status="failure",
- provider=provider_name or "unknown",
- model=model_name or "unknown",
+ provider=request.provider or "unknown",
+ model=request.model or "unknown",
)
+ save_agent_result(translation)
+ logger.exception("Unexpected translation failure for %s: %s", request.entry_id, exc)
+ return translation
+
+ def _build_agent(self, request: TranslationRequest) -> TranslationAgent:
+ if (
+ self.agent_factory is not TranslationAgent
+ and not request.provider
+ and not request.model
+ ):
+ return _instantiate_agent(self.agent_factory)
+
+ provider_name = (request.provider or "").strip().lower()
+ if provider_name == "mock" or _use_mock_llm():
+ return _instantiate_agent(self.agent_factory, use_mock=True)
+
+ provider = _resolve_provider(request)
+ if request.model:
+ provider = ModelOverrideProvider(provider, request.model)
+ return _instantiate_agent(self.agent_factory, provider=provider)
def _resolve_content(self, entry_id: str, reader_html: str) -> str:
- """
- Resolve article content, preferring cleaned content over raw HTML.
-
- Content resolution priority:
- 1. Cleaned markdown (highest quality)
- 2. Plain text (fallback)
- 3. Stripped raw HTML (last resort)
-
- Args:
- entry_id: Article ID for database lookup
- reader_html: Raw HTML from feed (fallback)
-
- Returns:
- Best available content string, or empty string if none available
- """
- # Try to get cleaned content from database
stored_content = get_article_content(entry_id)
if stored_content is not None:
if stored_content.cleaned_markdown.strip():
@@ -209,28 +159,204 @@ def _resolve_content(self, entry_id: str, reader_html: str) -> str:
if stored_content.plain_text.strip():
return stored_content.plain_text.strip()
- # Fallback: strip raw HTML
if reader_html.strip():
- return self._strip_html(reader_html)
-
+ return _strip_html(reader_html)
return ""
- @staticmethod
- def _strip_html(html: str) -> str:
- """
- Simple HTML tag removal.
-
- Removes HTML tags and normalizes whitespace.
- Used as fallback when cleaned content unavailable.
-
- Args:
- html: Raw HTML string
-
- Returns:
- Plain text with normalized whitespace
- """
- # Remove HTML tags
- text = re.sub(r"<[^>]+>", " ", html)
- # Normalize whitespace
- text = re.sub(r"\s+", " ", text)
- return text.strip()
+ async def _translate_via_legacy_client(self, content: str, target_lang: str) -> dict:
+ legacy_config = _load_translation_legacy_config()
+ client = LLMClient(
+ api_key=legacy_config["api_key"],
+ base_url=legacy_config["base_url"],
+ model=legacy_config["model"],
+ )
+ prompt = (
+ "You are a professional translator specializing in Chinese <-> English translation. "
+ "Translate the provided article content into the requested target language.\n"
+ "- If the target language is English, translate every Chinese sentence into English.\n"
+ "- If the target language is Chinese, translate every English sentence into Chinese.\n"
+ "- Text already written in the target language must be left exactly as-is.\n"
+ "- Never echo the source text back untranslated, and never answer in the "
+ "source language when a translation is requested.\n\n"
+ "Preserve meaning, nuance, structure, formatting, technical terms, and tone. "
+ "Remove raw HTML tags and output clean readable Markdown only.\n\n"
+ f"Please translate to {target_lang}:\n\n{content}"
+ )
+ try:
+ response = await client.chat_with_usage(prompt)
+ except LLMClientError as exc:
+ raise LLMProviderError(exc.detail) from exc
+
+ return {
+ "translated_text": response.text,
+ "provider": "legacy-env",
+ "model": client.model,
+ "usage": {
+ "prompt_tokens": response.prompt_tokens,
+ "completion_tokens": response.completion_tokens,
+ },
+ }
+
+
+class ModelOverrideProvider:
+ """Wrap an LLMProvider while allowing the request to override the model name."""
+
+ def __init__(self, provider: LLMProvider, model: str):
+ self._provider = provider
+ self._model = model
+
+ @property
+ def name(self) -> str:
+ return self._provider.name
+
+ @property
+ def model(self) -> str:
+ return self._model
+
+ async def chat(self, messages, *, stream=False, options=None):
+ return await self._provider.chat(messages, stream=stream, options=options)
+
+
+class SummaryClientProvider:
+ """Adapter that makes agent_summary's legacy LLMClient usable as an LLMProvider."""
+
+ def __init__(self, request: TranslationRequest):
+ self._client = LLMClient(model=request.model or None)
+ self._provider_name = "legacy-env"
+
+ @property
+ def name(self) -> str:
+ return self._provider_name
+
+ @property
+ def model(self) -> str:
+ return self._client.model
+
+ async def chat(
+ self,
+ messages: list[ChatMessage],
+ *,
+ stream: bool = False,
+ options: ChatOptions | None = None,
+ ) -> ChatCompletion:
+ if stream:
+ raise LLMProviderError("Streaming is not supported by SummaryClientProvider")
+
+ prompt = _messages_to_prompt(messages, options=options)
+ try:
+ response = await self._client.chat_with_usage(prompt)
+ except LLMClientError as exc:
+ raise LLMProviderError(exc.detail) from exc
+
+ return ChatCompletion(
+ content=response.text,
+ model=self._client.model,
+ usage=TokenUsage(
+ prompt_tokens=response.prompt_tokens,
+ completion_tokens=response.completion_tokens,
+ ),
+ )
+
+
+def _strip_html(value: str) -> str:
+ return re.sub(r"\s+", " ", re.sub(r"<[^>]+>", " ", value)).strip()
+
+
+def _use_mock_llm() -> bool:
+ return False
+
+
+def _resolve_provider(request: TranslationRequest) -> LLMProvider:
+ provider_name = request.provider if request.provider else None
+ if not provider_name:
+ return SummaryClientProvider(request)
+
+ try:
+ return get_provider(provider_name)
+ except ProviderNotFoundError:
+ return _provider_from_legacy_env(request)
+
+
+def _provider_from_legacy_env(request: TranslationRequest) -> LLMProvider:
+ legacy_config = _load_translation_legacy_config()
+ api_key = legacy_config["api_key"]
+ if not api_key:
+ raise ProviderNotFoundError("No providers configured")
+
+ base_url = legacy_config["base_url"]
+ model = (request.model or legacy_config["model"] or "ecnu-max").strip()
+ config = ProviderConfig(
+ name=request.provider or "legacy-env",
+ kind=ProviderKind.OPENAI_COMPATIBLE,
+ model=model,
+ base_url=base_url,
+ api_key=SecretStr(api_key),
+ )
+ return OpenAICompatibleProvider(config)
+
+
+def _messages_to_prompt(
+ messages: list[ChatMessage],
+ *,
+ options: ChatOptions | None = None,
+) -> str:
+ prompt_parts: list[str] = []
+ for message in messages:
+ if message.role == "system":
+ prompt_parts.append(message.content.strip())
+ continue
+ label = "User" if message.role == "user" else "Assistant"
+ prompt_parts.append(f"{label}:\n{message.content.strip()}")
+
+ if options is not None and options.temperature is not None:
+ prompt_parts.append(f"Temperature: {options.temperature}")
+
+ return "\n\n".join(part for part in prompt_parts if part)
+
+
+def _candidate_env_paths() -> list[Path]:
+ home = Path.home()
+ return [
+ Path(__file__).parent / ".env",
+ Path.cwd() / ".env",
+ home / ".mercury" / ".env",
+ home / ".mercury" / "agent_translation.env",
+ ]
+
+
+def _load_legacy_env() -> None:
+ for env_path in _candidate_env_paths():
+ if not env_path.exists():
+ continue
+ with open(env_path, encoding="utf-8") as handle:
+ for line in handle:
+ line = line.strip()
+ if line and not line.startswith("#") and "=" in line:
+ key, value = line.split("=", 1)
+ os.environ.setdefault(key.strip(), value.strip())
+
+
+def _load_translation_legacy_config() -> dict[str, str]:
+ env_values: dict[str, str] = {}
+ for env_path in _candidate_env_paths():
+ if not env_path.exists():
+ continue
+ with open(env_path, encoding="utf-8") as handle:
+ for line in handle:
+ line = line.strip()
+ if line and not line.startswith("#") and "=" in line:
+ key, value = line.split("=", 1)
+ env_values.setdefault(key.strip(), value.strip())
+
+ return {
+ "api_key": env_values.get("LLM_API_KEY", "").strip(),
+ "base_url": env_values.get("LLM_BASE_URL", "https://chat.ecnu.edu.cn/open/api/v1").strip(),
+ "model": env_values.get("LLM_MODEL", "ecnu-max").strip(),
+ }
+
+
+def _instantiate_agent(agent_factory, **kwargs):
+ try:
+ return agent_factory(**kwargs)
+ except TypeError:
+ return agent_factory()
diff --git a/backend/agent_translation/tests/test_agent_resilience.py b/backend/agent_translation/tests/test_agent_resilience.py
index 8c0a88d..80183b8 100644
--- a/backend/agent_translation/tests/test_agent_resilience.py
+++ b/backend/agent_translation/tests/test_agent_resilience.py
@@ -96,3 +96,23 @@ async def test_single_chunk_failure_does_not_abort_when_others_succeed() -> None
# The article still returns; the failed chunk falls back to its original text.
assert "translated" in result["translated_text"]
assert result["provider"] == "fake"
+
+
+@pytest.mark.asyncio
+async def test_bilingual_translation_keeps_one_pair_per_paragraph() -> None:
+ provider = FakeProvider(["第一段译文", "第二段译文"])
+ agent = TranslationAgent(provider=provider)
+
+ result = await agent.translate(
+ "First paragraph.\n\nSecond paragraph.",
+ target_lang="Chinese",
+ bilingual=True,
+ )
+
+ assert provider.calls == 2
+ assert result["translated_text"].count('class="bilingual-original"') == 2
+ assert result["translated_text"].count('class="bilingual-translation"') == 2
+ assert '
First paragraph.
' in result["translated_text"]
+ assert '
第一段译文
' in result["translated_text"]
+ assert '
Second paragraph.
' in result["translated_text"]
+ assert '
第二段译文
' in result["translated_text"]
diff --git a/backend/tests/test_translation_http_contract.py b/backend/tests/test_translation_http_contract.py
new file mode 100644
index 0000000..0ab724f
--- /dev/null
+++ b/backend/tests/test_translation_http_contract.py
@@ -0,0 +1,204 @@
+from __future__ import annotations
+
+from collections.abc import Iterator
+
+import pytest
+from fastapi.testclient import TestClient
+
+from app.config import settings
+from app.main import app
+from app.schemas.entry import Entry
+from app.schemas.feed import Feed
+from db import get_article, init_db, save_article, save_article_content, save_feed
+
+
+@pytest.fixture
+def api_db(tmp_path, monkeypatch) -> Iterator[str]:
+ db_path = tmp_path / "mercury-translation-http.db"
+ monkeypatch.setattr(settings, "db_path", db_path)
+ init_db(db_path)
+
+ save_feed(
+ Feed(
+ id="feed-1",
+ title="Mercury Blog",
+ site_url="https://example.com",
+ feed_url="https://example.com/feed.xml",
+ unread_count=0,
+ status="idle",
+ ),
+ db_path,
+ )
+ save_article(_article("article-1", reader_html="
Hello Mercury
"), db_path)
+ save_article_content(
+ article_id="article-1",
+ raw_html="
Hello Mercury
",
+ cleaned_html="
Hello Mercury
",
+ cleaned_markdown="Hello Mercury",
+ plain_text="Hello Mercury",
+ db_path=db_path,
+ )
+
+ yield str(db_path)
+
+
+@pytest.fixture
+def api_client(api_db: str) -> Iterator[TestClient]:
+ with TestClient(app) as client:
+ yield client
+
+
+def test_translate_article_uses_mock_fallback_and_persists_result(
+ api_client: TestClient,
+ monkeypatch,
+) -> None:
+ import agent_translation.service as translation_service_module
+ from agent_translation.http import router as translation_router
+ from agent_translation.service import TranslationService
+
+ monkeypatch.setattr(translation_service_module, "_use_mock_llm", lambda: True)
+ monkeypatch.setattr(
+ translation_router,
+ "get_translation_service",
+ lambda: TranslationService(),
+ )
+
+ response = api_client.post(
+ "/agents/translation",
+ json={"entry_id": "article-1", "target_lang": "Chinese"},
+ )
+
+ assert response.status_code == 200
+ payload = response.json()
+ assert payload["status"] == "success"
+ assert payload["provider"] == "mock"
+ assert '
Hello Mercury
' in payload["translation_html"]
+ assert (
+ '
[Mock Chinese] Hello Mercury
'
+ in payload["translation_html"]
+ )
+
+ saved = get_article("article-1")
+ assert saved is not None
+ assert payload["translation_html"] == saved.translation_html
+
+
+def test_translate_article_returns_404_for_missing_entry(api_client: TestClient) -> None:
+ response = api_client.post(
+ "/agents/translation",
+ json={"entry_id": "missing", "target_lang": "Chinese"},
+ )
+
+ assert response.status_code == 404
+ assert response.json() == {"detail": "Entry not found"}
+
+
+def test_translate_article_returns_409_when_no_content_is_available(api_client: TestClient) -> None:
+ save_article(_article("article-empty", reader_html=""), settings.db_path)
+
+ response = api_client.post(
+ "/agents/translation",
+ json={"entry_id": "article-empty", "target_lang": "Chinese"},
+ )
+
+ assert response.status_code == 409
+ assert response.json() == {"detail": "Entry has no content to translate"}
+
+
+def test_translate_article_returns_failure_payload_for_internal_errors(
+ api_client: TestClient, monkeypatch
+) -> None:
+ from agent_translation.http import router as translation_router
+ from agent_translation.service import TranslationService
+
+ class FailingAgent:
+ async def translate(
+ self,
+ content: str,
+ target_lang: str,
+ temperature: float = 0.3,
+ bilingual: bool = False,
+ ) -> dict:
+ raise RuntimeError("boom")
+
+ monkeypatch.setattr(
+ translation_router,
+ "get_translation_service",
+ lambda: TranslationService(agent_factory=FailingAgent),
+ )
+
+ response = api_client.post(
+ "/agents/translation",
+ json={"entry_id": "article-1", "target_lang": "Chinese"},
+ )
+
+ assert response.status_code == 200
+ assert response.json() == {
+ "entry_id": "article-1",
+ "target_lang": "Chinese",
+ "translation_html": "",
+ "status": "failure",
+ "provider": "unknown",
+ "model": "unknown",
+ }
+
+
+def test_translate_article_falls_back_to_legacy_env_provider_when_registry_is_empty(
+ api_client: TestClient, monkeypatch
+) -> None:
+ from agent_translation.http import router as translation_router
+ from agent_translation.service import TranslationService
+
+ async def fake_translate_via_legacy_client(self, content: str, target_lang: str) -> dict:
+ assert content == "Hello Mercury"
+ assert target_lang == "Chinese"
+ return {
+ "translated_text": "Translated from legacy env",
+ "provider": "legacy-env",
+ "model": "legacy-model",
+ "usage": {"prompt_tokens": 1, "completion_tokens": 1},
+ }
+
+ monkeypatch.setattr(
+ TranslationService,
+ "_translate_via_legacy_client",
+ fake_translate_via_legacy_client,
+ )
+ monkeypatch.setattr(translation_router, "get_translation_service", lambda: TranslationService())
+
+ response = api_client.post(
+ "/agents/translation",
+ json={"entry_id": "article-1", "target_lang": "Chinese"},
+ )
+
+ assert response.status_code == 200
+ assert response.json() == {
+ "entry_id": "article-1",
+ "target_lang": "Chinese",
+ "translation_html": "Translated from legacy env",
+ "status": "success",
+ "provider": "legacy-env",
+ "model": "legacy-model",
+ }
+
+
+def _article(article_id: str, reader_html: str) -> Entry:
+ return Entry(
+ id=article_id,
+ feed_id="feed-1",
+ title=f"{article_id} title",
+ summary="Short summary",
+ author="Mercury Team",
+ url=f"https://example.com/{article_id}",
+ published_at="2026-05-25T08:00:00Z",
+ is_read=False,
+ is_starred=False,
+ tag_ids=[],
+ reader_html=reader_html,
+ web_preview="",
+ related_entry_ids=[],
+ note="",
+ summary_text="",
+ translation_html=None,
+ translation_status="idle",
+ )
diff --git a/packages/ui/src/App.tsx b/packages/ui/src/App.tsx
index 348f3b8..81ec5c9 100644
--- a/packages/ui/src/App.tsx
+++ b/packages/ui/src/App.tsx
@@ -1105,14 +1105,23 @@ function ReaderDetail(props: {
import("./services/api").then(({ translateArticle }) => {
translateArticle(entry.id, detectTargetLang(entry.readerHtml))
.then((result) => {
- props.onUpdateEntry(entry.id, (current) => ({
- ...current,
- translationHtml: result.translation_html,
- translationStatus: result.status === "success" ? "success" : "failure"
- }));
+ if (result.status === "success" && result.translation_html.trim()) {
+ props.onUpdateEntry(entry.id, (current) => ({
+ ...current,
+ translationHtml: result.translation_html,
+ translationStatus: "success"
+ }));
+ return;
+ }
+
+ props.onUpdateEntry(entry.id, (current) => ({ ...current, translationStatus: "failure" }));
+ props.onNotice(t("translationFailed"));
+ setTranslationMode("original");
})
- .catch(() => {
+ .catch((error) => {
props.onUpdateEntry(entry.id, (current) => ({ ...current, translationStatus: "failure" }));
+ props.onNotice(getApiErrorMessage(error));
+ setTranslationMode("original");
});
});
return;
@@ -2540,10 +2549,10 @@ function capitalize(value: string): string {
// article read in an English UI still gets a Chinese translation rather than
// a no-op "English -> English" round trip.
function detectTargetLang(content: string): string {
- const chineseChars = (content.match(/[一-鿿]/g) ?? []).length;
+ const chineseChars = Array.from(content.matchAll(/\p{Script=Han}/gu)).length;
const asciiLetters = (content.match(/[A-Za-z]/g) ?? []).length;
// Treat the text as Chinese when CJK characters clearly dominate.
- return chineseChars > asciiLetters ? "English" : "中文";
+ return chineseChars > asciiLetters ? "English" : "Chinese";
}
function emptyProviderDraft(): ProviderDraft {