"""Public-first service guide query APIs.""" from __future__ import annotations from typing import Annotated, Any from fastapi import APIRouter, Depends, HTTPException, Query from opensearchpy.exceptions import NotFoundError from app.api.deps import get_es_client, get_neo4j_client, get_optional_permission_context from app.api.schemas.service_guide import ( ServiceGuideDetailResponse, ServiceGuideListItem, ServiceGuideListResponse, ) from app.config import settings from app.core.graph_query_service import GraphQueryService from app.core.permission import PermissionContext from app.infrastructure.es_client import ESClient from app.infrastructure.neo4j_client import Neo4jClient from app.schemas.service_guide_profile import build_compatible_service_guide_profile from app.utils.logger import get_logger router = APIRouter(prefix="/service-guides", tags=["service-guides"]) logger = get_logger(__name__) def _append_term_filter(filter_clauses: list[dict[str, Any]], field: str, value: str | bool | None) -> None: if isinstance(value, str): normalized = value.strip() if not normalized: return filter_clauses.append({"term": {field: normalized}}) return if value is not None: filter_clauses.append({"term": {field: value}}) def _build_acl_filter(perm: PermissionContext | None) -> dict[str, Any]: if perm is None: return PermissionContext.public_es_filter() return perm.build_es_filter() def _has_acl_access(doc_acl_ids: list[str], perm: PermissionContext | None) -> bool: if not doc_acl_ids: return True if perm is None: return False return perm.has_acl_access(doc_acl_ids) def _to_list_response( hits: list[dict[str, Any]], *, total: int, page: int, page_size: int, ) -> ServiceGuideListResponse: return ServiceGuideListResponse( total=total, page=page, page_size=page_size, items=[ ServiceGuideListItem( profile_id=hit["_source"].get("profile_id", hit.get("_id", "")), doc_id=hit["_source"].get("doc_id", ""), matter_name=hit["_source"].get("matter_name", ""), colloquial_names=hit["_source"].get("colloquial_names", []), implementation_code=hit["_source"].get("implementation_code", ""), matter_version=hit["_source"].get("matter_version", ""), handled_org_names=hit["_source"].get("handled_org_names", []), region_names=hit["_source"].get("region_names", []), express_supported=hit["_source"].get("express_supported"), reservation_supported=hit["_source"].get("reservation_supported"), needs_review=bool(hit["_source"].get("needs_review", False)), ) for hit in hits ], ) def _to_detail_response(source: dict[str, Any], *, fallback_profile_id: str) -> ServiceGuideDetailResponse: profile, used_fallback = build_compatible_service_guide_profile(source) if used_fallback: logger.warning( "service_guide_profile_compat_fallback_used", profile_id=source.get("profile_id", fallback_profile_id), doc_id=source.get("doc_id", ""), ) return ServiceGuideDetailResponse( profile_id=source.get("profile_id", fallback_profile_id), doc_id=source.get("doc_id", ""), scene_type=source.get("scene_type", ""), guide_version=source.get("guide_version", ""), profile=profile, ) @router.get("", response_model=ServiceGuideListResponse) async def list_service_guides( es_client: Annotated[ESClient, Depends(get_es_client)], perm: Annotated[PermissionContext | None, Depends(get_optional_permission_context)], q: str = Query(default="", alias="query", description="关键词，匹配事项名/俗称/检索文本"), legacy_q: str = Query(default="", alias="q", include_in_schema=False, description="兼容旧参数 q"), doc_id: str = Query(default="", description="按文档 ID 过滤"), implementation_code: str = Query(default="", description="按实施编码过滤"), basic_code: str = Query(default="", description="按基本编码过滤"), matter_type: str = Query(default="", description="按事项类型过滤"), handled_org: str = Query(default="", description="按受理机构过滤"), region: str = Query(default="", description="按地区过滤"), express_supported: bool | None = Query(default=None, description="按是否支持物流快递过滤"), reservation_supported: bool | None = Query(default=None, description="按是否支持预约办理过滤"), must_onsite: bool | None = Query(default=None, description="按是否必须现场办理过滤"), matter_name: str = Query(default="", description="按事项名称过滤"), page: int = Query(default=1, ge=1), page_size: int = Query(default=20, ge=1, le=100), ) -> ServiceGuideListResponse: must: list[dict[str, Any]] = [{"term": {"is_current": True}}] filter_clauses: list[dict[str, Any]] = [_build_acl_filter(perm)] keyword = q.strip() or legacy_q.strip() if keyword: must.append( { "multi_match": { "query": keyword, "fields": [ "matter_name^4", "colloquial_names^3", "implementing_subject^2", "guide_search_text^1.5", ], "type": "best_fields", } } ) if doc_id: filter_clauses.append({"term": {"doc_id": doc_id}}) if implementation_code: filter_clauses.append({"term": {"implementation_code": implementation_code}}) _append_term_filter(filter_clauses, "basic_code", basic_code) _append_term_filter(filter_clauses, "matter_type", matter_type) _append_term_filter(filter_clauses, "handled_org_names", handled_org) _append_term_filter(filter_clauses, "region_names", region) _append_term_filter(filter_clauses, "express_supported", express_supported) _append_term_filter(filter_clauses, "reservation_supported", reservation_supported) _append_term_filter(filter_clauses, "must_onsite", must_onsite) if matter_name: must.append({"match_phrase": {"matter_name": matter_name}}) resp = await es_client.raw.search( index=settings.es_service_guide_index, body={ "query": { "bool": { "must": must, "filter": filter_clauses, } }, "from": (page - 1) * page_size, "size": page_size, "sort": [ {"completeness_score": {"order": "desc", "missing": "_last"}}, {"confidence_score": {"order": "desc", "missing": "_last"}}, {"updated_at": {"order": "desc"}}, ], "_source": [ "profile_id", "doc_id", "matter_name", "colloquial_names", "implementation_code", "matter_version", "handled_org_names", "region_names", "express_supported", "reservation_supported", "needs_review", ], }, ) raw = resp if isinstance(resp, dict) else resp.body hits = raw.get("hits", {}).get("hits", []) total = raw.get("hits", {}).get("total", {}) total_value = total.get("value", 0) if isinstance(total, dict) else int(total or 0) return _to_list_response( hits, total=total_value, page=page, page_size=page_size, ) @router.get("/by-matter/{matter_id}", response_model=ServiceGuideListResponse) async def list_service_guides_by_matter( matter_id: str, es_client: Annotated[ESClient, Depends(get_es_client)], perm: Annotated[PermissionContext | None, Depends(get_optional_permission_context)], neo4j: Annotated[Neo4jClient, Depends(get_neo4j_client)], page: int = Query(default=1, ge=1), page_size: int = Query(default=20, ge=1, le=100), ) -> ServiceGuideListResponse: graph = GraphQueryService(neo4j) matter = await graph.get_matter_card( matter_id, acl_tokens=perm.acl_tokens if perm is not None else [], ) if matter is None: raise HTTPException(status_code=404, detail=f"Matter '{matter_id}' not found") matter_name = (matter.get("name") or "").strip() should_clauses: list[dict[str, Any]] = [{"term": {"linked_matter_ids": matter_id}}] if matter_name: should_clauses.append({"match_phrase": {"matter_name": matter_name}}) resp = await es_client.raw.search( index=settings.es_service_guide_index, body={ "query": { "bool": { "must": [{"term": {"is_current": True}}], "filter": [_build_acl_filter(perm)], "should": should_clauses, "minimum_should_match": 1, } }, "from": (page - 1) * page_size, "size": page_size, "sort": [ {"completeness_score": {"order": "desc", "missing": "_last"}}, {"confidence_score": {"order": "desc", "missing": "_last"}}, {"updated_at": {"order": "desc"}}, ], "_source": [ "profile_id", "doc_id", "matter_name", "colloquial_names", "implementation_code", "matter_version", "handled_org_names", "region_names", "express_supported", "reservation_supported", "needs_review", ], }, ) raw = resp if isinstance(resp, dict) else resp.body hits = raw.get("hits", {}).get("hits", []) total = raw.get("hits", {}).get("total", {}) total_value = total.get("value", 0) if isinstance(total, dict) else int(total or 0) return _to_list_response( hits, total=total_value, page=page, page_size=page_size, ) @router.get("/by-doc/{doc_id}", response_model=ServiceGuideDetailResponse) async def get_service_guide_by_doc( doc_id: str, es_client: Annotated[ESClient, Depends(get_es_client)], perm: Annotated[PermissionContext | None, Depends(get_optional_permission_context)], ) -> ServiceGuideDetailResponse: resp = await es_client.raw.search( index=settings.es_service_guide_index, body={ "query": { "bool": { "must": [ {"term": {"doc_id": doc_id}}, {"term": {"is_current": True}}, ], "filter": [_build_acl_filter(perm)], } }, "size": 1, "sort": [{"updated_at": {"order": "desc"}}], }, ) raw = resp if isinstance(resp, dict) else resp.body hits = raw.get("hits", {}).get("hits", []) if not hits: raise HTTPException(status_code=404, detail=f"Service guide for doc '{doc_id}' not found") source = hits[0].get("_source", {}) return _to_detail_response(source, fallback_profile_id=hits[0].get("_id", "")) @router.get("/{profile_id}", response_model=ServiceGuideDetailResponse) async def get_service_guide_detail( profile_id: str, es_client: Annotated[ESClient, Depends(get_es_client)], perm: Annotated[PermissionContext | None, Depends(get_optional_permission_context)], ) -> ServiceGuideDetailResponse: try: resp = await es_client.raw.get(index=settings.es_service_guide_index, id=profile_id) except NotFoundError as exc: raise HTTPException(status_code=404, detail=f"Service guide '{profile_id}' not found") from exc raw = resp if isinstance(resp, dict) else resp.body source = raw.get("_source", {}) if not _has_acl_access(source.get("acl_ids", []), perm): raise HTTPException(status_code=404, detail=f"Service guide '{profile_id}' not found") return _to_detail_response(source, fallback_profile_id=profile_id)