tmp

ASP/settings.py

@@ -102,7 +102,7 @@ LOGGING = {
                 'file',
             ],
             'level': 'DEBUG',
-            'propagate': True,
+            'propagate': False,
         },
 
     }

MODULES/Cloud-01-AWS-IAM-Privilege-Escalation-via-AttachUserPolicy.py

@@ -7,7 +7,7 @@ from Lib.basemodule import BaseModule
 from PLUGINS.SIRP.correlation import Correlation
 from PLUGINS.SIRP.sirpapi import Alert, Case
 from PLUGINS.SIRP.sirpcoremodel import ArtifactType, ArtifactRole, Severity, Impact, Disposition, AlertAction, Confidence, AlertAnalyticType, ProductCategory, \
-    AlertPolicyType, AlertRiskLevel, AlertStatus, CasePriority, ArtifactModel, AlertModel, CaseModel, EnrichmentModel
+    AlertPolicyType, AlertRiskLevel, AlertStatus, CasePriority, ArtifactModel, AlertModel, CaseModel, EnrichmentModel, CaseStatus
 
 
 class Module(BaseModule):
@@ -178,7 +178,6 @@ class Module(BaseModule):
 
         # 保存告警
         saved_alert_row_id = Alert.create(alert_model)
-        self.logger.info(f"Alert created: {saved_alert_row_id}")
 
         # 5. Case 处理 (Case Management)
         try:
@@ -194,6 +193,7 @@ class Module(BaseModule):
                 # 根据 Alert 计算 Case字段
                 new_case = CaseModel(
                     title=f"Potential IAM Privilege Escalation in Account {account_id}",
+                    status=CaseStatus.NEW,  # 创建时显式设置为New
                     severity=severity,
                     impact=Impact.HIGH if outcome == "success" else Impact.MEDIUM,
                     priority=CasePriority.HIGH if outcome == "success" else CasePriority.MEDIUM,

PLUGINS/SIRP/nocolyapi.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+import time
 from typing import Dict
 from typing import List, Any
 
@@ -27,6 +28,24 @@ adapter = HTTPAdapter(
 HTTP_SESSION.mount('http://', adapter)
 HTTP_SESSION.mount('https://', adapter)
 
+
+def _request_with_timing(method: str, url: str, **kwargs):
+    started_at = time.perf_counter()
+    try:
+        response = HTTP_SESSION.request(method=method, url=url, **kwargs)
+        elapsed_ms = (time.perf_counter() - started_at) * 1000
+        logger.debug(
+            f"[SIRP] request method={method} url={url} status={response.status_code} elapsed_ms={elapsed_ms:.2f}"
+        )
+        return response
+    except Exception as exc:
+        elapsed_ms = (time.perf_counter() - started_at) * 1000
+        logger.debug(
+            f"[SIRP] request failed method={method} url={url} elapsed_ms={elapsed_ms:.2f} error={exc}"
+        )
+        raise
+
+
 SYSTEM_FIELDS = ['_initiatedBy', '_owner', '_updatedAt', '_createdAt', '_remainingTime', '_createdBy', '_updatedBy', '_processName', '_nodeAssignees',
                  '_initiatedAt', '_nodeStartedAt', '_approvalCompletedAt', '_dueAt', '_processStatus']
 
@@ -43,7 +62,8 @@ class Worksheet(object):
 
         url = f"{SIRP_URL}/api/v3/app/worksheets/{worksheet_id}"
 
-        response = HTTP_SESSION.get(
+        response = _request_with_timing(
+            "GET",
             url
         )
         response.raise_for_status()
@@ -169,7 +189,8 @@ class WorksheetRow(object):
     def get(worksheet_id: str, row_id: str, include_system_fields=True) -> dict:
         url = f"{SIRP_URL}/api/v3/app/worksheets/{worksheet_id}/rows/{row_id}"
         fields = Worksheet.get_fields(worksheet_id)
-        response = HTTP_SESSION.get(
+        response = _request_with_timing(
+            "GET",
             url,
             timeout=SIRP_REQUEST_TIMEOUT,
             params={"includeSystemFields": include_system_fields}
@@ -210,10 +231,11 @@ class WorksheetRow(object):
                 "pageIndex": page_index
             }
 
-            response = HTTP_SESSION.post(url,
-                                         timeout=SIRP_REQUEST_TIMEOUT,
-                                         headers=HEADERS,
-                                         json=data)
+            response = _request_with_timing("POST",
+                                            url,
+                                            timeout=SIRP_REQUEST_TIMEOUT,
+                                            headers=HEADERS,
+                                            json=data)
             response.raise_for_status()
             response_data = response.json()
 
@@ -250,9 +272,10 @@ class WorksheetRow(object):
         }
 
         try:
-            response = HTTP_SESSION.post(url,
-                                         timeout=SIRP_REQUEST_TIMEOUT,
-                                         json=data)
+            response = _request_with_timing("POST",
+                                            url,
+                                            timeout=SIRP_REQUEST_TIMEOUT,
+                                            json=data)
             response.raise_for_status()
 
             response_data = response.json()
@@ -274,9 +297,10 @@ class WorksheetRow(object):
             "triggerWorkflow": trigger_workflow,
             "fields": fields
         }
-        response = HTTP_SESSION.patch(url,
-                                      timeout=SIRP_REQUEST_TIMEOUT,
-                                      json=data)
+        response = _request_with_timing("PATCH",
+                                        url,
+                                        timeout=SIRP_REQUEST_TIMEOUT,
+                                        json=data)
         response.raise_for_status()
 
         response_data = response.json()
@@ -304,9 +328,10 @@ class WorksheetRow(object):
         }
 
         try:
-            response = HTTP_SESSION.post(url,
-                                         timeout=SIRP_REQUEST_TIMEOUT,
-                                         json=data)
+            response = _request_with_timing("POST",
+                                            url,
+                                            timeout=SIRP_REQUEST_TIMEOUT,
+                                            json=data)
             response.raise_for_status()
 
             response_data = response.json()
@@ -335,9 +360,10 @@ class WorksheetRow(object):
         }
 
         try:
-            response = HTTP_SESSION.patch(url,
-                                          timeout=SIRP_REQUEST_TIMEOUT,
-                                          json=data)
+            response = _request_with_timing("PATCH",
+                                            url,
+                                            timeout=SIRP_REQUEST_TIMEOUT,
+                                            json=data)
             response.raise_for_status()
 
             response_data = response.json()
@@ -357,9 +383,10 @@ class WorksheetRow(object):
             "triggerWorkflow": trigger_workflow,
         }
 
-        response = HTTP_SESSION.delete(url,
-                                       timeout=SIRP_REQUEST_TIMEOUT,
-                                       json=data)
+        response = _request_with_timing("DELETE",
+                                        url,
+                                        timeout=SIRP_REQUEST_TIMEOUT,
+                                        json=data)
         response.raise_for_status()
 
         response_data = response.json()
@@ -378,9 +405,10 @@ class WorksheetRow(object):
             "triggerWorkflow": trigger_workflow,
         }
 
-        response = HTTP_SESSION.delete(url,
-                                       timeout=SIRP_REQUEST_TIMEOUT,
-                                       json=data)
+        response = _request_with_timing("DELETE",
+                                        url,
+                                        timeout=SIRP_REQUEST_TIMEOUT,
+                                        json=data)
         response.raise_for_status()
 
         response_data = response.json()
@@ -395,8 +423,9 @@ class WorksheetRow(object):
     def get_discussions(worksheet_id: str, row_id: str) -> List[Dict]:
         url = f"{SIRP_URL}/api/v3/app/worksheets/{worksheet_id}/rows/{row_id}/discussions"
 
-        response = HTTP_SESSION.get(url,
-                                    timeout=SIRP_REQUEST_TIMEOUT)
+        response = _request_with_timing("GET",
+                                        url,
+                                        timeout=SIRP_REQUEST_TIMEOUT)
         response.raise_for_status()
 
         response_data = response.json()
@@ -437,9 +466,10 @@ class WorksheetRow(object):
         if include_system_fields is not None:
             params["isReturnSystemFields"] = include_system_fields
 
-        response = HTTP_SESSION.get(url,
-                                    timeout=SIRP_REQUEST_TIMEOUT,
-                                    params=params)
+        response = _request_with_timing("GET",
+                                        url,
+                                        timeout=SIRP_REQUEST_TIMEOUT,
+                                        params=params)
         response.raise_for_status()
 
         response_data = response.json()
@@ -465,8 +495,9 @@ class OptionSet(object):
             return cached_optionsets
         url = f"{SIRP_URL}/api/v3/app/optionsets"
 
-        response = HTTP_SESSION.get(url,
-                                    timeout=SIRP_REQUEST_TIMEOUT)
+        response = _request_with_timing("GET",
+                                        url,
+                                        timeout=SIRP_REQUEST_TIMEOUT)
         response.raise_for_status()
 
         response_data = response.json()

PLUGINS/SIRP/sirpapi.py

@@ -1,20 +1,338 @@
 import json
-from typing import List, Union, Annotated
+from abc import ABC
+from typing import List, Union, Annotated, Dict, Any, TypeVar, Generic, Type
 
 import requests
 from langchain_core.documents import Document
+from pydantic import BaseModel
 
 from Lib.log import logger
 from PLUGINS.Embeddings.embeddings_qdrant import embedding_api_singleton_qdrant, SIRP_KNOWLEDGE_COLLECTION
 from PLUGINS.SIRP.CONFIG import SIRP_NOTICE_WEBHOOK
 from PLUGINS.SIRP.nocolyapi import WorksheetRow
 from PLUGINS.SIRP.nocolymodel import Condition, Group, Operator
-from PLUGINS.SIRP.sirpbase import BaseWorksheetEntity
-from PLUGINS.SIRP.sirpbasemodel import AutoAccount
+from PLUGINS.SIRP.sirpbasemodel import AutoAccount, BaseSystemModel
 from PLUGINS.SIRP.sirpcoremodel import Severity, Confidence, EnrichmentModel, TicketModel, ArtifactModel, AlertModel, CaseModel
 from PLUGINS.SIRP.sirpextramodel import PlaybookType, PlaybookJobStatus, KnowledgeAction, MessageModel, PlaybookModel, KnowledgeModel
 
 
+def model_to_fields(model_instance: BaseModel) -> List[Dict[str, Any]]:
+    fields = []
+    model_data = model_instance.model_dump(mode='json', exclude_unset=True)
+    for key, value in model_data.items():
+        field_info = model_instance.model_fields.get(key)
+        field_item = {
+            'id': key,
+            'value': value
+        }
+        if field_info and field_info.json_schema_extra:
+            field_item.update(field_info.json_schema_extra)
+        fields.append(field_item)
+    return fields
+
+
+T = TypeVar('T', bound=BaseSystemModel)
+
+
+class BaseSimpleEntity(ABC):
+    """简化的工作表实体基类（不使用模型）"""
+
+    WORKSHEET_ID: str
+
+    @classmethod
+    def list(cls, filter_dict: dict) -> List[Dict]:
+        """列表查询
+
+        Args:
+            filter_dict: 过滤条件字典
+
+        Returns:
+            字典列表
+        """
+        return WorksheetRow.list(cls.WORKSHEET_ID, filter_dict, include_system_fields=False)
+
+    @classmethod
+    def get(cls, row_id: str) -> Dict:
+        """获取单条记录
+
+        Args:
+            row_id: 记录ID
+
+        Returns:
+            字典
+        """
+        return WorksheetRow.get(cls.WORKSHEET_ID, row_id, include_system_fields=False)
+
+    @classmethod
+    def create(cls, fields: List[Dict]) -> str:
+        """创建记录
+
+        Args:
+            fields: 字段列表
+
+        Returns:
+            新创建的记录ID
+        """
+        return WorksheetRow.create(cls.WORKSHEET_ID, fields)
+
+    @classmethod
+    def update(cls, row_id: str, fields: List[Dict]) -> str:
+        """更新记录
+
+        Args:
+            row_id: 记录ID
+            fields: 字段列表
+
+        Returns:
+            更新的记录ID
+        """
+        return WorksheetRow.update(cls.WORKSHEET_ID, row_id, fields)
+
+
+class BaseWorksheetEntity(ABC, Generic[T]):
+    """通用工作表实体基类 - 支持泛型和关联加载"""
+
+    WORKSHEET_ID: str
+    MODEL_CLASS: Type[T]
+
+    @classmethod
+    def get(
+            cls,
+            row_id: str,
+            include_system_fields: bool = True,
+            lazy_load: bool = False
+    ) -> T:
+        """获取单条记录
+
+        Args:
+            row_id: 记录ID
+            include_system_fields: 是否包含系统字段
+            lazy_load: 是否延迟加载关联数据（True时不加载关联）
+
+        Returns:
+            模型实例
+        """
+        result = WorksheetRow.get(
+            cls.WORKSHEET_ID,
+            row_id,
+            include_system_fields=include_system_fields
+        )
+        model = cls.MODEL_CLASS(**result)
+
+        if not lazy_load:
+            model = cls._load_relations(model, include_system_fields)
+
+        return model
+
+    @classmethod
+    def list(
+            cls,
+            filter_model: Group,
+            include_system_fields: bool = True,
+            lazy_load: bool = False
+    ) -> List[T]:
+        """按过滤条件列表查询
+
+        Args:
+            filter_model: 过滤条件Group对象
+            include_system_fields: 是否包含系统字段
+            lazy_load: 是否延迟加载关联数据（True时不加载关联）
+
+        Returns:
+            模型实例列表
+        """
+        if filter_model.children:
+            filter_dict = filter_model.model_dump()
+        else:
+            filter_dict = {}
+        result = WorksheetRow.list(
+            cls.WORKSHEET_ID,
+            filter_dict,
+            include_system_fields=include_system_fields
+        )
+
+        model_list = []
+        for item in result:
+            model_obj = cls.MODEL_CLASS(**item)
+            if not lazy_load:
+                model_obj = cls._load_relations(model_obj, include_system_fields)
+            model_list.append(model_obj)
+
+        return model_list
+
+    @classmethod
+    def update_by_filter(cls,
+                         filter_model: Group,
+                         model: T,
+                         include_system_fields: bool = True) -> dict:
+        filter_dict = filter_model.model_dump()
+        result = WorksheetRow.list(
+            cls.WORKSHEET_ID,
+            filter_dict,
+            fields=["row_id"],
+            include_system_fields=include_system_fields
+        )
+        row_ids = []
+        for item in result:
+            row_ids.append(item["row_id"])
+
+        model = cls._prepare_for_save(model)
+
+        fields = model_to_fields(model)
+        result = WorksheetRow.batch_update(cls.WORKSHEET_ID, row_ids, fields)
+        return result
+
+    @classmethod
+    def list_by_row_ids(
+            cls,
+            row_ids: List[Any],
+            include_system_fields: bool = True,
+            lazy_load: bool = False
+    ) -> Union[List[T], List[str], None]:
+        """按ID列表查询
+
+        Args:
+            row_ids: 记录ID列表
+            include_system_fields: 是否包含系统字段
+            lazy_load: 是否延迟加载关联数据
+
+        Returns:
+            模型实例列表或原始row_ids列表
+        """
+
+        if row_ids is not None and row_ids != []:
+            if isinstance(row_ids[0], BaseSystemModel):
+                return row_ids
+
+            filter_model = Group(
+                logic="AND",
+                children=[
+                    Condition(
+                        field="row_id",
+                        operator=Operator.IN,
+                        value=row_ids
+                    )
+                ]
+            )
+            return cls.list(filter_model, include_system_fields=include_system_fields, lazy_load=lazy_load)
+        return row_ids
+
+    @classmethod
+    def create(cls, model: T) -> str:
+        """创建记录
+
+        Args:
+            model: 模型实例
+
+        Returns:
+            新创建的记录ID
+        """
+        model = cls._prepare_for_save(model)
+
+        fields = model_to_fields(model)
+        row_id = WorksheetRow.create(cls.WORKSHEET_ID, fields)
+        return row_id
+
+    @classmethod
+    def update(cls, model: T) -> str:
+        """更新记录
+
+        Args:
+            model: 模型实例（必须包含row_id）
+
+        Returns:
+            更新的记录ID
+
+        Raises:
+            ValueError: 当row_id为None时
+        """
+        if model.row_id is None:
+            raise ValueError(f"{cls.__name__} row_id is None, cannot update.")
+
+        model = cls._prepare_for_save(model)
+
+        fields = model_to_fields(model)
+        row_id = WorksheetRow.update(cls.WORKSHEET_ID, model.row_id, fields)
+        return row_id
+
+    @classmethod
+    def update_or_create(cls, model: T) -> str:
+        """更新或创建记录
+
+        Args:
+            model: 模型实例
+
+        Returns:
+            记录ID
+        """
+        model = cls._prepare_for_save(model)
+
+        fields = model_to_fields(model)
+
+        if model.row_id is None:
+            row_id = WorksheetRow.create(cls.WORKSHEET_ID, fields)
+        else:
+            row_id = WorksheetRow.update(cls.WORKSHEET_ID, model.row_id, fields)
+
+        return row_id
+
+    @classmethod
+    def batch_update_or_create(cls, model_list: List[Union[T, str]]) -> Union[List[str], None]:
+        """批量更新
+
+        Args:
+            model_list: 模型实例或ID字符串的列表
+
+        Returns:
+            更新后的记录ID列表
+
+        Raises:
+            TypeError: 当列表中包含不支持的类型时
+        """
+        if model_list is None:
+            return model_list
+
+        row_ids = []
+        for model in model_list:
+            if isinstance(model, str):
+                row_ids.append(model)  # just link
+            elif isinstance(model, cls.MODEL_CLASS):
+                row_id = cls.update_or_create(model)
+                row_ids.append(row_id)
+            else:
+                raise TypeError(
+                    f"Unsupported {cls.__name__} data type: {type(model).__name__}. "
+                    f"Expected str or {cls.MODEL_CLASS.__name__}"
+                )
+
+        return row_ids
+
+    @classmethod
+    def _load_relations(cls, model: T, include_system_fields: bool = True) -> T:
+        """加载关联数据（子类可覆盖）
+
+        Args:
+            model: 模型实例
+            include_system_fields: 是否包含系统字段
+
+        Returns:
+            加载了关联数据的模型实例
+        """
+        return model
+
+    @classmethod
+    def _prepare_for_save(cls, model: T) -> T:
+        """保存前准备（子类可覆盖）
+
+        Args:
+            model: 模型实例
+
+        Returns:
+            准备好的模型实例
+        """
+        return model
+
+
 class Enrichment(BaseWorksheetEntity[EnrichmentModel]):
     """Enrichment 实体类"""
     WORKSHEET_ID = "enrichment"

PLUGINS/SIRP/sirpbase.py

@@ -1,326 +0,0 @@
-from abc import ABC
-from typing import TypeVar, Generic, Type, List, Dict, Union, Any
-
-from pydantic import BaseModel
-
-from PLUGINS.SIRP.nocolyapi import WorksheetRow
-from PLUGINS.SIRP.nocolymodel import Condition, Group, Operator
-from PLUGINS.SIRP.sirpbasemodel import BaseSystemModel
-
-
-def model_to_fields(model_instance: BaseModel) -> List[Dict[str, Any]]:
-    fields = []
-    model_data = model_instance.model_dump(mode='json', exclude_unset=True)
-    for key, value in model_data.items():
-        field_info = model_instance.model_fields.get(key)
-        field_item = {
-            'id': key,
-            'value': value
-        }
-        if field_info and field_info.json_schema_extra:
-            field_item.update(field_info.json_schema_extra)
-        fields.append(field_item)
-    return fields
-
-
-# 定义泛型类型
-T = TypeVar('T', bound=BaseSystemModel)
-
-
-class BaseWorksheetEntity(ABC, Generic[T]):
-    """通用工作表实体基类 - 支持泛型和关联加载"""
-
-    WORKSHEET_ID: str
-    MODEL_CLASS: Type[T]
-
-    @classmethod
-    def get(
-            cls,
-            row_id: str,
-            include_system_fields: bool = True,
-            lazy_load: bool = False
-    ) -> T:
-        """获取单条记录
-
-        Args:
-            row_id: 记录ID
-            include_system_fields: 是否包含系统字段
-            lazy_load: 是否延迟加载关联数据（True时不加载关联）
-
-        Returns:
-            模型实例
-        """
-        result = WorksheetRow.get(
-            cls.WORKSHEET_ID,
-            row_id,
-            include_system_fields=include_system_fields
-        )
-        model = cls.MODEL_CLASS(**result)
-
-        if not lazy_load:
-            model = cls._load_relations(model, include_system_fields)
-
-        return model
-
-    @classmethod
-    def list(
-            cls,
-            filter_model: Group,
-            include_system_fields: bool = True,
-            lazy_load: bool = False
-    ) -> List[T]:
-        """按过滤条件列表查询
-
-        Args:
-            filter_model: 过滤条件Group对象
-            include_system_fields: 是否包含系统字段
-            lazy_load: 是否延迟加载关联数据（True时不加载关联）
-
-        Returns:
-            模型实例列表
-        """
-        if filter_model.children:
-            filter_dict = filter_model.model_dump()
-        else:
-            filter_dict = {}
-        result = WorksheetRow.list(
-            cls.WORKSHEET_ID,
-            filter_dict,
-            include_system_fields=include_system_fields
-        )
-
-        model_list = []
-        for item in result:
-            model_obj = cls.MODEL_CLASS(**item)
-            if not lazy_load:
-                model_obj = cls._load_relations(model_obj, include_system_fields)
-            model_list.append(model_obj)
-
-        return model_list
-
-    @classmethod
-    def update_by_filter(cls,
-                         filter_model: Group,
-                         model: T,
-                         include_system_fields: bool = True) -> dict:
-        filter_dict = filter_model.model_dump()
-        result = WorksheetRow.list(
-            cls.WORKSHEET_ID,
-            filter_dict,
-            fields=["row_id"],
-            include_system_fields=include_system_fields
-        )
-        row_ids = []
-        for item in result:
-            row_ids.append(item["row_id"])
-
-        model = cls._prepare_for_save(model)
-
-        fields = model_to_fields(model)
-        result = WorksheetRow.batch_update(cls.WORKSHEET_ID, row_ids, fields)
-        return result
-
-    @classmethod
-    def list_by_row_ids(
-            cls,
-            row_ids: List[Any],
-            include_system_fields: bool = True,
-            lazy_load: bool = False
-    ) -> Union[List[T], List[str], None]:
-        """按ID列表查询
-
-        Args:
-            row_ids: 记录ID列表
-            include_system_fields: 是否包含系统字段
-            lazy_load: 是否延迟加载关联数据
-
-        Returns:
-            模型实例列表或原始row_ids列表
-        """
-
-        if row_ids is not None and row_ids != []:
-            if isinstance(row_ids[0], BaseSystemModel):
-                return row_ids
-
-            filter_model = Group(
-                logic="AND",
-                children=[
-                    Condition(
-                        field="row_id",
-                        operator=Operator.IN,
-                        value=row_ids
-                    )
-                ]
-            )
-            return cls.list(filter_model, include_system_fields=include_system_fields, lazy_load=lazy_load)
-        return row_ids
-
-    @classmethod
-    def create(cls, model: T) -> str:
-        """创建记录
-
-        Args:
-            model: 模型实例
-
-        Returns:
-            新创建的记录ID
-        """
-        model = cls._prepare_for_save(model)
-
-        fields = model_to_fields(model)
-        row_id = WorksheetRow.create(cls.WORKSHEET_ID, fields)
-        return row_id
-
-    @classmethod
-    def update(cls, model: T) -> str:
-        """更新记录
-
-        Args:
-            model: 模型实例（必须包含row_id）
-
-        Returns:
-            更新的记录ID
-
-        Raises:
-            ValueError: 当row_id为None时
-        """
-        if model.row_id is None:
-            raise ValueError(f"{cls.__name__} row_id is None, cannot update.")
-
-        model = cls._prepare_for_save(model)
-
-        fields = model_to_fields(model)
-        row_id = WorksheetRow.update(cls.WORKSHEET_ID, model.row_id, fields)
-        return row_id
-
-    @classmethod
-    def update_or_create(cls, model: T) -> str:
-        """更新或创建记录
-
-        Args:
-            model: 模型实例
-
-        Returns:
-            记录ID
-        """
-        model = cls._prepare_for_save(model)
-
-        fields = model_to_fields(model)
-
-        if model.row_id is None:
-            row_id = WorksheetRow.create(cls.WORKSHEET_ID, fields)
-        else:
-            row_id = WorksheetRow.update(cls.WORKSHEET_ID, model.row_id, fields)
-
-        return row_id
-
-    @classmethod
-    def batch_update_or_create(cls, model_list: List[Union[T, str]]) -> Union[List[str], None]:
-        """批量更新
-
-        Args:
-            model_list: 模型实例或ID字符串的列表
-
-        Returns:
-            更新后的记录ID列表
-
-        Raises:
-            TypeError: 当列表中包含不支持的类型时
-        """
-        if model_list is None:
-            return model_list
-
-        row_ids = []
-        for model in model_list:
-            if isinstance(model, str):
-                row_ids.append(model)  # just link
-            elif isinstance(model, cls.MODEL_CLASS):
-                row_id = cls.update_or_create(model)
-                row_ids.append(row_id)
-            else:
-                raise TypeError(
-                    f"Unsupported {cls.__name__} data type: {type(model).__name__}. "
-                    f"Expected str or {cls.MODEL_CLASS.__name__}"
-                )
-
-        return row_ids
-
-    @classmethod
-    def _load_relations(cls, model: T, include_system_fields: bool = True) -> T:
-        """加载关联数据（子类可覆盖）
-
-        Args:
-            model: 模型实例
-            include_system_fields: 是否包含系统字段
-
-        Returns:
-            加载了关联数据的模型实例
-        """
-        return model
-
-    @classmethod
-    def _prepare_for_save(cls, model: T) -> T:
-        """保存前准备（子类可覆盖）
-
-        Args:
-            model: 模型实例
-
-        Returns:
-            准备好的模型实例
-        """
-        return model
-
-
-class BaseSimpleEntity(ABC):
-    """简化的工作表实体基类（不使用模型）"""
-
-    WORKSHEET_ID: str
-
-    @classmethod
-    def list(cls, filter_dict: dict) -> List[Dict]:
-        """列表查询
-
-        Args:
-            filter_dict: 过滤条件字典
-
-        Returns:
-            字典列表
-        """
-        return WorksheetRow.list(cls.WORKSHEET_ID, filter_dict, include_system_fields=False)
-
-    @classmethod
-    def get(cls, row_id: str) -> Dict:
-        """获取单条记录
-
-        Args:
-            row_id: 记录ID
-
-        Returns:
-            字典
-        """
-        return WorksheetRow.get(cls.WORKSHEET_ID, row_id, include_system_fields=False)
-
-    @classmethod
-    def create(cls, fields: List[Dict]) -> str:
-        """创建记录
-
-        Args:
-            fields: 字段列表
-
-        Returns:
-            新创建的记录ID
-        """
-        return WorksheetRow.create(cls.WORKSHEET_ID, fields)
-
-    @classmethod
-    def update(cls, row_id: str, fields: List[Dict]) -> str:
-        """更新记录
-
-        Args:
-            row_id: 记录ID
-            fields: 字段列表
-
-        Returns:
-            更新的记录ID
-        """
-        return WorksheetRow.update(cls.WORKSHEET_ID, row_id, fields)

PLUGINS/SIRP/sirpcoremodel.py

@@ -418,8 +418,7 @@ class CaseModel(BaseSystemModel):
 
     # 创建记录填写字段
     title: Optional[str] = Field(default="", description="Case title (案例标题)")
-    severity: Optional[Severity] = Field(default=None,
-                                         description="Analyst-assessed severity (严重程度)")
+    severity: Optional[Severity] = Field(default=None, description="Analyst-assessed severity (严重程度)")
     impact: Optional[Impact] = Field(default=None, description="Analyst-assessed impact (影响)")
     priority: Optional[CasePriority] = Field(default=None, description="Response priority (响应优先级)")