From bd348854b522f1a2e6be2794cf0f650076e313dc Mon Sep 17 00:00:00 2001 From: rockxsj Date: Thu, 19 Mar 2026 17:57:40 +0800 Subject: [PATCH] =?UTF-8?q?=E8=A7=A3=E5=86=B3cpa=E6=B5=8B=E8=AF=95?= =?UTF-8?q?=E9=80=9A=E8=BF=87=E5=AE=9E=E9=99=85=E4=B8=8A=E4=BC=A0=E5=A4=B1?= =?UTF-8?q?=E8=B4=A5=E7=9A=84=E9=97=AE=E9=A2=98?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- src/core/upload/cpa_upload.py | 137 +++++++++++++++++++------- templates/settings.html | 1 + tests/test_cpa_upload.py | 110 +++++++++++++++++++++ tests/test_static_asset_versioning.py | 28 ++++++ 4 files changed, 241 insertions(+), 35 deletions(-) create mode 100644 tests/test_cpa_upload.py create mode 100644 tests/test_static_asset_versioning.py 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 @@
+

支持填写根地址、`/v0/management` 或完整的 `/v0/management/auth-files` 地址

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