diff --git a/src/core/upload/cpa_upload.py b/src/core/upload/cpa_upload.py
index d61758c..900cff6 100644
--- a/src/core/upload/cpa_upload.py
+++ b/src/core/upload/cpa_upload.py
@@ -6,6 +6,7 @@ import json
import logging
from typing import List, Dict, Any, Tuple, Optional
from datetime import datetime
+from urllib.parse import quote
from curl_cffi import requests as cffi_requests
from curl_cffi import CurlMime
@@ -17,6 +18,77 @@ from ...config.settings import get_settings
logger = logging.getLogger(__name__)
+def _normalize_cpa_auth_files_url(api_url: str) -> str:
+ """将用户填写的 CPA 地址规范化为 auth-files 接口地址。"""
+ normalized = (api_url or "").strip().rstrip("/")
+ lower_url = normalized.lower()
+
+ if not normalized:
+ return ""
+
+ if lower_url.endswith("/auth-files"):
+ return normalized
+
+ if lower_url.endswith("/v0/management") or lower_url.endswith("/management"):
+ return f"{normalized}/auth-files"
+
+ if lower_url.endswith("/v0"):
+ return f"{normalized}/management/auth-files"
+
+ return f"{normalized}/v0/management/auth-files"
+
+
+def _build_cpa_headers(api_token: str, content_type: Optional[str] = None) -> dict:
+ headers = {
+ "Authorization": f"Bearer {api_token}",
+ }
+ if content_type:
+ headers["Content-Type"] = content_type
+ return headers
+
+
+def _extract_cpa_error(response) -> str:
+ error_msg = f"上传失败: HTTP {response.status_code}"
+ try:
+ error_detail = response.json()
+ if isinstance(error_detail, dict):
+ error_msg = error_detail.get("message", error_msg)
+ except Exception:
+ error_msg = f"{error_msg} - {response.text[:200]}"
+ return error_msg
+
+
+def _post_cpa_auth_file_multipart(upload_url: str, filename: str, file_content: bytes, api_token: str):
+ mime = CurlMime()
+ mime.addpart(
+ name="file",
+ data=file_content,
+ filename=filename,
+ content_type="application/json",
+ )
+
+ return cffi_requests.post(
+ upload_url,
+ multipart=mime,
+ headers=_build_cpa_headers(api_token),
+ proxies=None,
+ timeout=30,
+ impersonate="chrome110",
+ )
+
+
+def _post_cpa_auth_file_raw_json(upload_url: str, filename: str, file_content: bytes, api_token: str):
+ raw_upload_url = f"{upload_url}?name={quote(filename)}"
+ return cffi_requests.post(
+ raw_upload_url,
+ data=file_content,
+ headers=_build_cpa_headers(api_token, content_type="application/json"),
+ proxies=None,
+ timeout=30,
+ impersonate="chrome110",
+ )
+
+
def generate_token_json(account: Account) -> dict:
"""
生成 CPA 格式的 Token JSON
@@ -73,45 +145,35 @@ def upload_to_cpa(
if not effective_token:
return False, "CPA API Token 未配置"
- api_url = effective_url.rstrip("/")
- upload_url = f"{api_url}/v0/management/auth-files"
+ upload_url = _normalize_cpa_auth_files_url(effective_url)
filename = f"{token_data['email']}.json"
file_content = json.dumps(token_data, ensure_ascii=False, indent=2).encode("utf-8")
- headers = {
- "Authorization": f"Bearer {effective_token}",
- }
-
try:
- mime = CurlMime()
- mime.addpart(
- name="file",
- data=file_content,
- filename=filename,
- content_type="application/json",
- )
-
- response = cffi_requests.post(
+ response = _post_cpa_auth_file_multipart(
upload_url,
- multipart=mime,
- headers=headers,
- proxies=None,
- timeout=30,
- impersonate="chrome110",
+ filename,
+ file_content,
+ effective_token,
)
if response.status_code in (200, 201):
return True, "上传成功"
- error_msg = f"上传失败: HTTP {response.status_code}"
- try:
- error_detail = response.json()
- if isinstance(error_detail, dict):
- error_msg = error_detail.get("message", error_msg)
- except Exception:
- error_msg = f"{error_msg} - {response.text[:200]}"
- return False, error_msg
+ if response.status_code in (404, 405, 415):
+ logger.warning("CPA multipart 上传失败，尝试原始 JSON 回退: %s", response.status_code)
+ fallback_response = _post_cpa_auth_file_raw_json(
+ upload_url,
+ filename,
+ file_content,
+ effective_token,
+ )
+ if fallback_response.status_code in (200, 201):
+ return True, "上传成功"
+ response = fallback_response
+
+ return False, _extract_cpa_error(response)
except Exception as e:
logger.error(f"CPA 上传异常: {e}")
@@ -217,12 +279,11 @@ def test_cpa_connection(api_url: str, api_token: str, proxy: str = None) -> Tupl
if not api_token:
return False, "API Token 不能为空"
- api_url = api_url.rstrip("/")
- test_url = f"{api_url}/v0/management/auth-files"
- headers = {"Authorization": f"Bearer {api_token}"}
+ test_url = _normalize_cpa_auth_files_url(api_url)
+ headers = _build_cpa_headers(api_token)
try:
- response = cffi_requests.options(
+ response = cffi_requests.get(
test_url,
headers=headers,
proxies=None,
@@ -230,10 +291,16 @@ def test_cpa_connection(api_url: str, api_token: str, proxy: str = None) -> Tupl
impersonate="chrome110",
)
- if response.status_code in (200, 204, 401, 403, 405):
- if response.status_code == 401:
- return False, "连接成功，但 API Token 无效"
+ if response.status_code == 200:
return True, "CPA 连接测试成功"
+ if response.status_code == 401:
+ return False, "连接成功，但 API Token 无效"
+ if response.status_code == 403:
+ return False, "连接成功，但服务端未启用远程管理或当前 Token 无权限"
+ if response.status_code == 404:
+ return False, "未找到 CPA auth-files 接口，请检查 API URL 是否填写为根地址、/v0/management 或完整 auth-files 地址"
+ if response.status_code == 503:
+ return False, "连接成功，但服务端认证管理器不可用"
return False, f"服务器返回异常状态码: {response.status_code}"
diff --git a/templates/settings.html b/templates/settings.html
index 2cd20b8..e925ce3 100644
--- a/templates/settings.html
+++ b/templates/settings.html
@@ -389,6 +389,7 @@
diff --git a/tests/test_cpa_upload.py b/tests/test_cpa_upload.py
new file mode 100644
index 0000000..82125bc
--- /dev/null
+++ b/tests/test_cpa_upload.py
@@ -0,0 +1,110 @@
+from src.core.upload import cpa_upload
+
+
+class FakeResponse:
+ def __init__(self, status_code=200, payload=None, text=""):
+ self.status_code = status_code
+ self._payload = payload
+ self.text = text
+
+ def json(self):
+ if self._payload is None:
+ raise ValueError("no json payload")
+ return self._payload
+
+
+class FakeMime:
+ def __init__(self):
+ self.parts = []
+
+ def addpart(self, **kwargs):
+ self.parts.append(kwargs)
+
+
+def test_upload_to_cpa_accepts_management_root_url(monkeypatch):
+ calls = []
+
+ def fake_post(url, **kwargs):
+ calls.append({"url": url, "kwargs": kwargs})
+ return FakeResponse(status_code=201)
+
+ monkeypatch.setattr(cpa_upload, "CurlMime", FakeMime)
+ monkeypatch.setattr(cpa_upload.cffi_requests, "post", fake_post)
+
+ success, message = cpa_upload.upload_to_cpa(
+ {"email": "tester@example.com"},
+ api_url="https://cpa.example.com/v0/management",
+ api_token="token-123",
+ )
+
+ assert success is True
+ assert message == "上传成功"
+ assert calls[0]["url"] == "https://cpa.example.com/v0/management/auth-files"
+
+
+def test_upload_to_cpa_does_not_double_append_full_endpoint(monkeypatch):
+ calls = []
+
+ def fake_post(url, **kwargs):
+ calls.append({"url": url, "kwargs": kwargs})
+ return FakeResponse(status_code=201)
+
+ monkeypatch.setattr(cpa_upload, "CurlMime", FakeMime)
+ monkeypatch.setattr(cpa_upload.cffi_requests, "post", fake_post)
+
+ success, _ = cpa_upload.upload_to_cpa(
+ {"email": "tester@example.com"},
+ api_url="https://cpa.example.com/v0/management/auth-files",
+ api_token="token-123",
+ )
+
+ assert success is True
+ assert calls[0]["url"] == "https://cpa.example.com/v0/management/auth-files"
+
+
+def test_upload_to_cpa_falls_back_to_raw_json_when_multipart_returns_404(monkeypatch):
+ calls = []
+ responses = [
+ FakeResponse(status_code=404, text="404 page not found"),
+ FakeResponse(status_code=200, payload={"status": "ok"}),
+ ]
+
+ def fake_post(url, **kwargs):
+ calls.append({"url": url, "kwargs": kwargs})
+ return responses.pop(0)
+
+ monkeypatch.setattr(cpa_upload, "CurlMime", FakeMime)
+ monkeypatch.setattr(cpa_upload.cffi_requests, "post", fake_post)
+
+ success, message = cpa_upload.upload_to_cpa(
+ {"email": "tester@example.com", "type": "codex"},
+ api_url="https://cpa.example.com",
+ api_token="token-123",
+ )
+
+ assert success is True
+ assert message == "上传成功"
+ assert calls[0]["kwargs"]["multipart"] is not None
+ assert calls[1]["url"] == "https://cpa.example.com/v0/management/auth-files?name=tester%40example.com.json"
+ assert calls[1]["kwargs"]["headers"]["Content-Type"] == "application/json"
+ assert calls[1]["kwargs"]["data"].startswith(b"{")
+
+
+def test_test_cpa_connection_uses_get_and_normalized_url(monkeypatch):
+ calls = []
+
+ def fake_get(url, **kwargs):
+ calls.append({"url": url, "kwargs": kwargs})
+ return FakeResponse(status_code=200, payload={"files": []})
+
+ monkeypatch.setattr(cpa_upload.cffi_requests, "get", fake_get)
+
+ success, message = cpa_upload.test_cpa_connection(
+ "https://cpa.example.com/v0/management",
+ "token-123",
+ )
+
+ assert success is True
+ assert message == "CPA 连接测试成功"
+ assert calls[0]["url"] == "https://cpa.example.com/v0/management/auth-files"
+ assert calls[0]["kwargs"]["headers"]["Authorization"] == "Bearer token-123"
diff --git a/tests/test_static_asset_versioning.py b/tests/test_static_asset_versioning.py
new file mode 100644
index 0000000..ba23cf9
--- /dev/null
+++ b/tests/test_static_asset_versioning.py
@@ -0,0 +1,28 @@
+from pathlib import Path
+import importlib
+
+web_app = importlib.import_module("src.web.app")
+
+
+def test_static_asset_version_is_non_empty_string():
+ version = web_app._build_static_asset_version(web_app.STATIC_DIR)
+
+ assert isinstance(version, str)
+ assert version
+ assert version.isdigit()
+
+
+def test_email_services_template_uses_versioned_static_assets():
+ template = Path("templates/email_services.html").read_text(encoding="utf-8")
+
+ assert '/static/css/style.css?v={{ static_version }}' in template
+ assert '/static/js/utils.js?v={{ static_version }}' in template
+ assert '/static/js/email_services.js?v={{ static_version }}' in template
+
+
+def test_index_template_uses_versioned_static_assets():
+ template = Path("templates/index.html").read_text(encoding="utf-8")
+
+ assert '/static/css/style.css?v={{ static_version }}' in template
+ assert '/static/js/utils.js?v={{ static_version }}' in template
+ assert '/static/js/app.js?v={{ static_version }}' in template