feat: +pylint class view

metagpt/utils/common.py

@@ -17,8 +17,8 @@ import inspect
 import os
 import platform
 import re
-import typing
-from typing import List, Tuple, Union
+from pathlib import Path
+from typing import Callable, List, Tuple, Union
 
 import aiofiles
 import loguru
@@ -332,7 +332,7 @@ def get_class_name(cls) -> str:
     return f"{cls.__module__}.{cls.__name__}"
 
 
-def any_to_str(val: str | typing.Callable) -> str:
+def any_to_str(val: str | Callable) -> str:
     """Return the class name or the class name of the object, or 'val' if it's a string type."""
     if isinstance(val, str):
         return val
@@ -443,3 +443,20 @@ async def aread(file_path: str) -> str:
     async with aiofiles.open(str(file_path), mode="r") as reader:
         content = await reader.read()
     return content
+
+
+async def read_file_block(filename: str | Path, lineno: int, end_lineno: int):
+    if not Path(filename).exists():
+        return ""
+    lines = []
+    async with aiofiles.open(str(filename), mode="r") as reader:
+        ix = 0
+        while ix < end_lineno:
+            ix += 1
+            line = await reader.readline()
+            if ix < lineno:
+                continue
+            if ix > end_lineno:
+                break
+            lines.append(line)
+    return "".join(lines)

metagpt/utils/di_graph_repository.py

@@ -10,11 +10,12 @@ from __future__ import annotations
 
 import json
 from pathlib import Path
+from typing import List
 
 import aiofiles
 import networkx
 
-from metagpt.utils.graph_repository import GraphRepository
+from metagpt.utils.graph_repository import SPO, GraphRepository
 
 
 class DiGraphRepository(GraphRepository):
@@ -31,6 +32,18 @@ class DiGraphRepository(GraphRepository):
     async def update(self, subject: str, predicate: str, object_: str):
         pass
 
+    async def select(self, subject: str = None, predicate: str = None, object_: str = None) -> List[SPO]:
+        result = []
+        for s, o, p in self._repo.edges(data="predicate"):
+            if subject and subject != s:
+                continue
+            if predicate and predicate != p:
+                continue
+            if object_ and object_ != o:
+                continue
+            result.append(SPO(subject=s, predicate=p, object_=o))
+        return result
+
     def json(self) -> str:
         m = networkx.node_link_data(self._repo)
         data = json.dumps(m)
@@ -53,10 +66,12 @@ class DiGraphRepository(GraphRepository):
 
     @staticmethod
     async def load_from(pathname: str | Path) -> GraphRepository:
-        name = Path(pathname).with_suffix("").name
-        root = Path(pathname).parent
+        pathname = Path(pathname)
+        name = pathname.with_suffix("").name
+        root = pathname.parent
         graph = DiGraphRepository(name=name, root=root)
-        await graph.load(pathname=pathname)
+        if pathname.exists():
+            await graph.load(pathname=pathname)
         return graph
 
     @property

metagpt/utils/graph_repository.py

@@ -6,18 +6,38 @@
 @File    : graph_repository.py
 @Desc    : Superclass for graph repository.
 """
+
 from abc import ABC, abstractmethod
-from enum import Enum
+from pathlib import Path
+from typing import List
+
+from pydantic import BaseModel
+
+from metagpt.repo_parser import ClassInfo, RepoFileInfo
+from metagpt.utils.common import concat_namespace
 
 
-class GraphKeyword(Enum):
+class GraphKeyword:
     IS = "is"
     CLASS = "class"
     FUNCTION = "function"
+    SOURCE_CODE = "source_code"
+    NULL = "<null>"
     GLOBAL_VARIABLE = "global_variable"
     CLASS_FUNCTION = "class_function"
     CLASS_PROPERTY = "class_property"
     HAS_CLASS = "has_class"
+    HAS_PAGE_INFO = "has_page_info"
+    HAS_CLASS_VIEW = "has_class_view"
+    HAS_SEQUENCE_VIEW = "has_sequence_view"
+    HAS_ARGS_DESC = "has_args_desc"
+    HAS_TYPE_DESC = "has_type_desc"
+
+
+class SPO(BaseModel):
+    subject: str
+    predicate: str
+    object_: str
 
 
 class GraphRepository(ABC):
@@ -37,6 +57,94 @@ class GraphRepository(ABC):
     async def update(self, subject: str, predicate: str, object_: str):
         pass
 
+    @abstractmethod
+    async def select(self, subject: str = None, predicate: str = None, object_: str = None) -> List[SPO]:
+        pass
+
     @property
     def name(self) -> str:
         return self._repo_name
+
+    @staticmethod
+    async def update_graph_db_with_file_info(graph_db: "GraphRepository", file_info: RepoFileInfo):
+        await graph_db.insert(subject=file_info.file, predicate=GraphKeyword.IS, object_=GraphKeyword.SOURCE_CODE)
+        file_types = {".py": "python", ".js": "javascript"}
+        file_type = file_types.get(Path(file_info.file).suffix, GraphKeyword.NULL)
+        await graph_db.insert(subject=file_info.file, predicate=GraphKeyword.IS, object_=file_type)
+        for c in file_info.classes:
+            class_name = c.get("name", "")
+            await graph_db.insert(
+                subject=file_info.file,
+                predicate=GraphKeyword.HAS_CLASS,
+                object_=concat_namespace(file_info.file, class_name),
+            )
+            await graph_db.insert(
+                subject=concat_namespace(file_info.file, class_name),
+                predicate=GraphKeyword.IS,
+                object_=GraphKeyword.CLASS,
+            )
+            methods = c.get("methods", [])
+            for fn in methods:
+                await graph_db.insert(
+                    subject=concat_namespace(file_info.file, class_name, fn),
+                    predicate=GraphKeyword.IS,
+                    object_=GraphKeyword.CLASS_FUNCTION,
+                )
+        for f in file_info.functions:
+            await graph_db.insert(
+                subject=concat_namespace(file_info.file, f), predicate=GraphKeyword.IS, object_=GraphKeyword.FUNCTION
+            )
+        for g in file_info.globals:
+            await graph_db.insert(
+                subject=concat_namespace(file_info.file, g),
+                predicate=GraphKeyword.IS,
+                object_=GraphKeyword.GLOBAL_VARIABLE,
+            )
+        for code_block in file_info.page_info:
+            if code_block.tokens:
+                await graph_db.insert(
+                    subject=concat_namespace(file_info.file, *code_block.tokens),
+                    predicate=GraphKeyword.HAS_PAGE_INFO,
+                    object_=code_block.json(ensure_ascii=False),
+                )
+            for k, v in code_block.properties.items():
+                await graph_db.insert(
+                    subject=concat_namespace(file_info.file, k, v),
+                    predicate=GraphKeyword.HAS_PAGE_INFO,
+                    object_=code_block.json(ensure_ascii=False),
+                )
+
+    @staticmethod
+    async def update_graph_db_with_class_views(graph_db: "GraphRepository", class_views: List[ClassInfo]):
+        for c in class_views:
+            filename, class_name = c.package.split(":", 1)
+            await graph_db.insert(subject=filename, predicate=GraphKeyword.IS, object_=GraphKeyword.SOURCE_CODE)
+            file_types = {".py": "python", ".js": "javascript"}
+            file_type = file_types.get(Path(filename).suffix, GraphKeyword.NULL)
+            await graph_db.insert(subject=filename, predicate=GraphKeyword.IS, object_=file_type)
+            await graph_db.insert(subject=filename, predicate=GraphKeyword.HAS_CLASS, object_=class_name)
+            await graph_db.insert(
+                subject=c.package,
+                predicate=GraphKeyword.IS,
+                object_=GraphKeyword.CLASS,
+            )
+            for vn, vt in c.attributes.items():
+                await graph_db.insert(
+                    subject=concat_namespace(c.package, vn),
+                    predicate=GraphKeyword.IS,
+                    object_=GraphKeyword.CLASS_PROPERTY,
+                )
+                await graph_db.insert(
+                    subject=concat_namespace(c.package, vn), predicate=GraphKeyword.HAS_TYPE_DESC, object_=vt
+                )
+            for fn, desc in c.methods.items():
+                await graph_db.insert(
+                    subject=concat_namespace(c.package, fn),
+                    predicate=GraphKeyword.IS,
+                    object_=GraphKeyword.CLASS_FUNCTION,
+                )
+                await graph_db.insert(
+                    subject=concat_namespace(c.package, fn),
+                    predicate=GraphKeyword.HAS_ARGS_DESC,
+                    object_=desc,
+                )