chengma/cmshoppe
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: complete T-530 cmhub URL normalization

Browse Source
This commit is contained in:
chengma
2026-07-06 10:31:36 +08:00
parent 42dd8a993f
commit b99019e954
14 changed files with 142 additions and 31 deletions
Download Patch File Download Diff File Expand all files Collapse all files
app/ai.py
+12 -1
View File
@@ -771,7 +771,7 @@ def _cmhub_error_from_response(data, response, api_key):
error = data.get("error") if isinstance(data, dict) else None
if not isinstance(error, dict):
error = {}
code = str(error.get("code") or _cmhub_code_for_status(status))
code = _normalize_cmhub_error_code(error.get("code") or _cmhub_code_for_status(status), status)
raw_message = (error.get("message") or data.get("message")) if isinstance(data, dict) else ""
if not raw_message:
raw_message = getattr(response, "text", "")[:500]
@@ -786,12 +786,20 @@ def _cmhub_error_from_response(data, response, api_key):
)
def _normalize_cmhub_error_code(code, status=None):
normalized = str(code or "unknown").strip().lower().replace("-", "_")
if status == 404 or normalized in {"notfound", "not_found"}:
return "not_found"
return normalized or "unknown"
def _cmhub_code_for_status(status):
return {
400: "bad_request",
401: "unauthorized",
402: "insufficient_points",
403: "account_disabled",
404: "not_found",
429: "rate_limited",
502: "upstream_error",
}.get(status, "unknown")
@@ -812,8 +820,11 @@ def _cmhub_user_message(code, message):
"content_blocked": "cmhub 内容安全策略拒绝本次生成",
"upstream_error": "cmhub 上游生成失败，请稍后重试",
"rate_limited": "cmhub 请求过于频繁，请稍后重试",
"not_found": "cmhub 接口不存在，请检查 Base URL 或该实例是否已部署 /api/v1/models",
}
default = defaults.get(str(code))
if str(code) == "not_found":
return default
if default and message and str(message) not in default:
return "%s：%s" % (default, message)
return default or str(message or code)
app/appconfig.py
+33 -2
View File
@@ -9,6 +9,7 @@ import copy
import json
import os
import urllib.error
import urllib.parse
import urllib.request
@@ -129,6 +130,17 @@ def _deep_merge(defaults, loaded):
return merged
def _normalize_config_values(config):
if not isinstance(config, dict):
return config
ai = config.get("ai")
if isinstance(ai, dict):
cmhub = ai.get("cmhub")
if isinstance(cmhub, dict):
cmhub["base_url"] = normalize_cmhub_base_url(cmhub.get("base_url", ""))
return config
def _assert_no_secrets(config):
def visit(value, path):
if isinstance(value, dict):
@@ -247,6 +259,7 @@ def save_config(config, path=CONFIG_PATH) -> dict:
"""Persist config to JSON and return the normalized config."""
normalized = _deep_merge(DEFAULT_CONFIG, config)
_normalize_config_values(normalized)
_assert_no_secrets(normalized)
directory = os.path.dirname(os.path.abspath(path))
if directory:
@@ -268,6 +281,7 @@ def load_config(path=CONFIG_PATH) -> dict:
except json.JSONDecodeError as exc:
raise ConfigError(f"配置文件不是有效 JSON: {path}") from exc
normalized = _deep_merge(DEFAULT_CONFIG, loaded)
_normalize_config_values(normalized)
_assert_no_secrets(normalized)
return normalized
@@ -334,7 +348,7 @@ def cmhub_config(config=None) -> dict:
if not isinstance(value, dict):
raise ConfigError("ai.cmhub 必须是对象")
merged = _deep_merge(DEFAULT_CONFIG["ai"]["cmhub"], value)
merged["base_url"] = str(merged.get("base_url", "") or "").strip()
merged["base_url"] = normalize_cmhub_base_url(merged.get("base_url", ""))
merged["title_alias"] = str(merged.get("title_alias", "") or "").strip()
merged["image_alias"] = str(merged.get("image_alias", "") or "").strip()
merged["connect_timeout"] = int(merged.get("connect_timeout", 10) or 10)
@@ -344,8 +358,25 @@ def cmhub_config(config=None) -> dict:
return merged
def normalize_cmhub_base_url(base_url) -> str:
"""Return the cmhub gateway root URL without path/query/fragment."""
text = str(base_url or "").strip()
if not text:
return ""
parts = urllib.parse.urlsplit(text)
if parts.scheme and parts.netloc:
return urllib.parse.urlunsplit((parts.scheme, parts.netloc, "", "", ""))
if parts.netloc:
return urllib.parse.urlunsplit((parts.scheme, parts.netloc, "", "", ""))
without_query = text.split("?", 1)[0].split("#", 1)[0].strip().strip("/")
if "://" not in without_query:
return without_query.split("/", 1)[0]
return text.rstrip("/")
def cmhub_request_url(base_url, endpoint) -> str:
base = str(base_url or "").strip().rstrip("/")
base = normalize_cmhub_base_url(base_url)
path = "/" + str(endpoint or "").strip().lstrip("/")
if not base:
return path
app/gui/tabs/settings.py
+10 -4
View File
@@ -71,7 +71,7 @@ class SettingsTab(QWidget):
self.backend_combo.setVisible(False)
self.cmhub_base_url_edit = QLineEdit()
self.cmhub_base_url_edit.setObjectName("cmhubBaseUrlEdit")
self.cmhub_base_url_edit.setPlaceholderText("https://<cmhub>")
self.cmhub_base_url_edit.setPlaceholderText("https://host（不要带 /api 或 /api/v1）")
self.cmhub_api_key_edit = QLineEdit()
self.cmhub_api_key_edit.setObjectName("cmhubApiKeyEdit")
self.cmhub_api_key_edit.setEchoMode(QLineEdit.Password)
@@ -93,6 +93,9 @@ class SettingsTab(QWidget):
self.cmhub_result_label = QLabel("")
self.cmhub_result_label.setObjectName("cmhubResultLabel")
self.cmhub_result_label.setWordWrap(True)
self.cmhub_base_url_hint_label = QLabel("Base URL 只填网关根，如 https://host；不要带 /api 或 /api/v1。")
self.cmhub_base_url_hint_label.setObjectName("cmhubBaseUrlHintLabel")
self.cmhub_base_url_hint_label.setWordWrap(True)
self.cmhub_key_hint_label = QLabel("API Key 仅在 cmhub 网页端创建时显示一次；复制到此处后会本地明文保存并打码显示。")
self.cmhub_key_hint_label.setObjectName("cmhubKeyHintLabel")
self.cmhub_key_hint_label.setWordWrap(True)
@@ -302,6 +305,7 @@ class SettingsTab(QWidget):
cmhub_panel_layout.setContentsMargins(0, 0, 0, 0)
cmhub_panel_layout.setSpacing(8)
cmhub_panel_layout.addLayout(cmhub_form)
cmhub_panel_layout.addWidget(self.cmhub_base_url_hint_label)
cmhub_panel_layout.addWidget(self.cmhub_key_hint_label)
cmhub_panel_layout.addLayout(cmhub_action_layout)
cmhub_panel_layout.addWidget(self.cmhub_result_label)
@@ -673,7 +677,7 @@ class SettingsTab(QWidget):
def _cmhub_settings_values(self, backend):
current = appconfig.cmhub_config(self.config)
values = {
"base_url": self.cmhub_base_url_edit.text().strip(),
"base_url": appconfig.normalize_cmhub_base_url(self.cmhub_base_url_edit.text()),
"title_alias": self.cmhub_title_alias_combo.currentData() or "",
"image_alias": self.cmhub_image_alias_combo.currentData() or "",
"connect_timeout": self.cmhub_connect_timeout_spin.value(),
@@ -700,7 +704,7 @@ class SettingsTab(QWidget):
ai_cfg = appconfig.ai_config(self.config)
self._set_combo_by_data(self.backend_combo, "cmhub")
cmhub_cfg = appconfig.cmhub_config(self.config)
self.cmhub_base_url_edit.setText(cmhub_cfg.get("base_url", ""))
self.cmhub_base_url_edit.setText(appconfig.normalize_cmhub_base_url(cmhub_cfg.get("base_url", "")))
self._loaded_cmhub_api_key = appconfig.get_cmhub_api_key(path=self.cmhub_config_path)
self.cmhub_api_key_edit.setText(self._loaded_cmhub_api_key)
self.cmhub_connect_timeout_spin.setValue(
@@ -937,7 +941,9 @@ class SettingsTab(QWidget):
if self.cmhub_thread is not None:
self._set_status("cmhub 检测正在进行...")
return
base_url = self.cmhub_base_url_edit.text().strip()
base_url = appconfig.normalize_cmhub_base_url(self.cmhub_base_url_edit.text())
if base_url != self.cmhub_base_url_edit.text().strip():
self.cmhub_base_url_edit.setText(base_url)
api_key = self.cmhub_api_key_edit.text()
missing = []
if not base_url: