TanWenyan/leaudit-platform-backend
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: stabilize rule config and cross-review backend

Browse Source
This commit is contained in:
wren
2026-05-11 02:03:01 +08:00
parent 900fc2e8a2
commit 32fb2a4812
14 changed files with 444 additions and 46 deletions
Download Patch File Download Diff File Expand all files Collapse all files
fastapi_modules/fastapi_leaudit/controllers/ruleConfigController.py
+6 -6
View File
@@ -9,7 +9,7 @@ from fastapi_common.fastapi_common_security.security import verify_access_token
from fastapi_common.fastapi_common_web.controller import BaseController
from fastapi_modules.fastapi_leaudit.services.impl.permissionServiceImpl import PermissionServiceImpl
from fastapi_modules.fastapi_leaudit.services.impl.ruleConfigServiceImpl import RuleConfigServiceImpl
from fastapi_modules.fastapi_leaudit.services.impl.ruleConfigServiceImpl import GetRuleConfigServiceSingleton
from fastapi_modules.fastapi_leaudit.services.permissionService import IPermissionService
from fastapi_modules.fastapi_leaudit.services.ruleConfigService import IRuleConfigService
@@ -19,7 +19,7 @@ class RuleConfigController(BaseController):
def __init__(self):
super().__init__(prefix="/v3/rule-config-packs", tags=["规则配置"])
self.RuleConfigService: IRuleConfigService = RuleConfigServiceImpl()
self.RuleConfigService: IRuleConfigService = GetRuleConfigServiceSingleton()
self.PermissionService: IPermissionService = PermissionServiceImpl()
@self.router.get("")
@@ -42,7 +42,7 @@ class RuleConfigController(BaseController):
return JSONResponse(status_code=200, content={"code": 200, "message": "success", "data": data.model_dump()})
async def _check_permission(self, user_id: int) -> bool:
for permission_key in ("rules:list:read", "rules:content:read", "evaluation_group:list:read"):
if await self.PermissionService.CheckPermission(user_id, permission_key):
return True
return False
return await self.PermissionService.HasAnyPermission(
user_id,
["rules:list:read", "rules:content:read", "evaluation_group:list:read"],
)
fastapi_modules/fastapi_leaudit/controllers/ruleController.py
+2 -2
View File
@@ -18,7 +18,7 @@ from fastapi_modules.fastapi_leaudit.domian.vo.ruleVo import (
RuleVersionVO,
)
from fastapi_modules.fastapi_leaudit.services import IRuleService
from fastapi_modules.fastapi_leaudit.services.impl.ruleServiceImpl import RuleServiceImpl
from fastapi_modules.fastapi_leaudit.services.impl.ruleServiceImpl import GetRuleServiceSingleton
class RuleController(BaseController):
@@ -26,7 +26,7 @@ class RuleController(BaseController):
def __init__(self):
super().__init__(prefix="/rule-sets", tags=["规则管理"])
self.RuleService: IRuleService = RuleServiceImpl()
self.RuleService: IRuleService = GetRuleServiceSingleton()
@self.router.get("", response_model=Result[list[RuleSetVO]])
async def ListRuleSets():
fastapi_modules/fastapi_leaudit/services/impl/authServiceImpl.py
+8 -3
View File
@@ -27,6 +27,8 @@ class AuthServiceImpl(IAuthService):
"""账密登录。
现阶段仍兼容旧库明文密码，后续应迁移到哈希校验。
登录标识同时兼容旧系统常见的 `sub` 与 `username`，
避免前端展示用户名为 `admin`、实际登录只能输入 `000`。
"""
async with GetAsyncSession() as session:
from sqlalchemy import text
@@ -36,14 +38,17 @@ class AuthServiceImpl(IAuthService):
"SELECT id, sub, username, nick_name, phone_number, email, "
"ou_id, ou_name, is_leader, password, status, deleted_at, "
"try_count, try_login_time, area, tenant_name, dep_name, dep_short_name "
"FROM sso_users WHERE sub = :sub"
"FROM sso_users "
"WHERE deleted_at IS NULL AND (sub = :identifier OR username = :identifier) "
"ORDER BY CASE WHEN sub = :identifier THEN 0 ELSE 1 END, id ASC "
"LIMIT 1"
),
{"sub": Sub},
{"identifier": Sub},
)
row = result.fetchone()
if not row:
logger.warning("登录失败: 用户不存在 - sub=%s", Sub)
logger.warning("登录失败: 用户不存在 - identifier=%s", Sub)
raise LeauditException(StatusCodeEnum.HTTP_401_UNAUTHORIZED, "账号或密码错误")
user = dict(row._mapping)
fastapi_modules/fastapi_leaudit/services/impl/crossReviewServiceImpl.py
+132
View File
@@ -44,6 +44,102 @@ from fastapi_modules.fastapi_leaudit.services.impl.documentServiceImpl import Do
class CrossReviewServiceImpl(ICrossReviewService):
"""交叉评查服务实现。"""
_SCHEMA_BOOTSTRAP_STATEMENTS: tuple[str, ...] = (
"""
CREATE TABLE IF NOT EXISTS leaudit_cross_review_tasks (
id BIGINT GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
task_name VARCHAR(255) NOT NULL,
task_type VARCHAR(32) NOT NULL,
doc_type_id BIGINT,
doc_type_code VARCHAR(64),
assigner_id BIGINT NOT NULL,
status VARCHAR(32) NOT NULL DEFAULT 'in_progress',
create_time TIMESTAMPTZ NOT NULL DEFAULT NOW(),
update_time TIMESTAMPTZ NOT NULL DEFAULT NOW(),
delete_time TIMESTAMPTZ
)
""",
"CREATE INDEX IF NOT EXISTS idx_lcr_tasks_assigner_id ON leaudit_cross_review_tasks (assigner_id)",
"CREATE INDEX IF NOT EXISTS idx_lcr_tasks_status ON leaudit_cross_review_tasks (status)",
"CREATE INDEX IF NOT EXISTS idx_lcr_tasks_doc_type_id ON leaudit_cross_review_tasks (doc_type_id)",
"""
CREATE TABLE IF NOT EXISTS leaudit_cross_review_task_members (
id BIGINT GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
task_id BIGINT NOT NULL,
user_id BIGINT NOT NULL,
member_role VARCHAR(32) NOT NULL DEFAULT 'participant',
create_time TIMESTAMPTZ NOT NULL DEFAULT NOW(),
update_time TIMESTAMPTZ NOT NULL DEFAULT NOW(),
delete_time TIMESTAMPTZ
)
""",
"CREATE INDEX IF NOT EXISTS idx_lcr_task_members_task_id ON leaudit_cross_review_task_members (task_id)",
"CREATE INDEX IF NOT EXISTS idx_lcr_task_members_user_id ON leaudit_cross_review_task_members (user_id)",
"CREATE INDEX IF NOT EXISTS idx_lcr_task_members_role ON leaudit_cross_review_task_members (member_role)",
"""
CREATE UNIQUE INDEX IF NOT EXISTS uq_lcr_task_members_task_user_active
ON leaudit_cross_review_task_members (task_id, user_id)
WHERE delete_time IS NULL
""",
"""
CREATE TABLE IF NOT EXISTS leaudit_cross_review_task_documents (
id BIGINT GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
task_id BIGINT NOT NULL,
document_id BIGINT NOT NULL,
audit_status INTEGER NOT NULL DEFAULT 0,
create_time TIMESTAMPTZ NOT NULL DEFAULT NOW(),
update_time TIMESTAMPTZ NOT NULL DEFAULT NOW(),
delete_time TIMESTAMPTZ
)
""",
"CREATE INDEX IF NOT EXISTS idx_lcr_task_documents_task_id ON leaudit_cross_review_task_documents (task_id)",
"CREATE INDEX IF NOT EXISTS idx_lcr_task_documents_document_id ON leaudit_cross_review_task_documents (document_id)",
"CREATE INDEX IF NOT EXISTS idx_lcr_task_documents_task_status ON leaudit_cross_review_task_documents (task_id, audit_status)",
"""
CREATE UNIQUE INDEX IF NOT EXISTS uq_lcr_task_documents_task_document_active
ON leaudit_cross_review_task_documents (task_id, document_id)
WHERE delete_time IS NULL
""",
"""
CREATE TABLE IF NOT EXISTS leaudit_cross_review_proposals (
id BIGINT GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
task_id BIGINT NOT NULL,
document_id BIGINT NOT NULL,
rule_result_id BIGINT NOT NULL,
proposer_id BIGINT NOT NULL,
proposed_score_delta NUMERIC(10, 2) NOT NULL,
reason TEXT NOT NULL,
status VARCHAR(32) NOT NULL DEFAULT 'pending',
create_time TIMESTAMPTZ NOT NULL DEFAULT NOW(),
update_time TIMESTAMPTZ NOT NULL DEFAULT NOW(),
delete_time TIMESTAMPTZ
)
""",
"CREATE INDEX IF NOT EXISTS idx_lcr_proposals_task_id ON leaudit_cross_review_proposals (task_id)",
"CREATE INDEX IF NOT EXISTS idx_lcr_proposals_document_id ON leaudit_cross_review_proposals (document_id)",
"CREATE INDEX IF NOT EXISTS idx_lcr_proposals_rule_result_id ON leaudit_cross_review_proposals (rule_result_id)",
"CREATE INDEX IF NOT EXISTS idx_lcr_proposals_proposer_id ON leaudit_cross_review_proposals (proposer_id)",
"CREATE INDEX IF NOT EXISTS idx_lcr_proposals_status ON leaudit_cross_review_proposals (status)",
"""
CREATE TABLE IF NOT EXISTS leaudit_cross_review_votes (
id BIGINT GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
proposal_id BIGINT NOT NULL,
voter_id BIGINT NOT NULL,
vote_type VARCHAR(16) NOT NULL,
create_time TIMESTAMPTZ NOT NULL DEFAULT NOW(),
update_time TIMESTAMPTZ NOT NULL DEFAULT NOW(),
delete_time TIMESTAMPTZ
)
""",
"CREATE INDEX IF NOT EXISTS idx_lcr_votes_proposal_id ON leaudit_cross_review_votes (proposal_id)",
"CREATE INDEX IF NOT EXISTS idx_lcr_votes_voter_id ON leaudit_cross_review_votes (voter_id)",
"""
CREATE UNIQUE INDEX IF NOT EXISTS uq_lcr_votes_proposal_voter_active
ON leaudit_cross_review_votes (proposal_id, voter_id)
WHERE delete_time IS NULL
""",
)
def __init__(self):
self.DocumentService: IDocumentService = DocumentServiceImpl()
@@ -56,6 +152,7 @@ class CrossReviewServiceImpl(ICrossReviewService):
principalUserIds = self._unique_int_list(Body.principalUserIds)
documentIds = self._unique_int_list(Body.documentIds)
await self._reset_transaction_for_write(session)
async with session.begin():
taskRow = (
await session.execute(
@@ -400,6 +497,7 @@ class CrossReviewServiceImpl(ICrossReviewService):
if not permission.canConfirm:
raise LeauditException(StatusCodeEnum.HTTP_403_FORBIDDEN, permission.reason)
await self._reset_transaction_for_write(session)
async with session.begin():
mapping = (
await session.execute(
@@ -517,6 +615,7 @@ class CrossReviewServiceImpl(ICrossReviewService):
if Body.deductionScore > 0 and currentScore >= fullScore:
raise LeauditException(StatusCodeEnum.HTTP_400_BAD_REQUEST, "当前分值已满分，不能继续加分")
await self._reset_transaction_for_write(session)
async with session.begin():
proposalRow = (
await session.execute(
@@ -576,6 +675,7 @@ class CrossReviewServiceImpl(ICrossReviewService):
if str(proposal["status"]) in {"approved", "rejected", "cancelled"}:
raise LeauditException(StatusCodeEnum.HTTP_400_BAD_REQUEST, "当前提案状态不允许继续投票")
await self._reset_transaction_for_write(session)
async with session.begin():
if voteType == "cancel":
deleted = await session.execute(
@@ -638,6 +738,7 @@ class CrossReviewServiceImpl(ICrossReviewService):
if str(proposal["status"]) not in {"pending"}:
raise LeauditException(StatusCodeEnum.HTTP_400_BAD_REQUEST, "当前提案状态不允许撤销")
await self._reset_transaction_for_write(session)
async with session.begin():
await session.execute(
text(
@@ -745,6 +846,7 @@ class CrossReviewServiceImpl(ICrossReviewService):
async with GetAsyncSession() as session:
await self._ensure_tables_ready(session)
await self._reset_transaction_for_write(session)
async with session.begin():
exists = bool(
await session.scalar(
@@ -1153,6 +1255,31 @@ class CrossReviewServiceImpl(ICrossReviewService):
"leaudit_cross_review_proposals",
"leaudit_cross_review_votes",
]
missing_tables: list[str] = []
for tableName in required:
exists = bool(
await session.scalar(
text(
"""
SELECT EXISTS (
SELECT 1
FROM information_schema.tables
WHERE table_schema = current_schema()
AND table_name = :table_name
)
"""
),
{"table_name": tableName},
)
)
if not exists:
missing_tables.append(tableName)
if missing_tables:
for statement in self._SCHEMA_BOOTSTRAP_STATEMENTS:
await session.execute(text(statement))
await session.commit()
for tableName in required:
exists = bool(
await session.scalar(
@@ -1195,6 +1322,11 @@ class CrossReviewServiceImpl(ICrossReviewService):
if not exists:
raise LeauditException(StatusCodeEnum.HTTP_403_FORBIDDEN, "当前用户不是交叉评查任务成员")
async def _reset_transaction_for_write(self, session) -> None:
"""显式写事务前清理查询阶段开启的隐式事务。"""
if session.in_transaction():
await session.rollback()
def _unique_int_list(self, values: list[int]) -> list[int]:
"""去重并保留原顺序。"""
seen: set[int] = set()
fastapi_modules/fastapi_leaudit/services/impl/evaluationPointGroupServiceImpl.py
+2 -2
View File
@@ -39,14 +39,14 @@ from fastapi_modules.fastapi_leaudit.services.impl.ruleGroupSupport import (
ensure_rule_group_schema,
sync_doc_type_bindings_from_group,
)
from fastapi_modules.fastapi_leaudit.services.impl.ruleServiceImpl import RuleServiceImpl
from fastapi_modules.fastapi_leaudit.services.impl.ruleServiceImpl import GetRuleServiceSingleton
class EvaluationPointGroupServiceImpl(IEvaluationPointGroupService):
"""评查点分组服务实现。"""
def __init__(self) -> None:
self.RuleService = RuleServiceImpl()
self.RuleService = GetRuleServiceSingleton()
async def ListGroups(
self,
fastapi_modules/fastapi_leaudit/services/impl/homeServiceImpl.py
+2 -4
View File
@@ -21,6 +21,7 @@ class HomeServiceImpl(IHomeService):
"/files/upload",
"/documents",
"/chat-with-llm/chat",
"/cross-checking",
)
def __init__(self) -> None:
@@ -162,7 +163,7 @@ class HomeServiceImpl(IHomeService):
if not self._isAllowedTargetPath(targetPath, allowedPaths):
continue
requiresDocumentTypes = targetPath not in {"/chat-with-llm/chat"}
requiresDocumentTypes = targetPath not in {"/chat-with-llm/chat", "/cross-checking"}
modules.append(
HomeEntryModuleVO(
@@ -210,9 +211,6 @@ class HomeServiceImpl(IHomeService):
if RawPath == "/contract-template/search" and HasDocumentTypes:
return "/files/upload"
if RawPath == "/cross-checking":
return None
if any(
RawPath == enabledPath or RawPath.startswith(f"{enabledPath}/")
for enabledPath in self._MINIMAL_ENABLED_TARGETS
fastapi_modules/fastapi_leaudit/services/impl/permissionServiceImpl.py
+10
View File
@@ -9,6 +9,7 @@
from fastapi_common.fastapi_common_logger import logger
from fastapi_common.fastapi_common_sqlalchemy.database import GetAsyncSession
import time
from fastapi_modules.fastapi_leaudit.services.permissionService import IPermissionService
@@ -16,6 +17,9 @@ from fastapi_modules.fastapi_leaudit.services.permissionService import IPermissi
class PermissionServiceImpl(IPermissionService):
"""权限检查服务实现。"""
def __init__(self) -> None:
self._permission_cache: dict[int, tuple[float, tuple[set[str], set[str]]]] = {}
async def CheckPermission(self, UserId: int, PermissionKey: str) -> bool:
"""检查用户是否拥有指定权限。
@@ -74,6 +78,11 @@ class PermissionServiceImpl(IPermissionService):
async def _getUserPermissions(self, UserId: int) -> tuple[set[str], set[str]]:
"""从数据库查询用户的 GRANT 和 DENY 权限集合。"""
cached = self._permission_cache.get(UserId)
now = time.monotonic()
if cached and now - cached[0] <= 60:
return cached[1]
grants: set[str] = set()
denies: set[str] = set()
@@ -107,6 +116,7 @@ class PermissionServiceImpl(IPermissionService):
denies.add(permKey)
logger.debug(f"用户权限: user={UserId}, grants={len(grants)}, denies={len(denies)}")
self._permission_cache[UserId] = (now, (grants, denies))
return grants, denies
@staticmethod
fastapi_modules/fastapi_leaudit/services/impl/rbacAdminServiceImpl.py
+1 -1
View File
@@ -124,7 +124,7 @@ class RbacAdminServiceImpl(IRbacAdminService):
"route_path": "/cross-checking/result",
"route_name": "cross-checking-result",
"component": "cross-checking.result",
"route_title": "评查结果",
"route_title": "评查任务列表",
"icon": "ri-file-list-3-line",
"sort_order": 2,
"parent_path": "/cross-checking",
fastapi_modules/fastapi_leaudit/services/impl/ruleConfigServiceImpl.py
+67 -5
View File
@@ -5,6 +5,7 @@ from __future__ import annotations
import asyncio
from collections import defaultdict
import re
import time
from typing import Any
from fastapi_common.fastapi_common_sqlalchemy.database import GetAsyncSession
@@ -20,17 +21,49 @@ from fastapi_modules.fastapi_leaudit.domian.vo.ruleConfigVo import (
from fastapi_modules.fastapi_leaudit.leaudit_bridge.ruleValidator import RuleValidator
from fastapi_modules.fastapi_leaudit.services import IOssService
from fastapi_modules.fastapi_leaudit.services.impl.ossServiceImpl import OssServiceImpl
from fastapi_modules.fastapi_leaudit.services.impl.ruleServiceImpl import RuleServiceImpl
from fastapi_modules.fastapi_leaudit.services.impl.ruleServiceImpl import GetRuleServiceSingleton
from fastapi_modules.fastapi_leaudit.services.ruleConfigService import IRuleConfigService
class RuleConfigServiceImpl(IRuleConfigService):
"""规则配置页聚合服务实现。"""
_GLOBAL_YAML_SUMMARY_CACHE: dict[int, tuple[float, dict[str, Any]]] = {}
_GLOBAL_PACK_SUMMARY_CACHE: tuple[float, list[RuleConfigPackListVO]] | None = None
_GLOBAL_WARM_LOCK: asyncio.Lock | None = None
def __init__(self, OssService: IOssService | None = None) -> None:
self.OssService = OssService or OssServiceImpl()
self.RuleService = RuleServiceImpl(self.OssService)
self.RuleService = GetRuleServiceSingleton()
self.Validator = RuleValidator()
self._yaml_summary_cache = self.__class__._GLOBAL_YAML_SUMMARY_CACHE
self._pack_summary_cache = self.__class__._GLOBAL_PACK_SUMMARY_CACHE
@classmethod
def _set_pack_summary_cache(cls, value: tuple[float, list[RuleConfigPackListVO]] | None) -> None:
cls._GLOBAL_PACK_SUMMARY_CACHE = value
@classmethod
def _get_warm_lock(cls) -> asyncio.Lock:
if cls._GLOBAL_WARM_LOCK is None:
cls._GLOBAL_WARM_LOCK = asyncio.Lock()
return cls._GLOBAL_WARM_LOCK
def InvalidateSummaryCaches(self, version_ids: list[int] | None = None) -> None:
"""清理规则摘要缓存；version_ids 为空时清空全部。"""
self.__class__._set_pack_summary_cache(None)
if version_ids is None:
self._yaml_summary_cache.clear()
return
for version_id in {int(item) for item in version_ids if item is not None}:
self._yaml_summary_cache.pop(version_id, None)
async def WarmPackSummaries(self, force: bool = False) -> list[RuleConfigPackListVO]:
"""预热规则列表摘要缓存。"""
async with self.__class__._get_warm_lock():
if force:
self.InvalidateSummaryCaches()
return await self.ListPackSummaries()
async def ListPacks(self) -> list[RuleConfigPackVO]:
"""列出规则配置页所需的全部 pack。"""
@@ -40,6 +73,11 @@ class RuleConfigServiceImpl(IRuleConfigService):
async def ListPackSummaries(self) -> list[RuleConfigPackListVO]:
"""列出规则列表页所需的轻量 pack。"""
cached = self.__class__._GLOBAL_PACK_SUMMARY_CACHE
now = time.monotonic()
if cached and now - cached[0] <= 60:
return cached[1]
rows = await self._load_pack_rows()
if not rows:
return []
@@ -121,6 +159,9 @@ class RuleConfigServiceImpl(IRuleConfigService):
item["usableRuleCount"] = len(rules)
packs.append(RuleConfigPackListVO(**item))
cache_value = (time.monotonic(), packs)
self.__class__._set_pack_summary_cache(cache_value)
self._pack_summary_cache = cache_value
return packs
async def GetPack(self, PackId: int) -> RuleConfigPackVO:
@@ -471,10 +512,16 @@ class RuleConfigServiceImpl(IRuleConfigService):
async def _load_one(version_id: int, oss_url: str):
if not oss_url:
return version_id, {"loaded": False, "yaml_name": "", "rules": []}
cached = self._yaml_summary_cache.get(version_id)
now = time.monotonic()
if cached and now - cached[0] <= 120:
return version_id, cached[1]
try:
yaml_text = (await self.OssService.DownloadBytes(oss_url)).decode("utf-8")
if not yaml_text.strip():
return version_id, {"loaded": False, "yaml_name": "", "rules": []}
summary = {"loaded": False, "yaml_name": "", "rules": []}
self._yaml_summary_cache[version_id] = (time.monotonic(), summary)
return version_id, summary
rules_file = self.Validator.ParseValidated(yaml_text)
groups: dict[str, str] = {}
try:
@@ -540,13 +587,17 @@ class RuleConfigServiceImpl(IRuleConfigService):
seen_dependencies.add(normalized_dependency)
merged_dependencies.append(normalized_dependency)
item["dependencies"] = merged_dependencies
return version_id, {
summary = {
"loaded": True,
"yaml_name": str(getattr(getattr(rules_file, "metadata", None), "name", "") or ""),
"rules": summaries,
}
self._yaml_summary_cache[version_id] = (time.monotonic(), summary)
return version_id, summary
except Exception:
return version_id, {"loaded": False, "yaml_name": "", "rules": []}
summary = {"loaded": False, "yaml_name": "", "rules": []}
self._yaml_summary_cache[version_id] = (time.monotonic(), summary)
return version_id, summary
results = await asyncio.gather(*[_load_one(version_id, oss_url) for version_id, oss_url in version_oss_map.items()])
return dict(results)
@@ -592,3 +643,14 @@ class RuleConfigServiceImpl(IRuleConfigService):
if value is None:
return None
return int(value)
_RULE_CONFIG_SERVICE_SINGLETON: RuleConfigServiceImpl | None = None
def GetRuleConfigServiceSingleton() -> RuleConfigServiceImpl:
"""返回共享的规则配置服务实例，供控制器和预热任务共用缓存。"""
global _RULE_CONFIG_SERVICE_SINGLETON
if _RULE_CONFIG_SERVICE_SINGLETON is None:
_RULE_CONFIG_SERVICE_SINGLETON = RuleConfigServiceImpl()
return _RULE_CONFIG_SERVICE_SINGLETON
fastapi_modules/fastapi_leaudit/services/impl/ruleServiceImpl.py
+59 -2
View File
@@ -3,6 +3,8 @@
from __future__ import annotations
import hashlib
import logging
import time
from fastapi_common.fastapi_common_sqlalchemy.database import GetAsyncSession
from fastapi_common.fastapi_common_web.domain.responses import StatusCodeEnum
from fastapi_common.fastapi_common_web.exception.LeauditException import LeauditException
@@ -25,6 +27,9 @@ from fastapi_modules.fastapi_leaudit.services.impl.ruleGroupSupport import sync_
class RuleServiceImpl(IRuleService):
"""规则服务实现。"""
_GLOBAL_LIST_SETS_CACHE: tuple[float, list[RuleSetVO]] | None = None
_GLOBAL_RULE_COUNT_CACHE: dict[int, tuple[float, int]] = {}
def __init__(
self,
OssService: IOssService | None = None,
@@ -32,6 +37,21 @@ class RuleServiceImpl(IRuleService):
) -> None:
self.OssService = OssService or OssServiceImpl()
self.Validator = Validator or RuleValidator()
self._list_sets_cache = self.__class__._GLOBAL_LIST_SETS_CACHE
self._rule_count_cache = self.__class__._GLOBAL_RULE_COUNT_CACHE
@classmethod
def _set_list_sets_cache(cls, value: tuple[float, list[RuleSetVO]] | None) -> None:
cls._GLOBAL_LIST_SETS_CACHE = value
def InvalidateCaches(self, version_ids: list[int] | None = None) -> None:
"""清理规则集及规则数量缓存。"""
self.__class__._set_list_sets_cache(None)
if version_ids is None:
self._rule_count_cache.clear()
return
for version_id in {int(item) for item in version_ids if item is not None}:
self._rule_count_cache.pop(version_id, None)
async def _resolve_unique_child_group_id(self, Session, DocTypeId: int) -> int | None:
"""仅当文档类型唯一对应一个二级分组时，返回该分组ID。"""
@@ -71,6 +91,11 @@ class RuleServiceImpl(IRuleService):
async def ListSets(self) -> list[RuleSetVO]:
"""列出所有规则集。"""
now = time.monotonic()
cached = self.__class__._GLOBAL_LIST_SETS_CACHE
if cached and now - cached[0] <= 30:
return cached[1]
async with GetAsyncSession() as Session:
Result = await Session.execute(
text(
@@ -114,7 +139,7 @@ class RuleServiceImpl(IRuleService):
if usable_version_id is not None and int(usable_version_id) not in usable_counts:
usable_counts[int(usable_version_id)] = await self._GetRuleCountByVersionId(int(usable_version_id))
return [
items = [
RuleSetVO(
id=int(Row["id"]),
ruleType=Row["rule_type"],
@@ -130,9 +155,18 @@ class RuleServiceImpl(IRuleService):
)
for Row in rows
]
cache_value = (time.monotonic(), items)
self.__class__._set_list_sets_cache(cache_value)
self._list_sets_cache = cache_value
return items
async def _GetRuleCountByVersionId(self, VersionId: int) -> int:
"""读取指定可用规则版本的规则数。"""
cached = self._rule_count_cache.get(VersionId)
now = time.monotonic()
if cached and now - cached[0] <= 120:
return cached[1]
async with GetAsyncSession() as Session:
Result = await Session.execute(
text(
@@ -153,7 +187,9 @@ class RuleServiceImpl(IRuleService):
try:
yaml_text = (await self.OssService.DownloadBytes(Row["oss_url"])).decode("utf-8")
validation = self.Validator.ValidateYaml(yaml_text)
return int(validation.ruleCount or 0)
count = int(validation.ruleCount or 0)
self._rule_count_cache[VersionId] = (time.monotonic(), count)
return count
except Exception:
return 0
@@ -786,6 +822,16 @@ class RuleServiceImpl(IRuleService):
)
await Session.commit()
self.InvalidateCaches()
try:
from fastapi_modules.fastapi_leaudit.services.impl.ruleConfigServiceImpl import GetRuleConfigServiceSingleton
RuleConfigService = GetRuleConfigServiceSingleton()
RuleConfigService.InvalidateSummaryCaches()
await RuleConfigService.WarmPackSummaries(force=False)
except Exception as exc:
logging.getLogger("RULE").warning("刷新规则配置摘要缓存失败: %s", exc)
VersionRow = await self._GetVersion(Session, VersionId)
return self._BuildRuleVersionVo(VersionRow)
@@ -848,3 +894,14 @@ class RuleServiceImpl(IRuleService):
"legal_doc": "legal_doc",
}
return Mapping.get(Prefix, Prefix or "unknown")
_RULE_SERVICE_SINGLETON: RuleServiceImpl | None = None
def GetRuleServiceSingleton() -> RuleServiceImpl:
"""返回共享规则服务实例，供控制器与聚合服务复用缓存。"""
global _RULE_SERVICE_SINGLETON
if _RULE_SERVICE_SINGLETON is None:
_RULE_SERVICE_SINGLETON = RuleServiceImpl()
return _RULE_SERVICE_SINGLETON