add 过滤规则测试API

app/api/endpoints/system.py

@@ -9,12 +9,14 @@ from fastapi.responses import StreamingResponse
 from sqlalchemy.orm import Session
 
 from app import schemas
+from app.chain.search import SearchChain
 from app.core.config import settings
 from app.core.security import verify_token
 from app.db import get_db
 from app.db.systemconfig_oper import SystemConfigOper
 from app.helper.message import MessageHelper
 from app.helper.progress import ProgressHelper
+from app.schemas.types import SystemConfigKey
 from app.utils.http import RequestUtils
 from version import APP_VERSION
 
@@ -166,3 +168,33 @@ def latest_version(_: schemas.TokenPayload = Depends(verify_token)):
         if ver_json:
             return schemas.Response(success=True, data=ver_json)
     return schemas.Response(success=False)
+
+
+@router.get("/ruletest", summary="过滤规则测试", response_model=schemas.Response)
+def ruletest(title: str,
+             subtitle: str = None,
+             ruletype: str = None,
+             db: Session = Depends(get_db),
+             _: schemas.TokenPayload = Depends(verify_token)):
+    """
+    过滤规则测试，规则类型 1-订阅，2-洗版
+    """
+    torrent = schemas.TorrentInfo(
+        title=title,
+        description=subtitle,
+    )
+    if ruletype == "2":
+        rule_string = SystemConfigOper(db).get(SystemConfigKey.FilterRules2)
+    else:
+        rule_string = SystemConfigOper(db).get(SystemConfigKey.FilterRules)
+    if not rule_string:
+        return schemas.Response(success=False, message="过滤规则未设置！")
+
+    # 过滤
+    result = SearchChain(db).filter_torrents(rule_string=rule_string,
+                                             torrent_list=[torrent])
+    if not result:
+        return schemas.Response(success=False, message="不符合过滤规则！")
+    return schemas.Response(success=True, data={
+        "priority": 100 - result[0].pri_order + 1
+    })