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

feat: cmhub shared Session + connection pool + proxy handling (T-546)

Browse Source 
- app/ai.py: 模块级共享 requests.Session + HTTPAdapter 连接池
  (CMHUB_HTTP_POOL_SIZE=32)；_apply_cmhub_proxy() 切换 trust_env；
  生成/下载/models/balance 统一复用该 Session。
- app/appconfig.py: ai.cmhub.use_system_proxy 默认 false（绕过系统代理）。
- tests/test_ai.py: 补 Session 复用/代理/连接池测试；原 18 处 cmhub mock
  从模块级 requests 改到共享 Session。
- docs: troubleshooting 加「图片下载很慢」排障节；06-tasks 标 T-546 DONE。

附带：把 app/ai.py 行尾从 CRLF 归一为 LF；并一并纳入工作区中此前未提交、
已在运行构建里的 cmhub 下载诊断代码（调试图片 URL 日志、下载重试事件），
无法与 T-546 在同一文件内拆分单独提交。243 单测全绿。

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
chengma
2026-07-08 00:15:43 +08:00
co-authored by Claude Opus 4.8
parent b257fb5c83
commit 5a09c6662e
5 changed files with 412 additions and 23 deletions
Download Patch File Download Diff File Expand all files Collapse all files
app/ai.py
+152 -8
View File
@@ -16,6 +16,7 @@ import urllib.request
import uuid
import requests
from requests.adapters import HTTPAdapter
from . import appconfig, db, diagnostics, image_paths
from . import prompts as prompt_module
@@ -40,6 +41,50 @@ CMHUB_IMAGE_MAX_BYTES = 20 * 1024 * 1024
CMHUB_TITLE_READ_TIMEOUT_SECONDS = 600
CMHUB_IMAGE_READ_TIMEOUT_SECONDS = 650
CMHUB_IMAGE_CONCURRENCY_LIMIT = 5
CMHUB_IMAGE_DOWNLOAD_ATTEMPTS = 3
CMHUB_IMAGE_SLOW_DOWNLOAD_SECONDS = 20.0
# 连接池要覆盖“生图并发 + 下载并发”（各上限 5）再留余量，避免连接不足排队或
# urllib3 "Connection pool is full" 警告。
CMHUB_HTTP_POOL_SIZE = 32
_CMHUB_SESSION = None
_CMHUB_SESSION_LOCK = threading.Lock()
def _cmhub_session():
"""返回 cmhub 专用的共享 `requests.Session`（连接复用 + 连接池）。
所有 cmhub HTTP（生成/下载/models/balance）都走同一个 Session，避免每次调用
新建 TCP+TLS 造成握手风暴与 connect_timeout；跨线程发请求安全，连接池足够大即可。
"""
global _CMHUB_SESSION
if _CMHUB_SESSION is None:
with _CMHUB_SESSION_LOCK:
if _CMHUB_SESSION is None:
session = requests.Session()
adapter = HTTPAdapter(
pool_connections=CMHUB_HTTP_POOL_SIZE,
pool_maxsize=CMHUB_HTTP_POOL_SIZE,
max_retries=0,
)
session.mount("http://", adapter)
session.mount("https://", adapter)
_CMHUB_SESSION = session
return _CMHUB_SESSION
def _apply_cmhub_proxy(use_system_proxy):
"""设置 cmhub Session 是否读取系统代理环境变量。
`use_system_proxy=False` → `trust_env=False`，忽略 `HTTP(S)_PROXY`/`ALL_PROXY` 直连；
公网 cmhub 网关默认直连，避免误走慢代理导致明文图片下载卡住。需要代理的环境可在
`config.json` 的 `ai.cmhub.use_system_proxy` 打开。
"""
session = _cmhub_session()
session.trust_env = bool(use_system_proxy)
return session
_RESOLUTION_SIZES = {
"512": (512, 512),
@@ -233,7 +278,16 @@ def _notify_step(callback, step):
pass
def _notify_step_event(callback, step, result="success", detail=None, level="info"):
def _notify_step_event(
callback,
step,
result="success",
detail=None,
level="info",
debug_only=False,
attempt=None,
attempts=None,
):
if callback is None:
return
payload = {
@@ -243,6 +297,12 @@ def _notify_step_event(callback, step, result="success", detail=None, level="inf
}
if detail is not None:
payload["detail"] = detail
if attempt is not None:
payload["attempt"] = attempt
if attempts is not None:
payload["attempts"] = attempts
if debug_only:
payload["debug_only"] = True
try:
callback(payload)
except Exception:
@@ -421,6 +481,7 @@ def generate_batch(tasks, prompts, ai_cfg=None, on_progress=None, should_stop=No
attempt=event.get("attempt"),
attempts=event.get("attempts"),
metadata=event.get("metadata"),
debug_only=event.get("debug_only"),
)
return
set_step(task, event)
@@ -703,7 +764,7 @@ def generate_batch(tasks, prompts, ai_cfg=None, on_progress=None, should_stop=No
return summary
def fetch_cmhub_models(base_url, api_key, connect_timeout=10, read_timeout=30):
def fetch_cmhub_models(base_url, api_key, connect_timeout=10, read_timeout=30, use_system_proxy=False):
"""Fetch cmhub model aliases for settings UI."""
base_url = str(base_url or "").strip()
@@ -712,6 +773,7 @@ def fetch_cmhub_models(base_url, api_key, connect_timeout=10, read_timeout=30):
raise CMHubError("cmhub_not_configured", "请去⑤设置配置 cmhub Base URL")
if not api_key:
raise CMHubError("cmhub_not_configured", "请去⑤设置配置 cmhub API Key")
_apply_cmhub_proxy(use_system_proxy)
data = _cmhub_call_with_retry(
"GET",
appconfig.cmhub_request_url(base_url, "/api/v1/models"),
@@ -728,7 +790,7 @@ def fetch_cmhub_models(base_url, api_key, connect_timeout=10, read_timeout=30):
return [copy.deepcopy(model) for model in models if isinstance(model, dict)]
def fetch_cmhub_balance(base_url, api_key, connect_timeout=10, read_timeout=30):
def fetch_cmhub_balance(base_url, api_key, connect_timeout=10, read_timeout=30, use_system_proxy=False):
"""Fetch cmhub point balance for settings UI."""
base_url = str(base_url or "").strip()
@@ -737,6 +799,7 @@ def fetch_cmhub_balance(base_url, api_key, connect_timeout=10, read_timeout=30):
raise CMHubError("cmhub_not_configured", "请去⑤设置配置 cmhub Base URL")
if not api_key:
raise CMHubError("cmhub_not_configured", "请去⑤设置配置 cmhub API Key")
_apply_cmhub_proxy(use_system_proxy)
data = _cmhub_call_with_retry(
"GET",
appconfig.cmhub_request_url(base_url, "/api/v1/balance"),
@@ -894,6 +957,15 @@ def _request_cmhub_cover_image(
"cover_request",
detail="cmhub 已返回 image_url，耗时 %s" % _format_seconds(request_elapsed),
)
if _debug_cmhub_image_url_enabled():
_notify_step_event(
on_step,
"cover_image_url",
result="debug",
detail="cmhub 图片 URL：%s" % image_url,
level="warning",
debug_only=True,
)
return {
"image_url": image_url,
"connect_timeout": runtime["connect_timeout"],
@@ -912,13 +984,12 @@ def _download_and_save_cmhub_cover(request_result, on_step=None):
resolution = request_result["resolution"]
quality = request_result["quality"]
_notify_step(on_step, "cover_download")
download_started = time.perf_counter()
image_bytes = _download_cmhub_image(
image_bytes, download_elapsed = _download_cmhub_image_with_retry(
image_url,
connect_timeout=connect_timeout,
read_timeout=read_timeout,
on_step=on_step,
)
download_elapsed = time.perf_counter() - download_started
_notify_step_event(
on_step,
"cover_download",
@@ -955,11 +1026,14 @@ def _cmhub_runtime(config, operation, cmhub_config_path):
"请去⑤设置配置 cmhub：缺少 " + "、".join(missing),
retryable=False,
)
use_system_proxy = bool(hub.get("use_system_proxy", False))
_apply_cmhub_proxy(use_system_proxy)
return {
"base_url": hub["base_url"].rstrip("/"),
"api_key": api_key,
"alias": hub[alias_key],
"connect_timeout": max(1, int(hub.get("connect_timeout", 10) or 10)),
"use_system_proxy": use_system_proxy,
}
@@ -992,6 +1066,73 @@ def _cmhub_read_timeout(config, resolution):
return CMHUB_IMAGE_READ_TIMEOUT_SECONDS
def _debug_cmhub_image_url_enabled():
value = str(os.environ.get("CMSHOPEE_DEBUG_CMHUB_IMAGE_URL", "") or "")
return value.strip().lower() in {"1", "true", "yes", "on", "debug"}
def _download_cmhub_image_with_retry(
url,
connect_timeout,
read_timeout,
on_step=None,
attempts=CMHUB_IMAGE_DOWNLOAD_ATTEMPTS,
slow_threshold=CMHUB_IMAGE_SLOW_DOWNLOAD_SECONDS,
):
total_attempts = max(1, int(attempts or 1))
total_started = time.perf_counter()
last_exc = None
for index in range(total_attempts):
try:
image_bytes = _download_cmhub_image(
url,
connect_timeout=connect_timeout,
read_timeout=read_timeout,
)
elapsed = time.perf_counter() - total_started
if elapsed >= float(slow_threshold or 0):
_notify_step_event(
on_step,
"cover_download",
result="warning",
detail="图片下载较慢，已用 %s，大小 %s"
% (_format_seconds(elapsed), _format_bytes(len(image_bytes))),
level="warning",
)
return image_bytes, elapsed
except Exception as exc:
last_exc = exc
if index + 1 >= total_attempts or not _cmhub_download_retryable(exc):
break
_notify_step_event(
on_step,
"cover_download",
result="retry",
detail=str(exc),
level="warning",
attempt=index + 1,
attempts=total_attempts,
)
time.sleep(min(2.0, 0.5 * (index + 1)))
if total_attempts > 1 and _cmhub_download_retryable(last_exc):
raise AIError(
"下载 cmhub 图片失败（已尝试 %s 次）: %s"
% (total_attempts, str(last_exc))
) from last_exc
raise last_exc
def _cmhub_download_retryable(exc):
if exc is None:
return False
message = str(exc or "")
if "图片超过大小上限" in message:
return False
if "HTTP 4" in message:
return False
return message.startswith("下载 cmhub 图片失败")
def _cmhub_call_with_retry(
method,
url,
@@ -1039,7 +1180,7 @@ def _cmhub_call_once(method, url, api_key, payload, connect_timeout, read_timeou
if str(method).upper() != "GET":
request_kwargs["json"] = payload or {}
try:
response = requests.request(str(method).upper(), url, **request_kwargs)
response = _cmhub_session().request(str(method).upper(), url, **request_kwargs)
except requests.exceptions.ConnectTimeout as exc:
raise CMHubError("connect_timeout", "连接 cmhub 超时", retryable=True) from exc
except requests.exceptions.ReadTimeout as exc:
@@ -1208,7 +1349,7 @@ def _format_bytes(size):
def _download_cmhub_image(url, connect_timeout, read_timeout, max_bytes=CMHUB_IMAGE_MAX_BYTES):
_assert_public_http_url(url)
try:
response = requests.get(
response = _cmhub_session().get(
url,
stream=True,
timeout=(max(1, int(connect_timeout)), max(1, int(read_timeout))),
@@ -1379,6 +1520,7 @@ def _emit_generation_event(
attempt=None,
attempts=None,
metadata=None,
debug_only=None,
):
if callback is None:
return
@@ -1397,6 +1539,8 @@ def _emit_generation_event(
payload["attempts"] = attempts
if metadata is not None:
payload["metadata"] = appconfig.sanitize_for_log(metadata)
if debug_only:
payload["debug_only"] = True
try:
callback(payload)
except Exception:
app/appconfig.py
+1
View File
@@ -78,6 +78,7 @@ DEFAULT_CONFIG = {
"title_alias": "",
"image_alias": "",
"connect_timeout": 10,
"use_system_proxy": False,
"check_balance_before_batch": False,
},
"title_concurrency": 4,