"""Mock OA endpoints for integration testing. These endpoints simulate the OA (Office Automation) system's side of the integration, letting developers: - Generate JWT tokens with custom user context (office/dept/area/roles) - Login with predefined test accounts - Check task status and list recent tasks - List ingested documents in Elasticsearch - Delete test documents to clean up WARNING: These endpoints are for **development / testing only**. Disable by setting MOCK_ENABLED=false in production. """ from __future__ import annotations from pathlib import Path from typing import Annotated, Any from fastapi import APIRouter, Depends, HTTPException, Request from pydantic import BaseModel, Field from app.api.deps import get_es_client from app.api.v1.auth import ensure_prefix, PREFIX_USER, PREFIX_OFFICE, PREFIX_DEPT, PREFIX_AREA, PREFIX_ROLE from app.config import settings from app.core.security import TokenClaims, create_access_token from app.utils.logger import get_logger logger = get_logger(__name__) router = APIRouter(prefix="/mock", tags=["mock-oa"]) # ── Predefined test users ───────────────────────────────────────────────────── def _load_test_users() -> dict[str, dict]: """从配置文件加载测试用户数据。""" import json _config_path = Path(__file__).resolve().parent.parent.parent / "config" / "test_users.json" with open(_config_path, encoding="utf-8") as f: return json.load(f) _TEST_USERS: dict[str, dict] = _load_test_users() # ── Schemas ────────────────────────────────────────────────────────────────── class TokenRequest(BaseModel): """Request body for generating a test JWT token.""" user_id: str = Field(default="test_user_001", description="User ID (JWT sub claim)") user_name: str = Field(default="", description="用户姓名") office_id: str = Field(default="O_01", description="科室 ID") office_name: str = Field(default="", description="科室名称") dept_id: str = Field(default="D_01", description="部门 ID") dept_name: str = Field(default="", description="部门名称") area_id: str = Field(default="A_01", description="地区 ID") area_name: str = Field(default="", description="地区名称") role_ids: list[str] = Field(default_factory=list, description="角色 ID 列表") expires_minutes: int = Field(default=480, description="Token validity in minutes (default 8h)") class TokenResponse(BaseModel): """Generated JWT token with decoded info for display.""" access_token: str token_type: str = "bearer" expires_in: int user_id: str office_id: str dept_id: str area_id: str role_ids: list[str] class DocListItem(BaseModel): """Summary of a document in Elasticsearch.""" doc_id: str content_hash: str = "" title: str = "" doc_number: str = "" issuing_org: str = "" doc_type: str = "" publish_date: str = "" chunk_count: int = 0 acl_ids: list[str] = [] status: str = "" indexed_at: str = "" class TaskStatusResponse(BaseModel): """Celery task status.""" task_id: str status: str progress: float = 0.0 result: dict[str, Any] | None = None error: str | None = None class LoginRequest(BaseModel): """Username / password login for predefined test accounts.""" username: str password: str class LoginResponse(BaseModel): """Successful login response.""" access_token: str token_type: str = "bearer" expires_in: int user_id: str username: str user_name: str office_id: str office_name: str = "" dept_id: str dept_name: str = "" area_id: str area_name: str = "" role_ids: list[str] role: str class TestUserInfo(BaseModel): """Public info about a predefined test user (no password).""" username: str user_name: str office_id: str office_name: str = "" dept_id: str dept_name: str = "" area_id: str area_name: str = "" role_ids: list[str] role: str # ── Endpoints ───────────────────────────────────────────────────────────────── @router.post("/token", response_model=TokenResponse) async def generate_test_token(body: TokenRequest) -> TokenResponse: """Generate a signed JWT token for testing. The token is signed with the same secret as the main application so it works with all authenticated endpoints (search, research, etc.). No authentication required – **for testing only**. """ # 为原始 ID 添加类型前缀 p_user_id = ensure_prefix(body.user_id, PREFIX_USER) p_office_id = ensure_prefix(body.office_id, PREFIX_OFFICE) p_dept_id = ensure_prefix(body.dept_id, PREFIX_DEPT) p_area_id = ensure_prefix(body.area_id, PREFIX_AREA) p_role_ids = [ensure_prefix(r, PREFIX_ROLE) for r in body.role_ids] claims = TokenClaims( user_id=p_user_id, user_name=body.user_name, office_id=p_office_id, office_name=body.office_name, dept_id=p_dept_id, dept_name=body.dept_name, area_id=p_area_id, area_name=body.area_name, role_ids=p_role_ids, ) result = create_access_token(claims, expires_minutes=body.expires_minutes, issuer="zm-rag-mock") logger.info( "mock_token_generated", user_id=p_user_id, office_id=p_office_id, dept_id=p_dept_id, area_id=p_area_id, ) return TokenResponse( access_token=result.access_token, expires_in=result.expires_in, user_id=p_user_id, office_id=p_office_id, dept_id=p_dept_id, area_id=p_area_id, role_ids=p_role_ids, ) @router.post("/login", response_model=LoginResponse) async def login(body: LoginRequest) -> LoginResponse: """Login with a predefined test account. Predefined accounts (原始 ID → 自动加前缀) ------------------------------------------- admin / admin123 – 地区领导 (A_01, D_01, O_01, R_01) zhang_san / user123 – 科室用户 (A_01, D_05, O_17) li_si / user123 – 科室用户 (A_02, D_08, O_22) wang_wu / manager123 – 部门领导 (A_01, D_05, O_01, R_03) """ user = _TEST_USERS.get(body.username) if user is None or user["password"] != body.password: raise HTTPException(status_code=401, detail="用户名或密码错误") # 为原始 ID 添加类型前缀 p_user_id = ensure_prefix(user["user_id"], PREFIX_USER) p_office_id = ensure_prefix(user["office_id"], PREFIX_OFFICE) p_dept_id = ensure_prefix(user["dept_id"], PREFIX_DEPT) p_area_id = ensure_prefix(user["area_id"], PREFIX_AREA) p_role_ids = [ensure_prefix(r, PREFIX_ROLE) for r in user["role_ids"]] claims = TokenClaims( user_id=p_user_id, user_name=user["user_name"], office_id=p_office_id, office_name=user["office_name"], dept_id=p_dept_id, dept_name=user["dept_name"], area_id=p_area_id, area_name=user["area_name"], role_ids=p_role_ids, ) result = create_access_token(claims, issuer="zm-rag-mock") logger.info("mock_login", username=body.username, user_id=p_user_id) return LoginResponse( access_token=result.access_token, expires_in=result.expires_in, user_id=p_user_id, username=body.username, user_name=user["user_name"], office_id=p_office_id, office_name=user["office_name"], dept_id=p_dept_id, dept_name=user["dept_name"], area_id=p_area_id, area_name=user["area_name"], role_ids=p_role_ids, role=user["role"], ) @router.get("/users", response_model=list[TestUserInfo]) async def list_test_users() -> list[TestUserInfo]: """Return the list of predefined test users (passwords not included).""" return [ TestUserInfo( username=uname, user_name=info["user_name"], office_id=ensure_prefix(info["office_id"], PREFIX_OFFICE), office_name=info.get("office_name", ""), dept_id=ensure_prefix(info["dept_id"], PREFIX_DEPT), dept_name=info.get("dept_name", ""), area_id=ensure_prefix(info["area_id"], PREFIX_AREA), area_name=info.get("area_name", ""), role_ids=[ensure_prefix(r, PREFIX_ROLE) for r in info["role_ids"]], role=info["role"], ) for uname, info in _TEST_USERS.items() ] @router.get("/docs", response_model=list[DocListItem]) async def list_docs( request: Request, size: int = 20, from_: int = 0, keyword: str = "", ) -> list[DocListItem]: """List documents currently indexed in Elasticsearch. Returns document-level metadata (from the ``gov_doc_meta`` index). No authentication required for testing convenience. """ es = get_es_client(request) query: dict[str, Any] = {"match_all": {}} if keyword.strip(): query = { "multi_match": { "query": keyword.strip(), "fields": ["title", "doc_number", "issuing_org"], } } try: resp = await es.raw.search( index=settings.es_meta_index, body={ "query": query, "size": min(size, 100), "from": from_, "sort": [{"_score": "desc"}, {"publish_date": {"order": "desc"}}], }, ) raw = resp if isinstance(resp, dict) else resp.body hits = raw.get("hits", {}).get("hits", []) results = [] for hit in hits: src = hit.get("_source", {}) results.append( DocListItem( doc_id=src.get("doc_id") or hit["_id"], content_hash=src.get("content_hash") or "", title=src.get("title") or "", doc_number=src.get("doc_number") or "", issuing_org=src.get("issuing_org") or "", doc_type=src.get("doc_type") or "", publish_date=src.get("publish_date") or "", chunk_count=src.get("chunk_count") or 0, acl_ids=src.get("acl_ids") or [], status=src.get("status") or "", indexed_at=src.get("created_at") or "", ) ) return results except Exception as exc: # 安全修复：不向客户端暴露内部异常详情 logger.exception("mock_list_docs_error") raise HTTPException(status_code=500, detail="Internal server error") @router.get("/task/{task_id}", response_model=TaskStatusResponse) async def get_task_status(task_id: str) -> TaskStatusResponse: """Poll the status of a Celery ingest task. No authentication required for testing convenience. """ from celery.result import AsyncResult from app.tasks.ingest_task import ingest_document_task result = AsyncResult(task_id, app=ingest_document_task.app) if result.state == "PENDING": return TaskStatusResponse(task_id=task_id, status="PENDING", progress=0.0) elif result.state in ("STARTED", "PROCESSING"): info = result.info or {} return TaskStatusResponse( task_id=task_id, status="PROCESSING", progress=info.get("progress", 0.1) if isinstance(info, dict) else 0.1, ) elif result.state == "SUCCESS": info = result.result or {} # 原三元表达式两个分支都返回 "COMPLETED"，属于死代码，简化为直接赋值 status_str = "COMPLETED" return TaskStatusResponse( task_id=task_id, status=status_str, progress=1.0, result=info if isinstance(info, dict) else {"raw": str(info)}, ) elif result.state == "FAILURE": return TaskStatusResponse( task_id=task_id, status="FAILED", progress=0.0, error=str(result.result), ) else: return TaskStatusResponse(task_id=task_id, status=result.state, progress=0.0) @router.delete("/doc/{doc_id}") async def delete_test_doc( doc_id: str, request: Request, ) -> dict[str, Any]: """Delete a test document from ES (handles shared content) and Neo4j. Uses ESClient.delete_document() which correctly handles shared content: - If last reference: deletes chunks + meta - If shared content: removes meta, recomputes chunk ACL No authentication required for testing convenience. """ es = get_es_client(request) result = await es.delete_document(doc_id) # Delete from Neo4j (best-effort) try: neo4j = request.app.state.neo4j_client async with neo4j.driver.session(database=settings.neo4j_database) as session: await session.run( "MATCH (d:Document {doc_id: $doc_id}) DETACH DELETE d", doc_id=doc_id, ) result["neo4j"] = 1 except Exception as exc: logger.warning("mock_delete_neo4j_error", doc_id=doc_id, error=str(exc)) result["neo4j"] = 0 logger.info("mock_doc_deleted", doc_id=doc_id, **result) return {"doc_id": doc_id, **result, "status": "ok"} @router.get("/stats") async def system_stats(request: Request) -> dict[str, Any]: """Return basic system stats: ES doc counts, Redis info. No authentication required for testing convenience. """ stats: dict[str, Any] = {} # ES counts try: es = get_es_client(request) chunks_resp = await es.raw.count(index=settings.es_chunk_index) meta_resp = await es.raw.count(index=settings.es_meta_index) cr = chunks_resp if isinstance(chunks_resp, dict) else chunks_resp.body mr = meta_resp if isinstance(meta_resp, dict) else meta_resp.body stats["es"] = { "chunks": cr.get("count", 0), "documents": mr.get("count", 0), } except Exception as exc: stats["es"] = {"error": str(exc)} # Redis ping try: redis_client = request.app.state.redis_client await redis_client.ping() stats["redis"] = {"status": "ok"} except Exception as exc: stats["redis"] = {"status": "error", "error": str(exc)} # Neo4j node counts try: neo4j = request.app.state.neo4j_client async with neo4j.driver.session(database=settings.neo4j_database) as session: result = await session.run("MATCH (n) RETURN labels(n)[0] AS label, count(*) AS cnt") records = await result.data() stats["neo4j"] = {r["label"]: r["cnt"] for r in records if r["label"]} except Exception as exc: stats["neo4j"] = {"error": str(exc)} return stats