from datetime import UTC, datetime from typing import Any from app.core.contracts import RequestContext from app.core.errors import ResourceNotFoundError from app.service.authorization_service import AuthorizationService class KnowledgeReferenceService: async def resolve(self, context: RequestContext, token: str) -> dict[str, Any]: await AuthorizationService.require(context, "knowledge:reference:read") pieces = token.split(".") if len(pieces) != 5 or pieces[0] != "kr1" or pieces[1] != context.user_id: raise ResourceNotFoundError("引用不存在") try: if int(pieces[3]) < int(datetime.now(UTC).timestamp()): raise ResourceNotFoundError("引用不存在") except ValueError as exc: raise ResourceNotFoundError("引用不存在") from exc raise ResourceNotFoundError("引用不存在")