2 Commits

Author SHA1 Message Date
tliu93 70b0cf08ee Add AI search query expansion
test / pytest (push) Successful in 1m20s
docker-image / build-and-push (push) Successful in 5m6s
2026-06-01 21:28:29 +02:00
tliu93 d36b940981 Add LLM settings integration
test / pytest (push) Successful in 1m13s
Add app_settings migration, settings UI, and OpenAI-compatible httpx LLM client with mocked tests.

Preserve API keys on blank form submissions, require a fresh key when base_url changes, and keep AI search settings untouched for step 3.

Update docs/design LLM integration and step 3 AI search notes, including prompt contract and extra-hints planning.
2026-06-01 20:06:22 +02:00
15 changed files with 2273 additions and 93 deletions
+272
View File
@@ -0,0 +1,272 @@
"""LLM client module — all network egress is concentrated here.
Uses ``httpx`` (already in requirements) to call OpenAI-compatible endpoints.
No ``openai`` SDK dependency. Sync functions are fine: FastAPI runs sync
handlers in a threadpool.
Public API:
- ``is_configured(cfg)`` — returns True when the client can make calls.
- ``test_connection(cfg)`` — minimal request to verify credentials.
- ``expand_query(cfg, query)`` — query-term expansion (step 3 consumer).
Returns ``ExpansionResult`` with ``terms`` and optional ``error``.
- ``analyze_image(...)`` — **reserved stub, not implemented**.
All calls go through ``_call_chat_completion()`` so tests can mock a single
boundary.
"""
from __future__ import annotations
import json
import re
from dataclasses import dataclass
from typing import Any
import httpx
from app.settings_store import LLMConfig
# Sensible defaults
_TIMEOUT_SECONDS = 30
# ── Prompt for query expansion (Step 3) ──────────────────────────────────
_EXPAND_QUERY_SYSTEM_PROMPT = (
"你是搬家物品搜索助手。用户在搜索自己打包的箱子与物品(家居/搬家场景)。"
"给定一个搜索词,列出用户可能用来命名同一类物品的相关词:"
"近义词、常见别称、上位类别、具体品类。"
"规则:用与查询相同的语言;"
"只给与该物品紧密相关、有助于在清单里找到它的词;"
"不要解释、不要造无关词;最多 8 个;"
"只输出一个 JSON 字符串数组,例如 "
'`["炒锅","平底锅","汤锅","厨具"]`。'
)
@dataclass
class LLMResult:
"""Uniform result wrapper for LLM calls."""
success: bool
message: str
data: Any = None
@dataclass
class ExpansionResult:
"""Structured result from ``expand_query``.
``terms`` is always a list (may be empty).
``error`` is ``None`` on success (including legitimate empty results);
on failure (timeout, network error, HTTP error) it contains a
human-friendly error message.
"""
terms: list[str]
error: str | None = None
def is_configured(cfg: LLMConfig) -> bool:
"""Return True only when the LLM is enabled AND has required fields."""
return bool(cfg.enabled and cfg.model and cfg.api_key)
def test_connection(cfg: LLMConfig) -> LLMResult:
"""Send a minimal chat-completion request to verify the config.
Uses a tiny prompt to minimise cost. Returns an ``LLMResult`` indicating
success or failure with a human-readable message.
"""
if not is_configured(cfg):
return LLMResult(
success=False,
message="LLM 未配置或未启用(缺少 model 或 api_key)。",
)
try:
response = _call_chat_completion(
cfg,
messages=[{"role": "user", "content": "Hi"}],
max_tokens=1,
)
return LLMResult(
success=True,
message=f"连接成功(模型:{cfg.model})。",
data=response,
)
except httpx.HTTPStatusError as exc:
status = exc.response.status_code
return LLMResult(
success=False,
message=f"连接失败(HTTP {status})。请检查 base_url、model 和 api_key。",
)
except httpx.ConnectError:
return LLMResult(
success=False,
message="无法连接到服务器。请检查 base_url 是否正确。",
)
except httpx.TimeoutException:
return LLMResult(
success=False,
message="连接超时。请检查网络或 base_url 是否可达。",
)
except Exception as exc: # noqa: BLE001 — graceful degradation
return LLMResult(
success=False,
message=f"未知错误:{exc}",
)
def expand_query(
cfg: LLMConfig,
query: str,
extra_hints: str = "",
) -> ExpansionResult:
"""Expand a search query into multiple synonymous terms via LLM.
Returns an ``ExpansionResult``. On success ``terms`` contains the expanded
terms (possibly empty) and ``error`` is ``None``. On failure (network
error, timeout, HTTP error) ``terms`` is ``[]`` and ``error`` contains a
human-friendly message.
"""
if not is_configured(cfg):
return ExpansionResult(terms=[])
system_prompt = _EXPAND_QUERY_SYSTEM_PROMPT
if extra_hints and extra_hints.strip():
system_prompt += "\n" + extra_hints.strip()
try:
response = _call_chat_completion(
cfg,
messages=[
{"role": "system", "content": system_prompt},
{"role": "user", "content": query},
],
max_tokens=200,
temperature=0,
)
except httpx.TimeoutException:
return ExpansionResult(
terms=[],
error="AI 搜索请求超时,请稍后再试。",
)
except httpx.ConnectError:
return ExpansionResult(
terms=[],
error="无法连接到 AI 服务,请检查网络或设置。",
)
except httpx.HTTPStatusError:
return ExpansionResult(
terms=[],
error="AI 服务返回错误,请检查配置。",
)
except Exception: # noqa: BLE001 — graceful degradation
return ExpansionResult(
terms=[],
error="AI 搜索暂时不可用,请稍后再试。",
)
choices = response.get("choices", [])
if not choices:
return ExpansionResult(terms=[])
content = choices[0].get("message", {}).get("content", "")
return ExpansionResult(terms=_parse_json_string_array(content))
# ── Constants for output contract enforcement ────────────────────────────
_MAX_EXPANSION_TERMS = 8
_MAX_TERM_LENGTH = 30
def _parse_json_string_array(content: str) -> list[str]:
"""Parse LLM output into a list of strings.
Strict contract enforcement:
1. Strip markdown code fences;
2. Try ``json.loads`` — only accept a JSON **array of strings**;
3. Anything else (prose, JSON objects, bad JSON) → return ``[]``.
This ensures the output contract is enforced by code: no matter what
the model returns or what ``ai_search_extra_hints`` contains, only a
valid JSON string array is accepted.
"""
text = content.strip()
if not text:
return []
# Strip markdown code fences
text = re.sub(r"^```(?:json)?\s*", "", text)
text = re.sub(r"\s*```$", "", text)
text = text.strip()
# Attempt JSON parse — strictly require a list
try:
parsed = json.loads(text)
except (json.JSONDecodeError, ValueError):
return []
if not isinstance(parsed, list):
return []
# Validate every element is a string; reject non-string items
terms: list[str] = []
for item in parsed:
if not isinstance(item, str):
return []
cleaned = item.strip()
if cleaned and len(cleaned) <= _MAX_TERM_LENGTH:
terms.append(cleaned)
# Cap total count
return terms[:_MAX_EXPANSION_TERMS]
def analyze_image(cfg: LLMConfig, image_data: bytes, prompt: str) -> LLMResult:
"""Analyze an image via LLM vision API.
.. note:: **Reserved stub — not implemented.** Will be filled in a future
round for image analysis. The signature is fixed so callers can
depend on it.
"""
# TODO: Implement in future round for image analysis.
return LLMResult(
success=False,
message="图片分析功能尚未实现。",
)
# ------------------------------------------------------------------
# Internal boundary — all network calls go through this single function
# ------------------------------------------------------------------
def _call_chat_completion(
cfg: LLMConfig,
*,
messages: list[dict[str, str]],
max_tokens: int = 1,
temperature: float | None = None,
) -> dict:
"""Call the OpenAI-compatible ``/chat/completions`` endpoint.
Returns the parsed JSON response body on success (status 2xx).
Raises ``httpx.HTTPStatusError`` on non-2xx, or other ``httpx`` exceptions
on network failures — callers handle these for graceful degradation.
"""
url = cfg.base_url.rstrip("/") + "/chat/completions"
payload: dict[str, Any] = {
"model": cfg.model,
"messages": messages,
"max_tokens": max_tokens,
}
if temperature is not None:
payload["temperature"] = temperature
headers = {
"Authorization": f"Bearer {cfg.api_key}",
"Content-Type": "application/json",
}
with httpx.Client(timeout=_TIMEOUT_SECONDS) as client:
response = client.post(url, json=payload, headers=headers)
response.raise_for_status()
return response.json()
+298 -34
View File
@@ -5,12 +5,15 @@ from fastapi import Depends, FastAPI, File, Form, HTTPException, Request, Upload
from fastapi.responses import FileResponse, RedirectResponse, Response from fastapi.responses import FileResponse, RedirectResponse, Response
from fastapi.staticfiles import StaticFiles from fastapi.staticfiles import StaticFiles
from fastapi.templating import Jinja2Templates from fastapi.templating import Jinja2Templates
from sqlalchemy import func, or_ from sqlalchemy import func, false, or_
from sqlalchemy.orm import Session from sqlalchemy.orm import Session
from app.db import get_db, init_db from app.db import get_db, init_db
from app.images import process_upload from app.images import process_upload
from app.llm import expand_query, is_configured, test_connection
from app.llm import LLMResult
from app.models import Box, Item, SubItem from app.models import Box, Item, SubItem
from app.settings_store import LLMConfig, get_app_settings, save_app_settings
templates = Jinja2Templates(directory="app/templates") templates = Jinja2Templates(directory="app/templates")
STATIC_DIR = Path("app/static") STATIC_DIR = Path("app/static")
@@ -88,6 +91,46 @@ def _wants_add_next(submit_action: str | None) -> bool:
return submit_action == "save_and_add_next" return submit_action == "save_and_add_next"
def _validate_settings_origin(request: Request) -> str | None:
"""Check Origin/Referer for same-host browser requests.
Returns an error message if validation fails, or None if OK.
Missing both headers (e.g. curl, API call) is allowed for now.
"""
origin = request.headers.get("origin")
referer = request.headers.get("referer")
if origin:
host = request.headers.get("host", "")
# origin includes scheme, host only has host:port
from urllib.parse import urlparse
parsed = urlparse(origin)
origin_host = parsed.netloc
if origin_host != host:
return "请求来源与当前站点不一致,操作被拒绝。"
elif referer:
host = request.headers.get("host", "")
from urllib.parse import urlparse
parsed = urlparse(referer)
referer_host = parsed.netloc
if referer_host != host:
return "请求来源与当前站点不一致,操作被拒绝。"
return None
def _validate_base_url_scheme(base_url: str) -> str | None:
"""Return an error message if base_url scheme is not allowed, else None."""
from urllib.parse import urlparse
parsed = urlparse(base_url)
if parsed.scheme not in ("https", "http"):
return "Base URL 必须以 http:// 或 https:// 开头。"
return None
def _format_average(total: int, divisor: int) -> str: def _format_average(total: int, divisor: int) -> str:
if divisor == 0: if divisor == 0:
return "0.0" return "0.0"
@@ -117,24 +160,41 @@ def _build_boxes_overview_summary(db: Session) -> dict[str, int | str]:
} }
def _build_search_results(db: Session, query: str) -> list[dict]: def _build_search_results(db: Session, query: str | list[str]) -> list[dict]:
keyword = f"%{query.lower()}%" """Search Box / Item / SubItem by name and note using case-insensitive LIKE.
Accepts either a single query string or a list of keywords.
When multiple keywords are given, they are combined with OR — a match on
*any* keyword is sufficient.
"""
keywords = [query] if isinstance(query, str) else query
patterns = [f"%{kw.lower()}%" for kw in keywords]
def _or_like(column, note_column):
"""Build an OR filter that matches any pattern on either column."""
conditions = []
for pat in patterns:
conditions.append(func.lower(column).like(pat))
conditions.append(func.lower(func.coalesce(note_column, "")).like(pat))
return or_(false(), *conditions) if conditions else false()
results: list[dict] = [] results: list[dict] = []
seen_ids: set[tuple[str, int]] = set()
def _add(result_type: str, obj_id: int, entry: dict) -> None:
key = (result_type, obj_id)
if key not in seen_ids:
seen_ids.add(key)
results.append(entry)
box_matches = ( box_matches = (
db.query(Box) db.query(Box)
.filter( .filter(_or_like(Box.name, Box.note))
or_(
func.lower(Box.name).like(keyword),
func.lower(func.coalesce(Box.note, "")).like(keyword),
)
)
.order_by(Box.id.desc()) .order_by(Box.id.desc())
.all() .all()
) )
for box in box_matches: for box in box_matches:
results.append( _add("Box", box.id, {
{
"type": "Box", "type": "Box",
"name": box.name, "name": box.name,
"note": box.note, "note": box.note,
@@ -145,24 +205,17 @@ def _build_search_results(db: Session, query: str) -> list[dict]:
"path": "顶层箱子", "path": "顶层箱子",
"is_container": None, "is_container": None,
"image_url": f"/boxes/{box.id}/image" if box.image_blob else None, "image_url": f"/boxes/{box.id}/image" if box.image_blob else None,
} })
)
item_matches = ( item_matches = (
db.query(Item) db.query(Item)
.join(Item.box) .join(Item.box)
.filter( .filter(_or_like(Item.name, Item.note))
or_(
func.lower(Item.name).like(keyword),
func.lower(func.coalesce(Item.note, "")).like(keyword),
)
)
.order_by(Item.id.desc()) .order_by(Item.id.desc())
.all() .all()
) )
for item in item_matches: for item in item_matches:
results.append( _add("Item", item.id, {
{
"type": "Item", "type": "Item",
"name": item.name, "name": item.name,
"note": item.note, "note": item.note,
@@ -173,25 +226,18 @@ def _build_search_results(db: Session, query: str) -> list[dict]:
"path": f"位于箱子:{item.box.name}", "path": f"位于箱子:{item.box.name}",
"is_container": item.is_container, "is_container": item.is_container,
"image_url": f"/items/{item.id}/image" if item.image_blob else None, "image_url": f"/items/{item.id}/image" if item.image_blob else None,
} })
)
subitem_matches = ( subitem_matches = (
db.query(SubItem) db.query(SubItem)
.join(SubItem.parent_item) .join(SubItem.parent_item)
.join(Item.box) .join(Item.box)
.filter( .filter(_or_like(SubItem.name, SubItem.note))
or_(
func.lower(SubItem.name).like(keyword),
func.lower(func.coalesce(SubItem.note, "")).like(keyword),
)
)
.order_by(SubItem.id.desc()) .order_by(SubItem.id.desc())
.all() .all()
) )
for subitem in subitem_matches: for subitem in subitem_matches:
results.append( _add("SubItem", subitem.id, {
{
"type": "SubItem", "type": "SubItem",
"name": subitem.name, "name": subitem.name,
"note": subitem.note, "note": subitem.note,
@@ -205,12 +251,39 @@ def _build_search_results(db: Session, query: str) -> list[dict]:
), ),
"is_container": None, "is_container": None,
"image_url": f"/subitems/{subitem.id}/image" if subitem.image_blob else None, "image_url": f"/subitems/{subitem.id}/image" if subitem.image_blob else None,
} })
)
return results return results
def _ai_search(db: Session, cfg: "LLMConfig", query: str) -> tuple[list[str], list[dict], str | None]:
"""Swappable AI search seam.
Returns ``(expanded_terms, results, error_message)``.
- On success: expanded terms + broadened results, ``error_message`` is ``None``.
- On failure (timeout, network error, HTTP error): empty terms + normal LIKE
results + friendly error message.
- On empty expansion (model returned ``[]`` legitimately): empty terms + normal
results, ``error_message`` is ``None``.
"""
result = expand_query(cfg, query, extra_hints=cfg.ai_search_extra_hints)
if result.error:
# Real failure (timeout / network / HTTP) → show error + fallback
results = _build_search_results(db, query)
return [], results, result.error
if not result.terms:
# Legitimate empty expansion → normal results, no error
results = _build_search_results(db, query)
return [], results, None
# Deduplicate: original query + expanded terms
all_terms = list(dict.fromkeys([query] + result.terms))
results = _build_search_results(db, all_terms)
return result.terms, results, None
def create_app() -> FastAPI: def create_app() -> FastAPI:
@asynccontextmanager @asynccontextmanager
async def lifespan(app: FastAPI): async def lifespan(app: FastAPI):
@@ -242,10 +315,28 @@ def create_app() -> FastAPI:
def search_page( def search_page(
request: Request, request: Request,
q: str | None = None, q: str | None = None,
ai: str | None = None,
db: Session = Depends(get_db), db: Session = Depends(get_db),
): ):
query = (q or "").strip() query = (q or "").strip()
results = _build_search_results(db, query) if query else [] cfg = get_app_settings(db)
ai_requested = ai == "1"
ai_available = cfg.ai_search_enabled and is_configured(cfg)
expanded_terms: list[str] = []
ai_error: str | None = None
if query:
if ai_requested and ai_available:
try:
expanded_terms, results, ai_error = _ai_search(db, cfg, query)
except Exception: # noqa: BLE001 — graceful degradation
ai_error = "AI 搜索暂时不可用,已回退到普通搜索。"
results = _build_search_results(db, query)
else:
results = _build_search_results(db, query)
else:
results = []
return templates.TemplateResponse( return templates.TemplateResponse(
request=request, request=request,
name="search/index.html", name="search/index.html",
@@ -254,6 +345,10 @@ def create_app() -> FastAPI:
"query": query, "query": query,
"results": results, "results": results,
"searched": bool(query), "searched": bool(query),
"ai_activated": ai_requested and ai_available and bool(query),
"expanded_terms": expanded_terms,
"ai_error": ai_error,
"ai_available": ai_available,
}, },
) )
@@ -267,6 +362,175 @@ def create_app() -> FastAPI:
context={"page_title": "箱子", "boxes": boxes, "summary": summary}, context={"page_title": "箱子", "boxes": boxes, "summary": summary},
) )
# ------------------------------------------------------------------
# Settings
# ------------------------------------------------------------------
@app.get("/settings")
def settings_page(request: Request, db: Session = Depends(get_db)):
cfg = get_app_settings(db)
return templates.TemplateResponse(
request=request,
name="settings/form.html",
context={
"page_title": "设置",
"config": cfg,
"api_key_configured": bool(cfg.api_key),
"test_result": None,
},
)
@app.post("/settings")
def save_settings(
request: Request,
enabled: str | None = Form(default=None),
base_url: str | None = Form(default=None),
model: str | None = Form(default=None),
api_key: str | None = Form(default=None),
ai_search_enabled: str | None = Form(default=None),
ai_search_extra_hints: str | None = Form(default=None),
db: Session = Depends(get_db),
) -> Response:
# Origin/Referer check for browser requests
origin_error = _validate_settings_origin(request)
if origin_error:
raise HTTPException(status_code=403, detail=origin_error)
resolved_base_url = _clean_text(base_url) or "https://api.openai.com/v1"
# Validate base_url scheme
scheme_error = _validate_base_url_scheme(resolved_base_url)
if scheme_error:
raise HTTPException(status_code=400, detail=scheme_error)
resolved_model = _clean_text(model) or ""
# Only base_url change counts as an endpoint change — model switches
# under the same base_url do not require a new key.
existing_cfg = get_app_settings(db)
submitted_key = _clean_text(api_key)
base_url_changed = resolved_base_url != existing_cfg.base_url
if base_url_changed and submitted_key is None:
# base_url changed but no new key provided — refuse to save,
# return to settings page with a clear error message.
return templates.TemplateResponse(
request=request,
name="settings/form.html",
context={
"page_title": "设置",
"config": LLMConfig(
enabled=enabled == "on",
base_url=resolved_base_url,
model=resolved_model,
api_key=existing_cfg.api_key,
ai_search_enabled=ai_search_enabled == "on",
ai_search_extra_hints=_clean_text(ai_search_extra_hints) or "",
),
"api_key_configured": bool(existing_cfg.api_key),
"test_result": LLMResult(
success=False,
message="Base URL 已变更,请重新输入 API Key 后保存。",
),
},
)
# submitted_key is None → keep old key; str (including "") → use new value
resolved_api_key = submitted_key
resolved_extra_hints = _clean_text(ai_search_extra_hints) or ""
save_app_settings(
db,
enabled=enabled == "on",
base_url=resolved_base_url,
model=resolved_model,
api_key=resolved_api_key,
ai_search_enabled=ai_search_enabled == "on",
ai_search_extra_hints=resolved_extra_hints,
)
return RedirectResponse(url="/settings", status_code=status.HTTP_303_SEE_OTHER)
@app.post("/settings/test")
def test_settings_connection(
request: Request,
enabled: str | None = Form(default=None),
base_url: str | None = Form(default=None),
model: str | None = Form(default=None),
api_key: str | None = Form(default=None),
ai_search_enabled: str | None = Form(default=None),
ai_search_extra_hints: str | None = Form(default=None),
db: Session = Depends(get_db),
):
# Origin/Referer check for browser requests
origin_error = _validate_settings_origin(request)
if origin_error:
raise HTTPException(status_code=403, detail=origin_error)
resolved_base_url = _clean_text(base_url) or "https://api.openai.com/v1"
# Validate base_url scheme
scheme_error = _validate_base_url_scheme(resolved_base_url)
if scheme_error:
raise HTTPException(status_code=400, detail=scheme_error)
resolved_model = _clean_text(model) or ""
# Only reuse stored key if base_url matches saved config. Model switches
# under the same base_url can use the same key; a base_url change cannot.
existing_cfg = get_app_settings(db)
submitted_key = _clean_text(api_key)
base_url_matches = resolved_base_url == existing_cfg.base_url
if base_url_matches and submitted_key is None:
resolved_api_key = existing_cfg.api_key
elif submitted_key is not None:
resolved_api_key = submitted_key
else:
# base_url changed but no key provided — refuse to test
return templates.TemplateResponse(
request=request,
name="settings/form.html",
context={
"page_title": "设置",
"config": LLMConfig(
enabled=enabled == "on",
base_url=resolved_base_url,
model=resolved_model,
api_key="",
ai_search_enabled=ai_search_enabled == "on",
ai_search_extra_hints=_clean_text(ai_search_extra_hints) or "",
),
"api_key_configured": bool(existing_cfg.api_key),
"test_result": LLMResult(
success=False,
message="Base URL 已变更,请重新输入 API Key 后再测试。",
),
},
)
test_cfg = LLMConfig(
enabled=enabled == "on",
base_url=resolved_base_url,
model=resolved_model,
api_key=resolved_api_key or "",
ai_search_enabled=ai_search_enabled == "on",
ai_search_extra_hints=_clean_text(ai_search_extra_hints) or "",
)
result = test_connection(test_cfg)
return templates.TemplateResponse(
request=request,
name="settings/form.html",
context={
"page_title": "设置",
"config": test_cfg,
"api_key_configured": bool(test_cfg.api_key),
"test_result": result,
},
)
@app.get("/boxes/new") @app.get("/boxes/new")
def new_box_page(request: Request): def new_box_page(request: Request):
return templates.TemplateResponse( return templates.TemplateResponse(
+7
View File
@@ -90,3 +90,10 @@ class SubItem(Base):
) )
parent_item: Mapped[Item] = relationship(back_populates="subitems") parent_item: Mapped[Item] = relationship(back_populates="subitems")
class AppSetting(Base):
__tablename__ = "app_settings"
key: Mapped[str] = mapped_column(Text, primary_key=True)
value: Mapped[str | None] = mapped_column(Text, nullable=True)
+93
View File
@@ -0,0 +1,93 @@
"""Settings read/write helpers for the ``app_settings`` KV table.
Provides a typed ``LLMConfig`` dataclass and two helpers:
- ``get_app_settings(db) -> LLMConfig`` — reads KV rows (or returns defaults).
- ``save_app_settings(db, ...) -> None`` — writes KV rows; API key left blank
means "keep the old value".
"""
from __future__ import annotations
from dataclasses import dataclass
from sqlalchemy.orm import Session
from app.models import AppSetting
@dataclass
class LLMConfig:
"""All settings consumed by the LLM client and settings UI."""
enabled: bool = False
base_url: str = "https://api.openai.com/v1"
model: str = ""
api_key: str = ""
ai_search_enabled: bool = False
ai_search_extra_hints: str = ""
def _get_value(rows: dict[str, str], key: str, default: str) -> str:
return rows.get(key, default)
def _get_bool(rows: dict[str, str], key: str, default: bool) -> bool:
return rows.get(key, str(default).lower()) == "true"
def get_app_settings(db: Session) -> LLMConfig:
"""Read all settings from ``app_settings`` and return an ``LLMConfig``."""
rows: dict[str, str] = {}
for row in db.query(AppSetting).all():
if row.value is not None:
rows[row.key] = row.value
return LLMConfig(
enabled=_get_bool(rows, "llm_enabled", False),
base_url=_get_value(rows, "llm_base_url", "https://api.openai.com/v1"),
model=_get_value(rows, "llm_model", ""),
api_key=_get_value(rows, "llm_api_key", ""),
ai_search_enabled=_get_bool(rows, "ai_search_enabled", False),
ai_search_extra_hints=_get_value(rows, "ai_search_extra_hints", ""),
)
def save_app_settings(
db: Session,
*,
enabled: bool | None = None,
base_url: str | None = None,
model: str | None = None,
api_key: str | None = None,
ai_search_enabled: bool | None = None,
ai_search_extra_hints: str | None = None,
) -> None:
"""Write settings to ``app_settings``.
If ``api_key`` is ``None`` (form field left blank), the existing key is
preserved. All other fields are written as-is.
"""
updates: dict[str, str | None] = {}
if enabled is not None:
updates["llm_enabled"] = str(enabled).lower()
if base_url is not None:
updates["llm_base_url"] = base_url
if model is not None:
updates["llm_model"] = model
if api_key is not None:
updates["llm_api_key"] = api_key
if ai_search_enabled is not None:
updates["ai_search_enabled"] = str(ai_search_enabled).lower()
if ai_search_extra_hints is not None:
updates["ai_search_extra_hints"] = ai_search_extra_hints
for key, value in updates.items():
existing = db.get(AppSetting, key)
if existing is not None:
existing.value = value
else:
db.add(AppSetting(key=key, value=value))
db.commit()
+1
View File
@@ -19,6 +19,7 @@
<nav class="top-nav"> <nav class="top-nav">
<a href="/boxes">箱子</a> <a href="/boxes">箱子</a>
<a href="/search">搜索</a> <a href="/search">搜索</a>
<a href="/settings">设置</a>
</nav> </nav>
{% block content %}{% endblock %} {% block content %}{% endblock %}
</main> </main>
+24
View File
@@ -20,7 +20,31 @@
</form> </form>
</section> </section>
{% if query and ai_available %}
<section class="card" style="margin-top: 8px;">
{% if ai_activated %}
<span class="muted">AI 搜索已启用</span>
{% else %}
<a href="/search?q={{ query | urlencode }}&ai=1" class="button button-secondary" style="display:inline-block; text-decoration:none;">
AI 智能搜索
</a>
{% endif %}
</section>
{% endif %}
{% if searched %} {% if searched %}
{% if ai_error %}
<section class="card" style="margin-top: 8px; border-color: #b42318;">
<p style="margin:0; color: #b42318;"><strong>{{ ai_error }}</strong></p>
</section>
{% endif %}
{% if ai_activated and expanded_terms %}
<section class="card" style="margin-top: 8px; border-color: #0b57d0;">
<p style="margin:0; color: #0b57d0;"><strong>AI 帮你扩展了:</strong>{{ expanded_terms | join('、') }}</p>
</section>
{% endif %}
{% if results %} {% if results %}
<section class="stack"> <section class="stack">
<p class="muted">共找到 {{ results|length }} 条结果。</p> <p class="muted">共找到 {{ results|length }} 条结果。</p>
+69
View File
@@ -0,0 +1,69 @@
{% extends "base.html" %}
{% block content %}
<div class="breadcrumb">
<a href="/boxes">箱子</a>
<span>/</span>
<strong>设置</strong>
</div>
<div class="page-header">
<div>
<h1>设置</h1>
<p class="muted">配置 LLM 连接参数。未配置时,整站行为不受影响。</p>
</div>
</div>
{% if test_result %}
<section class="card" style="margin-bottom: 16px; border-color: {% if test_result.success %}#2f6b1f{% else %}#b42318{% endif %};">
<p style="margin:0; color: {% if test_result.success %}#2f6b1f{% else %}#b42318{% endif %};">
<strong>{{ "✓ " if test_result.success else "✗ " }}{{ test_result.message }}</strong>
</p>
</section>
{% endif %}
<form method="post" action="/settings" class="stack form-panel">
<label class="form-field checkbox-row">
<input type="checkbox" name="enabled" {% if config.enabled %}checked{% endif %}>
启用 LLM
</label>
<p class="checkbox-help">开启后,AI 相关功能将使用下方配置连接 LLM 服务。</p>
<label class="form-field">
Base URL
<input type="text" name="base_url" value="{{ config.base_url }}" placeholder="https://api.openai.com/v1">
</label>
<label class="form-field">
模型名称
<input type="text" name="model" value="{{ config.model }}" placeholder="例如 gpt-4o-mini">
</label>
<label class="form-field">
API Key
{% if api_key_configured %}
<input type="password" name="api_key" value="" placeholder="已配置,留空=不修改">
{% else %}
<input type="password" name="api_key" value="" placeholder="输入 API Key">
{% endif %}
</label>
<hr style="border:none;border-top:1px solid #ddd;margin:16px 0;">
<label class="form-field checkbox-row">
<input type="checkbox" name="ai_search_enabled" {% if config.ai_search_enabled %}checked{% endif %}>
启用 AI 智能搜索
</label>
<p class="checkbox-help">开启后,搜索页将显示「AI 智能搜索」按钮,通过查询词扩展增强搜索结果。</p>
<label class="form-field">
额外领域提示(可选)
<textarea name="ai_search_extra_hints" rows="3" placeholder="例如:用户物品主要涉及厨房用品和电子产品">{% if config.ai_search_extra_hints %}{{ config.ai_search_extra_hints }}{% endif %}</textarea>
</label>
<p class="checkbox-help">追加到 AI 搜索提示词末尾,帮助模型理解你的物品领域。留空则使用默认提示词。</p>
<div class="form-actions">
<button type="submit" class="button button-primary">保存设置</button>
<button type="submit" class="button button-secondary" formaction="/settings/test" formmethod="post">测试连接</button>
</div>
</form>
{% endblock %}
+27 -7
View File
@@ -199,6 +199,7 @@ More settings are coming; adding columns to an *existing* table costs a migratio
| `llm_model` | 模型名 / model name | (空 / empty | | `llm_model` | 模型名 / model name | (空 / empty |
| `llm_api_key` | API Key(明文 / plaintext,见 §7 | (空 / empty | | `llm_api_key` | API Key(明文 / plaintext,见 §7 | (空 / empty |
| `ai_search_enabled` | AI 搜索功能开关 / AI-search feature toggle | `false` | | `ai_search_enabled` | AI 搜索功能开关 / AI-search feature toggle | `false` |
| `ai_search_extra_hints` | AI 搜索:可选「额外领域提示」,追加到默认系统提示词(step 3 引入)/ optional extra domain hints appended to the default prompt | (空 / empty |
> 读写封装 / Access helpers`get_app_settings(db) -> LLMConfig`dataclass 视图)与 `save_app_settings(db, ...)`,供路由与 `app/llm.py` 复用。 > 读写封装 / Access helpers`get_app_settings(db) -> LLMConfig`dataclass 视图)与 `save_app_settings(db, ...)`,供路由与 `app/llm.py` 复用。
> Helpers `get_app_settings(db) -> LLMConfig` and `save_app_settings(db, ...)`, reused by routes and `app/llm.py`. > Helpers `get_app_settings(db) -> LLMConfig` and `save_app_settings(db, ...)`, reused by routes and `app/llm.py`.
@@ -209,7 +210,7 @@ OpenAI 兼容的薄客户端,基于 `httpx`**无新依赖** / A thin OpenAI
- `is_configured(cfg) -> bool`:开关开启且 `model`/`api_key` 齐全。 - `is_configured(cfg) -> bool`:开关开启且 `model`/`api_key` 齐全。
- `test_connection(cfg) -> Result`:发一个最小请求验证 `base_url`/`model`/`api_key`,供配置页"测试连接"用。 - `test_connection(cfg) -> Result`:发一个最小请求验证 `base_url`/`model`/`api_key`,供配置页"测试连接"用。
- `expand_query(cfg, query) -> list[str]`:把查询词扩成一批近义/相关词(本轮 AI 搜索用,见 §5)。 - `expand_query(cfg, query, extra_hints="") -> ExpansionResult`:把查询词扩成一批近义/相关词;`terms` 为扩展词列表(不含原词),`error` 用于区分超时/网络/HTTP 等真实调用失败(提示词与输出契约见 §5.2)。
- `analyze_image(...)`:**本轮不实现**,仅在文档中预留为未来接口(图片分析轮次)。Reserved for a future round, not implemented now. - `analyze_image(...)`:**本轮不实现**,仅在文档中预留为未来接口(图片分析轮次)。Reserved for a future round, not implemented now.
要点 / Notes 要点 / Notes
@@ -246,28 +247,46 @@ When disabled/unconfigured: the settings page still works; the AI-search button
- **常驻动作 / Persistent action** 搜索页**始终**提供「AI 智能搜索」,**不以"零结果"为前提**——即便普通搜索已出结果,用户不满意时也能点。 - **常驻动作 / Persistent action** 搜索页**始终**提供「AI 智能搜索」,**不以"零结果"为前提**——即便普通搜索已出结果,用户不满意时也能点。
The "AI search" action is **always** present on the search page, **not gated on zero results** — usable even when normal results exist. The "AI search" action is **always** present on the search page, **not gated on zero results** — usable even when normal results exist.
- **流程 / Flow** 普通 `LIKE` 照常先出结果 → 用户触发 AI → `expand_query` 把查询词扩成近义/相关词 → 用「原词 + 扩展词」对 `name`/`note` 做 OR `LIKE` 重搜 → 展示,并用横幅标注「AI 帮你扩展了:…」。 - **流程 / Flow** 普通 `LIKE` 照常先出结果 → 用户触发 AI → `expand_query` 返回 `ExpansionResult`(扩展词 `terms` 不含原词;调用失败写入 `error`)→ `ai_search` 用「原词 + 扩展词」对 `name`/`note` 做 OR `LIKE` 重搜 → 展示,并用横幅标注「AI 帮你扩展了:…」。
Normal `LIKE` first → user triggers AI → `expand_query` OR-`LIKE` over name/note with the original + expanded terms → render with a banner listing the expansion. Normal `LIKE` first → user triggers AI → `expand_query` returns an `ExpansionResult` (`terms` exclude the original query; failures are represented by `error`) → `ai_search` OR-`LIKE`s over name/note with the original + expanded terms → render with a banner listing the expansion.
- **只把查询词发出去 / Only the query leaves**,不外泄物品清单;token 恒定、不随上千件物品增长。 - **只把查询词发出去 / Only the query leaves**,不外泄物品清单;token 恒定、不随上千件物品增长。
Only the query is sent; the inventory is not. Token cost is constant and does not grow with thousands of items. Only the query is sent; the inventory is not. Token cost is constant and does not grow with thousands of items.
### 5.2 实现接口 / Implementation seam ### 5.2 提示词与输出契约 / Prompt & Output Contract
`expand_query` 的**质量**取决于提示词,**集成稳定性**取决于输出契约——两者都在代码侧掌控(决策 C)。
Quality hinges on the prompt; integration stability hinges on the output contract — both are code-controlled (decision C).
- **基础系统提示词写死在 `app/llm.py`(用户改不坏)/ Base system prompt hardcoded** 框定搬家/家居场景,要求"列出用户可能用来命名同一物品的相关词(近义、别称、上位类别、具体品类)";语言跟随查询;最多约 8 个;不解释、不造无关词。
Frames the moving/household domain, asks for related naming terms, follows the query's language, caps the count, no prose.
- **可选「额外领域提示」/ Optional extra hints** KV `ai_search_extra_hints`(设置页一个多行输入,默认空)。非空时**追加**到基础提示词之后,供业主微调倾向(如"厨房用品多,偏向厨具类")。**它只能补充,不能改写输出格式。**
An optional free-text setting appended to the base prompt; it can only add guidance, never alter the output format.
- **输出契约(代码强制,与提示词解耦)/ Output contract (code-enforced)** 要求模型只返回 **JSON 字符串数组**;解析时去掉 ` ```json ` 围栏 → `json.loads` → 只接受字符串数组 → 过滤空串/过长词 → 最多 8 个。散文、坏 JSON、JSON object、非字符串数组都视为**合法空扩展**(`terms=[]`, `error=None`);网络错误、HTTP 错误、超时等真实调用失败写入 `ExpansionResult.error``expand_query``terms` 只包含扩展词;**原词由 `ai_search` 并入并去重**。
Require a JSON string array; strip code fences, `json.loads`, accept only string arrays, filter empty/overlong terms, and cap to 8 terms. Prose, bad JSON, JSON objects, and non-string arrays are successful empty expansions (`terms=[]`, `error=None`); network/HTTP/timeout failures are represented by `ExpansionResult.error`. `expand_query.terms` contains only expanded terms; `ai_search` adds the original term and dedupes.
- **客户端参数 / Client params** 低 temperature、较小 max_tokens、设超时。Low temperature, small max_tokens, a timeout.
- **措辞留松 / Wording left loose** 默认提示词的具体字句可在 step-3 实测中迭代,不在文档里冻死。
Exact default wording can be iterated during step-3 testing.
### 5.3 实现接口 / Implementation seam
- 路由层扩展现有 `GET /search`:增加 `ai=1` 触发位(如 `GET /search?q=锅&ai=1`),保持单页、可收藏、SSR 友好。 - 路由层扩展现有 `GET /search`:增加 `ai=1` 触发位(如 `GET /search?q=锅&ai=1`),保持单页、可收藏、SSR 友好。
Extend the existing `GET /search` with an `ai=1` trigger (e.g. `/search?q=…&ai=1`), staying single-page and bookmarkable. Extend the existing `GET /search` with an `ai=1` trigger (e.g. `/search?q=…&ai=1`), staying single-page and bookmarkable.
- 内部定义可替换的检索 seam,例如 `ai_search(db, query) -> (expanded_terms, results)` - 内部定义可替换的检索 seam,例如 `ai_search(db, query) -> (expanded_terms, results, error_message)`
Define a replaceable retrieval seam, e.g. `ai_search(db, query) -> (expanded_terms, results)`: Define a replaceable retrieval seam, e.g. `ai_search(db, query) -> (expanded_terms, results, error_message)`:
- **本轮 / now:** 内部=查询词扩展 + 本地 `LIKE` - **本轮 / now:** 内部=查询词扩展 + 本地 `LIKE`
- **未来 / later:** 换成向量嵌入 + 相似度检索,**路由与模板不变**。 - **未来 / later:** 换成向量嵌入 + 相似度检索,**路由与模板不变**。
Swap to embeddings + similarity later **without changing the route or template**. Swap to embeddings + similarity later **without changing the route or template**.
- 本轮检索范围=`name` + `note``image_description` 本轮不存在)。 - 本轮检索范围=`name` + `note``image_description` 本轮不存在)。
Search scope this round = `name` + `note` (no `image_description` yet). Search scope this round = `name` + `note` (no `image_description` yet).
### 5.3 降级 / Degradation ### 5.4 降级 / Degradation
AI 关闭/未配置 → 不显示按钮(或提示去 `/settings`);调用失败 → 友好提示并回退到普通结果。 AI 关闭/未配置 → 不显示按钮(或提示去 `/settings`);调用失败 → 友好提示并回退到普通结果。
AI off/unconfigured → no button (or a hint to `/settings`); on failure → a friendly message, fall back to normal results. AI off/unconfigured → no button (or a hint to `/settings`); on failure → a friendly message, fall back to normal results.
合法空扩展(模型返回 `[]` 或输出无法通过严格 JSON 字符串数组契约)不视为调用失败:回退普通结果,不显示故障提示。
A legitimate empty expansion (model returns `[]` or output fails the strict JSON-string-array contract) is not treated as a call failure: fall back to normal results without an error banner.
--- ---
## 6. 数据模型与路由变更 / Data Model & Route Changes ## 6. 数据模型与路由变更 / Data Model & Route Changes
@@ -334,3 +353,4 @@ AI off/unconfigured → no button (or a hint to `/settings`); on failure → a f
| D8 | AI 搜索 v1=查询词扩展 / query-term expansion | 上千件物品下可扩展、不外泄清单、token 恒定。 | | D8 | AI 搜索 v1=查询词扩展 / query-term expansion | 上千件物品下可扩展、不外泄清单、token 恒定。 |
| D9 | 检索做成可替换 seam / pluggable retrieval | 未来换嵌入式语义搜索时上层不动。 | | D9 | 检索做成可替换 seam / pluggable retrieval | 未来换嵌入式语义搜索时上层不动。 |
| D10 | 图片分析不在本轮 / image analysis deferred | 业主本轮三件事不含它;架构预留接口。 | | D10 | 图片分析不在本轮 / image analysis deferred | 业主本轮三件事不含它;架构预留接口。 |
| D11 | AI 搜索提示词:默认写死 + 可选「额外领域提示」;输出契约由代码强制 / hardcoded default prompt + optional extra-hints, code-enforced JSON contract | 保证解析稳定(用户改不坏),又给业主一点不改代码即可微调的空间。 |
+1 -1
View File
@@ -55,7 +55,7 @@ A settings page to enter & test the LLM config, persisted to `app_settings`, plu
- [ ] 新增 `app/llm.py`(基于 `httpx`): - [ ] 新增 `app/llm.py`(基于 `httpx`):
- [ ] `is_configured(cfg) -> bool` - [ ] `is_configured(cfg) -> bool`
- [ ] `test_connection(cfg) -> Result`(发最小请求验证 `base_url`/`model`/`api_key`)。 - [ ] `test_connection(cfg) -> Result`(发最小请求验证 `base_url`/`model`/`api_key`)。
- [ ] `expand_query(cfg, query) -> list[str]`(查询词扩展;**步骤 3 会用**,本步先落地+单测)。 - [ ] `expand_query(cfg, query) -> ExpansionResult`(查询词扩展;**步骤 3 会校准提示词与输出契约**`terms` 为扩展词列表,`error` 用于区分超时/网络/HTTP 等真实调用失败)。
- [ ] 统一超时 + 错误处理;失败优雅降级。 - [ ] 统一超时 + 错误处理;失败优雅降级。
- [ ] **(预留,不实现)** `analyze_image(...)`:仅留 TODO/签名占位 + 注释指向"未来图片分析轮次"。Reserved, not implemented. - [ ] **(预留,不实现)** `analyze_image(...)`:仅留 TODO/签名占位 + 注释指向"未来图片分析轮次"。Reserved, not implemented.
- [ ] 把所有网络调用收敛到**单一函数边界**,便于测试整体 mock。 - [ ] 把所有网络调用收敛到**单一函数边界**,便于测试整体 mock。
+25 -8
View File
@@ -17,7 +17,9 @@ A **persistent** "AI search" action on the search page that broadens results via
- 现有搜索:`app/main.py::_build_search_results(db, query)``Box`/`Item`/`SubItem``name``note` 做大小写不敏感 `LIKE`,返回结果列表;路由 `GET /search`(函数 `search_page`,参数 `q`)渲染 `app/templates/search/index.html` - 现有搜索:`app/main.py::_build_search_results(db, query)``Box`/`Item`/`SubItem``name``note` 做大小写不敏感 `LIKE`,返回结果列表;路由 `GET /search`(函数 `search_page`,参数 `q`)渲染 `app/templates/search/index.html`
Existing search: `_build_search_results(db, query)` does case-insensitive `LIKE` over name/note; route `GET /search` renders `search/index.html`. Existing search: `_build_search_results(db, query)` does case-insensitive `LIKE` over name/note; route `GET /search` renders `search/index.html`.
- 步骤 2 已提供:`app/llm.py::expand_query(cfg, query) -> list[str]`、配置读取 `get_app_settings(db)`、开关 `ai_search_enabled``is_configured(cfg)` - 步骤 2 已提供:`app/llm.py::expand_query` 的基础能力、配置读取 `get_app_settings(db)`、开关 `ai_search_enabled``is_configured(cfg)`、设置页 `app/templates/settings/form.html`;本步将 `expand_query` 校准为返回结构化 `ExpansionResult(terms, error)`
- 本步**新增**配置项 `ai_search_extra_hints`(可选「额外领域提示」)并在设置页加一个多行输入——这是本步**唯一**触及设置页之处。
This step adds the `ai_search_extra_hints` setting + a textarea on the settings page (the only settings-page change here).
- 本轮检索范围=`name` + `note``image_description` 本轮不存在,属未来图片分析轮次)。 - 本轮检索范围=`name` + `note``image_description` 本轮不存在,属未来图片分析轮次)。
Search scope = `name` + `note` (no `image_description` this round). Search scope = `name` + `note` (no `image_description` this round).
@@ -25,18 +27,29 @@ A **persistent** "AI search" action on the search page that broadens results via
- **常驻、不依赖零结果。** 普通 `LIKE` 照常先出结果;AI 动作始终可用(开启且已配置时)。 - **常驻、不依赖零结果。** 普通 `LIKE` 照常先出结果;AI 动作始终可用(开启且已配置时)。
Persistent and not gated on zero results. Persistent and not gated on zero results.
- **流程:** 触发 AI → `expand_query` 得到"原词 + 一批近义/相关词" → 用这组词`name`/`note` 做 OR `LIKE` 重搜 → 展示,并用横幅标注「AI 帮你扩展了:…」。**只把查询词发出去**,不外泄物品清单。 - **流程:** 触发 AI → `expand_query` 返回 `ExpansionResult`(扩展词 `terms` 不含原词,调用失败写入 `error`)→ `ai_search` 合并「原词 + 扩展词」并`name`/`note` 做 OR `LIKE` 重搜 → 展示,并用横幅标注「AI 帮你扩展了:…」。**只把查询词发出去**,不外泄物品清单。
Trigger → expand OR-`LIKE` over the original + expanded terms → render with a banner of the expansion. Only the query leaves. Trigger → `expand_query` returns an `ExpansionResult` (`terms` exclude the original query; failures are represented by `error`) → `ai_search` OR-`LIKE`s over the original + expanded terms → render with a banner of the expansion. Only the query leaves.
- **可替换的检索 seam。** 把 AI 检索抽成一个函数(如 `ai_search(db, query) -> (expanded_terms, results)`),本轮内部=查询词扩展 + 本地 `LIKE`;**未来换成向量嵌入 + 相似度时,路由与模板不变**。 - **可替换的检索 seam。** 把 AI 检索抽成一个函数(如 `ai_search(db, query) -> (expanded_terms, results, error_message)`),本轮内部=查询词扩展 + 本地 `LIKE`;**未来换成向量嵌入 + 相似度时,路由与模板不变**。
Wrap AI retrieval behind a swappable seam so embeddings can replace it later without touching route/template. Wrap AI retrieval behind a swappable seam so embeddings can replace it later without touching route/template.
- **提示词(决策 C,详见设计 §5.2)。** 基础系统提示词**写死在 `app/llm.py`**;设置页可选的 `ai_search_extra_hints` 非空时**追加**到其后;**输出契约由代码强制**(只接受 JSON 字符串数组;散文/坏 JSON/非字符串数组解析为合法空扩展;网络/超时/HTTP 失败写入 `ExpansionResult.error`),用户改 hints 也改不坏解析。
Base prompt hardcoded; optional extra hints appended; output contract enforced in code: only a JSON string array is accepted; prose/bad JSON/non-string arrays become a successful empty expansion; network/timeout/HTTP failures are represented by `ExpansionResult.error`.
- **优雅降级。** AI 关闭/未配置 → 不显示按钮(或提示去 `/settings`);调用失败 → 友好提示 + 回退普通结果。 - **优雅降级。** AI 关闭/未配置 → 不显示按钮(或提示去 `/settings`);调用失败 → 友好提示 + 回退普通结果。
--- ---
## 任务 / Tasks ## 任务 / Tasks
- [ ] 实现检索 seam:在 `app/main.py`(或抽一个小搜索模块 `app/search.py`)加 `ai_search(db, query) -> (expanded_terms, results)` - [ ] **落地/校准 `expand_query` 的提示词(按设计 §5.2)**
- `expand_query(cfg, query)` 得到扩展词; - 基础系统提示词写死在 `app/llm.py`(搬家/家居场景、列相关命名词、跟随查询语言、≤ ~8 个、不解释、不造无关词)。默认提示词起点(**可迭代** / a starting point, tune during testing):
> 你是搬家物品搜索助手。用户在搜索自己打包的箱子与物品(家居/搬家场景)。给定一个搜索词,列出用户可能用来命名同一类物品的相关词:近义词、常见别称、上位类别、具体品类。规则:用与查询相同的语言;只给与该物品紧密相关、有助于在清单里找到它的词;不要解释、不要造无关词;最多 8 个;只输出一个 JSON 字符串数组,例如 `["炒锅","平底锅","汤锅","厨具"]`。
- 读取 `ai_search_extra_hints`,非空则**追加**到基础提示词之后(只补充,不改格式)。
- **返回契约**`expand_query(cfg, query, extra_hints="") -> ExpansionResult`,其中 `terms` 是扩展词列表(**不含原词**),`error` 在成功时为 `None`
- **输出契约**:要求模型只回 JSON 字符串数组;解析去 ` ```json ` 围栏 → `json.loads` → 只接受字符串数组 → 过滤空串/过长词 → 最多 8 个;散文、坏 JSON、JSON object、非字符串数组都返回 `terms=[]``error=None`(合法空扩展);网络错误、HTTP 错误、超时等调用失败返回 `terms=[]``error=<友好错误>`;不向上抛 500。
- [ ] **新增配置项 `ai_search_extra_hints`**KV 默认空;纳入 `get_app_settings` / `save_app_settings`;设置页 `app/templates/settings/form.html` 加一个多行输入(沿用 step 2 风格)。
- [ ] 实现检索 seam:在 `app/main.py`(或抽一个小搜索模块 `app/search.py`)加 `ai_search(db, query) -> (expanded_terms, results, error_message)`
-`expand_query(cfg, query)` 得到 `ExpansionResult`
-`result.error` 非空:回退普通搜索,并把友好错误传给模板;
-`result.terms` 为空且无错误:视为合法空扩展,回退普通搜索,不显示故障提示;
- 用「原词 + 扩展词」对 `name`/`note` 做 OR `LIKE`**复用现有 `_build_search_results` 的匹配逻辑**,避免重复实现),去重。 - 用「原词 + 扩展词」对 `name`/`note` 做 OR `LIKE`**复用现有 `_build_search_results` 的匹配逻辑**,避免重复实现),去重。
- 注意:现有 `_build_search_results(db, query)` 只接收单个查询词;建议把它泛化为接收一组关键词(对多个词做 OR),让 AI 搜索与普通搜索共用同一套匹配逻辑,避免分叉。 - 注意:现有 `_build_search_results(db, query)` 只接收单个查询词;建议把它泛化为接收一组关键词(对多个词做 OR),让 AI 搜索与普通搜索共用同一套匹配逻辑,避免分叉。
Note: `_build_search_results` currently takes a single query — generalize it to accept multiple keywords so AI and normal search share one matching path. Note: `_build_search_results` currently takes a single query — generalize it to accept multiple keywords so AI and normal search share one matching path.
@@ -52,13 +65,17 @@ A **persistent** "AI search" action on the search page that broadens results via
- [ ] 扩展词驱动命中:原词 `LIKE` 搜不到、扩展后能搜到。 - [ ] 扩展词驱动命中:原词 `LIKE` 搜不到、扩展后能搜到。
- [ ] 已有结果时点 AI 仍可用,且结果集被扩大(含原结果)。 - [ ] 已有结果时点 AI 仍可用,且结果集被扩大(含原结果)。
- [ ] 按钮可见性随 `ai_search_enabled` + `is_configured()` 门控。 - [ ] 按钮可见性随 `ai_search_enabled` + `is_configured()` 门控。
- [ ] 调用失败 → 回退普通结果、页面不报错。 - [ ] 调用失败(超时/网络/HTTP→ 回退普通结果、显示友好提示、页面不报错。
- [ ] `expand_query` 输出解析:模型回合法 JSON 数组 → 正确解析;回散文/坏 JSON/非字符串数组 → `terms=[]``error=None`;超时/网络/HTTP 失败 → `terms=[]``error` 非空;均不抛错。
Output parsing: valid JSON array → parsed; prose/bad JSON/non-string arrays → `terms=[]`, `error=None`; timeout/network/HTTP failures → `terms=[]`, non-empty `error`; no raise.
- [ ] `ai_search_extra_hints` 非空时确被追加进请求(可对构造的请求体断言)。
Extra hints, when set, are appended to the request.
--- ---
## 涉及文件 / Files ## 涉及文件 / Files
`app/main.py`、(可选 `app/search.py`)、`app/templates/search/index.html``tests/` `app/llm.py``app/main.py`、(可选 `app/search.py`)、`app/templates/search/index.html``app/templates/settings/form.html`、配置读写 helperstep 2 的 settings store)、`tests/`
--- ---
@@ -0,0 +1,32 @@
"""V2 app_settings
Revision ID: a1b2c3d4e5f6
Revises: 57af90893f55
Create Date: 2026-06-01 14:00:00.000000
"""
from typing import Sequence, Union
from alembic import op
import sqlalchemy as sa
# revision identifiers, used by Alembic.
revision: str = 'a1b2c3d4e5f6'
down_revision: Union[str, Sequence[str], None] = '57af90893f55'
branch_labels: Union[str, Sequence[str], None] = None
depends_on: Union[str, Sequence[str], None] = None
def upgrade() -> None:
"""Upgrade schema."""
op.create_table('app_settings',
sa.Column('key', sa.Text(), nullable=False),
sa.Column('value', sa.Text(), nullable=True),
sa.PrimaryKeyConstraint('key')
)
def downgrade() -> None:
"""Downgrade schema."""
op.drop_table('app_settings')
+2
View File
@@ -1,4 +1,6 @@
[pytest] [pytest]
pythonpath = . pythonpath = .
testpaths = tests
norecursedirs = app .venv
filterwarnings = filterwarnings =
ignore:'asyncio\.iscoroutinefunction' is deprecated and slated for removal in Python 3\.16; use inspect\.iscoroutinefunction\(\) instead:DeprecationWarning:fastapi\.routing ignore:'asyncio\.iscoroutinefunction' is deprecated and slated for removal in Python 3\.16; use inspect\.iscoroutinefunction\(\) instead:DeprecationWarning:fastapi\.routing
+707
View File
@@ -0,0 +1,707 @@
"""Tests for AI search (Step 3).
All LLM calls are mocked — CI never touches the network.
Coverage areas:
- expand_query JSON output parsing (valid, fenced, prose, bad JSON, timeout)
- Output contract enforcement (strict JSON array only)
- Expansion term count cap and length cap
- ai_search seam function
- GET /search with ai=1 trigger
- AI button visibility on search page
- Graceful degradation on failure
- ai_search_extra_hints appended to prompt
- ai_search_enabled toggle
"""
from unittest.mock import patch
import httpx
import pytest
from app.llm import (
_MAX_EXPANSION_TERMS,
_MAX_TERM_LENGTH,
ExpansionResult,
LLMResult,
_parse_json_string_array,
expand_query,
is_configured,
)
from app.main import _ai_search, _build_search_results
from app.models import AppSetting, Box, Item, SubItem
from app.settings_store import LLMConfig, get_app_settings, save_app_settings
# ---------------------------------------------------------------------------
# Helper: configure AI search for route tests
# ---------------------------------------------------------------------------
_AI_CFG = LLMConfig(
enabled=True,
base_url="https://api.example.com/v1",
model="gpt-4o-mini",
api_key="sk-test-key",
ai_search_enabled=True,
)
def _enable_ai_search(client, db_session):
"""Persist a fully-configured AI search setup via the settings route."""
client.post(
"/settings",
data={
"enabled": "on",
"base_url": "https://api.example.com/v1",
"model": "gpt-4o-mini",
"api_key": "sk-test-key",
"ai_search_enabled": "on",
},
follow_redirects=False,
)
# ---------------------------------------------------------------------------
# _parse_json_string_array: strict JSON contract enforcement
# ---------------------------------------------------------------------------
class TestParseJsonStringArray:
def test_valid_json_array(self):
result = _parse_json_string_array('["炒锅","平底锅","汤锅"]')
assert result == ["炒锅", "平底锅", "汤锅"]
def test_json_array_with_code_fence(self):
result = _parse_json_string_array('```json\n["","铲子"]\n```')
assert result == ["", "铲子"]
def test_json_array_with_code_fence_no_lang(self):
result = _parse_json_string_array('```\n["","铲子"]\n```')
assert result == ["", "铲子"]
def test_empty_string_returns_empty(self):
assert _parse_json_string_array("") == []
assert _parse_json_string_array(" ") == []
def test_prose_returns_empty(self):
"""Prose text does NOT become expansion terms — strict contract."""
assert _parse_json_string_array("I cannot help with that.") == []
def test_prose_newlines_returns_empty(self):
"""Line-separated prose does NOT become expansion terms."""
assert _parse_json_string_array("炒锅\n平底锅\n汤锅") == []
def test_prose_commas_returns_empty(self):
"""Comma-separated prose does NOT become expansion terms."""
assert _parse_json_string_array("炒锅, 平底锅, 汤锅") == []
def test_bad_json_returns_empty(self):
"""Invalid JSON returns empty — no fallback."""
assert _parse_json_string_array("{invalid json") == []
def test_json_object_returns_empty(self):
"""JSON object (non-array) returns empty."""
assert _parse_json_string_array('{"terms":["","厨具"]}') == []
def test_json_array_with_numbers_returns_empty(self):
"""Non-string items in array cause rejection — strict contract."""
assert _parse_json_string_array('[1, 2, 3]') == []
def test_json_array_with_mixed_types_returns_empty(self):
"""Mixed string/number array is rejected."""
assert _parse_json_string_array('["", 1]') == []
def test_empty_json_array(self):
result = _parse_json_string_array('[]')
assert result == []
def test_capped_at_max_terms(self):
"""More than _MAX_EXPANSION_TERMS items are truncated."""
terms = [f"{i}" for i in range(20)]
json_str = "[" + ",".join(f'"{t}"' for t in terms) + "]"
result = _parse_json_string_array(json_str)
assert len(result) == _MAX_EXPANSION_TERMS
def test_long_terms_filtered_out(self):
"""Terms exceeding _MAX_TERM_LENGTH are silently dropped."""
short = ""
long_term = "A" * (_MAX_TERM_LENGTH + 1)
json_str = f'["{short}", "{long_term}"]'
result = _parse_json_string_array(json_str)
assert result == [""]
def test_whitespace_stripped(self):
result = _parse_json_string_array('["", " 平底锅 "]')
assert result == ["", "平底锅"]
def test_empty_strings_filtered(self):
result = _parse_json_string_array('["", "", " ", "平底锅"]')
assert result == ["", "平底锅"]
# ---------------------------------------------------------------------------
# expand_query: prompt, hints, graceful degradation
# ---------------------------------------------------------------------------
class TestExpandQueryNew:
def test_returns_empty_when_not_configured(self):
cfg = LLMConfig(enabled=False)
result = expand_query(cfg, "")
assert result.terms == []
assert result.error is None
@patch("app.llm._call_chat_completion")
def test_parses_valid_json_response(self, mock_call):
mock_call.return_value = {
"choices": [{"message": {"content": '["炒锅","平底锅","汤锅","厨具"]'}}]
}
cfg = LLMConfig(enabled=True, model="gpt-4o", api_key="sk-key")
result = expand_query(cfg, "")
assert "炒锅" in result.terms
assert "平底锅" in result.terms
assert "厨具" in result.terms
assert result.error is None
@patch("app.llm._call_chat_completion")
def test_handles_json_with_code_fence(self, mock_call):
mock_call.return_value = {
"choices": [
{"message": {"content": '```json\n["炒锅","平底锅"]\n```'}}
]
}
cfg = LLMConfig(enabled=True, model="gpt-4o", api_key="sk-key")
result = expand_query(cfg, "")
assert "炒锅" in result.terms
assert result.error is None
@patch("app.llm._call_chat_completion")
def test_prose_response_returns_empty_no_error(self, mock_call):
"""Prose from model → empty terms, no error (successful call, unparseable output)."""
mock_call.return_value = {
"choices": [{"message": {"content": "I cannot help with that."}}]
}
cfg = LLMConfig(enabled=True, model="gpt-4o", api_key="sk-key")
result = expand_query(cfg, "")
assert result.terms == []
assert result.error is None
@patch("app.llm._call_chat_completion")
def test_json_object_response_returns_empty_no_error(self, mock_call):
"""JSON object (non-array) → empty terms, no error."""
mock_call.return_value = {
"choices": [{"message": {"content": '{"terms":["","厨具"]}'}}]
}
cfg = LLMConfig(enabled=True, model="gpt-4o", api_key="sk-key")
result = expand_query(cfg, "")
assert result.terms == []
assert result.error is None
@patch("app.llm._call_chat_completion")
def test_timeout_returns_error(self, mock_call):
"""Timeout → empty terms + error message."""
mock_call.side_effect = httpx.TimeoutException("timeout")
cfg = LLMConfig(enabled=True, model="gpt-4o", api_key="sk-key")
result = expand_query(cfg, "")
assert result.terms == []
assert result.error is not None
assert "超时" in result.error
@patch("app.llm._call_chat_completion")
def test_network_error_returns_error(self, mock_call):
"""Network error → empty terms + error message."""
mock_call.side_effect = httpx.ConnectError("refused")
cfg = LLMConfig(enabled=True, model="gpt-4o", api_key="sk-key")
result = expand_query(cfg, "")
assert result.terms == []
assert result.error is not None
assert "无法连接" in result.error
@patch("app.llm._call_chat_completion")
def test_http_error_returns_error(self, mock_call):
"""HTTP error → empty terms + error message."""
mock_call.side_effect = httpx.HTTPStatusError(
"401",
request=httpx.Request("POST", "http://x"),
response=httpx.Response(401),
)
cfg = LLMConfig(enabled=True, model="gpt-4o", api_key="sk-key")
result = expand_query(cfg, "")
assert result.terms == []
assert result.error is not None
assert "错误" in result.error
@patch("app.llm._call_chat_completion")
def test_returns_empty_on_empty_choices(self, mock_call):
mock_call.return_value = {"choices": []}
cfg = LLMConfig(enabled=True, model="gpt-4o", api_key="sk-key")
result = expand_query(cfg, "")
assert result.terms == []
assert result.error is None
@patch("app.llm._call_chat_completion")
def test_extra_hints_appended_to_system_prompt(self, mock_call):
"""When extra_hints is non-empty, it should be appended to the system prompt."""
mock_call.return_value = {
"choices": [{"message": {"content": '["扩展词"]'}}]
}
cfg = LLMConfig(enabled=True, model="gpt-4o", api_key="sk-key")
expand_query(cfg, "", extra_hints="用户物品主要涉及厨房用品")
# Verify the system prompt includes the extra hints
call_args = mock_call.call_args
messages = call_args[1]["messages"] if "messages" in call_args[1] else call_args[0][1]
system_content = messages[0]["content"]
assert "用户物品主要涉及厨房用品" in system_content
@patch("app.llm._call_chat_completion")
def test_extra_hints_ignored_when_empty(self, mock_call):
"""When extra_hints is empty, system prompt should not change."""
mock_call.return_value = {
"choices": [{"message": {"content": '["扩展词"]'}}]
}
cfg = LLMConfig(enabled=True, model="gpt-4o", api_key="sk-key")
expand_query(cfg, "", extra_hints="")
call_args = mock_call.call_args
messages = call_args[1]["messages"] if "messages" in call_args[1] else call_args[0][1]
system_content = messages[0]["content"]
# Should be the base prompt only
assert "搬家物品搜索助手" in system_content
assert "JSON 字符串数组" in system_content
@patch("app.llm._call_chat_completion")
def test_temperature_zero_passed(self, mock_call):
"""expand_query should pass temperature=0 for deterministic output."""
mock_call.return_value = {
"choices": [{"message": {"content": '["扩展词"]'}}]
}
cfg = LLMConfig(enabled=True, model="gpt-4o", api_key="sk-key")
expand_query(cfg, "")
call_args = mock_call.call_args
assert call_args[1]["temperature"] == 0
# ---------------------------------------------------------------------------
# _ai_search: seam function
# ---------------------------------------------------------------------------
class TestAiSearchSeam:
@patch("app.main.expand_query")
def test_returns_expanded_terms_and_results(self, mock_expand, client, db_session):
"""AI search returns expanded terms and broader results."""
box = Box(name="厨房箱", note="装了炒锅和铲子")
db_session.add(box)
db_session.commit()
mock_expand.return_value = ExpansionResult(terms=["炒锅", "平底锅", "汤锅"])
cfg = get_app_settings(db_session)
expanded, results, error = _ai_search(db_session, cfg, "平底锅")
assert "炒锅" in expanded
assert error is None
assert len(results) >= 1
assert any("厨房箱" in r["name"] or "炒锅" in (r.get("note") or "") for r in results)
@patch("app.main.expand_query")
def test_includes_original_query_in_search(self, mock_expand, client, db_session):
"""AI search includes the original query term in the search."""
box = Box(name="冬季衣物箱")
db_session.add(box)
db_session.commit()
mock_expand.return_value = ExpansionResult(terms=["羽绒服"])
cfg = get_app_settings(db_session)
expanded, results, error = _ai_search(db_session, cfg, "衣物")
assert error is None
assert any("冬季衣物箱" in r["name"] for r in results)
@patch("app.main.expand_query")
def test_empty_expansion_returns_normal_results_no_error(self, mock_expand, client, db_session):
"""Legitimate empty expansion (no synonyms found) → normal results, no error."""
box = Box(name="书房箱")
db_session.add(box)
db_session.commit()
mock_expand.return_value = ExpansionResult(terms=[])
cfg = get_app_settings(db_session)
expanded, results, error = _ai_search(db_session, cfg, "书房")
assert expanded == []
assert error is None
assert any("书房箱" in r["name"] for r in results)
@patch("app.main.expand_query")
def test_llm_failure_returns_normal_results_with_error(self, mock_expand, client, db_session):
"""When expand_query signals failure, seam returns normal results + error message."""
box = Box(name="厨房箱", note="装了炒锅")
db_session.add(box)
db_session.commit()
mock_expand.return_value = ExpansionResult(terms=[], error="AI 搜索请求超时,请稍后再试。")
cfg = get_app_settings(db_session)
expanded, results, error = _ai_search(db_session, cfg, "厨房")
assert expanded == []
assert error is not None
assert "超时" in error
assert len(results) >= 1
# ---------------------------------------------------------------------------
# _build_search_results: multi-keyword support
# ---------------------------------------------------------------------------
class TestBuildSearchResultsMultiKeyword:
def test_single_keyword_works_as_before(self, db_session):
box = Box(name="厨房箱")
db_session.add(box)
db_session.commit()
results = _build_search_results(db_session, "厨房")
assert len(results) == 1
assert results[0]["name"] == "厨房箱"
def test_multiple_keywords_match_any(self, db_session):
box1 = Box(name="厨房箱")
box2 = Box(name="卧室箱")
db_session.add_all([box1, box2])
db_session.commit()
results = _build_search_results(db_session, ["厨房", "卧室"])
assert len(results) == 2
def test_multiple_keywords_dedupes_results(self, db_session):
"""A box matching multiple keywords appears only once."""
box = Box(name="厨房箱", note="装了厨房用品")
db_session.add(box)
db_session.commit()
results = _build_search_results(db_session, ["厨房", "用品"])
assert len(results) == 1
def test_empty_keywords_returns_empty(self, db_session):
results = _build_search_results(db_session, [])
assert results == []
# ---------------------------------------------------------------------------
# Routes: GET /search with ai=1
# ---------------------------------------------------------------------------
class TestSearchRouteAI:
@patch("app.llm._call_chat_completion")
def test_ai_search_finds_more_results(self, mock_call, client, db_session):
"""Original query misses, but expanded term finds items."""
box = Box(name="杂物箱")
item = Item(name="炒锅", box=box, is_container=False)
db_session.add_all([box, item])
db_session.commit()
mock_call.return_value = {
"choices": [{"message": {"content": '["炒锅","平底锅","汤锅"]'}}]
}
_enable_ai_search(client, db_session)
# Normal search for "平底锅" — no results
response = client.get("/search?q=平底锅")
assert "没有找到匹配结果" in response.text
# AI search for "平底锅" — finds "炒锅" via expansion
response = client.get("/search?q=平底锅&ai=1")
assert response.status_code == 200
assert "炒锅" in response.text
assert "AI 帮你扩展了" in response.text
@patch("app.llm._call_chat_completion")
def test_ai_search_includes_original_results(self, mock_call, client, db_session):
"""AI search should also include results from original query."""
box = Box(name="厨房箱")
item1 = Item(name="锅铲", box=box, is_container=False)
item2 = Item(name="平底锅", box=box, is_container=False)
db_session.add_all([box, item1, item2])
db_session.commit()
mock_call.return_value = {
"choices": [{"message": {"content": '["炒锅","汤锅"]'}}]
}
_enable_ai_search(client, db_session)
response = client.get("/search?q=锅&ai=1")
assert response.status_code == 200
# Original result "平底锅" should still be there
assert "平底锅" in response.text
@patch("app.llm._call_chat_completion")
def test_ai_search_shows_expansion_banner(self, mock_call, client, db_session):
"""When AI search is activated, a banner shows expanded terms."""
box = Box(name="厨房箱")
db_session.add(box)
db_session.commit()
mock_call.return_value = {
"choices": [{"message": {"content": '["炒锅","平底锅"]'}}]
}
_enable_ai_search(client, db_session)
response = client.get("/search?q=锅&ai=1")
assert response.status_code == 200
assert "AI 帮你扩展了" in response.text
assert "炒锅" in response.text
def test_ai_search_without_flag_does_normal_search(self, client, db_session):
"""Without ai=1, search behaves normally even when AI is configured."""
box = Box(name="厨房箱")
db_session.add(box)
db_session.commit()
_enable_ai_search(client, db_session)
response = client.get("/search?q=厨房")
assert response.status_code == 200
assert "厨房箱" in response.text
assert "AI 帮你扩展了" not in response.text
@patch("app.llm._call_chat_completion")
def test_ai_search_without_configuration_ignores_flag(self, mock_call, client, db_session):
"""ai=1 is ignored when AI is not configured."""
box = Box(name="厨房箱")
db_session.add(box)
db_session.commit()
response = client.get("/search?q=厨房&ai=1")
assert response.status_code == 200
assert "厨房箱" in response.text
assert "AI 帮你扩展了" not in response.text
mock_call.assert_not_called()
@patch("app.llm._call_chat_completion")
def test_ai_search_graceful_degradation_on_llm_failure(self, mock_call, client, db_session):
"""LLM failure (timeout) → normal results + friendly error banner."""
box = Box(name="厨房箱", note="装了炒锅")
db_session.add(box)
db_session.commit()
# expand_query catches timeout and returns ExpansionResult with error
mock_call.side_effect = httpx.TimeoutException("timeout")
_enable_ai_search(client, db_session)
response = client.get("/search?q=厨房&ai=1")
assert response.status_code == 200
assert "厨房箱" in response.text
# Should show error banner — timeout is a real failure
assert "超时" in response.text or "不可用" in response.text
def test_ai_search_empty_query_does_nothing(self, client, db_session):
"""ai=1 with empty query does not trigger AI."""
_enable_ai_search(client, db_session)
response = client.get("/search?ai=1")
assert response.status_code == 200
assert "AI 帮你扩展了" not in response.text
@patch("app.llm._call_chat_completion")
def test_ai_search_disabled_ignores_flag(self, mock_call, client, db_session):
"""ai=1 is ignored when ai_search_enabled is False."""
box = Box(name="厨房箱")
db_session.add(box)
db_session.commit()
# Enable LLM but NOT ai_search_enabled
client.post(
"/settings",
data={
"enabled": "on",
"base_url": "https://api.example.com/v1",
"model": "gpt-4o-mini",
"api_key": "sk-test-key",
},
follow_redirects=False,
)
response = client.get("/search?q=厨房&ai=1")
assert response.status_code == 200
assert "厨房箱" in response.text
assert "AI 帮你扩展了" not in response.text
mock_call.assert_not_called()
# ---------------------------------------------------------------------------
# Button visibility on search page
# ---------------------------------------------------------------------------
class TestAIButtonVisibility:
@patch("app.llm._call_chat_completion")
def test_button_visible_when_configured_and_enabled(self, mock_call, client, db_session):
"""AI search button is visible when ai_search_enabled and configured."""
_enable_ai_search(client, db_session)
response = client.get("/search?q=测试")
assert response.status_code == 200
assert "AI 智能搜索" in response.text
def test_button_hidden_when_not_configured(self, client, db_session):
"""AI search button is hidden when LLM is not configured."""
response = client.get("/search?q=测试")
assert response.status_code == 200
assert "AI 智能搜索" not in response.text
def test_button_hidden_when_ai_search_disabled(self, client, db_session):
"""AI search button is hidden when ai_search_enabled is False."""
client.post(
"/settings",
data={
"enabled": "on",
"base_url": "https://api.example.com/v1",
"model": "gpt-4o-mini",
"api_key": "sk-test-key",
},
follow_redirects=False,
)
response = client.get("/search?q=测试")
assert "AI 智能搜索" not in response.text
@patch("app.llm._call_chat_completion")
def test_button_hidden_on_empty_query(self, mock_call, client, db_session):
"""AI search button is not shown when there's no query."""
_enable_ai_search(client, db_session)
response = client.get("/search")
assert "AI 智能搜索" not in response.text
@patch("app.llm._call_chat_completion")
def test_button_link_includes_current_query(self, mock_call, client, db_session):
"""AI button link includes the current query parameter."""
_enable_ai_search(client, db_session)
response = client.get("/search?q=锅")
assert response.status_code == 200
assert "ai=1" in response.text
from urllib.parse import quote
assert f"q={quote('')}" in response.text or "q=锅" in response.text
@patch("app.llm._call_chat_completion")
def test_no_button_when_ai_already_activated(self, mock_call, client, db_session):
"""When AI is already activated, show status text instead of button."""
mock_call.return_value = {
"choices": [{"message": {"content": '["炒锅"]'}}]
}
_enable_ai_search(client, db_session)
response = client.get("/search?q=锅&ai=1")
assert response.status_code == 200
assert "AI 搜索已启用" in response.text
# ---------------------------------------------------------------------------
# Settings: ai_search_extra_hints
# ---------------------------------------------------------------------------
class TestExtraHintsSettings:
def test_extra_hints_defaults_to_empty(self, db_session):
cfg = get_app_settings(db_session)
assert cfg.ai_search_extra_hints == ""
def test_save_extra_hints(self, db_session):
save_app_settings(db_session, ai_search_extra_hints="用户物品主要涉及厨房")
cfg = get_app_settings(db_session)
assert cfg.ai_search_extra_hints == "用户物品主要涉及厨房"
def test_save_extra_hints_empty_string(self, db_session):
save_app_settings(db_session, ai_search_extra_hints="厨房用品")
save_app_settings(db_session, ai_search_extra_hints="")
cfg = get_app_settings(db_session)
assert cfg.ai_search_extra_hints == ""
def test_settings_page_has_extra_hints_textarea(self, client):
response = client.get("/settings")
assert response.status_code == 200
assert 'name="ai_search_extra_hints"' in response.text
assert "额外领域提示" in response.text
def test_settings_page_has_ai_search_checkbox(self, client):
response = client.get("/settings")
assert response.status_code == 200
assert 'name="ai_search_enabled"' in response.text
assert "启用 AI 智能搜索" in response.text
def test_save_ai_search_settings_via_route(self, client, db_session):
client.post(
"/settings",
data={
"enabled": "on",
"base_url": "https://api.example.com/v1",
"model": "gpt-4o-mini",
"api_key": "sk-key",
"ai_search_enabled": "on",
"ai_search_extra_hints": "用户物品主要涉及厨房用品",
},
follow_redirects=False,
)
cfg = get_app_settings(db_session)
assert cfg.ai_search_enabled is True
assert cfg.ai_search_extra_hints == "用户物品主要涉及厨房用品"
def test_save_preserves_extra_hints_on_other_changes(self, client, db_session):
"""Changing LLM settings should not clear extra hints."""
client.post(
"/settings",
data={
"enabled": "on",
"base_url": "https://api.example.com/v1",
"model": "gpt-4o-mini",
"api_key": "sk-key",
"ai_search_enabled": "on",
"ai_search_extra_hints": "厨房用品和电子产品",
},
follow_redirects=False,
)
client.post(
"/settings",
data={
"enabled": "on",
"base_url": "https://api.example.com/v1",
"model": "gpt-4o",
"api_key": "",
"ai_search_enabled": "on",
"ai_search_extra_hints": "厨房用品和电子产品",
},
follow_redirects=False,
)
cfg = get_app_settings(db_session)
assert cfg.ai_search_extra_hints == "厨房用品和电子产品"
assert cfg.model == "gpt-4o"
# ---------------------------------------------------------------------------
# Regression: existing features still work without AI
# ---------------------------------------------------------------------------
class TestRegressionWithoutAI:
def test_normal_search_still_works(self, client, db_session):
box = Box(name="测试箱")
db_session.add(box)
db_session.commit()
response = client.get("/search?q=测试")
assert response.status_code == 200
assert "测试箱" in response.text
def test_search_page_no_results(self, client):
response = client.get("/search?q=不存在")
assert "没有找到匹配结果" in response.text
def test_search_empty_query(self, client):
response = client.get("/search")
assert "输入关键词后" in response.text
+26 -10
View File
@@ -28,6 +28,7 @@ from app.db import Base, SessionLocal, configure_database
from app.migrate import ( from app.migrate import (
V1_REVISION, V1_REVISION,
_detect_db_state, _detect_db_state,
_make_alembic_config,
run_migrations, run_migrations,
verify_schema_is_current, verify_schema_is_current,
) )
@@ -35,6 +36,18 @@ from app.main import create_app
from fastapi.testclient import TestClient from fastapi.testclient import TestClient
def _get_head_revision() -> str:
"""Resolve the current Alembic head revision from migration scripts."""
from alembic.script import ScriptDirectory
cfg = _make_alembic_config("sqlite:///") # URL is unused for script lookup
script = ScriptDirectory.from_config(cfg)
return script.get_current_head()
HEAD_REVISION = _get_head_revision()
@pytest.fixture() @pytest.fixture()
def tmp_db_path(tmp_path): def tmp_db_path(tmp_path):
"""Provide a temporary SQLite database path.""" """Provide a temporary SQLite database path."""
@@ -55,7 +68,7 @@ def tmp_db_url(tmp_db_path):
class TestFreshDBMigration: class TestFreshDBMigration:
"""Empty database gets all tables created by migration.""" """Empty database gets all tables created by migration."""
def test_creates_all_three_tables(self, tmp_db_url): def test_creates_all_tables(self, tmp_db_url):
run_migrations(tmp_db_url) run_migrations(tmp_db_url)
eng = create_engine(tmp_db_url) eng = create_engine(tmp_db_url)
tables = set(inspect(eng).get_table_names()) tables = set(inspect(eng).get_table_names())
@@ -63,6 +76,7 @@ class TestFreshDBMigration:
assert "boxes" in tables assert "boxes" in tables
assert "items" in tables assert "items" in tables
assert "subitems" in tables assert "subitems" in tables
assert "app_settings" in tables
def test_creates_alembic_version_table(self, tmp_db_url): def test_creates_alembic_version_table(self, tmp_db_url):
run_migrations(tmp_db_url) run_migrations(tmp_db_url)
@@ -77,7 +91,7 @@ class TestFreshDBMigration:
with eng.begin() as conn: with eng.begin() as conn:
version = conn.execute(text("SELECT version_num FROM alembic_version")).scalar() version = conn.execute(text("SELECT version_num FROM alembic_version")).scalar()
eng.dispose() eng.dispose()
assert version == V1_REVISION assert version == HEAD_REVISION
def test_boxes_table_has_all_columns(self, tmp_db_url): def test_boxes_table_has_all_columns(self, tmp_db_url):
run_migrations(tmp_db_url) run_migrations(tmp_db_url)
@@ -147,9 +161,11 @@ class TestUnmanagedDBAdoption2a:
"""Database with existing tables matching baseline gets adopted.""" """Database with existing tables matching baseline gets adopted."""
def _create_old_db(self, db_url: str) -> None: def _create_old_db(self, db_url: str) -> None:
"""Simulate a pre-Alembic DB: create_all + insert data.""" """Simulate a pre-Alembic DB: create V1 tables only + insert data."""
eng = create_engine(db_url) eng = create_engine(db_url)
Base.metadata.create_all(bind=eng) # Only create V1 tables (boxes, items, subitems) — not app_settings
for table_name in ("boxes", "items", "subitems"):
Base.metadata.tables[table_name].create(bind=eng)
with eng.begin() as conn: with eng.begin() as conn:
conn.execute(text( conn.execute(text(
"INSERT INTO boxes (name, room, status, created_at, updated_at) " "INSERT INTO boxes (name, room, status, created_at, updated_at) "
@@ -170,7 +186,7 @@ class TestUnmanagedDBAdoption2a:
with eng.begin() as conn: with eng.begin() as conn:
version = conn.execute(text("SELECT version_num FROM alembic_version")).scalar() version = conn.execute(text("SELECT version_num FROM alembic_version")).scalar()
eng.dispose() eng.dispose()
assert version == V1_REVISION assert version == HEAD_REVISION
def test_data_preserved_after_adoption(self, tmp_db_url): def test_data_preserved_after_adoption(self, tmp_db_url):
self._create_old_db(tmp_db_url) self._create_old_db(tmp_db_url)
@@ -187,14 +203,14 @@ class TestUnmanagedDBAdoption2a:
assert item_count == 1 assert item_count == 1
assert box_name == "Kitchen Box" assert box_name == "Kitchen Box"
def test_no_extra_tables_created(self, tmp_db_url): def test_no_extra_tables_beyond_migrations(self, tmp_db_url):
self._create_old_db(tmp_db_url) self._create_old_db(tmp_db_url)
run_migrations(tmp_db_url) run_migrations(tmp_db_url)
eng = create_engine(tmp_db_url) eng = create_engine(tmp_db_url)
tables = set(inspect(eng).get_table_names()) tables = set(inspect(eng).get_table_names())
eng.dispose() eng.dispose()
assert tables == {"alembic_version", "boxes", "items", "subitems"} assert tables == {"alembic_version", "boxes", "items", "subitems", "app_settings"}
def test_adoption_is_idempotent(self, tmp_db_url): def test_adoption_is_idempotent(self, tmp_db_url):
"""Running run_migrations twice does not error or duplicate data.""" """Running run_migrations twice does not error or duplicate data."""
@@ -209,7 +225,7 @@ class TestUnmanagedDBAdoption2a:
eng.dispose() eng.dispose()
assert box_count == 1 assert box_count == 1
assert version == V1_REVISION assert version == HEAD_REVISION
# --------------------------------------------------------------------------- # ---------------------------------------------------------------------------
@@ -396,7 +412,7 @@ class TestManagedDBMigration:
with eng.begin() as conn: with eng.begin() as conn:
version = conn.execute(text("SELECT version_num FROM alembic_version")).scalar() version = conn.execute(text("SELECT version_num FROM alembic_version")).scalar()
eng.dispose() eng.dispose()
assert version == V1_REVISION assert version == HEAD_REVISION
# --------------------------------------------------------------------------- # ---------------------------------------------------------------------------
@@ -639,7 +655,7 @@ class TestProdDBCopyAdoption:
subitems_after = conn.execute(text("SELECT COUNT(*) FROM subitems")).scalar() subitems_after = conn.execute(text("SELECT COUNT(*) FROM subitems")).scalar()
eng.dispose() eng.dispose()
assert version == V1_REVISION assert version == HEAD_REVISION
assert boxes_after == boxes_before assert boxes_after == boxes_before
assert items_after == items_before assert items_after == items_before
assert subitems_after == subitems_before assert subitems_after == subitems_before
+656
View File
@@ -0,0 +1,656 @@
"""Tests for the settings store, LLM client, and settings routes.
All LLM calls are mocked — CI never touches the network.
"""
from unittest.mock import patch
import pytest
import app.llm as llm_module
from app.llm import LLMResult, expand_query, is_configured
from app.llm import ExpansionResult
from app.models import AppSetting
from app.settings_store import LLMConfig, get_app_settings, save_app_settings
# Alias to avoid pytest collecting it as a test function
_test_connection = llm_module.test_connection
# ---------------------------------------------------------------------------
# LLMConfig dataclass defaults
# ---------------------------------------------------------------------------
class TestLLMConfigDefaults:
def test_default_values(self):
cfg = LLMConfig()
assert cfg.enabled is False
assert cfg.base_url == "https://api.openai.com/v1"
assert cfg.model == ""
assert cfg.api_key == ""
assert cfg.ai_search_enabled is False
assert cfg.ai_search_extra_hints == ""
# ---------------------------------------------------------------------------
# settings_store: get_app_settings
# ---------------------------------------------------------------------------
class TestGetAppSettings:
def test_returns_defaults_when_no_rows(self, db_session):
cfg = get_app_settings(db_session)
assert cfg.enabled is False
assert cfg.base_url == "https://api.openai.com/v1"
assert cfg.model == ""
assert cfg.api_key == ""
assert cfg.ai_search_enabled is False
def test_reads_stored_values(self, db_session):
db_session.add(AppSetting(key="llm_enabled", value="true"))
db_session.add(AppSetting(key="llm_base_url", value="https://custom.api/v1"))
db_session.add(AppSetting(key="llm_model", value="gpt-4o"))
db_session.add(AppSetting(key="llm_api_key", value="sk-test-key"))
db_session.add(AppSetting(key="ai_search_enabled", value="true"))
db_session.commit()
cfg = get_app_settings(db_session)
assert cfg.enabled is True
assert cfg.base_url == "https://custom.api/v1"
assert cfg.model == "gpt-4o"
assert cfg.api_key == "sk-test-key"
assert cfg.ai_search_enabled is True
def test_handles_null_value_as_default(self, db_session):
db_session.add(AppSetting(key="llm_model", value=None))
db_session.commit()
cfg = get_app_settings(db_session)
assert cfg.model == ""
# ---------------------------------------------------------------------------
# settings_store: save_app_settings
# ---------------------------------------------------------------------------
class TestSaveAppSettings:
def test_saves_new_settings(self, db_session):
save_app_settings(
db_session,
enabled=True,
base_url="https://my-api.com/v1",
model="gpt-4o-mini",
api_key="sk-new-key",
)
cfg = get_app_settings(db_session)
assert cfg.enabled is True
assert cfg.base_url == "https://my-api.com/v1"
assert cfg.model == "gpt-4o-mini"
assert cfg.api_key == "sk-new-key"
def test_updates_existing_settings(self, db_session):
save_app_settings(db_session, enabled=True, model="old-model", api_key="key1")
save_app_settings(db_session, model="new-model")
cfg = get_app_settings(db_session)
assert cfg.model == "new-model"
# enabled was not passed in second save, so it stays unchanged
assert cfg.enabled is True
def test_api_key_none_preserves_old_key(self, db_session):
save_app_settings(db_session, api_key="sk-original")
save_app_settings(db_session, model="gpt-4o", api_key=None)
cfg = get_app_settings(db_session)
assert cfg.api_key == "sk-original"
assert cfg.model == "gpt-4o"
def test_api_key_empty_string_overwrites(self, db_session):
save_app_settings(db_session, api_key="sk-original")
save_app_settings(db_session, api_key="")
cfg = get_app_settings(db_session)
assert cfg.api_key == ""
def test_partial_save_only_updates_specified_fields(self, db_session):
save_app_settings(db_session, enabled=True, model="gpt-4o")
save_app_settings(db_session, base_url="https://new.url/v1")
cfg = get_app_settings(db_session)
assert cfg.enabled is True
assert cfg.model == "gpt-4o"
assert cfg.base_url == "https://new.url/v1"
# ---------------------------------------------------------------------------
# is_configured
# ---------------------------------------------------------------------------
class TestIsConfigured:
def test_false_when_disabled(self):
cfg = LLMConfig(enabled=False, model="gpt-4o", api_key="sk-key")
assert is_configured(cfg) is False
def test_false_when_no_model(self):
cfg = LLMConfig(enabled=True, model="", api_key="sk-key")
assert is_configured(cfg) is False
def test_false_when_no_api_key(self):
cfg = LLMConfig(enabled=True, model="gpt-4o", api_key="")
assert is_configured(cfg) is False
def test_true_when_all_set(self):
cfg = LLMConfig(enabled=True, model="gpt-4o", api_key="sk-key")
assert is_configured(cfg) is True
# ---------------------------------------------------------------------------
# test_connection (mocked)
# ---------------------------------------------------------------------------
class TestTestConnection:
def test_returns_failure_when_not_configured(self):
cfg = LLMConfig(enabled=False)
result = _test_connection(cfg)
assert result.success is False
assert "未配置" in result.message
@patch("app.llm._call_chat_completion")
def test_success_when_configured(self, mock_call):
mock_call.return_value = {"choices": [{"message": {"content": "Hi"}}]}
cfg = LLMConfig(enabled=True, model="gpt-4o", api_key="sk-key")
result = _test_connection(cfg)
assert result.success is True
assert "连接成功" in result.message
assert "gpt-4o" in result.message
@patch("app.llm._call_chat_completion")
def test_handles_http_error(self, mock_call):
import httpx
mock_call.side_effect = httpx.HTTPStatusError(
"401",
request=httpx.Request("POST", "http://x"),
response=httpx.Response(401),
)
cfg = LLMConfig(enabled=True, model="gpt-4o", api_key="sk-bad")
result = _test_connection(cfg)
assert result.success is False
assert "401" in result.message
@patch("app.llm._call_chat_completion")
def test_handles_connect_error(self, mock_call):
import httpx
mock_call.side_effect = httpx.ConnectError("refused")
cfg = LLMConfig(enabled=True, model="gpt-4o", api_key="sk-key")
result = _test_connection(cfg)
assert result.success is False
assert "无法连接" in result.message
@patch("app.llm._call_chat_completion")
def test_handles_timeout(self, mock_call):
import httpx
mock_call.side_effect = httpx.TimeoutException("timeout")
cfg = LLMConfig(enabled=True, model="gpt-4o", api_key="sk-key")
result = _test_connection(cfg)
assert result.success is False
assert "超时" in result.message
# ---------------------------------------------------------------------------
# expand_query (mocked)
# ---------------------------------------------------------------------------
class TestExpandQuery:
def test_returns_empty_when_not_configured(self):
cfg = LLMConfig(enabled=False)
result = expand_query(cfg, "")
assert result.terms == []
assert result.error is None
@patch("app.llm._call_chat_completion")
def test_expands_query_successfully(self, mock_call):
mock_call.return_value = {
"choices": [
{"message": {"content": '["平底锅","炒锅","锅具","厨房锅"]'}}
]
}
cfg = LLMConfig(enabled=True, model="gpt-4o", api_key="sk-key")
result = expand_query(cfg, "")
assert "平底锅" in result.terms
assert "炒锅" in result.terms
assert result.error is None
@patch("app.llm._call_chat_completion")
def test_fallback_on_api_failure(self, mock_call):
mock_call.side_effect = Exception("network down")
cfg = LLMConfig(enabled=True, model="gpt-4o", api_key="sk-key")
result = expand_query(cfg, "")
assert result.terms == []
assert result.error is not None
@patch("app.llm._call_chat_completion")
def test_fallback_on_empty_response(self, mock_call):
mock_call.return_value = {"choices": [{"message": {"content": ""}}]}
cfg = LLMConfig(enabled=True, model="gpt-4o", api_key="sk-key")
result = expand_query(cfg, "")
assert result.terms == []
assert result.error is None
# ---------------------------------------------------------------------------
# Routes: GET /settings
# ---------------------------------------------------------------------------
class TestSettingsPage:
def test_settings_page_returns_200(self, client):
response = client.get("/settings")
assert response.status_code == 200
def test_settings_page_has_form_elements(self, client):
response = client.get("/settings")
assert "设置" in response.text
assert 'name="enabled"' in response.text
assert 'name="base_url"' in response.text
assert 'name="model"' in response.text
assert 'name="api_key"' in response.text
assert "保存设置" in response.text
assert "测试连接" in response.text
def test_settings_page_shows_nav_link(self, client):
response = client.get("/boxes")
assert "设置" in response.text
assert 'href="/settings"' in response.text
def test_settings_page_no_api_key_echoed(self, client, db_session):
save_app_settings(db_session, api_key="sk-super-secret-key-12345")
response = client.get("/settings")
assert "sk-super-secret-key-12345" not in response.text
assert "已配置" in response.text
def test_settings_page_shows_placeholder_when_no_key(self, client):
response = client.get("/settings")
assert "输入 API Key" in response.text
def test_settings_page_shows_default_base_url(self, client):
response = client.get("/settings")
assert "https://api.openai.com/v1" in response.text
# ---------------------------------------------------------------------------
# Routes: POST /settings
# ---------------------------------------------------------------------------
class TestSaveSettingsRoute:
def test_save_settings_redirects(self, client):
response = client.post(
"/settings",
data={
"enabled": "on",
"base_url": "https://my-api.com/v1",
"model": "gpt-4o-mini",
"api_key": "sk-test-key",
},
follow_redirects=False,
)
assert response.status_code == 303
assert response.headers["location"] == "/settings"
def test_saved_settings_persist(self, client, db_session):
client.post(
"/settings",
data={
"enabled": "on",
"base_url": "https://my-api.com/v1",
"model": "gpt-4o-mini",
"api_key": "sk-test-key",
},
follow_redirects=False,
)
cfg = get_app_settings(db_session)
assert cfg.enabled is True
assert cfg.base_url == "https://my-api.com/v1"
assert cfg.model == "gpt-4o-mini"
assert cfg.api_key == "sk-test-key"
def test_save_with_blank_api_key_preserves_old(self, client, db_session):
# First save with a key
client.post(
"/settings",
data={"enabled": "on", "model": "gpt-4o", "api_key": "sk-original"},
follow_redirects=False,
)
# Second save without key (blank)
client.post(
"/settings",
data={"enabled": "on", "model": "gpt-4o", "api_key": ""},
follow_redirects=False,
)
cfg = get_app_settings(db_session)
assert cfg.api_key == "sk-original"
def test_save_disabled_state(self, client, db_session):
# First enable
client.post(
"/settings",
data={"enabled": "on", "model": "gpt-4o", "api_key": "sk-key"},
follow_redirects=False,
)
# Then disable (no 'enabled' checkbox)
client.post(
"/settings",
data={"model": "gpt-4o", "api_key": ""},
follow_redirects=False,
)
cfg = get_app_settings(db_session)
assert cfg.enabled is False
def test_save_settings_no_api_key_in_redirect_page(self, client):
client.post(
"/settings",
data={"enabled": "on", "model": "gpt-4o", "api_key": "sk-secret-key"},
follow_redirects=False,
)
response = client.get("/settings")
assert "sk-secret-key" not in response.text
def test_save_refuses_when_base_url_changes_and_key_blank(self, client, db_session):
"""P1 fix: if base_url changes and api_key is blank, refuse save with error."""
client.post(
"/settings",
data={
"enabled": "on",
"base_url": "https://old-api.com/v1",
"model": "gpt-4o",
"api_key": "sk-old-key",
},
follow_redirects=False,
)
# Try saving with different base_url, no key
response = client.post(
"/settings",
data={
"enabled": "on",
"base_url": "https://new-api.com/v1",
"model": "gpt-4o",
"api_key": "", # blank + base_url changed → refuse
},
)
assert response.status_code == 200
assert "请重新输入 API Key 后保存" in response.text
# Old config should be unchanged — nothing was saved
cfg = get_app_settings(db_session)
assert cfg.base_url == "https://old-api.com/v1"
assert cfg.api_key == "sk-old-key"
def test_save_preserves_key_when_endpoint_unchanged_and_key_blank(self, client, db_session):
"""P1 fix: if endpoint is unchanged and api_key is blank, keep old key."""
client.post(
"/settings",
data={
"enabled": "on",
"base_url": "https://api.openai.com/v1",
"model": "gpt-4o",
"api_key": "sk-original",
},
follow_redirects=False,
)
# Re-save same endpoint, blank key
client.post(
"/settings",
data={
"enabled": "on",
"base_url": "https://api.openai.com/v1",
"model": "gpt-4o",
"api_key": "",
},
follow_redirects=False,
)
cfg = get_app_settings(db_session)
assert cfg.api_key == "sk-original"
def test_save_preserves_key_when_only_model_changes_and_key_blank(self, client, db_session):
"""Model change alone should not clear the key — same base_url, different model."""
client.post(
"/settings",
data={
"enabled": "on",
"base_url": "https://api.openai.com/v1",
"model": "gpt-4o",
"api_key": "sk-original",
},
follow_redirects=False,
)
# Change only model, leave api_key blank
client.post(
"/settings",
data={
"enabled": "on",
"base_url": "https://api.openai.com/v1",
"model": "gpt-4o-mini",
"api_key": "",
},
follow_redirects=False,
)
cfg = get_app_settings(db_session)
assert cfg.model == "gpt-4o-mini"
assert cfg.api_key == "sk-original"
def test_save_includes_ai_search_enabled_checkbox(self, client, db_session):
"""Saving settings now also persists the ai_search_enabled checkbox."""
# Set ai_search_enabled to true first
db_session.add(AppSetting(key="ai_search_enabled", value="true"))
db_session.commit()
# Save without the checkbox → ai_search_enabled is set to False
client.post(
"/settings",
data={"enabled": "on", "model": "gpt-4o", "api_key": "sk-key"},
follow_redirects=False,
)
cfg = get_app_settings(db_session)
assert cfg.ai_search_enabled is False
def test_save_preserves_ai_search_enabled_when_checked(self, client, db_session):
"""Saving settings with ai_search_enabled checked persists it."""
client.post(
"/settings",
data={
"enabled": "on",
"model": "gpt-4o",
"api_key": "sk-key",
"ai_search_enabled": "on",
},
follow_redirects=False,
)
cfg = get_app_settings(db_session)
assert cfg.ai_search_enabled is True
# ---------------------------------------------------------------------------
# Routes: POST /settings/test
# ---------------------------------------------------------------------------
class TestTestConnectionRoute:
@patch("app.llm._call_chat_completion")
def test_test_connection_success(self, mock_call, client):
mock_call.return_value = {"choices": [{"message": {"content": "Hi"}}]}
response = client.post(
"/settings/test",
data={
"enabled": "on",
"base_url": "https://api.openai.com/v1",
"model": "gpt-4o",
"api_key": "sk-test",
},
)
assert response.status_code == 200
assert "连接成功" in response.text
assert "gpt-4o" in response.text
@patch("app.llm._call_chat_completion")
def test_test_connection_failure(self, mock_call, client):
import httpx
mock_call.side_effect = httpx.HTTPStatusError(
"401",
request=httpx.Request("POST", "http://x"),
response=httpx.Response(401),
)
response = client.post(
"/settings/test",
data={
"enabled": "on",
"base_url": "https://api.openai.com/v1",
"model": "gpt-4o",
"api_key": "sk-bad",
},
)
assert response.status_code == 200
assert "连接失败" in response.text
assert "401" in response.text
def test_test_connection_not_configured(self, client):
response = client.post(
"/settings/test",
data={
"enabled": "", # not checked
"base_url": "https://api.openai.com/v1",
"model": "",
"api_key": "",
},
)
assert response.status_code == 200
assert "未配置" in response.text
@patch("app.llm._call_chat_completion")
def test_test_connection_uses_stored_key_when_endpoint_matches(self, mock_call, client, db_session):
"""When api_key is blank but base_url and model match saved config, the stored key should be used."""
mock_call.return_value = {"choices": [{"message": {"content": "Hi"}}]}
# Store a config first
save_app_settings(
db_session,
enabled=True,
base_url="https://api.openai.com/v1",
model="gpt-4o",
api_key="sk-stored-key",
)
response = client.post(
"/settings/test",
data={
"enabled": "on",
"base_url": "https://api.openai.com/v1",
"model": "gpt-4o",
"api_key": "", # blank → use stored key (endpoint matches)
},
)
assert response.status_code == 200
assert "连接成功" in response.text
@patch("app.llm._call_chat_completion")
def test_test_connection_uses_stored_key_when_only_model_changes(self, mock_call, client, db_session):
"""Model changes under the same base_url can reuse the stored key."""
captured = {}
def fake_call(cfg, **kwargs):
captured["base_url"] = cfg.base_url
captured["model"] = cfg.model
captured["api_key"] = cfg.api_key
return {"choices": [{"message": {"content": "Hi"}}]}
mock_call.side_effect = fake_call
save_app_settings(
db_session,
enabled=True,
base_url="https://api.openai.com/v1",
model="gpt-4o",
api_key="sk-stored-key",
)
response = client.post(
"/settings/test",
data={
"enabled": "on",
"base_url": "https://api.openai.com/v1",
"model": "gpt-4o-mini",
"api_key": "",
},
)
assert response.status_code == 200
assert "连接成功" in response.text
assert captured == {
"base_url": "https://api.openai.com/v1",
"model": "gpt-4o-mini",
"api_key": "sk-stored-key",
}
def test_test_connection_refuses_stored_key_when_endpoint_changed(self, client, db_session):
"""When base_url changed and api_key is blank, refuse to test."""
save_app_settings(
db_session,
enabled=True,
base_url="https://api.openai.com/v1",
model="gpt-4o",
api_key="sk-stored-key",
)
response = client.post(
"/settings/test",
data={
"enabled": "on",
"base_url": "https://attacker.example/v1", # different endpoint
"model": "gpt-4o",
"api_key": "", # blank → refuse
},
)
assert response.status_code == 200
assert "请重新输入 API Key" in response.text
def test_test_connection_result_shows_on_settings_page(self, client):
"""Test result is rendered on the same settings page."""
response = client.post(
"/settings/test",
data={
"enabled": "",
"base_url": "https://api.openai.com/v1",
"model": "",
"api_key": "",
},
)
assert response.status_code == 200
assert "设置" in response.text
assert "保存设置" in response.text
# ---------------------------------------------------------------------------
# Graceful degradation: unconfigured LLM does not affect existing features
# ---------------------------------------------------------------------------
class TestGracefulDegradation:
def test_boxes_page_works_without_llm_config(self, client):
response = client.get("/boxes")
assert response.status_code == 200
def test_search_page_works_without_llm_config(self, client):
response = client.get("/search?q=test")
assert response.status_code == 200
def test_crud_works_without_llm_config(self, client, db_session):
from app.models import Box
response = client.post(
"/boxes",
data={"name": "No LLM Box"},
follow_redirects=False,
)
assert response.status_code == 303
assert db_session.query(Box).count() == 1