diff --git a/app/core/config.py b/app/core/config.py index 8eb4923..b9568ce 100644 --- a/app/core/config.py +++ b/app/core/config.py @@ -54,6 +54,8 @@ class Settings(BaseSettings): somni_mongo_answers_collection: str = "somni_quiz_answers" sim_threshold: float = 0.7 # 内容形态向量模糊命中阈值（规范 §五-2） + # GetAudio query_text 与根标签向量相似度下限 + get_audio_root_tag_sim_threshold: float = 0.85 # 多路文本检索厌恶硬剔除阈值；≥ 该值 penalty=1.0 丢弃候选 strong_dislike_sim_threshold: float = 0.85 search_sleep_stage_filter_enabled: bool = True # 检索步骤 1 是否按睡眠阶段过滤 @@ -93,6 +95,7 @@ class Settings(BaseSettings): redis_max_connections: int = 512 search_cache_max_size: int = 2048 search_cache_ttl_sec: int = 604800 # 7 天 + somni_audio_catalog_cache_ttl_sec: float = 60.0 # CUD 后延时重建睡眠阶段候选缓存，窗口内多次写入只重建一次 sleep_stage_cache_rewarm_delay_sec: float = 5.0 diff --git a/app/es/search.py b/app/es/search.py index 14c4325..a7a3129 100644 --- a/app/es/search.py +++ b/app/es/search.py @@ -97,9 +97,20 @@ def _candidate_search_body(query: dict[str, Any], *, size: int = 1000) -> dict[s class EsSearch: """封装检索相关的 ES 查询与文档解析。""" - def __init__(self, client: AsyncElasticsearch, settings: Settings) -> None: + def __init__( + self, + client: AsyncElasticsearch, + settings: Settings, + *, + audio_index: str | None = None, + tag_dictionary_index: str | None = None, + ) -> None: self._client = client self._settings = settings + self._audio_index = audio_index or settings.es_audio_index + self._tag_dictionary_index = ( + tag_dictionary_index or settings.es_tag_vectors_index + ) self._content_tag_vectors_cache: list[dict[str, Any]] | None = None self._content_tag_vectors_lock = asyncio.Lock() # 按 tag_id 缓存 name_vector，避免每请求 mget（内容准入模糊路径） @@ -108,11 +119,11 @@ def __init__(self, client: AsyncElasticsearch, settings: Settings) -> None: @property def audio_index(self) -> str: - return self._settings.es_audio_index + return self._audio_index @property def tag_dictionary_index(self) -> str: - return self._settings.es_tag_vectors_index + return self._tag_dictionary_index @property def tag_vectors_index(self) -> str: @@ -321,6 +332,7 @@ async def _fetch_content_tag_vectors(self, *, size: int) -> list[dict[str, Any]] "label": label, "dimension": source.get("type", ""), "vector": vector, + "parent_tag_id": str(source.get("parent_tag_id") or ""), } ) return tags @@ -380,6 +392,18 @@ def content_tag_ids(tags: AudioTags) -> list[str]: ids.extend(item.vector_id for item in dim) return ids + async def list_audio_catalog_docs(self, *, size: int) -> list[dict[str, Any]]: + """量产 GetAudio：音频全量（不含 embedding），供内存过滤。""" + response = await self._client.search( + index=self.audio_index, + body={ + "query": {"match_all": {}}, + "size": max(1, size), + "_source": {"excludes": ["embedding", "description_vector"]}, + }, + ) + return [_document_from_hit(hit) for hit in response["hits"]["hits"]] + async def migrate_legacy_indices(self) -> None: """删除旧版 audio_materials / tag_vectors 索引。""" for index in LEGACY_INDICES: diff --git a/app/main.py b/app/main.py index c0808af..7607580 100644 --- a/app/main.py +++ b/app/main.py @@ -63,6 +63,7 @@ def _bootstrap_dev_entry() -> None: from app.server.bootstrap import GrpcServers, start_grpc_servers, stop_grpc_servers from app.server.handboard.audio.service import AudioService from app.server.handboard.audio.store import MaterialsStore, create_materials_store +from app.server.somni.audio.catalog import AudioCatalogService as SomniAudioService from app.server.somni.quiz.service import QuizService as SomniQuizService from app.server.somni.report.service import ReportService as SomniReportService from app.services.retrieval import RetrievalService @@ -73,6 +74,7 @@ def _bootstrap_dev_entry() -> None: class AppState: settings: Settings es_client: AsyncElasticsearch | None = None + somni_es_client: AsyncElasticsearch | None = None encoder: Encoder | None = None materials_store: MaterialsStore | None = None somni_mongo_client: AsyncIOMotorClient | None = None @@ -82,6 +84,7 @@ class AppState: audio_service: AudioService | None = None somni_quiz_service: SomniQuizService | None = None somni_report_service: SomniReportService | None = None + somni_audio_service: SomniAudioService | None = None search_cache: AudioSearchCache | None = None sleep_stage_cache: SleepStageCandidateCache | None = None grpc_servers: GrpcServers | None = None @@ -171,10 +174,27 @@ async def lifespan(app: FastAPI) -> AsyncIterator[None]: somni_mongo = AsyncIOMotorClient(settings.somni_mongo_uri) _app_state.somni_mongo_client = somni_mongo else: - logger.warning("未配置 SOMNI_MONGO_URI，量产问卷将不可用") + logger.warning("未配置 SOMNI_MONGO_URI，量产问卷与音频查询将不可用") _app_state.somni_quiz_service = SomniQuizService(somni_mongo, settings) _app_state.somni_report_service = SomniReportService() + somni_es_client = create_es_client( + settings, + node=settings.effective_somni_es_node, + ) + _app_state.somni_es_client = somni_es_client + somni_es_search = EsSearch( + somni_es_client, + settings, + audio_index=settings.somni_es_audio_index, + tag_dictionary_index=settings.somni_es_tag_vectors_index, + ) + _app_state.somni_audio_service = SomniAudioService( + somni_mongo, + settings, + es_search=somni_es_search, + encoder=encoder, + ) start_sync_scheduler(_app_state, settings) _app_state.grpc_servers = await start_grpc_servers(_app_state, settings) @@ -191,6 +211,7 @@ async def lifespan(app: FastAPI) -> AsyncIterator[None]: materials_store.close() if somni_mongo is not None: somni_mongo.close() + await somni_es_client.close() await es_client.close() diff --git a/app/server/bootstrap.py b/app/server/bootstrap.py index 1346245..cf79b4d 100644 --- a/app/server/bootstrap.py +++ b/app/server/bootstrap.py @@ -6,13 +6,20 @@ from typing import TYPE_CHECKING import grpc +from grpc_reflection.v1alpha import reflection from loguru import logger from app.server.handboard.audio.rpc import AudioRpc as HandboardAudioRpc from app.server.handboard.quiz.rpc import QuizRpc as HandboardQuizRpc +from app.server.somni.audio.rpc import AudioRpc as SomniAudioRpc from app.server.somni.quiz.rpc import QuizRpc as SomniQuizRpc from app.server.somni.report.rpc import ReportRpc as SomniReportRpc -from app.uburnode_grpc.grpc_gen import uburnode_pb2_grpc, uburnode_somni_pb2_grpc +from app.uburnode_grpc.grpc_gen import ( + uburnode_pb2, + uburnode_pb2_grpc, + uburnode_somni_pb2, + uburnode_somni_pb2_grpc, +) if TYPE_CHECKING: from app.core.config import Settings @@ -55,6 +62,7 @@ async def _start_handboard(state: AppState, settings: Settings) -> grpc.aio.Serv server, ) uburnode_pb2_grpc.add_QuizServiceServicer_to_server(HandboardQuizRpc(), server) + _enable_reflection(server, uburnode_pb2) bind = f"{settings.grpc_host}:{settings.grpc_port}" _bind(server, bind, "功能手板") await server.start() @@ -71,12 +79,23 @@ async def _start_somni(state: AppState, settings: Settings) -> grpc.aio.Server: SomniReportRpc(getattr(state, "somni_report_service", None)), server, ) + uburnode_somni_pb2_grpc.add_AudioServiceServicer_to_server( + SomniAudioRpc(getattr(state, "somni_audio_service", None)), + server, + ) + _enable_reflection(server, uburnode_somni_pb2) bind = f"{settings.grpc_host}:{settings.somni_grpc_port}" _bind(server, bind, "量产") await server.start() return server +def _enable_reflection(server: grpc.aio.Server, proto_module) -> None: + names = [reflection.SERVICE_NAME] + names.extend(svc.full_name for svc in proto_module.DESCRIPTOR.services_by_name.values()) + reflection.enable_server_reflection(tuple(names), server) + + def _bind(server: grpc.aio.Server, bind: str, label: str) -> None: if server.add_insecure_port(bind) == 0: raise RuntimeError(f"{label} gRPC 无法绑定 {bind}") diff --git a/app/server/somni/audio/catalog.py b/app/server/somni/audio/catalog.py new file mode 100644 index 0000000..16a8963 --- /dev/null +++ b/app/server/somni/audio/catalog.py @@ -0,0 +1,255 @@ +"""量产音频目录查询：标签词典 + 音频原料。""" + +from __future__ import annotations + +import asyncio +import math +from time import monotonic +from typing import Any + +from motor.motor_asyncio import AsyncIOMotorClient, AsyncIOMotorCollection + +from app.core.bson_util import bson_to_jsonable +from app.core.codes import HttpStatus +from app.core.config import Settings +from app.core.exceptions import AppError, EncoderNotReadyError +from app.embedding.encoder import Encoder +from app.es.search import EsSearch + +_TAG_ENABLED = "启用" +_CONTENT_FORM = "content_form" + + +class InvalidAudioQueryError(AppError): + def __init__(self, message: str) -> None: + super().__init__(message=message, status_code=HttpStatus.BAD_REQUEST) + + +class AudioCatalogService: + def __init__( + self, + client: AsyncIOMotorClient | None, + settings: Settings, + *, + es_search: EsSearch | None = None, + encoder: Encoder | None = None, + ) -> None: + self._client = client + self._settings = settings + self._es_search = es_search + self._encoder = encoder + self._audio_cache: dict[bool, tuple[float, list[dict[str, Any]]]] = {} + self._audio_cache_lock = asyncio.Lock() + + async def get_audio_tag(self) -> dict[str, Any]: + collection = self._tags() + query = _root_tag_query() + total = await collection.count_documents(query) + self._reject_over_limit(total) + cursor = collection.find(query, {"type": 1, "code": 1, "name": 1, "name_en": 1}) + docs = [bson_to_jsonable(doc) async for doc in cursor] + return {"tags": [_map_tag_dict(doc) for doc in docs]} + + async def get_audio( + self, + *, + page: int | None, + page_size: int | None, + fetch_all: bool, + query_text: str, + tag_code: str, + ) -> dict[str, Any]: + text = query_text.strip() + code = tag_code.strip() + docs = await self._load_audios(from_es=bool(text)) + if code: + docs = [doc for doc in docs if _has_content_form_code(doc, code)] + if text: + tag_ids = await self._root_tag_ids_by_text(text) + docs = [doc for doc in docs if _has_root_content_form_id(doc, tag_ids)] + return _paginate_docs(docs, page, page_size, fetch_all, self._settings) + + async def get_hot(self) -> None: + return None + + async def _load_audios(self, *, from_es: bool) -> list[dict[str, Any]]: + now = monotonic() + cached = self._audio_cache.get(from_es) + if cached is not None and self._is_cache_fresh(cached[0], now): + return cached[1] + async with self._audio_cache_lock: + cached = self._audio_cache.get(from_es) + if cached is not None and self._is_cache_fresh(cached[0], now): + return cached[1] + raw = await self._fetch_audios_es() if from_es else await self._fetch_audios_mongo() + docs = [_map_material(doc) for doc in raw] + self._audio_cache[from_es] = (now, docs) + return docs + + def _is_cache_fresh(self, loaded_at: float, now: float) -> bool: + ttl = self._settings.somni_audio_catalog_cache_ttl_sec + return ttl > 0 and now - loaded_at < ttl + + async def _fetch_audios_mongo(self) -> list[dict[str, Any]]: + collection = self._materials() + total = await collection.count_documents({}) + self._reject_over_limit(total) + cursor = collection.find({}, {"embedding": 0}) + return [bson_to_jsonable(doc) async for doc in cursor] + + async def _fetch_audios_es(self) -> list[dict[str, Any]]: + if self._es_search is None: + raise AppError( + message="Elasticsearch 未就绪，无法按搜索词查询音频", + status_code=HttpStatus.SERVICE_UNAVAILABLE, + ) + docs = await self._es_search.list_audio_catalog_docs( + size=self._settings.fetch_all_hard_limit + 1, + ) + self._reject_over_limit(len(docs)) + return docs + + async def _root_tag_ids_by_text(self, text: str) -> set[str]: + if self._encoder is None or not self._encoder.is_loaded: + raise EncoderNotReadyError() + if self._es_search is None: + raise AppError( + message="Elasticsearch 未就绪，无法按搜索词匹配标签", + status_code=HttpStatus.SERVICE_UNAVAILABLE, + ) + query_vector = await self._encoder.encode_one(text) + tags = await self._es_search.list_content_tag_vectors() + threshold = self._settings.get_audio_root_tag_sim_threshold + matched: set[str] = set() + for tag in tags: + if not _is_root_content_form_dict(tag): + continue + vector = tag.get("vector") + if not isinstance(vector, list) or not vector: + continue + if _cosine_similarity(query_vector, vector) <= threshold: + continue + tag_id = str(tag.get("id") or "").strip() + if tag_id: + matched.add(tag_id) + return matched + + def _tags(self) -> AsyncIOMotorCollection: + return self._db()[self._settings.somni_mongo_tag_dictionary_collection] + + def _materials(self) -> AsyncIOMotorCollection: + return self._db()[self._settings.somni_mongo_materials_collection] + + def _db(self): + if self._client is None: + raise AppError( + message="量产 Mongo 未配置（SOMNI_MONGO_URI），无法查询音频", + status_code=HttpStatus.SERVICE_UNAVAILABLE, + ) + return self._client[self._settings.somni_mongo_db] + + def _reject_over_limit(self, total: int) -> None: + limit = self._settings.fetch_all_hard_limit + if total > limit: + raise InvalidAudioQueryError(f"全量条数超过上限 {limit}") + + +def _root_tag_query() -> dict[str, Any]: + return { + "status": _TAG_ENABLED, + "$or": [ + {"parent_tag_id": {"$exists": False}}, + {"parent_tag_id": None}, + {"parent_tag_id": ""}, + ], + } + + +def _paginate_docs( + docs: list[dict[str, Any]], + page: int | None, + page_size: int | None, + fetch_all: bool, + settings: Settings, +) -> dict[str, Any]: + total = len(docs) + if fetch_all: + if total > settings.fetch_all_hard_limit: + raise InvalidAudioQueryError(f"全量条数超过上限 {settings.fetch_all_hard_limit}") + return {"materials": docs, "page": _page_info(1, len(docs), total, 1)} + cur_page, size = _page_window(page, page_size, settings) + start = (cur_page - 1) * size + chunk = docs[start : start + size] + pages = math.ceil(total / size) if size else 0 + return {"materials": chunk, "page": _page_info(cur_page, size, total, pages)} + + +def _page_window( + page: int | None, + page_size: int | None, + settings: Settings, +) -> tuple[int, int]: + cur_page = 1 if page is None else page + size = settings.default_page_size if page_size is None else page_size + if cur_page < 1 or size < 1: + raise InvalidAudioQueryError("page / page_size 须 ≥ 1") + return cur_page, min(size, settings.max_page_size) + + +def _page_info(page: int, page_size: int, total: int, total_pages: int) -> dict[str, int]: + return {"page": page, "page_size": page_size, "total": total, "total_pages": total_pages} + + +def _map_tag_dict(doc: dict[str, Any]) -> dict[str, str]: + return { + "type": str(doc.get("type") or ""), + "code": str(doc.get("code") or ""), + "name": str(doc.get("name") or ""), + "name_en": str(doc.get("name_en") or ""), + } + + +def _map_material(doc: dict[str, Any]) -> dict[str, Any]: + mapped = dict(doc) + mapped.pop("embedding", None) + mapped["id"] = str(doc.get("id") or doc.get("_id") or "") + mapped.pop("_id", None) + return mapped + + +def _is_blank(value: Any) -> bool: + return value is None or str(value).strip() in ("", "None") + + +def _is_root_content_form_dict(tag: dict[str, Any]) -> bool: + dimension = str(tag.get("dimension") or tag.get("type") or "") + return dimension == _CONTENT_FORM and _is_blank(tag.get("parent_tag_id")) + + +def _has_content_form_code(doc: dict[str, Any], tag_code: str) -> bool: + for item in doc.get("content_form_tags") or []: + if isinstance(item, dict) and str(item.get("code") or "") == tag_code: + return True + return False + + +def _has_root_content_form_id(doc: dict[str, Any], tag_ids: set[str]) -> bool: + if not tag_ids: + return False + for item in doc.get("content_form_tags") or []: + if not isinstance(item, dict) or not _is_blank(item.get("parent_tag_id")): + continue + if str(item.get("tag_id") or "") in tag_ids: + return True + return False + + +def _cosine_similarity(left: list[float], right: list[float]) -> float: + if len(left) != len(right) or not left: + return 0.0 + dot = sum(x * y for x, y in zip(left, right, strict=True)) + norm_left = math.sqrt(sum(x * x for x in left)) + norm_right = math.sqrt(sum(y * y for y in right)) + if norm_left == 0 or norm_right == 0: + return 0.0 + return dot / (norm_left * norm_right) diff --git a/app/server/somni/audio/rpc.py b/app/server/somni/audio/rpc.py index f3bd3f8..b0a4d94 100644 --- a/app/server/somni/audio/rpc.py +++ b/app/server/somni/audio/rpc.py @@ -4,104 +4,86 @@ from typing import TYPE_CHECKING, Any -from google.protobuf.json_format import ParseDict from google.protobuf.struct_pb2 import Struct from app.core.exceptions import ServiceNotReadyError -from app.server.errors import abort_from_app_error, abort_invalid, run_rpc_call +from app.server.errors import abort_from_app_error, run_rpc_call from app.uburnode_grpc.grpc_gen import uburnode_somni_pb2, uburnode_somni_pb2_grpc if TYPE_CHECKING: - from app.server.somni.audio.service import SomniAudioService + from app.server.somni.audio.catalog import AudioCatalogService class AudioRpc(uburnode_somni_pb2_grpc.AudioServiceServicer): - def __init__(self, service: SomniAudioService | None) -> None: + def __init__(self, service: AudioCatalogService | None) -> None: self._service = service - async def ListTags(self, request, context): + async def GetAudio(self, request, context): service = await self._require(context) async def _do(): - data = await service.list_tags( - page=request.page if request.HasField("page") else None, - page_size=request.page_size if request.HasField("page_size") else None, - fetch_all=bool(request.fetch_all) if request.HasField("fetch_all") else False, - type_=request.type if request.HasField("type") else None, - enabled_only=( - bool(request.enabled_only) if request.HasField("enabled_only") else False - ), - level=request.level if request.HasField("level") else 0, - ) - return _list_tags_res(data) + payload = await service.get_audio(**_get_audio_kwargs(request)) + return _to_audio_res(payload) return await run_rpc_call(context, _do) - async def ListAudios(self, request, context): + async def GetAudioTag(self, request, context): + del request service = await self._require(context) async def _do(): - data = await service.list_audios( - page=request.page if request.HasField("page") else None, - page_size=request.page_size if request.HasField("page_size") else None, - fetch_all=bool(request.fetch_all) if request.HasField("fetch_all") else False, - enabled_only=( - bool(request.enabled_only) if request.HasField("enabled_only") else False - ), - tags=list(request.tags), - ) - return _list_audios_res(data) + payload = await service.get_audio_tag() + return _to_tag_res(payload) return await run_rpc_call(context, _do) - async def SearchAudio(self, request, context): - if not request.query_text.strip(): - await abort_invalid(context, "query_text 不能为空") + async def GetHot(self, request, context): + del request service = await self._require(context) async def _do(): - top_k = request.top_k if request.HasField("top_k") else None - data = await service.search_audio(request.query_text, top_k) - res = uburnode_somni_pb2.SearchAudioRes() - for item in data.materials: - struct = Struct() - struct.update(item if isinstance(item, dict) else {}) - res.materials.append(struct) - return res + await service.get_hot() + return uburnode_somni_pb2.GetHotRes() return await run_rpc_call(context, _do) - async def _require(self, context) -> SomniAudioService: + async def _require(self, context) -> AudioCatalogService: if self._service is None: await abort_from_app_error(context, ServiceNotReadyError()) return self._service # type: ignore[return-value] -def _list_tags_res(data: dict[str, Any]) -> uburnode_somni_pb2.ListTagsRes: - res = uburnode_somni_pb2.ListTagsRes() - for tag in data.get("tags", []): - msg = uburnode_somni_pb2.Tag() - ParseDict(tag, msg, ignore_unknown_fields=True) - res.tags.append(msg) - page = data.get("page") or {} - res.page.CopyFrom( - uburnode_somni_pb2.PageInfo( - page=int(page.get("page") or 1), - page_size=int(page.get("page_size") or 0), - total=int(page.get("total") or 0), - total_pages=int(page.get("total_pages") or 0), +def _get_audio_kwargs(request) -> dict[str, Any]: + return { + "page": request.page if request.HasField("page") else None, + "page_size": request.page_size if request.HasField("page_size") else None, + "fetch_all": bool(request.fetch_all) if request.HasField("fetch_all") else False, + "query_text": request.query_text if request.HasField("query_text") else "", + "tag_code": request.tag_code if request.HasField("tag_code") else "", + } + + +def _to_tag_res(payload: dict[str, Any]) -> uburnode_somni_pb2.GetAudioTagRes: + res = uburnode_somni_pb2.GetAudioTagRes() + for item in payload.get("tags") or []: + res.tags.append( + uburnode_somni_pb2.TagDictItem( + type=str(item.get("type") or ""), + code=str(item.get("code") or ""), + name=str(item.get("name") or ""), + name_en=str(item.get("name_en") or ""), + ) ) - ) return res -def _list_audios_res(data: dict[str, Any]) -> uburnode_somni_pb2.ListAudiosRes: - res = uburnode_somni_pb2.ListAudiosRes() - for material in data.get("materials", []): - msg = uburnode_somni_pb2.AudioMaterial() - ParseDict(material, msg, ignore_unknown_fields=True) - res.materials.append(msg) - page = data.get("page") or {} +def _to_audio_res(payload: dict[str, Any]) -> uburnode_somni_pb2.GetAudioRes: + res = uburnode_somni_pb2.GetAudioRes() + for item in payload.get("materials") or []: + struct = Struct() + struct.update(item if isinstance(item, dict) else {}) + res.materials.append(struct) + page = payload.get("page") or {} res.page.CopyFrom( uburnode_somni_pb2.PageInfo( page=int(page.get("page") or 1), diff --git a/app/server/somni/quiz/rpc.py b/app/server/somni/quiz/rpc.py index fc17b47..593d1ed 100644 --- a/app/server/somni/quiz/rpc.py +++ b/app/server/somni/quiz/rpc.py @@ -2,11 +2,9 @@ from __future__ import annotations -import json from typing import TYPE_CHECKING, Any -from google.protobuf.json_format import Parse -from google.protobuf.struct_pb2 import Value +from google.protobuf.json_format import ParseDict from app.core.exceptions import ServiceNotReadyError from app.server.errors import abort_from_app_error, abort_invalid, run_rpc_call @@ -40,19 +38,18 @@ async def _require(self, context) -> QuizService: def _to_res(payload: dict[str, Any]) -> uburnode_somni_pb2.GetAnswerRes: res = uburnode_somni_pb2.GetAnswerRes() for item in payload.get("answers") or []: - answer = uburnode_somni_pb2.AnswerItem( - question_id=str(item.get("question_id") or ""), - input_type=str(item.get("input_type") or ""), - title=str(item.get("title") or ""), - tags=[str(tag) for tag in (item.get("tags") or [])], - extra_input=str(item.get("extra_input") or ""), - ) - _assign_value(answer.value, item.get("value")) - res.answers.append(answer) + res.answers.append(_to_item(item if isinstance(item, dict) else {})) return res -def _assign_value(target: Value, raw: Any) -> None: - if raw is None: - return - Parse(json.dumps(raw, ensure_ascii=False), target) +def _to_item(item: dict[str, Any]) -> uburnode_somni_pb2.AnswerItem: + answer = uburnode_somni_pb2.AnswerItem( + question_id=str(item.get("question_id") or ""), + input_type=str(item.get("input_type") or ""), + title=str(item.get("title") or ""), + extra_input=str(item.get("extra_input") or ""), + ) + raw = item.get("value") + if raw is not None: + ParseDict(raw, answer.value) + return answer diff --git a/app/server/somni/quiz/service.py b/app/server/somni/quiz/service.py index e213e30..31a97ab 100644 --- a/app/server/somni/quiz/service.py +++ b/app/server/somni/quiz/service.py @@ -49,14 +49,10 @@ def _normalize_answer(item: Any) -> dict[str, Any]: message="答卷明细格式非法", status_code=HttpStatus.INTERNAL_SERVER_ERROR, ) - tags = item.get("tags") or [] - if not isinstance(tags, list): - tags = [] return { "question_id": str(item.get("question_id") or ""), "input_type": str(item.get("input_type") or ""), "title": str(item.get("title") or ""), - "tags": [str(tag) for tag in tags], "value": item.get("value"), "extra_input": str(item.get("extra_input") or ""), } diff --git a/app/uburnode_grpc/grpc_gen/uburnode_somni_pb2.py b/app/uburnode_grpc/grpc_gen/uburnode_somni_pb2.py index 3bc6f84..9edcc40 100644 --- a/app/uburnode_grpc/grpc_gen/uburnode_somni_pb2.py +++ b/app/uburnode_grpc/grpc_gen/uburnode_somni_pb2.py @@ -25,7 +25,7 @@ from google.protobuf import struct_pb2 as google_dot_protobuf_dot_struct__pb2 -DESCRIPTOR = _descriptor_pool.Default().AddSerializedFile(b'\n\x14uburnode_somni.proto\x12\x11uburnode.somni.v1\x1a\x1cgoogle/protobuf/struct.proto\".\n\x0cGetAnswerReq\x12\x0b\n\x03uid\x18\x01 \x01(\t\x12\x11\n\tanswer_id\x18\x02 \x01(\t\"\x8e\x01\n\nAnswerItem\x12\x13\n\x0bquestion_id\x18\x01 \x01(\t\x12\x12\n\ninput_type\x18\x02 \x01(\t\x12\r\n\x05title\x18\x03 \x01(\t\x12\x0c\n\x04tags\x18\x04 \x03(\t\x12%\n\x05value\x18\x05 \x01(\x0b\x32\x16.google.protobuf.Value\x12\x13\n\x0b\x65xtra_input\x18\x06 \x01(\t\">\n\x0cGetAnswerRes\x12.\n\x07\x61nswers\x18\x01 \x03(\x0b\x32\x1d.uburnode.somni.v1.AnswerItem\"1\n\rReportDateReq\x12\x0b\n\x03uid\x18\x01 \x01(\t\x12\x13\n\x0brecord_date\x18\x02 \x01(\t\"\x0f\n\rGetSummaryRes\"\x0e\n\x0cGetEventsRes\"\x13\n\x11GetEnvironmentRes\"\x11\n\x0fGetStructureRes\"\x14\n\x12GetSleepQualityRes2\\\n\x0bQuizService\x12M\n\tGetAnswer\x12\x1f.uburnode.somni.v1.GetAnswerReq\x1a\x1f.uburnode.somni.v1.GetAnswerRes2\xbd\x03\n\rReportService\x12P\n\nGetSummary\x12 .uburnode.somni.v1.ReportDateReq\x1a .uburnode.somni.v1.GetSummaryRes\x12N\n\tGetEvents\x12 .uburnode.somni.v1.ReportDateReq\x1a\x1f.uburnode.somni.v1.GetEventsRes\x12X\n\x0eGetEnvironment\x12 .uburnode.somni.v1.ReportDateReq\x1a$.uburnode.somni.v1.GetEnvironmentRes\x12T\n\x0cGetStructure\x12 .uburnode.somni.v1.ReportDateReq\x1a\".uburnode.somni.v1.GetStructureRes\x12Z\n\x0fGetSleepQuality\x12 .uburnode.somni.v1.ReportDateReq\x1a%.uburnode.somni.v1.GetSleepQualityResb\x06proto3') +DESCRIPTOR = _descriptor_pool.Default().AddSerializedFile(b'\n\x14uburnode_somni.proto\x12\x11uburnode.somni.v1\x1a\x1cgoogle/protobuf/struct.proto\".\n\x0cGetAnswerReq\x12\x0b\n\x03uid\x18\x01 \x01(\t\x12\x11\n\tanswer_id\x18\x02 \x01(\t\"\x94\x01\n\nAnswerItem\x12\x13\n\x0bquestion_id\x18\x01 \x01(\t\x12\x12\n\ninput_type\x18\x02 \x01(\t\x12\r\n\x05title\x18\x03 \x01(\t\x12%\n\x05value\x18\x05 \x01(\x0b\x32\x16.google.protobuf.Value\x12\x13\n\x0b\x65xtra_input\x18\x06 \x01(\tJ\x04\x08\x04\x10\x05R\x04tagsR\x06values\">\n\x0cGetAnswerRes\x12.\n\x07\x61nswers\x18\x01 \x03(\x0b\x32\x1d.uburnode.somni.v1.AnswerItem\"1\n\rReportDateReq\x12\x0b\n\x03uid\x18\x01 \x01(\t\x12\x13\n\x0brecord_date\x18\x02 \x01(\t\"\x0f\n\rGetSummaryRes\"\x0e\n\x0cGetEventsRes\"\x13\n\x11GetEnvironmentRes\"\x11\n\x0fGetStructureRes\"\x14\n\x12GetSleepQualityRes\"\xc1\x01\n\x0bGetAudioReq\x12\x11\n\x04page\x18\x01 \x01(\x05H\x00\x88\x01\x01\x12\x16\n\tpage_size\x18\x02 \x01(\x05H\x01\x88\x01\x01\x12\x16\n\tfetch_all\x18\x03 \x01(\x08H\x02\x88\x01\x01\x12\x17\n\nquery_text\x18\x04 \x01(\tH\x03\x88\x01\x01\x12\x15\n\x08tag_code\x18\x05 \x01(\tH\x04\x88\x01\x01\x42\x07\n\x05_pageB\x0c\n\n_page_sizeB\x0c\n\n_fetch_allB\r\n\x0b_query_textB\x0b\n\t_tag_code\"O\n\x08PageInfo\x12\x0c\n\x04page\x18\x01 \x01(\x05\x12\x11\n\tpage_size\x18\x02 \x01(\x05\x12\r\n\x05total\x18\x03 \x01(\x05\x12\x13\n\x0btotal_pages\x18\x04 \x01(\x05\"d\n\x0bGetAudioRes\x12*\n\tmaterials\x18\x01 \x03(\x0b\x32\x17.google.protobuf.Struct\x12)\n\x04page\x18\x02 \x01(\x0b\x32\x1b.uburnode.somni.v1.PageInfo\"\x10\n\x0eGetAudioTagReq\"H\n\x0bTagDictItem\x12\x0c\n\x04type\x18\x01 \x01(\t\x12\x0c\n\x04\x63ode\x18\x02 \x01(\t\x12\x0c\n\x04name\x18\x03 \x01(\t\x12\x0f\n\x07name_en\x18\x04 \x01(\t\">\n\x0eGetAudioTagRes\x12,\n\x04tags\x18\x01 \x03(\x0b\x32\x1e.uburnode.somni.v1.TagDictItem\"\x0b\n\tGetHotReq\"\x0b\n\tGetHotRes2\\\n\x0bQuizService\x12M\n\tGetAnswer\x12\x1f.uburnode.somni.v1.GetAnswerReq\x1a\x1f.uburnode.somni.v1.GetAnswerRes2\xbd\x03\n\rReportService\x12P\n\nGetSummary\x12 .uburnode.somni.v1.ReportDateReq\x1a .uburnode.somni.v1.GetSummaryRes\x12N\n\tGetEvents\x12 .uburnode.somni.v1.ReportDateReq\x1a\x1f.uburnode.somni.v1.GetEventsRes\x12X\n\x0eGetEnvironment\x12 .uburnode.somni.v1.ReportDateReq\x1a$.uburnode.somni.v1.GetEnvironmentRes\x12T\n\x0cGetStructure\x12 .uburnode.somni.v1.ReportDateReq\x1a\".uburnode.somni.v1.GetStructureRes\x12Z\n\x0fGetSleepQuality\x12 .uburnode.somni.v1.ReportDateReq\x1a%.uburnode.somni.v1.GetSleepQualityRes2\xf5\x01\n\x0c\x41udioService\x12J\n\x08GetAudio\x12\x1e.uburnode.somni.v1.GetAudioReq\x1a\x1e.uburnode.somni.v1.GetAudioRes\x12S\n\x0bGetAudioTag\x12!.uburnode.somni.v1.GetAudioTagReq\x1a!.uburnode.somni.v1.GetAudioTagRes\x12\x44\n\x06GetHot\x12\x1c.uburnode.somni.v1.GetHotReq\x1a\x1c.uburnode.somni.v1.GetHotResb\x06proto3') _globals = globals() _builder.BuildMessageAndEnumDescriptors(DESCRIPTOR, _globals) @@ -35,23 +35,41 @@ _globals['_GETANSWERREQ']._serialized_start=73 _globals['_GETANSWERREQ']._serialized_end=119 _globals['_ANSWERITEM']._serialized_start=122 - _globals['_ANSWERITEM']._serialized_end=264 - _globals['_GETANSWERRES']._serialized_start=266 - _globals['_GETANSWERRES']._serialized_end=328 - _globals['_REPORTDATEREQ']._serialized_start=330 - _globals['_REPORTDATEREQ']._serialized_end=379 - _globals['_GETSUMMARYRES']._serialized_start=381 - _globals['_GETSUMMARYRES']._serialized_end=396 - _globals['_GETEVENTSRES']._serialized_start=398 - _globals['_GETEVENTSRES']._serialized_end=412 - _globals['_GETENVIRONMENTRES']._serialized_start=414 - _globals['_GETENVIRONMENTRES']._serialized_end=433 - _globals['_GETSTRUCTURERES']._serialized_start=435 - _globals['_GETSTRUCTURERES']._serialized_end=452 - _globals['_GETSLEEPQUALITYRES']._serialized_start=454 - _globals['_GETSLEEPQUALITYRES']._serialized_end=474 - _globals['_QUIZSERVICE']._serialized_start=476 - _globals['_QUIZSERVICE']._serialized_end=568 - _globals['_REPORTSERVICE']._serialized_start=571 - _globals['_REPORTSERVICE']._serialized_end=1016 + _globals['_ANSWERITEM']._serialized_end=270 + _globals['_GETANSWERRES']._serialized_start=272 + _globals['_GETANSWERRES']._serialized_end=334 + _globals['_REPORTDATEREQ']._serialized_start=336 + _globals['_REPORTDATEREQ']._serialized_end=385 + _globals['_GETSUMMARYRES']._serialized_start=387 + _globals['_GETSUMMARYRES']._serialized_end=402 + _globals['_GETEVENTSRES']._serialized_start=404 + _globals['_GETEVENTSRES']._serialized_end=418 + _globals['_GETENVIRONMENTRES']._serialized_start=420 + _globals['_GETENVIRONMENTRES']._serialized_end=439 + _globals['_GETSTRUCTURERES']._serialized_start=441 + _globals['_GETSTRUCTURERES']._serialized_end=458 + _globals['_GETSLEEPQUALITYRES']._serialized_start=460 + _globals['_GETSLEEPQUALITYRES']._serialized_end=480 + _globals['_GETAUDIOREQ']._serialized_start=483 + _globals['_GETAUDIOREQ']._serialized_end=676 + _globals['_PAGEINFO']._serialized_start=678 + _globals['_PAGEINFO']._serialized_end=757 + _globals['_GETAUDIORES']._serialized_start=759 + _globals['_GETAUDIORES']._serialized_end=859 + _globals['_GETAUDIOTAGREQ']._serialized_start=861 + _globals['_GETAUDIOTAGREQ']._serialized_end=877 + _globals['_TAGDICTITEM']._serialized_start=879 + _globals['_TAGDICTITEM']._serialized_end=951 + _globals['_GETAUDIOTAGRES']._serialized_start=953 + _globals['_GETAUDIOTAGRES']._serialized_end=1015 + _globals['_GETHOTREQ']._serialized_start=1017 + _globals['_GETHOTREQ']._serialized_end=1028 + _globals['_GETHOTRES']._serialized_start=1030 + _globals['_GETHOTRES']._serialized_end=1041 + _globals['_QUIZSERVICE']._serialized_start=1043 + _globals['_QUIZSERVICE']._serialized_end=1135 + _globals['_REPORTSERVICE']._serialized_start=1138 + _globals['_REPORTSERVICE']._serialized_end=1583 + _globals['_AUDIOSERVICE']._serialized_start=1586 + _globals['_AUDIOSERVICE']._serialized_end=1831 # @@protoc_insertion_point(module_scope) diff --git a/app/uburnode_grpc/grpc_gen/uburnode_somni_pb2_grpc.py b/app/uburnode_grpc/grpc_gen/uburnode_somni_pb2_grpc.py index 4402df1..b475e9e 100644 --- a/app/uburnode_grpc/grpc_gen/uburnode_somni_pb2_grpc.py +++ b/app/uburnode_grpc/grpc_gen/uburnode_somni_pb2_grpc.py @@ -339,3 +339,161 @@ def GetSleepQuality(request, timeout, metadata, _registered_method=True) + + +class AudioServiceStub: + """Missing associated documentation comment in .proto file.""" + + def __init__(self, channel): + """Constructor. + + Args: + channel: A grpc.Channel. + """ + self.GetAudio = channel.unary_unary( + '/uburnode.somni.v1.AudioService/GetAudio', + request_serializer=uburnode__somni__pb2.GetAudioReq.SerializeToString, + response_deserializer=uburnode__somni__pb2.GetAudioRes.FromString, + _registered_method=True) + self.GetAudioTag = channel.unary_unary( + '/uburnode.somni.v1.AudioService/GetAudioTag', + request_serializer=uburnode__somni__pb2.GetAudioTagReq.SerializeToString, + response_deserializer=uburnode__somni__pb2.GetAudioTagRes.FromString, + _registered_method=True) + self.GetHot = channel.unary_unary( + '/uburnode.somni.v1.AudioService/GetHot', + request_serializer=uburnode__somni__pb2.GetHotReq.SerializeToString, + response_deserializer=uburnode__somni__pb2.GetHotRes.FromString, + _registered_method=True) + + +class AudioServiceServicer: + """Missing associated documentation comment in .proto file.""" + + def GetAudio(self, request, context): + """Missing associated documentation comment in .proto file.""" + context.set_code(grpc.StatusCode.UNIMPLEMENTED) + context.set_details('Method not implemented!') + raise NotImplementedError('Method not implemented!') + + def GetAudioTag(self, request, context): + """Missing associated documentation comment in .proto file.""" + context.set_code(grpc.StatusCode.UNIMPLEMENTED) + context.set_details('Method not implemented!') + raise NotImplementedError('Method not implemented!') + + def GetHot(self, request, context): + """Missing associated documentation comment in .proto file.""" + context.set_code(grpc.StatusCode.UNIMPLEMENTED) + context.set_details('Method not implemented!') + raise NotImplementedError('Method not implemented!') + + +def add_AudioServiceServicer_to_server(servicer, server): + rpc_method_handlers = { + 'GetAudio': grpc.unary_unary_rpc_method_handler( + servicer.GetAudio, + request_deserializer=uburnode__somni__pb2.GetAudioReq.FromString, + response_serializer=uburnode__somni__pb2.GetAudioRes.SerializeToString, + ), + 'GetAudioTag': grpc.unary_unary_rpc_method_handler( + servicer.GetAudioTag, + request_deserializer=uburnode__somni__pb2.GetAudioTagReq.FromString, + response_serializer=uburnode__somni__pb2.GetAudioTagRes.SerializeToString, + ), + 'GetHot': grpc.unary_unary_rpc_method_handler( + servicer.GetHot, + request_deserializer=uburnode__somni__pb2.GetHotReq.FromString, + response_serializer=uburnode__somni__pb2.GetHotRes.SerializeToString, + ), + } + generic_handler = grpc.method_handlers_generic_handler( + 'uburnode.somni.v1.AudioService', rpc_method_handlers) + server.add_generic_rpc_handlers((generic_handler,)) + server.add_registered_method_handlers('uburnode.somni.v1.AudioService', rpc_method_handlers) + + + # This class is part of an EXPERIMENTAL API. +class AudioService: + """Missing associated documentation comment in .proto file.""" + + @staticmethod + def GetAudio(request, + target, + options=(), + channel_credentials=None, + call_credentials=None, + insecure=False, + compression=None, + wait_for_ready=None, + timeout=None, + metadata=None): + return grpc.experimental.unary_unary( + request, + target, + '/uburnode.somni.v1.AudioService/GetAudio', + uburnode__somni__pb2.GetAudioReq.SerializeToString, + uburnode__somni__pb2.GetAudioRes.FromString, + options, + channel_credentials, + insecure, + call_credentials, + compression, + wait_for_ready, + timeout, + metadata, + _registered_method=True) + + @staticmethod + def GetAudioTag(request, + target, + options=(), + channel_credentials=None, + call_credentials=None, + insecure=False, + compression=None, + wait_for_ready=None, + timeout=None, + metadata=None): + return grpc.experimental.unary_unary( + request, + target, + '/uburnode.somni.v1.AudioService/GetAudioTag', + uburnode__somni__pb2.GetAudioTagReq.SerializeToString, + uburnode__somni__pb2.GetAudioTagRes.FromString, + options, + channel_credentials, + insecure, + call_credentials, + compression, + wait_for_ready, + timeout, + metadata, + _registered_method=True) + + @staticmethod + def GetHot(request, + target, + options=(), + channel_credentials=None, + call_credentials=None, + insecure=False, + compression=None, + wait_for_ready=None, + timeout=None, + metadata=None): + return grpc.experimental.unary_unary( + request, + target, + '/uburnode.somni.v1.AudioService/GetHot', + uburnode__somni__pb2.GetHotReq.SerializeToString, + uburnode__somni__pb2.GetHotRes.FromString, + options, + channel_credentials, + insecure, + call_credentials, + compression, + wait_for_ready, + timeout, + metadata, + _registered_method=True) diff --git a/proto/uburnode_somni.proto b/proto/uburnode_somni.proto index 639475e..bfa813d 100644 --- a/proto/uburnode_somni.proto +++ b/proto/uburnode_somni.proto @@ -1,7 +1,7 @@ syntax = "proto3"; // 量产对外契约（见 docs/量产中间件接口文档.md） -// Quiz：GetAnswer；Report：GetSummary/GetEvents/GetEnvironment/GetStructure/GetSleepQuality +// Quiz / Report / Audio package uburnode.somni.v1; import "google/protobuf/struct.proto"; @@ -11,14 +11,15 @@ message GetAnswerReq { string answer_id = 2; // 必传：答卷 _id } -// 对齐 somni_quiz_answers.answers[]（见问卷测评四表-新结构） message AnswerItem { string question_id = 1; string input_type = 2; string title = 3; - repeated string tags = 4; - google.protobuf.Value value = 5; // 形态由 input_type 决定 - string extra_input = 6; // 题级补充文本；无则空 + reserved 4; + reserved "tags"; + reserved "values"; + google.protobuf.Value value = 5; + string extra_input = 6; } message GetAnswerRes { @@ -51,3 +52,46 @@ service ReportService { rpc GetStructure (ReportDateReq) returns (GetStructureRes); rpc GetSleepQuality (ReportDateReq) returns (GetSleepQualityRes); } + +message GetAudioReq { + optional int32 page = 1; // 页码，≥ 1；fetch_all 时忽略 + optional int32 page_size = 2; // 每页条数 + optional bool fetch_all = 3; // true 时分页字段忽略，拉全量 + optional string query_text = 4; // 搜索词，如「雨声」；空则不过滤 + optional string tag_code = 5; // 内容形态标签 code；空则不按标签过滤 +} + +message PageInfo { + int32 page = 1; + int32 page_size = 2; + int32 total = 3; + int32 total_pages = 4; +} + +message GetAudioRes { + repeated google.protobuf.Struct materials = 1; + PageInfo page = 2; +} + +message GetAudioTagReq {} + +message TagDictItem { + string type = 1; + string code = 2; + string name = 3; + string name_en = 4; +} + +message GetAudioTagRes { + repeated TagDictItem tags = 1; +} + +message GetHotReq {} + +message GetHotRes {} + +service AudioService { + rpc GetAudio (GetAudioReq) returns (GetAudioRes); + rpc GetAudioTag (GetAudioTagReq) returns (GetAudioTagRes); + rpc GetHot (GetHotReq) returns (GetHotRes); +} diff --git a/pyproject.toml b/pyproject.toml index fc2840b..c12ebcc 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -12,6 +12,7 @@ dependencies = [ "elasticsearch>=8.15.0,<9", "aiohttp>=3.11.0", "grpcio>=1.68.0", + "grpcio-reflection>=1.68.0", "onnxruntime>=1.19.0", "transformers>=4.40.0", "numpy>=1.26.0", diff --git a/tests/test_es_search.py b/tests/test_es_search.py index a39ed13..02063f8 100644 --- a/tests/test_es_search.py +++ b/tests/test_es_search.py @@ -8,6 +8,18 @@ from app.es.search import SEARCH_CANDIDATE_SOURCE_INCLUDES, EsSearch +def test_es_search_accepts_isolated_index_names() -> None: + search = EsSearch( + MagicMock(), + Settings(), + audio_index="somni-prod-audio", + tag_dictionary_index="somni-prod-tags", + ) + + assert search.audio_index == "somni-prod-audio" + assert search.tag_dictionary_index == "somni-prod-tags" + + @pytest.mark.asyncio async def test_get_dictionary_vectors_deduplicates_and_batches_ids() -> None: client = MagicMock() diff --git a/tests/test_grpc_quiz.py b/tests/test_grpc_quiz.py index 30bb548..5aeec4b 100644 --- a/tests/test_grpc_quiz.py +++ b/tests/test_grpc_quiz.py @@ -6,6 +6,8 @@ import grpc import pytest +from bson import ObjectId +from google.protobuf.json_format import MessageToDict from app.core.config import Settings from app.core.exceptions import AppError, MongoNotConfiguredError @@ -55,18 +57,41 @@ async def test_somni_get_answer_maps_answer_item() -> None: "question_id": "q1", "input_type": "radio", "title": "您的性别是？", - "tags": ["基础信息"], + "tags": [], "value": {"option_id": "A", "option_text": "先生"}, "extra_input": "", }, + { + "question_id": "q3", + "input_type": "checkbox", + "title": "关注阶段", + "value": [ + {"option_id": "A", "option_text": "早期"}, + {"option_id": "B", "option_text": "VC"}, + ], + "extra_input": "", + }, { "question_id": "q2", "input_type": "input_number", "title": "夜醒次数", - "tags": [], "value": 2, "extra_input": "", }, + { + "question_id": "q4", + "input_type": "input", + "title": "补充说明", + "value": "最近压力比较大", + "extra_input": "", + }, + { + "question_id": "q5", + "input_type": "switch", + "title": "是否启用", + "value": True, + "extra_input": "", + }, ] } ) @@ -75,12 +100,21 @@ async def test_somni_get_answer_maps_answer_item() -> None: uburnode_somni_pb2.GetAnswerReq(uid="u1", answer_id="a1"), _context(), ) - assert len(res.answers) == 2 - assert res.answers[0].question_id == "q1" - assert res.answers[0].input_type == "radio" - assert list(res.answers[0].tags) == ["基础信息"] - assert res.answers[0].value.struct_value.fields["option_id"].string_value == "A" - assert res.answers[1].value.number_value == 2 + payload = MessageToDict(res, preserving_proto_field_name=True) + assert len(payload["answers"]) == 5 + assert payload["answers"][0]["value"] == { + "option_id": "A", + "option_text": "先生", + } + assert payload["answers"][1]["value"] == [ + {"option_id": "A", "option_text": "早期"}, + {"option_id": "B", "option_text": "VC"}, + ] + assert payload["answers"][2]["value"] == 2 + assert payload["answers"][3]["value"] == "最近压力比较大" + assert payload["answers"][4]["value"] is True + assert "tags" not in payload["answers"][0] + assert "values" not in payload["answers"][0] @pytest.mark.asyncio @@ -93,9 +127,23 @@ async def test_somni_quiz_service_loads_from_collection() -> None: "question_id": "q1", "input_type": "radio", "title": "您的性别是？", - "tags": ["基础信息"], "value": {"option_id": "A", "option_text": "先生"}, }, + { + "question_id": "q3", + "input_type": "select", + "title": "常驻城市", + "value": {"option_id": "A", "option_text": "北京"}, + }, + { + "question_id": "q4", + "input_type": "checkbox", + "title": "关注阶段", + "value": [ + {"option_id": "A", "option_text": "早期"}, + {"option_id": "B", "option_text": "VC"}, + ], + }, { "question_id": "q2", "input_type": "input", @@ -123,9 +171,25 @@ async def test_somni_quiz_service_loads_from_collection() -> None: client.__getitem__.assert_called_with("Somni") db.__getitem__.assert_called_with("somni_quiz_answers") assert payload["answers"][0]["input_type"] == "radio" - assert payload["answers"][0]["value"]["option_id"] == "A" - assert payload["answers"][1]["extra_input"] == "备注" - assert payload["answers"][0]["tags"] == ["基础信息"] + assert payload["answers"][0]["value"] == { + "option_id": "A", + "option_text": "先生", + } + assert payload["answers"][1]["value"] == { + "option_id": "A", + "option_text": "北京", + } + assert payload["answers"][2]["value"] == [ + {"option_id": "A", "option_text": "早期"}, + {"option_id": "B", "option_text": "VC"}, + ] + assert payload["answers"][3]["extra_input"] == "备注" + assert "tags" not in payload["answers"][0] + query = collection.find_one.await_args.args[0] + assert query == { + "uid": "user-001", + "_id": ObjectId("69b10cc516d7472aedf6bb80"), + } @pytest.mark.asyncio diff --git a/tests/test_grpc_somni_audio.py b/tests/test_grpc_somni_audio.py new file mode 100644 index 0000000..48486c9 --- /dev/null +++ b/tests/test_grpc_somni_audio.py @@ -0,0 +1,113 @@ +"""量产 AudioRpc。""" + +from __future__ import annotations + +from unittest.mock import AsyncMock, MagicMock + +import pytest + +from app.server.somni.audio.rpc import AudioRpc +from app.uburnode_grpc.grpc_gen import uburnode_somni_pb2 + + +def _context() -> MagicMock: + return MagicMock() + + +def _make_rpc() -> tuple[AudioRpc, MagicMock]: + service = MagicMock() + service.get_audio = AsyncMock() + service.get_audio_tag = AsyncMock() + service.get_hot = AsyncMock() + return AudioRpc(service), service + + +@pytest.mark.asyncio +async def test_get_audio_passes_page_and_query() -> None: + rpc, service = _make_rpc() + service.get_audio = AsyncMock( + return_value={ + "materials": [{"id": "m1", "audio_name": "雨声"}], + "page": {"page": 1, "page_size": 20, "total": 1, "total_pages": 1}, + } + ) + req = uburnode_somni_pb2.GetAudioReq(page=1, page_size=20, query_text="雨声") + res = await rpc.GetAudio(req, _context()) + service.get_audio.assert_awaited_once_with( + page=1, + page_size=20, + fetch_all=False, + query_text="雨声", + tag_code="", + ) + assert res.materials[0]["audio_name"] == "雨声" + assert res.page.total == 1 + + +@pytest.mark.asyncio +async def test_get_audio_fetch_all() -> None: + rpc, service = _make_rpc() + service.get_audio = AsyncMock( + return_value={ + "materials": [], + "page": {"page": 1, "page_size": 0, "total": 0, "total_pages": 1}, + } + ) + req = uburnode_somni_pb2.GetAudioReq(fetch_all=True) + await rpc.GetAudio(req, _context()) + service.get_audio.assert_awaited_once_with( + page=None, + page_size=None, + fetch_all=True, + query_text="", + tag_code="", + ) + + +@pytest.mark.asyncio +async def test_get_audio_passes_tag_code() -> None: + rpc, service = _make_rpc() + service.get_audio = AsyncMock( + return_value={ + "materials": [], + "page": {"page": 1, "page_size": 20, "total": 0, "total_pages": 0}, + } + ) + req = uburnode_somni_pb2.GetAudioReq(tag_code="steady_rain") + await rpc.GetAudio(req, _context()) + service.get_audio.assert_awaited_once_with( + page=None, + page_size=None, + fetch_all=False, + query_text="", + tag_code="steady_rain", + ) + + +@pytest.mark.asyncio +async def test_get_audio_tag_maps_fields() -> None: + rpc, service = _make_rpc() + service.get_audio_tag = AsyncMock( + return_value={ + "tags": [ + { + "type": "content_form", + "code": "natural_sound", + "name": "自然声", + "name_en": "Natural Sound", + } + ] + } + ) + res = await rpc.GetAudioTag(uburnode_somni_pb2.GetAudioTagReq(), _context()) + service.get_audio_tag.assert_awaited_once_with() + assert res.tags[0].code == "natural_sound" + assert res.tags[0].name_en == "Natural Sound" + + +@pytest.mark.asyncio +async def test_get_hot_no_args() -> None: + rpc, service = _make_rpc() + res = await rpc.GetHot(uburnode_somni_pb2.GetHotReq(), _context()) + service.get_hot.assert_awaited_once_with() + assert res == uburnode_somni_pb2.GetHotRes() diff --git a/tests/test_somni_audio_catalog.py b/tests/test_somni_audio_catalog.py new file mode 100644 index 0000000..7939c99 --- /dev/null +++ b/tests/test_somni_audio_catalog.py @@ -0,0 +1,286 @@ +"""量产音频目录查询。""" + +from __future__ import annotations + +from unittest.mock import AsyncMock, MagicMock + +import pytest + +from app.core.config import Settings +from app.core.exceptions import AppError, EncoderNotReadyError +from app.server.somni.audio import catalog +from app.server.somni.audio.catalog import AudioCatalogService, InvalidAudioQueryError + + +class _Cursor: + def __init__(self, docs: list) -> None: + self._docs = list(docs) + self._index = 0 + + def skip(self, count: int) -> _Cursor: + self._docs = self._docs[count:] + return self + + def limit(self, count: int) -> _Cursor: + self._docs = self._docs[:count] + return self + + def __aiter__(self) -> _Cursor: + return self + + async def __anext__(self): + if self._index >= len(self._docs): + raise StopAsyncIteration + doc = self._docs[self._index] + self._index += 1 + return doc + + +_RAIN = { + "_id": "a1", + "audio_name": "雨夜", + "embedding": [0.1], + "content_form_tags": [ + { + "tag_id": "root-rain", + "code": "natural_sound", + "name": "自然声", + "parent_tag_id": None, + }, + { + "tag_id": "child-rain", + "code": "steady_rain", + "name": "中雨/稳定雨声", + "parent_tag_id": "root-rain", + }, + ], +} +_MUSIC = { + "_id": "a2", + "audio_name": "钢琴", + "content_form_tags": [ + { + "tag_id": "root-music", + "code": "music", + "name": "音乐", + "parent_tag_id": None, + } + ], +} + + +def _service( + collection: MagicMock, + *, + es_search: MagicMock | None = None, + encoder: MagicMock | None = None, + fetch_all_hard_limit: int = 50, + cache_ttl_sec: float = 60.0, +) -> AudioCatalogService: + db = MagicMock() + db.__getitem__ = MagicMock(return_value=collection) + client = MagicMock() + client.__getitem__ = MagicMock(return_value=db) + settings = Settings( + somni_mongo_db="Somni", + somni_mongo_materials_collection="somni_audio_materials", + default_page_size=1, + max_page_size=200, + fetch_all_hard_limit=fetch_all_hard_limit, + get_audio_root_tag_sim_threshold=0.85, + somni_audio_catalog_cache_ttl_sec=cache_ttl_sec, + ) + return AudioCatalogService( + client, + settings, + es_search=es_search, + encoder=encoder, + ) + + +def _mongo_collection(docs: list) -> MagicMock: + collection = MagicMock() + collection.count_documents = AsyncMock(return_value=len(docs)) + collection.find = MagicMock(return_value=_Cursor(docs)) + return collection + + +@pytest.mark.asyncio +async def test_get_audio_filters_content_form_code_then_pages() -> None: + svc = _service(_mongo_collection([_RAIN, _MUSIC])) + payload = await svc.get_audio( + page=1, + page_size=1, + fetch_all=False, + query_text="", + tag_code="steady_rain", + ) + assert [item["id"] for item in payload["materials"]] == ["a1"] + assert payload["page"]["total"] == 1 + assert "embedding" not in payload["materials"][0] + + +@pytest.mark.asyncio +async def test_get_audio_uses_cache_on_second_call() -> None: + collection = _mongo_collection([_RAIN, _MUSIC]) + svc = _service(collection) + await svc.get_audio(page=1, page_size=10, fetch_all=False, query_text="", tag_code="") + await svc.get_audio(page=1, page_size=10, fetch_all=False, query_text="", tag_code="music") + assert collection.find.call_count == 1 + + +@pytest.mark.asyncio +async def test_get_audio_keeps_mongo_and_es_caches_separate() -> None: + collection = _mongo_collection([_RAIN, _MUSIC]) + encoder = MagicMock() + encoder.is_loaded = True + encoder.encode_one = AsyncMock(return_value=[1.0, 0.0]) + es_search = MagicMock() + es_search.list_audio_catalog_docs = AsyncMock(return_value=[_RAIN]) + es_search.list_content_tag_vectors = AsyncMock( + return_value=[ + { + "id": "root-rain", + "dimension": "content_form", + "parent_tag_id": "", + "vector": [1.0, 0.0], + } + ] + ) + svc = _service(collection, es_search=es_search, encoder=encoder) + + await svc.get_audio( + page=1, page_size=10, fetch_all=False, query_text="", tag_code="" + ) + payload = await svc.get_audio( + page=1, page_size=10, fetch_all=False, query_text="雨声", tag_code="" + ) + + es_search.list_audio_catalog_docs.assert_awaited_once_with(size=51) + assert [item["id"] for item in payload["materials"]] == ["a1"] + + +@pytest.mark.asyncio +async def test_get_audio_cache_expires(monkeypatch) -> None: + times = iter([100.0, 111.0]) + monkeypatch.setattr(catalog, "monotonic", lambda: next(times), raising=False) + collection = _mongo_collection([_RAIN]) + svc = _service(collection, cache_ttl_sec=10.0) + + await svc.get_audio( + page=1, page_size=10, fetch_all=False, query_text="", tag_code="" + ) + await svc.get_audio( + page=1, page_size=10, fetch_all=False, query_text="", tag_code="" + ) + + assert collection.find.call_count == 2 + + +@pytest.mark.asyncio +async def test_get_audio_query_text_matches_root_tag_via_es() -> None: + collection = _mongo_collection([_RAIN]) + encoder = MagicMock() + encoder.is_loaded = True + encoder.encode_one = AsyncMock(return_value=[1.0, 0.0]) + es_search = MagicMock() + es_search.list_content_tag_vectors = AsyncMock( + return_value=[ + { + "id": "root-rain", + "dimension": "content_form", + "parent_tag_id": "", + "vector": [1.0, 0.0], + }, + { + "id": "child-rain", + "dimension": "content_form", + "parent_tag_id": "root-rain", + "vector": [1.0, 0.0], + }, + ] + ) + es_search.list_audio_catalog_docs = AsyncMock(return_value=[_RAIN, _MUSIC]) + svc = _service(collection, es_search=es_search, encoder=encoder) + payload = await svc.get_audio( + page=1, + page_size=10, + fetch_all=False, + query_text="雨声", + tag_code="", + ) + collection.find.assert_not_called() + es_search.list_audio_catalog_docs.assert_awaited_once_with(size=51) + assert [item["id"] for item in payload["materials"]] == ["a1"] + + +@pytest.mark.asyncio +async def test_get_audio_query_text_without_encoder_fails() -> None: + es_search = MagicMock() + es_search.list_audio_catalog_docs = AsyncMock(return_value=[_RAIN]) + svc = _service(_mongo_collection([_RAIN]), es_search=es_search) + with pytest.raises(EncoderNotReadyError): + await svc.get_audio( + page=1, + page_size=10, + fetch_all=False, + query_text="雨声", + tag_code="", + ) + + +@pytest.mark.asyncio +async def test_get_audio_rejects_invalid_page() -> None: + svc = _service(_mongo_collection([])) + with pytest.raises(InvalidAudioQueryError): + await svc.get_audio( + page=0, + page_size=20, + fetch_all=False, + query_text="", + tag_code="", + ) + + +@pytest.mark.asyncio +async def test_get_audio_mongo_load_rejects_over_limit() -> None: + collection = MagicMock() + collection.count_documents = AsyncMock(return_value=6) + svc = _service(collection, fetch_all_hard_limit=5) + with pytest.raises(InvalidAudioQueryError): + await svc.get_audio( + page=None, + page_size=None, + fetch_all=False, + query_text="", + tag_code="", + ) + + +@pytest.mark.asyncio +async def test_get_audio_tag_requires_mongo() -> None: + svc = AudioCatalogService(None, Settings()) + with pytest.raises(AppError) as exc: + await svc.get_audio_tag() + assert exc.value.status_code == 503 + + +@pytest.mark.asyncio +async def test_get_audio_tag_maps_root_fields() -> None: + collection = MagicMock() + collection.count_documents = AsyncMock(return_value=1) + collection.find = MagicMock( + return_value=_Cursor( + [ + { + "type": "content_form", + "code": "natural_sound", + "name": "自然声", + "name_en": "Natural Sound", + } + ] + ) + ) + svc = _service(collection) + payload = await svc.get_audio_tag() + assert payload["tags"][0]["code"] == "natural_sound" diff --git a/tests/test_uburnode_proto_import.py b/tests/test_uburnode_proto_import.py index a49c095..5a2deac 100644 --- a/tests/test_uburnode_proto_import.py +++ b/tests/test_uburnode_proto_import.py @@ -18,4 +18,11 @@ def test_somni_package() -> None: assert uburnode_somni_pb2.DESCRIPTOR.package == "uburnode.somni.v1" assert hasattr(uburnode_somni_pb2_grpc, "QuizServiceServicer") assert hasattr(uburnode_somni_pb2_grpc, "ReportServiceServicer") - assert not hasattr(uburnode_somni_pb2_grpc, "AudioServiceServicer") + assert hasattr(uburnode_somni_pb2_grpc, "AudioServiceServicer") + quiz = uburnode_somni_pb2.DESCRIPTOR.services_by_name["QuizService"] + assert quiz.full_name == "uburnode.somni.v1.QuizService" + assert "tags" not in uburnode_somni_pb2.AnswerItem.DESCRIPTOR.fields_by_name + assert "values" not in uburnode_somni_pb2.AnswerItem.DESCRIPTOR.fields_by_name + value_field = uburnode_somni_pb2.AnswerItem.DESCRIPTOR.fields_by_name["value"] + assert value_field.number == 5 + assert value_field.message_type.full_name == "google.protobuf.Value" diff --git a/uv.lock b/uv.lock index 2db0795..0f95076 100644 --- a/uv.lock +++ b/uv.lock @@ -441,6 +441,19 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/0c/e8/127dc2b246096ad50ef7c8d9b7b31d757787aeb796368bcdd4454e4204c4/grpcio-1.81.0-cp314-cp314-win_amd64.whl", hash = "sha256:b93cee313cae4e113fbb3a0ce1ea5633db6f63cfde2b2dc1d817429026b2a50b", size = 5070848, upload-time = "2026-06-01T05:56:19.735Z" }, ] +[[package]] +name = "grpcio-reflection" +version = "1.81.0" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "grpcio" }, + { name = "protobuf" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/c2/98/39a6972bb9a90750e32daabacaab7b4418e4384f7f2f686ff5af3d69094b/grpcio_reflection-1.81.0.tar.gz", hash = "sha256:5191db7aa6cab1b6981b0879fa44fdcdd43ba644f0301c40b976f813eb4eff06", size = 19192, upload-time = "2026-06-01T06:00:33.419Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/d5/61/e472357ff5484f67c802568e70b3182df3d8eb1d0c2de38199d9a9a28bb2/grpcio_reflection-1.81.0-py3-none-any.whl", hash = "sha256:85322a9c1ab62d9823b1262a9d78d653b1710b99b5764cdcef2673cfe352b9c1", size = 22907, upload-time = "2026-06-01T06:00:16.714Z" }, +] + [[package]] name = "grpcio-tools" version = "1.81.0" @@ -2131,6 +2144,7 @@ dependencies = [ { name = "elasticsearch" }, { name = "fastapi" }, { name = "grpcio" }, + { name = "grpcio-reflection" }, { name = "loguru" }, { name = "motor" }, { name = "numpy" }, @@ -2161,6 +2175,7 @@ requires-dist = [ { name = "elasticsearch", specifier = ">=8.15.0,<9" }, { name = "fastapi", specifier = ">=0.115.0,<0.116" }, { name = "grpcio", specifier = ">=1.68.0" }, + { name = "grpcio-reflection", specifier = ">=1.68.0" }, { name = "grpcio-tools", marker = "extra == 'dev'", specifier = ">=1.68.0" }, { name = "httpx", marker = "extra == 'dev'", specifier = ">=0.28.0" }, { name = "loguru", specifier = ">=0.7.0" },