Add Luxsin curve API client and validation functionality
- Introduced a new module `curve_client.py` for interacting with the Luxsin curve API, including functions for custom Base64 decoding and parametric EQ validation. - Implemented error handling for API responses and validation checks for curve data. - Removed outdated test files to streamline the codebase. - Updated brand creation and update logic in `brands.py` to ensure brand names are validated and checked for duplicates. - Enhanced model handling in `models.py` with new validation and push-to-search functionality, integrating curve validation before pushing to Meilisearch. - Updated frontend components to support new validation and push processes, including user feedback for validation results and progress tracking. - Changed favicon to SVG format for better scalability and appearance.
This commit is contained in:
@@ -0,0 +1,148 @@
|
||||
"""Luxsin 曲线 API:拉取、自定义 Base64 解码与 parametric_eq 校验。"""
|
||||
import base64
|
||||
import json
|
||||
import logging
|
||||
from typing import Any, Optional, Tuple
|
||||
from urllib.parse import urlencode
|
||||
|
||||
import requests
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
LUXSIN_CURVE_API_BASE = "https://api.luxsin.com.cn/audio/getCurve"
|
||||
|
||||
CUSTOM_CHARS = "KLMPQRSTUVWXYZABCGHdefIJjkNOlmnopqrstuvwxyzabcghiDEF34501289+67/"
|
||||
STANDARD_CHARS = "ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789+/"
|
||||
|
||||
TARGET_OVER_EAR = "Harman over-ear 2018"
|
||||
TARGET_IN_EAR = "Harman in-ear 2019"
|
||||
|
||||
FORM_TARGET_MAP = {
|
||||
"over-ear": TARGET_OVER_EAR,
|
||||
"in-ear": TARGET_IN_EAR,
|
||||
}
|
||||
|
||||
|
||||
def curve_target_for_form(form: Optional[str]) -> Optional[str]:
|
||||
if not form:
|
||||
return None
|
||||
return FORM_TARGET_MAP.get(form.strip().lower())
|
||||
|
||||
|
||||
def custom_base64_to_string(encoded: str) -> str:
|
||||
"""将 Luxsin 自定义 Base64 字母表映射为标准 Base64 后解码为 UTF-8 字符串。"""
|
||||
if not encoded or not isinstance(encoded, str):
|
||||
raise ValueError("曲线数据为空")
|
||||
|
||||
standard_b64 = []
|
||||
for c in encoded:
|
||||
idx = CUSTOM_CHARS.find(c)
|
||||
if idx != -1:
|
||||
standard_b64.append(STANDARD_CHARS[idx])
|
||||
else:
|
||||
standard_b64.append(c)
|
||||
|
||||
try:
|
||||
raw = base64.b64decode("".join(standard_b64), validate=False)
|
||||
except Exception as e:
|
||||
raise ValueError(f"Base64 解码失败:{e}") from e
|
||||
|
||||
try:
|
||||
return raw.decode("utf-8")
|
||||
except UnicodeDecodeError as e:
|
||||
raise ValueError(f"UTF-8 解码失败:{e}") from e
|
||||
|
||||
|
||||
def is_valid_parametric_eq_payload(data: Any) -> bool:
|
||||
if not isinstance(data, dict):
|
||||
return False
|
||||
peq = data.get("parametric_eq")
|
||||
if not isinstance(peq, dict):
|
||||
return False
|
||||
filters = peq.get("filters")
|
||||
return isinstance(filters, list) and len(filters) == 10
|
||||
|
||||
|
||||
def extract_encoded_payload(response: requests.Response) -> str:
|
||||
"""从 getCurve 响应中提取待解码字符串。"""
|
||||
text = (response.text or "").strip()
|
||||
if not text:
|
||||
raise ValueError("曲线接口响应为空")
|
||||
|
||||
try:
|
||||
body = response.json()
|
||||
except ValueError:
|
||||
return text
|
||||
|
||||
if isinstance(body, str):
|
||||
return body.strip()
|
||||
|
||||
if not isinstance(body, dict):
|
||||
raise ValueError("曲线接口响应格式异常")
|
||||
|
||||
for key in ("data", "curve", "result", "content", "body"):
|
||||
val = body.get(key)
|
||||
if isinstance(val, str) and val.strip():
|
||||
return val.strip()
|
||||
|
||||
nested = body.get("data")
|
||||
if isinstance(nested, dict):
|
||||
for key in ("curve", "data", "content", "encoded"):
|
||||
val = nested.get(key)
|
||||
if isinstance(val, str) and val.strip():
|
||||
return val.strip()
|
||||
|
||||
if isinstance(nested, str) and nested.strip():
|
||||
return nested.strip()
|
||||
|
||||
raise ValueError("曲线接口响应中未找到可解码数据")
|
||||
|
||||
|
||||
def fetch_and_validate_curve(
|
||||
brand: str,
|
||||
name: str,
|
||||
form: str,
|
||||
*,
|
||||
timeout: float = 20.0,
|
||||
) -> Tuple[bool, str]:
|
||||
"""
|
||||
拉取并校验曲线。返回 (是否通过, 失败原因);通过时原因为空字符串。
|
||||
"""
|
||||
brand = (brand or "").strip()
|
||||
name = (name or "").strip()
|
||||
if not brand or not name:
|
||||
return False, "品牌名称或型号名称为空"
|
||||
|
||||
target = curve_target_for_form(form)
|
||||
if not target:
|
||||
return False, "佩戴方式须为入耳式(in-ear)或头戴式(over-ear)才能校验曲线"
|
||||
|
||||
query = urlencode({"brand": brand, "name": name, "target": target})
|
||||
url = f"{LUXSIN_CURVE_API_BASE}?{query}"
|
||||
|
||||
try:
|
||||
resp = requests.get(url, timeout=timeout)
|
||||
except requests.RequestException as e:
|
||||
logger.warning("getCurve request failed: %s %s", url, e)
|
||||
return False, f"无法连接曲线接口:{e}"
|
||||
|
||||
if resp.status_code != 200:
|
||||
return False, f"曲线接口返回 HTTP {resp.status_code}"
|
||||
|
||||
try:
|
||||
encoded = extract_encoded_payload(resp)
|
||||
except ValueError as e:
|
||||
return False, str(e)
|
||||
|
||||
try:
|
||||
decoded_text = custom_base64_to_string(encoded)
|
||||
payload = json.loads(decoded_text)
|
||||
except json.JSONDecodeError:
|
||||
return False, "解码后的数据不是合法 JSON"
|
||||
except ValueError as e:
|
||||
return False, str(e)
|
||||
|
||||
if not is_valid_parametric_eq_payload(payload):
|
||||
return False, "曲线数据异常"
|
||||
|
||||
return True, ""
|
||||
Reference in New Issue
Block a user