v2

README.md

@@ -14,24 +14,41 @@ short_description: Patent translation demo via hiro-translation-api
 
 # HIRO Translation
 
-Patent translation demo powered by **hiro-translation-api** (stage).
+Patent translation demo for the public HIRO Translation API.
+
+- **stream** — SSE streaming (`textOriginal` / `textTranslated` per segment)
+- **fast** — single JSON with full `textTranslated`
+- Default gateway: `https://connect.zhihuiya.com/hiro_translation`
+- Auth: Bearer token via **Advanced → API Key** or env `HIRO_API_KEY`
+
+Request body (aligned with terminal curl):
+
+```json
+{
+  "content": "本发明公开了一种档案管理文件储存用分类标识装置。",
+  "sourceLanguageCode": "zh",
+  "targetLanguageCode": "en",
+  "mode": "stream"
+}
+```
+
+Internal stage URL (optional in Advanced):
 
-- **stream** — SSE streaming, segment-by-segment display
-- **fast** — single JSON response with full translation
-- Default backend: `stage-s-patsnaprd-hiro-translation-api-internet.patsnap.info`
+`http://stage-s-patsnaprd-hiro-translation-api-internet.patsnap.info/compute/hiro_translation_api`
 
 ## Files
 
 | File | Purpose |
 |------|---------|
 | `app.py` | Gradio UI |
-| `api_client.py` | HTTP client for `/compute/hiro_translation_api` |
+| `api_client.py` | HTTP client for `/translate` (camelCase request/response) |
 | `requirements.txt` | Python dependencies (`requests`; Gradio is pre-installed on Spaces) |
 
 ## Local run
 
 ```bash
 pip install -r requirements.txt
+export HIRO_API_KEY="sk-..."   # optional if set in UI
 python app.py
 ```
 

api_client.py

@@ -1,26 +1,75 @@
-"""HTTP client for hiro-translation-api (stage gateway)."""
+"""HTTP client for HIRO Translation API (public connect gateway or internal stage)."""
 
 from __future__ import annotations
 
 import json
+import os
 from collections.abc import Iterator
 from typing import Any
 
 import requests
 
-DEFAULT_BASE = (
-    "http://stage-s-patsnaprd-hiro-translation-api-internet.patsnap.info"
-)
-API_PREFIX = "/compute/hiro_translation_api"
+# Public gateway (matches curl examples against connect.zhihuiya.com).
+DEFAULT_BASE = "https://connect.zhihuiya.com/hiro_translation"
+
+API_KEY_ENV = "HIRO_API_KEY"
+
+
+def lang_to_codes(lang: str) -> tuple[str, str]:
+    src, _, tgt = lang.partition("2")
+    if not src or not tgt:
+        raise ValueError(f"invalid lang {lang!r}; expected format like zh2en")
+    return src, tgt
+
+
+def translate_payload(
+    content: str,
+    lang: str,
+    *,
+    mode: str,
+) -> dict[str, Any]:
+    source, target = lang_to_codes(lang)
+    return {
+        "content": content,
+        "sourceLanguageCode": source,
+        "targetLanguageCode": target,
+        "mode": mode,
+    }
 
 
 def _api_url(base_url: str, path: str) -> str:
-    return f"{base_url.rstrip('/')}{API_PREFIX}{path}"
+    return f"{base_url.rstrip('/')}{path}"
 
 
-def health_ok(base_url: str = DEFAULT_BASE) -> tuple[bool, str]:
+def _request_headers(api_key: str | None = None) -> dict[str, str]:
+    headers = {"Content-Type": "application/json"}
+    key = (api_key if api_key is not None else os.environ.get(API_KEY_ENV, "")).strip()
+    if key:
+        headers["Authorization"] = f"Bearer {key}"
+    return headers
+
+
+def _translated_text(data: dict[str, Any]) -> str:
+    value = data.get("textTranslated", data.get("text_translated", ""))
+    return value if isinstance(value, str) else ""
+
+
+def _original_text(data: dict[str, Any], fallback: str = "") -> str:
+    value = data.get("textOriginal", data.get("text_original", fallback))
+    return value if isinstance(value, str) else fallback
+
+
+def health_ok(
+    base_url: str = DEFAULT_BASE,
+    *,
+    api_key: str | None = None,
+) -> tuple[bool, str]:
     try:
-        r = requests.get(_api_url(base_url, "/health"), timeout=12)
+        r = requests.get(
+            _api_url(base_url, "/health"),
+            headers=_request_headers(api_key),
+            timeout=12,
+        )
         r.raise_for_status()
         data = r.json()
         if data.get("status") != "OK":
@@ -38,11 +87,13 @@ def translate_fast(
     lang: str,
     *,
     base_url: str = DEFAULT_BASE,
+    api_key: str | None = None,
     timeout: int = 1800,
 ) -> dict[str, Any]:
     r = requests.post(
         _api_url(base_url, "/translate"),
-        json={"text": text, "lang": lang, "mode": "fast"},
+        json=translate_payload(text, lang, mode="fast"),
+        headers=_request_headers(api_key),
         timeout=timeout,
     )
     if r.status_code >= 400:
@@ -51,7 +102,18 @@ def translate_fast(
         except json.JSONDecodeError:
             err = r.text[:500]
         raise RuntimeError(err)
-    return r.json()
+    data = r.json()
+    if data.get("state") not in (None, "success") and "error" in data:
+        raise RuntimeError(str(data.get("error", data)))
+    # Normalize to keys the UI already understands (snake_case aliases).
+    return {
+        "state": data.get("state", "success"),
+        "text_original": _original_text(data, text),
+        "text_translated": _translated_text(data),
+        "translated_character_count": data.get("translatedCharacterCount"),
+        "billing_amount": r.headers.get("X-Openapi-Amount"),
+        "raw": data,
+    }
 
 
 def _iter_sse_json(resp: requests.Response) -> Iterator[dict[str, Any]]:
@@ -67,7 +129,15 @@ def _iter_sse_json(resp: requests.Response) -> Iterator[dict[str, Any]]:
         chunk = json.loads(payload)
         if isinstance(chunk, dict) and chunk.get("error"):
             raise RuntimeError(str(chunk["error"]))
-        yield chunk
+        if isinstance(chunk, dict):
+            yield {
+                "state": chunk.get("state", "success"),
+                "text_original": _original_text(chunk),
+                "text_translated": _translated_text(chunk),
+                "translated_character_count": chunk.get("translatedCharacterCount"),
+                "progress": chunk.get("progress"),
+                "raw": chunk,
+            }
 
 
 def stream_translate(
@@ -75,11 +145,13 @@ def stream_translate(
     lang: str,
     *,
     base_url: str = DEFAULT_BASE,
+    api_key: str | None = None,
     timeout: int = 1800,
 ) -> Iterator[dict[str, Any]]:
     with requests.post(
         _api_url(base_url, "/translate"),
-        json={"text": text, "lang": lang, "mode": "stream"},
+        json=translate_payload(text, lang, mode="stream"),
+        headers=_request_headers(api_key),
         stream=True,
         timeout=timeout,
     ) as resp:

app.py

@@ -1,8 +1,10 @@
 """
-Gradio demo — HIRO Translation API (hiro-translation-api).
+Gradio demo — HIRO Translation API.
 
-Calls stage gateway by default:
-  http://stage-s-patsnaprd-hiro-translation-api-internet.patsnap.info
+Calls the public connect gateway by default:
+  https://connect.zhihuiya.com/hiro_translation/translate
+
+Set HIRO_API_KEY or paste a Bearer token in Advanced options.
 """
 
 from __future__ import annotations
@@ -228,19 +230,19 @@ LANG_OPTIONS = [
 ]
 
 EXAMPLES = [
-    ["一种低损耗换热器", "zh", "en"],
+    ["本发明公开了一种档案管理文件储存用分类标识装置。", "zh", "en"],
     [
-        "The present invention relates to a heat exchanger comprising a plurality of fluid passages.",
-        "en",
+        "本发明公开了一种档案管理文件储存用分类标识装置。\n我是智慧芽的研发",
         "zh",
+        "en",
     ],
     [
-        "1.一种蓝莓早促快发春梢的绿色生产技术，其特征在于，包括：(1)新主枝的培养、(2)调节树体结构。",
-        "zh",
+        "The present invention relates to a heat exchanger comprising a plurality of fluid passages.",
         "en",
+        "zh",
     ],
     [
-        "## 技术领域\n\n本发明涉及一种换热器。",
+        "1.一种蓝莓早促快发春梢的绿色生产技术，其特征在于，包括：(1)新主枝的培养、(2)调节树体结构。",
         "zh",
         "en",
     ],
@@ -254,9 +256,27 @@ def _resolve_base(api_base: str) -> str:
     return (api_base or DEFAULT_BASE).strip().rstrip("/")
 
 
-def check_service(api_base: str) -> str:
-    ok, msg = health_ok(_resolve_base(api_base))
+def _has_health_endpoint(base: str) -> bool:
+    """Only the internal gateway exposes /health; the public connect gateway does not.
+
+    The public OpenAPI gateway (connect.zhihuiya.com) only routes the purchased
+    endpoints (e.g. /translate); probing /health there returns a permission/quota
+    error, not a real health signal.
+    """
+    return "/compute/hiro_translation_api" in base
+
+
+def check_service(api_base: str, api_key: str = "") -> str:
     base = _resolve_base(api_base)
+
+    if not _has_health_endpoint(base):
+        return (
+            f'<span class="status-pill ok">● 公开网关<br>'
+            f'<span style="opacity:0.85">调用时以 Bearer 鉴权 · {html.escape(base)}</span>'
+            f"</span>"
+        )
+
+    ok, msg = health_ok(base, api_key=api_key or None)
     if ok:
         return (
             f'<span class="status-pill ok">● 在线<br>'
@@ -571,13 +591,14 @@ CLEAR_STREAM_JS = """
 """
 
 
-def run_translation(text: str, lang: str, mode: str, api_base: str):
+def run_translation(text: str, lang: str, mode: str, api_base: str, api_key: str = ""):
     text = (text or "").strip()
     if not text:
         yield _empty_outputs()
         return
 
     base = _resolve_base(api_base)
+    auth = api_key or None
     pairs: list[tuple[str, str]] = []
     t0 = time.perf_counter()
     last_progress = "0/0"
@@ -604,19 +625,19 @@ def run_translation(text: str, lang: str, mode: str, api_base: str):
 
     try:
         if mode == "fast":
-            data = translate_fast(text, lang, base_url=base)
+            data = translate_fast(text, lang, base_url=base, api_key=auth)
             orig = data.get("text_original", text)
             trans = data.get("text_translated", "")
-            engine = data.get("engine", "")
             elapsed = time.perf_counter() - t0
             pairs = [(orig, trans)]
             status = f"完成 · fast 模式 · {elapsed:.1f}s"
-            if engine:
-                status += f" · {engine}"
+            out_chars = data.get("translated_character_count")
+            if out_chars is not None:
+                status += f" · {out_chars} chars"
             yield emit(percent=100, status=status, force=True)
             return
 
-        for chunk in stream_translate(text, lang, base_url=base):
+        for chunk in stream_translate(text, lang, base_url=base, api_key=auth):
             orig = chunk.get("text_original") or ""
             trans = chunk.get("text_translated") or ""
             if orig or trans:
@@ -682,7 +703,7 @@ def build_ui() -> gr.Blocks:
     ) as demo:
         with gr.Sidebar():
             gr.HTML('<p class="main-title">HIRO Translation</p>')
-            gr.Markdown("专利翻译 · `hiro-translation-api`")
+            gr.Markdown("专利翻译 · `connect.zhihuiya.com` / `hiro-translation-api`")
 
             gateway_status = gr.HTML(value=check_service(DEFAULT_BASE))
             refresh_btn = gr.Button("检查网关", size="sm", variant="secondary")
@@ -711,6 +732,12 @@ def build_ui() -> gr.Blocks:
                     label="API Base URL",
                     value=DEFAULT_BASE,
                     placeholder=DEFAULT_BASE,
+                    info="公开网关默认 https://connect.zhihuiya.com/hiro_translation；内网可填 stage 地址",
+                )
+                api_key = gr.Textbox(
+                    label="API Key (Bearer)",
+                    type="password",
+                    placeholder="sk-... 或设置环境变量 HIRO_API_KEY",
                 )
 
             gr.Markdown("**快速示例**")
@@ -754,8 +781,9 @@ def build_ui() -> gr.Blocks:
                     elem_id="hiro-stream-state",
                 )
 
-        refresh_btn.click(check_service, inputs=[api_base], outputs=[gateway_status])
-        api_base.change(check_service, inputs=[api_base], outputs=[gateway_status])
+        refresh_btn.click(check_service, inputs=[api_base, api_key], outputs=[gateway_status])
+        api_base.change(check_service, inputs=[api_base, api_key], outputs=[gateway_status])
+        api_key.change(check_service, inputs=[api_base, api_key], outputs=[gateway_status])
 
         def _combine_lang(src: str, tgt: str) -> str:
             return f"{src}2{tgt}"
@@ -775,7 +803,7 @@ def build_ui() -> gr.Blocks:
 
         submit_btn.click(
             fn=run_translation,
-            inputs=[input_text, lang, mode, api_base],
+            inputs=[input_text, lang, mode, api_base, api_key],
             outputs=[stream_state, results_panel, copy_btn],
             js=RESET_STREAM_JS,
         )
@@ -817,7 +845,7 @@ def build_ui() -> gr.Blocks:
 
         demo.load(
             check_service,
-            inputs=[api_base],
+            inputs=[api_base, api_key],
             outputs=[gateway_status],
             js=INIT_STREAM_WATCH_JS,
         )