refactor: 收敛多处平行维护为单一事实来源，并修复审计发现的 P3/P4 问题

.dockerignore

@@ -5,3 +5,12 @@ tests
 docs
 __pycache__
 *.pyc
+.venv
+*.egg-info
+.pytest_cache
+.ruff_cache
+.coverage
+htmlcov
+*.db
+browser_profile
+uv.lock

.env.example

@@ -57,10 +57,16 @@ DEFAULT_CRAWL_JITTER_MINUTES=10
 # Docker 端口绑定地址：本机使用 127.0.0.1；NAS 需要局域网访问时可填 NAS 局域网 IP（不建议无认证暴露到公网）
 BIND_ADDRESS=127.0.0.1
 
-# SQLite 数据库文件路径
+# Docker 镜像标签：本地构建时用 second-eye:local；也可指向已发布镜像
+SECOND_EYE_IMAGE=second-eye:local
+
+# SQLite 数据库文件路径（Docker 部署时会被 compose.yml 的 environment 覆盖为
+# sqlite:////app/data/goodprice.db，容器内改这里不生效）
 DATABASE_URL=sqlite:///data/goodprice.db
 
 # Docker noVNC：默认关闭。临时登录时设置 ENABLE_NOVNC=1，并在 data/.novnc-password
 # 写入独立密码；登录完成后恢复为 0 并重启容器。
 ENABLE_NOVNC=0
 NOVNC_PASSWORD_FILE=/app/data/.novnc-password
+# noVNC 对外端口（compose 端口映射与登录页链接共用；需与 BIND_ADDRESS 一起对外可达）
+NOVNC_PORT=16080

README.md

@@ -82,7 +82,7 @@ Linux/macOS/NAS 可以使用仓库提供的跨平台辅助脚本：
 ./scripts/start-docker.sh --login
 ~~~
 
-脚本会生成 data/.novnc-password，启动容器并提示访问地址。默认访问：
+脚本会生成 data/.novnc-password，启动容器并提示访问地址。默认访问（端口可用 NOVNC_PORT 修改）：
 
 ~~~text
 http://127.0.0.1:16080/vnc.html

compose.yml

@@ -13,7 +13,7 @@ services:
     ports:
       # Bind to localhost by default: the app has no built-in authentication.
       - "${BIND_ADDRESS:-127.0.0.1}:18000:8000"
-      - "${BIND_ADDRESS:-127.0.0.1}:16080:6080"
+      - "${BIND_ADDRESS:-127.0.0.1}:${NOVNC_PORT:-16080}:6080"
     environment:
       DATABASE_URL: sqlite:////app/data/goodprice.db
       PROXY: ${PROXY:-}
@@ -36,7 +36,7 @@ services:
       start_period: 30s
 
   gotify:
-    image: gotify/server:latest
+    image: gotify/server:3.1.1
     container_name: second-eye-gotify
     restart: unless-stopped
     ports:

goodprice/__init__.py

@@ -1,4 +1,9 @@
 """闲鱼盯价助手：本地 Web 工具。"""
 
-__version__ = "0.1.0"
+from importlib.metadata import PackageNotFoundError
+from importlib.metadata import version as _pkg_version
 
+try:  # 版本号只在 pyproject.toml 定义一处，运行时从包元数据读取。
+    __version__ = _pkg_version("good-price")
+except PackageNotFoundError:  # 源码直接运行、未安装时
+    __version__ = "0.0.0"

goodprice/analysis/jev_typesafe.py

@@ -6,13 +6,15 @@
 from typing import Any, Callable, Optional
 
 from goodprice.analysis.judge import RequirementVerdict
+from goodprice.analysis.prompts import REQUIREMENT_CRITERION, VALUE_CRITERION
+from goodprice.constants import SCORE_MAX, SCORE_MIN
 
 logger = logging.getLogger(__name__)
 
-VALUE_LEVELS = [f"{i} 分" for i in range(1, 11)]  # Score 评分为索引空间 [0, 9]
+VALUE_LEVELS = [f"{i} 分" for i in range(SCORE_MIN, SCORE_MAX + 1)]  # SDK 返回 0–9 索引，+1 映射到 1–10
 
-REQUIREMENT_NOUL_INSTRUCTION = "该二手商品是否满足买家的硬性需求？"
-VALUE_SCORE_INSTRUCTION = "该商品按当前价格是否划算，1 分最不划算，10 分最划算"
+REQUIREMENT_NOUL_INSTRUCTION = f"该二手{REQUIREMENT_CRITERION}？"
+VALUE_SCORE_INSTRUCTION = f"该商品{VALUE_CRITERION}，1 分最不划算，10 分最划算"
 
 
 @dataclass
@@ -28,7 +30,7 @@ def enabled(self) -> bool:
 
     def analyze_requirement(
         self, title: str, description: str = "", requirement: str = ""
-    ) -> dict[str, Any]:
+    ) -> RequirementVerdict:
         if not self.enabled:
             raise RuntimeError("TypeSafe API Key 未配置")
         state = (
@@ -66,7 +68,7 @@ def analyze_batch_value(self, items: list[dict[str, Any]]) -> dict[str, Any]:
                 logger.warning("TypeSafe 性价比评估失败，跳过 %s: %s", external_id, exc)
                 continue
             raw = float(answer.score)
-            scores[external_id] = min(10, max(1, int(round(raw)) + 1))
+            scores[external_id] = min(SCORE_MAX, max(SCORE_MIN, int(round(raw)) + 1))
             confidences[external_id] = float(getattr(answer, "confidence", 0.0))
             reasons[external_id] = f"官方 Jev Score={raw:.1f}"
         best = None

goodprice/analysis/json_extract.py

@@ -5,8 +5,8 @@
 的括号与转义）；比原先的 ``find("{")``/``rfind("}")`` 更稳：后者一旦输出里出现
 多个对象、或字符串中含有大括号，就会取错范围。
 
-标准解析仍失败时，若安装了可选依赖 ``json-repair``，则交给它修复畸形或截断的
-JSON（例如输出被 max_tokens 截断）。
+标准解析仍失败时，交给 ``json-repair`` 修复畸形或截断的 JSON（例如输出被
+max_tokens 截断）。
 """
 from __future__ import annotations
 
@@ -64,7 +64,7 @@ def _remove_trailing_commas(text: str) -> str:
 
 
 def _try_repair(text: str) -> dict[str, Any] | None:
-    """标准解析失败时，尝试用可选依赖 json-repair 修复畸形/截断的 JSON。"""
+    """标准解析失败时，尝试用 json-repair 修复畸形/截断的 JSON。"""
     try:
         from json_repair import repair_json
     except ImportError:

goodprice/analysis/judge.py

@@ -7,26 +7,29 @@
 
 from goodprice.analysis.json_extract import extract_json_object
 from goodprice.analysis.llm import LLMClient
+from goodprice.analysis.parse_utils import clamp_score
+from goodprice.analysis.prompts import (
+    REQUIREMENT_CRITERION,
+    REQUIREMENT_SYSTEM_PREFIX,
+    VALUE_CRITERION,
+)
+from goodprice.analysis.prompts import (
+    REQUIREMENT_USER_TEMPLATE as REQUIREMENT_TYPED_USER_TEMPLATE,
+)
+from goodprice.constants import RISK_UNKNOWN
 
 logger = logging.getLogger(__name__)
 
 REQUIREMENT_TYPED_SYSTEM_PROMPT = (
-    "你是二手商品筛选助手。用户给出商品标题、卖家描述和买家需求，"
-    "请判断商品是否满足买家的硬性需求。只输出 JSON，不要输出其它文字，格式："
+    f"{REQUIREMENT_SYSTEM_PREFIX}"
+    f"请判断{REQUIREMENT_CRITERION}。只输出 JSON，不要输出其它文字，格式："
     '{"matched": true或false, "probability": 0到1的小数（该判断成立的概率，越接近1越确定）, '
     '"reason": "一句话理由"}'
 )
 
-REQUIREMENT_TYPED_USER_TEMPLATE = (
-    "商品标题：{title}\n"
-    "卖家描述：{description}\n"
-    "买家需求：{requirement}\n"
-    "请给出 JSON 结论。"
-)
-
 VALUE_TYPED_SYSTEM_PROMPT = (
     "你是二手商品性价比评估专家。用户给出单个商品（标题、价格、品相分、瑕疵、卖家风险）"
-    "与买家品相要求，请仅依据该商品本身独立评估按当前价格是否划算。只输出 JSON，不要输出其它文字，格式："
+    f"与买家品相要求，请仅依据该商品本身独立评估{VALUE_CRITERION}。只输出 JSON，不要输出其它文字，格式："
     '{"value_score": 1到10的整数（越高越划算）, "probability": 0到1的小数（该评分的可信度）, '
     '"reason": "一句话理由"}'
 )
@@ -54,7 +57,7 @@ def parse_requirement_typed(raw: str) -> dict[str, Any]:
 def parse_value_typed(raw: str) -> dict[str, Any]:
     data = extract_json_object(raw)
     try:
-        score = max(1, min(10, int(data.get("value_score", 0))))
+        score = clamp_score(data.get("value_score", 0))
     except (TypeError, ValueError):
         raise ValueError(f"类型化性价比缺少整数 value_score: {raw!r}") from None
     return {
@@ -127,7 +130,7 @@ def analyze_batch_value(self, items: list[dict[str, Any]]) -> dict[str, Any]:
                 price=it.get("price"),
                 condition_score=it.get("condition_score") or "未评估",
                 defects="、".join(str(d) for d in (it.get("defects") or [])[:5]) or "无",
-                seller_risk=it.get("seller_risk") or "未知",
+                seller_risk=it.get("seller_risk") or RISK_UNKNOWN,
             )
             try:
                 data = self._ask(VALUE_TYPED_SYSTEM_PROMPT, text, parse_value_typed)
@@ -151,8 +154,8 @@ def analyze_batch_value(self, items: list[dict[str, Any]]) -> dict[str, Any]:
         }
 
     def _ask(self, system: str, text: str, parser) -> dict[str, Any]:
-        payload = self.llm._payload([{"type": "text", "text": text}], system=system)
-        return self.llm._complete(payload, parser=parser)
+        payload = self.llm.payload([{"type": "text", "text": text}], system=system)
+        return self.llm.complete(payload, parser=parser)
 
 
 def build_judger(

goodprice/analysis/llm.py

@@ -6,6 +6,7 @@
 import httpx
 
 from goodprice.analysis.json_extract import extract_json_object
+from goodprice.analysis.parse_utils import clamp_score
 from goodprice.analysis.prompts import (
     BATCH_VALUE_SYSTEM_PROMPT,
     BATCH_VALUE_USER_TEMPLATE,
@@ -31,7 +32,7 @@
 
 def parse_analysis_json(raw: str) -> dict[str, Any]:
     data = extract_json_object(raw)
-    score = max(1, min(10, int(data.get("condition_score", 0))))
+    score = clamp_score(data.get("condition_score", 0))
     defects = [str(d) for d in data.get("defects", [])][:10]
     return {
         "condition_score": score,
@@ -61,7 +62,7 @@ def parse_batch_value_json(raw: str) -> dict[str, Any]:
         item_id = str(it.get("id", "")).strip()
         if not item_id:
             continue
-        score = max(1, min(10, int(it.get("value_score", 0))))
+        score = clamp_score(it.get("value_score", 0))
         scores[item_id] = score
         reasons[item_id] = str(it.get("reason", ""))[:200]
     best = str(data.get("best", "")).strip()
@@ -156,7 +157,7 @@ def analyze_batch_value(
             raise RuntimeError("LLM 未配置")
         if not items:
             return {"scores": {}, "best": None, "reasons": {}}
-        requirement = str(items[0].get("requirement") or "") if items else ""
+        requirement = str(items[0].get("requirement") or "")
         lines = []
         for i, it in enumerate(items, 1):
             defects = "、".join(str(d) for d in (it.get("defects") or [])[:5]) or "无"
@@ -187,6 +188,14 @@ def _payload(
             "temperature": 0.2,
         }
 
+    def payload(self, content: list[dict[str, Any]], system: str = "") -> dict[str, Any]:
+        """公开请求体构造，供判断层/路由复用，无需访问私有 ``_payload``。"""
+        return self._payload(content, system=system)
+
+    def complete(self, payload: dict[str, Any], parser=parse_analysis_json) -> dict[str, Any]:
+        """公开补全入口，按 ``api_format`` 自动选择 chat/responses。"""
+        return self._complete(payload, parser=parser)
+
     def _complete(
         self, payload: dict[str, Any], parser=parse_analysis_json
     ) -> dict[str, Any]:

goodprice/analysis/parse_utils.py

@@ -0,0 +1,9 @@
+"""解析结果的通用规整工具，避免各解析器重复 clamp 逻辑。"""
+from typing import Any
+
+from goodprice.constants import SCORE_MAX, SCORE_MIN
+
+
+def clamp_score(value: Any) -> int:
+    """把模型返回的评分规整到 [SCORE_MIN, SCORE_MAX]；非法值抛 ValueError。"""
+    return max(SCORE_MIN, min(SCORE_MAX, int(value)))

goodprice/analysis/prompts.py

@@ -1,3 +1,12 @@
+# 需求匹配与性价比的核心判据措辞：adapter（prompts/judge）与 typesafe
+# （jev_typesafe）两个判断后端共用，避免同一判据在多处各写一份而漂移。
+REQUIREMENT_CRITERION = "商品是否满足买家的硬性需求"
+VALUE_CRITERION = "按当前价格是否划算"
+
+# 需求判断系统提示前缀：adapter 与 typesafe 两个后端共用。
+REQUIREMENT_SYSTEM_PREFIX = "你是二手商品筛选助手。用户给出商品标题、卖家描述和买家需求，"
+
+
 CONDITION_SYSTEM_PROMPT = (
     "你是一位熟悉中国二手交易市场（闲鱼）的验货专家。用户给出商品标题、卖家描述和商品图片，"
     "请只依据图片与卖家描述判断商品实际品相（成色），并判断品相是否与卖家描述一致。"
@@ -16,8 +25,8 @@
 )
 
 REQUIREMENT_SYSTEM_PROMPT = (
-    "你是二手商品筛选助手。用户给出商品标题、卖家描述和买家需求，"
-    "请判断商品是否满足买家的硬性需求。只输出 JSON："
+    f"{REQUIREMENT_SYSTEM_PREFIX}"
+    f"请判断{REQUIREMENT_CRITERION}。只输出 JSON："
     '{"matched": true或false, "reason": "一句话理由"}'
 )
 
@@ -30,7 +39,8 @@
 
 BATCH_VALUE_SYSTEM_PROMPT = (
     "你是二手商品性价比分析专家。用户给出一批商品（标题、价格、品相分、瑕疵、卖家风险），"
-    "请在同一批内横向比较，判断每个商品按当前价格是否划算。只输出 JSON，不要输出其它文字，格式："
+    "请在同一批内横向比较，判断每个商品"
+    f"{VALUE_CRITERION}。只输出 JSON，不要输出其它文字，格式："
     '{"items": [{"id": "商品外部ID", "value_score": 1到10的整数（越高越划算）, "reason": "一句话理由"}], '
     '"best": "性价比最高商品的 id"}'
 )

goodprice/config.py

@@ -1,13 +1,38 @@
 from functools import lru_cache
 from pathlib import Path
 
+from pydantic import field_validator
 from pydantic_settings import BaseSettings, SettingsConfigDict
 
 PROJECT_ROOT = Path(__file__).resolve().parent.parent
 DEFAULT_DB = PROJECT_ROOT / "data" / "goodprice.db"
 
+# 敏感字段：设置页留空表示“保持原值”。保存逻辑与模板提示共用这一份清单，
+# 避免“留空白名单”在 routes.py 与 settings.html 两处各维护一份而漂移。
+SECRET_FIELDS: frozenset[str] = frozenset(
+    {
+        "llm_api_key",
+        "serverchan_sendkey",
+        "vision_api_key",
+        "wecom_webhook",
+        "feishu_webhook",
+        "feishu_secret",
+        "gotify_token",
+        "jev_api_key",
+    }
+)
+
+# Jev 后端可选值：保存校验与设置页下拉共用。
+JEV_BACKENDS: tuple[str, ...] = ("adapter", "typesafe")
+
 
 class Settings(BaseSettings):
+    """全部可配置字段的唯一事实来源。
+
+    运行时值（SettingsService）与部署模板（.env.example）都以本类为准；
+    新增字段后请同步 .env.example，tests/test_settings_consistency.py 会校验。
+    """
+
     model_config = SettingsConfigDict(
         env_file=PROJECT_ROOT / ".env",
         env_file_encoding="utf-8",
@@ -19,15 +44,15 @@ class Settings(BaseSettings):
     xianyu_cookie: str = ""
     llm_base_url: str = ""
     llm_api_key: str = ""
-    llm_model: str = "qwen-vl-max"
+    llm_model: str = "glm-4.7-flash"
     llm_api_format: str = "chat_completions"
     serverchan_sendkey: str = ""
     proxy: str = ""
     default_crawl_interval_minutes: int = 20
     default_crawl_jitter_minutes: int = 10
     vision_base_url: str = ""
     vision_api_key: str = ""
-    vision_model: str = "qwen-vl-max"
+    vision_model: str = "glm-4.6v-flash"
     vision_api_format: str = "chat_completions"
     wecom_webhook: str = ""
     feishu_webhook: str = ""
@@ -43,11 +68,18 @@ class Settings(BaseSettings):
     runtime_mode: str = "local"
     enable_novnc: bool = False
     novnc_password_file: str = "/app/data/.novnc-password"
+    novnc_port: int = 16080
     jev_enabled: bool = False
     jev_auto_threshold: float = 0.85
     jev_backend: str = "adapter"
     jev_api_key: str = ""
 
+    @field_validator("jev_backend")
+    @classmethod
+    def _normalize_jev_backend(cls, value: str) -> str:
+        """非法后端名回退到默认值，校验逻辑集中在 schema 一处。"""
+        return value if value in JEV_BACKENDS else JEV_BACKENDS[0]
+
 
 @lru_cache
 def get_settings() -> Settings:

goodprice/constants.py

@@ -0,0 +1,16 @@
+"""全局领域常量：评分刻度与卖家风险等级。
+
+这些字面量原先散落在 analysis 各解析器、services 与 web 路由中；集中定义后
+Python 侧共用同一份（模板仍按值渲染）。
+"""
+
+SCORE_MIN = 1
+SCORE_MAX = 10
+
+RISK_LOW = "低"
+RISK_MEDIUM = "中"
+RISK_HIGH = "高"
+RISK_UNKNOWN = "未知"
+
+# 卖家风险 → 满意度权重系数（未知按中等处理）
+RISK_FACTOR = {RISK_LOW: 1.0, RISK_MEDIUM: 0.5, RISK_HIGH: 0.0}