Merge pull request #5097 from wumode/refector-check_method

app/utils/object.py

@@ -1,5 +1,7 @@
+import ast
 import dis
 import inspect
+import textwrap
 from types import FunctionType
 from typing import Any, Callable, get_type_hints
 
@@ -39,45 +41,42 @@ class ObjectUtils:
         return len(list(parameters.keys()))
 
     @staticmethod
-    def check_method(func: FunctionType) -> bool:
+    def check_method(func: Callable[..., Any]) -> bool:
         """
         检查函数是否已实现
         """
         try:
-            # 尝试通过源代码分析
-            source = inspect.getsource(func)
-            in_comment = False
-            for line in source.split('\n'):
-                line = line.strip()
-                # 跳过空行
-                if not line:
-                    continue
-                # 处理"""单行注释
-                if (line.startswith(('"""', "'''"))
-                        and line.endswith(('"""', "'''"))
-                        and len(line) > 3):
-                    continue
-                # 处理"""多行注释
-                if line.startswith(('"""', "'''")):
-                    in_comment = not in_comment
-                    continue
-                # 在注释中则跳过
-                if in_comment:
-                    continue
-                # 跳过#注释、pass语句、装饰器、函数定义行
-                if (line.startswith('#')
-                        or line == "pass"
-                        or line.startswith('@')
-                        or line.startswith('def ')):
-                    continue
-                # 发现有效代码行
+            src = inspect.getsource(func)
+            tree = ast.parse(textwrap.dedent(src))
+            node = tree.body[0]
+            if not isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)):
+                return True
+            body = node.body
+
+            for stmt in body:
+                # 跳过 pass
+                if isinstance(stmt, ast.Pass):
+                    continue
+                # 跳过 docstring 或 ...
+                if isinstance(stmt, ast.Expr):
+                    expr = stmt.value
+                    if isinstance(expr, ast.Constant):
+                        if isinstance(expr.value, str) or expr.value is Ellipsis:
+                            continue
+                # 检查 raise NotImplementedError
+                if isinstance(stmt, ast.Raise):
+                    exc = stmt.exc
+                    if isinstance(exc, ast.Call) and getattr(exc.func, "id", None) == "NotImplementedError":
+                        continue
+                    if isinstance(exc, ast.Name) and exc.id == "NotImplementedError":
+                        continue
+
                 return True
-            # 没有有效代码行
             return False
         except Exception as err:
             print(err)
             # 源代码分析失败时，进行字节码分析
-            code_obj = func.__code__
+            code_obj = func.__code__  # type: ignore[attr-defined]
             instructions = list(dis.get_instructions(code_obj))
             # 检查是否为仅返回None的简单结构
             if len(instructions) == 2:

tests/run.py

@@ -1,6 +1,8 @@
 import unittest
 
 from tests.test_metainfo import MetaInfoTest
+from tests.test_object import ObjectUtilsTest
+
 
 if __name__ == '__main__':
     suite = unittest.TestSuite()
@@ -8,6 +10,7 @@ if __name__ == '__main__':
     # 测试名称识别
     suite.addTest(MetaInfoTest('test_metainfo'))
     suite.addTest(MetaInfoTest('test_emby_format_ids'))
+    suite.addTest(ObjectUtilsTest('test_check_method'))
 
     # 运行测试
     runner = unittest.TextTestRunner()

tests/test_object.py

@@ -0,0 +1,41 @@
+from unittest import TestCase
+
+from app.utils.object import ObjectUtils
+
+
+class ObjectUtilsTest(TestCase):
+
+    def test_check_method(self):
+        def implemented_function():
+            return "Hello"
+
+        def pass_function():
+            pass
+
+        def docstring_function():
+            """This is a docstring."""
+
+        def ellipsis_function():
+            ...
+
+        def not_implemented_function():
+            raise NotImplementedError
+
+        def not_implemented_function_with_call():
+            raise NotImplementedError()
+
+        async def multiple_lines_async_def(_param1: str,
+                                           _param2: str):
+            pass
+
+        def empty_function():
+            return
+
+        self.assertTrue(ObjectUtils.check_method(implemented_function))
+        self.assertFalse(ObjectUtils.check_method(pass_function))
+        self.assertFalse(ObjectUtils.check_method(docstring_function))
+        self.assertFalse(ObjectUtils.check_method(ellipsis_function))
+        self.assertFalse(ObjectUtils.check_method(not_implemented_function))
+        self.assertFalse(ObjectUtils.check_method(not_implemented_function_with_call))
+        self.assertFalse(ObjectUtils.check_method(multiple_lines_async_def))
+        self.assertTrue(ObjectUtils.check_method(empty_function))