1024 lines
35 KiB
Python
1024 lines
35 KiB
Python
#!/usr/bin/env python
|
|
# -*- coding: utf-8 -*-
|
|
"""
|
|
Build a symbols repository from source code.
|
|
|
|
This script is designed to create a symbols repository from the provided source code.
|
|
|
|
@Time : 2023/11/17 17:58
|
|
@Author : alexanderwu
|
|
@File : repo_parser.py
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import ast
|
|
import json
|
|
import re
|
|
import subprocess
|
|
from pathlib import Path
|
|
from typing import Dict, List, Optional
|
|
|
|
import pandas as pd
|
|
from pydantic import BaseModel, Field, field_validator
|
|
|
|
from metagpt.const import AGGREGATION, COMPOSITION, GENERALIZATION
|
|
from metagpt.logs import logger
|
|
from metagpt.utils.common import any_to_str, aread, remove_white_spaces
|
|
from metagpt.utils.exceptions import handle_exception
|
|
|
|
|
|
class RepoFileInfo(BaseModel):
|
|
"""
|
|
Repository data element that represents information about a file.
|
|
|
|
Attributes:
|
|
file (str): The name or path of the file.
|
|
classes (List): A list of class names present in the file.
|
|
functions (List): A list of function names present in the file.
|
|
globals (List): A list of global variable names present in the file.
|
|
page_info (List): A list of page-related information associated with the file.
|
|
"""
|
|
|
|
file: str
|
|
classes: List = Field(default_factory=list)
|
|
functions: List = Field(default_factory=list)
|
|
globals: List = Field(default_factory=list)
|
|
page_info: List = Field(default_factory=list)
|
|
|
|
|
|
class CodeBlockInfo(BaseModel):
|
|
"""
|
|
Repository data element representing information about a code block.
|
|
|
|
Attributes:
|
|
lineno (int): The starting line number of the code block.
|
|
end_lineno (int): The ending line number of the code block.
|
|
type_name (str): The type or category of the code block.
|
|
tokens (List): A list of tokens present in the code block.
|
|
properties (Dict): A dictionary containing additional properties associated with the code block.
|
|
"""
|
|
|
|
lineno: int
|
|
end_lineno: int
|
|
type_name: str
|
|
tokens: List = Field(default_factory=list)
|
|
properties: Dict = Field(default_factory=dict)
|
|
|
|
|
|
class DotClassAttribute(BaseModel):
|
|
"""
|
|
Repository data element representing a class attribute in dot format.
|
|
|
|
Attributes:
|
|
name (str): The name of the class attribute.
|
|
type_ (str): The type of the class attribute.
|
|
default_ (str): The default value of the class attribute.
|
|
description (str): A description of the class attribute.
|
|
compositions (List[str]): A list of compositions associated with the class attribute.
|
|
"""
|
|
|
|
name: str = ""
|
|
type_: str = ""
|
|
default_: str = ""
|
|
description: str
|
|
compositions: List[str] = Field(default_factory=list)
|
|
|
|
@classmethod
|
|
def parse(cls, v: str) -> "DotClassAttribute":
|
|
"""
|
|
Parses dot format text and returns a DotClassAttribute object.
|
|
|
|
Args:
|
|
v (str): Dot format text to be parsed.
|
|
|
|
Returns:
|
|
DotClassAttribute: An instance of the DotClassAttribute class representing the parsed data.
|
|
"""
|
|
val = ""
|
|
meet_colon = False
|
|
meet_equals = False
|
|
for c in v:
|
|
if c == ":":
|
|
meet_colon = True
|
|
elif c == "=":
|
|
meet_equals = True
|
|
if not meet_colon:
|
|
val += ":"
|
|
meet_colon = True
|
|
val += c
|
|
if not meet_colon:
|
|
val += ":"
|
|
if not meet_equals:
|
|
val += "="
|
|
|
|
cix = val.find(":")
|
|
eix = val.rfind("=")
|
|
name = val[0:cix].strip()
|
|
type_ = val[cix + 1 : eix]
|
|
default_ = val[eix + 1 :].strip()
|
|
|
|
type_ = remove_white_spaces(type_) # remove white space
|
|
if type_ == "NoneType":
|
|
type_ = ""
|
|
if "Literal[" in type_:
|
|
pre_l, literal, post_l = cls._split_literal(type_)
|
|
composition_val = pre_l + "Literal" + post_l # replace Literal[...] with Literal
|
|
type_ = pre_l + literal + post_l
|
|
else:
|
|
type_ = re.sub(r"['\"]+", "", type_) # remove '"
|
|
composition_val = type_
|
|
|
|
if default_ == "None":
|
|
default_ = ""
|
|
compositions = cls.parse_compositions(composition_val)
|
|
return cls(name=name, type_=type_, default_=default_, description=v, compositions=compositions)
|
|
|
|
@staticmethod
|
|
def parse_compositions(types_part) -> List[str]:
|
|
"""
|
|
Parses the type definition code block of source code and returns a list of compositions.
|
|
|
|
Args:
|
|
types_part: The type definition code block to be parsed.
|
|
|
|
Returns:
|
|
List[str]: A list of compositions extracted from the type definition code block.
|
|
"""
|
|
if not types_part:
|
|
return []
|
|
modified_string = re.sub(r"[\[\],\(\)]", "|", types_part)
|
|
types = modified_string.split("|")
|
|
filters = {
|
|
"str",
|
|
"frozenset",
|
|
"set",
|
|
"int",
|
|
"float",
|
|
"complex",
|
|
"bool",
|
|
"dict",
|
|
"list",
|
|
"Union",
|
|
"Dict",
|
|
"Set",
|
|
"Tuple",
|
|
"NoneType",
|
|
"None",
|
|
"Any",
|
|
"Optional",
|
|
"Iterator",
|
|
"Literal",
|
|
"List",
|
|
}
|
|
result = set()
|
|
for t in types:
|
|
t = re.sub(r"['\"]+", "", t.strip())
|
|
if t and t not in filters:
|
|
result.add(t)
|
|
return list(result)
|
|
|
|
@staticmethod
|
|
def _split_literal(v):
|
|
"""
|
|
Parses the literal definition code block and returns three parts: pre-part, literal-part, and post-part.
|
|
|
|
Args:
|
|
v: The literal definition code block to be parsed.
|
|
|
|
Returns:
|
|
Tuple[str, str, str]: A tuple containing the pre-part, literal-part, and post-part of the code block.
|
|
"""
|
|
tag = "Literal["
|
|
bix = v.find(tag)
|
|
eix = len(v) - 1
|
|
counter = 1
|
|
for i in range(bix + len(tag), len(v) - 1):
|
|
c = v[i]
|
|
if c == "[":
|
|
counter += 1
|
|
continue
|
|
if c == "]":
|
|
counter -= 1
|
|
if counter > 0:
|
|
continue
|
|
eix = i
|
|
break
|
|
pre_l = v[0:bix]
|
|
post_l = v[eix + 1 :]
|
|
pre_l = re.sub(r"['\"]", "", pre_l) # remove '"
|
|
pos_l = re.sub(r"['\"]", "", post_l) # remove '"
|
|
|
|
return pre_l, v[bix : eix + 1], pos_l
|
|
|
|
@field_validator("compositions", mode="after")
|
|
@classmethod
|
|
def sort(cls, lst: List) -> List:
|
|
"""
|
|
Auto-sorts a list attribute after making changes.
|
|
|
|
Args:
|
|
lst (List): The list attribute to be sorted.
|
|
|
|
Returns:
|
|
List: The sorted list.
|
|
"""
|
|
lst.sort()
|
|
return lst
|
|
|
|
|
|
class DotClassInfo(BaseModel):
|
|
"""
|
|
Repository data element representing information about a class in dot format.
|
|
|
|
Attributes:
|
|
name (str): The name of the class.
|
|
package (Optional[str]): The package to which the class belongs (optional).
|
|
attributes (Dict[str, DotClassAttribute]): A dictionary of attributes associated with the class.
|
|
methods (Dict[str, DotClassMethod]): A dictionary of methods associated with the class.
|
|
compositions (List[str]): A list of compositions associated with the class.
|
|
aggregations (List[str]): A list of aggregations associated with the class.
|
|
"""
|
|
|
|
name: str
|
|
package: Optional[str] = None
|
|
attributes: Dict[str, DotClassAttribute] = Field(default_factory=dict)
|
|
methods: Dict[str, DotClassMethod] = Field(default_factory=dict)
|
|
compositions: List[str] = Field(default_factory=list)
|
|
aggregations: List[str] = Field(default_factory=list)
|
|
|
|
@field_validator("compositions", "aggregations", mode="after")
|
|
@classmethod
|
|
def sort(cls, lst: List) -> List:
|
|
"""
|
|
Auto-sorts a list attribute after making changes.
|
|
|
|
Args:
|
|
lst (List): The list attribute to be sorted.
|
|
|
|
Returns:
|
|
List: The sorted list.
|
|
"""
|
|
lst.sort()
|
|
return lst
|
|
|
|
|
|
class DotClassRelationship(BaseModel):
|
|
"""
|
|
Repository data element representing a relationship between two classes in dot format.
|
|
|
|
Attributes:
|
|
src (str): The source class of the relationship.
|
|
dest (str): The destination class of the relationship.
|
|
relationship (str): The type or nature of the relationship.
|
|
label (Optional[str]): An optional label associated with the relationship.
|
|
"""
|
|
|
|
src: str = ""
|
|
dest: str = ""
|
|
relationship: str = ""
|
|
label: Optional[str] = None
|
|
|
|
|
|
class DotReturn(BaseModel):
|
|
"""
|
|
Repository data element representing a function or method return type in dot format.
|
|
|
|
Attributes:
|
|
type_ (str): The type of the return.
|
|
description (str): A description of the return type.
|
|
compositions (List[str]): A list of compositions associated with the return type.
|
|
"""
|
|
|
|
type_: str = ""
|
|
description: str
|
|
compositions: List[str] = Field(default_factory=list)
|
|
|
|
@classmethod
|
|
def parse(cls, v: str) -> "DotReturn" | None:
|
|
"""
|
|
Parses the return type part of dot format text and returns a DotReturn object.
|
|
|
|
Args:
|
|
v (str): The dot format text containing the return type part to be parsed.
|
|
|
|
Returns:
|
|
DotReturn | None: An instance of the DotReturn class representing the parsed return type,
|
|
or None if parsing fails.
|
|
"""
|
|
if not v:
|
|
return DotReturn(description=v)
|
|
type_ = remove_white_spaces(v)
|
|
compositions = DotClassAttribute.parse_compositions(type_)
|
|
return cls(type_=type_, description=v, compositions=compositions)
|
|
|
|
@field_validator("compositions", mode="after")
|
|
@classmethod
|
|
def sort(cls, lst: List) -> List:
|
|
"""
|
|
Auto-sorts a list attribute after making changes.
|
|
|
|
Args:
|
|
lst (List): The list attribute to be sorted.
|
|
|
|
Returns:
|
|
List: The sorted list.
|
|
"""
|
|
lst.sort()
|
|
return lst
|
|
|
|
|
|
class DotClassMethod(BaseModel):
|
|
name: str
|
|
args: List[DotClassAttribute] = Field(default_factory=list)
|
|
return_args: Optional[DotReturn] = None
|
|
description: str
|
|
aggregations: List[str] = Field(default_factory=list)
|
|
|
|
@classmethod
|
|
def parse(cls, v: str) -> "DotClassMethod":
|
|
"""
|
|
Parses a dot format method text and returns a DotClassMethod object.
|
|
|
|
Args:
|
|
v (str): The dot format text containing method information to be parsed.
|
|
|
|
Returns:
|
|
DotClassMethod: An instance of the DotClassMethod class representing the parsed method.
|
|
"""
|
|
bix = v.find("(")
|
|
eix = v.rfind(")")
|
|
rix = v.rfind(":")
|
|
if rix < 0 or rix < eix:
|
|
rix = eix
|
|
name_part = v[0:bix].strip()
|
|
args_part = v[bix + 1 : eix].strip()
|
|
return_args_part = v[rix + 1 :].strip()
|
|
|
|
name = cls._parse_name(name_part)
|
|
args = cls._parse_args(args_part)
|
|
return_args = DotReturn.parse(return_args_part)
|
|
aggregations = set()
|
|
for i in args:
|
|
aggregations.update(set(i.compositions))
|
|
aggregations.update(set(return_args.compositions))
|
|
|
|
return cls(name=name, args=args, description=v, return_args=return_args, aggregations=list(aggregations))
|
|
|
|
@staticmethod
|
|
def _parse_name(v: str) -> str:
|
|
"""
|
|
Parses the dot format method name part and returns the method name.
|
|
|
|
Args:
|
|
v (str): The dot format text containing the method name part to be parsed.
|
|
|
|
Returns:
|
|
str: The parsed method name.
|
|
"""
|
|
tags = [">", "</"]
|
|
if tags[0] in v:
|
|
bix = v.find(tags[0]) + len(tags[0])
|
|
eix = v.rfind(tags[1])
|
|
return v[bix:eix].strip()
|
|
return v.strip()
|
|
|
|
@staticmethod
|
|
def _parse_args(v: str) -> List[DotClassAttribute]:
|
|
"""
|
|
Parses the dot format method arguments part and returns the parsed arguments.
|
|
|
|
Args:
|
|
v (str): The dot format text containing the arguments part to be parsed.
|
|
|
|
Returns:
|
|
str: The parsed method arguments.
|
|
"""
|
|
if not v:
|
|
return []
|
|
parts = []
|
|
bix = 0
|
|
counter = 0
|
|
for i in range(0, len(v)):
|
|
c = v[i]
|
|
if c == "[":
|
|
counter += 1
|
|
continue
|
|
elif c == "]":
|
|
counter -= 1
|
|
continue
|
|
elif c == "," and counter == 0:
|
|
parts.append(v[bix:i].strip())
|
|
bix = i + 1
|
|
parts.append(v[bix:].strip())
|
|
|
|
attrs = []
|
|
for p in parts:
|
|
if p:
|
|
attr = DotClassAttribute.parse(p)
|
|
attrs.append(attr)
|
|
return attrs
|
|
|
|
|
|
class RepoParser(BaseModel):
|
|
"""
|
|
Tool to build a symbols repository from a project directory.
|
|
|
|
Attributes:
|
|
base_directory (Path): The base directory of the project.
|
|
"""
|
|
|
|
base_directory: Path = Field(default=None)
|
|
|
|
@classmethod
|
|
@handle_exception(exception_type=Exception, default_return=[])
|
|
def _parse_file(cls, file_path: Path) -> list:
|
|
"""
|
|
Parses a Python file in the repository.
|
|
|
|
Args:
|
|
file_path (Path): The path to the Python file to be parsed.
|
|
|
|
Returns:
|
|
list: A list containing the parsed symbols from the file.
|
|
"""
|
|
return ast.parse(file_path.read_text()).body
|
|
|
|
def extract_class_and_function_info(self, tree, file_path) -> RepoFileInfo:
|
|
"""
|
|
Extracts class, function, and global variable information from the Abstract Syntax Tree (AST).
|
|
|
|
Args:
|
|
tree: The Abstract Syntax Tree (AST) of the Python file.
|
|
file_path: The path to the Python file.
|
|
|
|
Returns:
|
|
RepoFileInfo: A RepoFileInfo object containing the extracted information.
|
|
"""
|
|
file_info = RepoFileInfo(file=str(file_path.relative_to(self.base_directory)))
|
|
for node in tree:
|
|
info = RepoParser.node_to_str(node)
|
|
if info:
|
|
file_info.page_info.append(info)
|
|
if isinstance(node, ast.ClassDef):
|
|
class_methods = [m.name for m in node.body if is_func(m)]
|
|
file_info.classes.append({"name": node.name, "methods": class_methods})
|
|
elif is_func(node):
|
|
file_info.functions.append(node.name)
|
|
elif isinstance(node, (ast.Assign, ast.AnnAssign)):
|
|
for target in node.targets if isinstance(node, ast.Assign) else [node.target]:
|
|
if isinstance(target, ast.Name):
|
|
file_info.globals.append(target.id)
|
|
return file_info
|
|
|
|
def generate_symbols(self) -> List[RepoFileInfo]:
|
|
"""
|
|
Builds a symbol repository from '.py' and '.js' files in the project directory.
|
|
|
|
Returns:
|
|
List[RepoFileInfo]: A list of RepoFileInfo objects containing the extracted information.
|
|
"""
|
|
files_classes = []
|
|
directory = self.base_directory
|
|
|
|
matching_files = []
|
|
extensions = ["*.py"]
|
|
for ext in extensions:
|
|
matching_files += directory.rglob(ext)
|
|
for path in matching_files:
|
|
tree = self._parse_file(path)
|
|
file_info = self.extract_class_and_function_info(tree, path)
|
|
files_classes.append(file_info)
|
|
|
|
return files_classes
|
|
|
|
def generate_json_structure(self, output_path: Path):
|
|
"""
|
|
Generates a JSON file documenting the repository structure.
|
|
|
|
Args:
|
|
output_path (Path): The path to the JSON file to be generated.
|
|
"""
|
|
files_classes = [i.model_dump() for i in self.generate_symbols()]
|
|
output_path.write_text(json.dumps(files_classes, indent=4))
|
|
|
|
def generate_dataframe_structure(self, output_path: Path):
|
|
"""
|
|
Generates a DataFrame documenting the repository structure and saves it as a CSV file.
|
|
|
|
Args:
|
|
output_path (Path): The path to the CSV file to be generated.
|
|
"""
|
|
files_classes = [i.model_dump() for i in self.generate_symbols()]
|
|
df = pd.DataFrame(files_classes)
|
|
df.to_csv(output_path, index=False)
|
|
|
|
def generate_structure(self, output_path: str | Path = None, mode="json") -> Path:
|
|
"""
|
|
Generates the structure of the repository in a specified format.
|
|
|
|
Args:
|
|
output_path (str | Path): The path to the output file or directory. Default is None.
|
|
mode (str): The output format mode. Options: "json" (default), "csv", etc.
|
|
|
|
Returns:
|
|
Path: The path to the generated output file or directory.
|
|
"""
|
|
output_file = self.base_directory / f"{self.base_directory.name}-structure.{mode}"
|
|
output_path = Path(output_path) if output_path else output_file
|
|
|
|
if mode == "json":
|
|
self.generate_json_structure(output_path)
|
|
elif mode == "csv":
|
|
self.generate_dataframe_structure(output_path)
|
|
return output_path
|
|
|
|
@staticmethod
|
|
def node_to_str(node) -> CodeBlockInfo | None:
|
|
"""
|
|
Parses and converts an Abstract Syntax Tree (AST) node to a CodeBlockInfo object.
|
|
|
|
Args:
|
|
node: The AST node to be converted.
|
|
|
|
Returns:
|
|
CodeBlockInfo | None: A CodeBlockInfo object representing the parsed AST node,
|
|
or None if the conversion fails.
|
|
"""
|
|
if isinstance(node, ast.Try):
|
|
return None
|
|
if any_to_str(node) == any_to_str(ast.Expr):
|
|
return CodeBlockInfo(
|
|
lineno=node.lineno,
|
|
end_lineno=node.end_lineno,
|
|
type_name=any_to_str(node),
|
|
tokens=RepoParser._parse_expr(node),
|
|
)
|
|
mappings = {
|
|
any_to_str(ast.Import): lambda x: [RepoParser._parse_name(n) for n in x.names],
|
|
any_to_str(ast.Assign): RepoParser._parse_assign,
|
|
any_to_str(ast.ClassDef): lambda x: x.name,
|
|
any_to_str(ast.FunctionDef): lambda x: x.name,
|
|
any_to_str(ast.ImportFrom): lambda x: {
|
|
"module": x.module,
|
|
"names": [RepoParser._parse_name(n) for n in x.names],
|
|
},
|
|
any_to_str(ast.If): RepoParser._parse_if,
|
|
any_to_str(ast.AsyncFunctionDef): lambda x: x.name,
|
|
any_to_str(ast.AnnAssign): lambda x: RepoParser._parse_variable(x.target),
|
|
}
|
|
func = mappings.get(any_to_str(node))
|
|
if func:
|
|
code_block = CodeBlockInfo(lineno=node.lineno, end_lineno=node.end_lineno, type_name=any_to_str(node))
|
|
val = func(node)
|
|
if isinstance(val, dict):
|
|
code_block.properties = val
|
|
elif isinstance(val, list):
|
|
code_block.tokens = val
|
|
elif isinstance(val, str):
|
|
code_block.tokens = [val]
|
|
else:
|
|
raise NotImplementedError(f"Not implement:{val}")
|
|
return code_block
|
|
logger.warning(f"Unsupported code block:{node.lineno}, {node.end_lineno}, {any_to_str(node)}")
|
|
return None
|
|
|
|
@staticmethod
|
|
def _parse_expr(node) -> List:
|
|
"""
|
|
Parses an expression Abstract Syntax Tree (AST) node.
|
|
|
|
Args:
|
|
node: The AST node representing an expression.
|
|
|
|
Returns:
|
|
List: A list containing the parsed information from the expression node.
|
|
"""
|
|
funcs = {
|
|
any_to_str(ast.Constant): lambda x: [any_to_str(x.value), RepoParser._parse_variable(x.value)],
|
|
any_to_str(ast.Call): lambda x: [any_to_str(x.value), RepoParser._parse_variable(x.value.func)],
|
|
any_to_str(ast.Tuple): lambda x: [any_to_str(x.value), RepoParser._parse_variable(x.value)],
|
|
}
|
|
func = funcs.get(any_to_str(node.value))
|
|
if func:
|
|
return func(node)
|
|
raise NotImplementedError(f"Not implement: {node.value}")
|
|
|
|
@staticmethod
|
|
def _parse_name(n):
|
|
"""
|
|
Gets the 'name' value of an Abstract Syntax Tree (AST) node.
|
|
|
|
Args:
|
|
n: The AST node.
|
|
|
|
Returns:
|
|
The 'name' value of the AST node.
|
|
"""
|
|
if n.asname:
|
|
return f"{n.name} as {n.asname}"
|
|
return n.name
|
|
|
|
@staticmethod
|
|
def _parse_if(n):
|
|
"""
|
|
Parses an 'if' statement Abstract Syntax Tree (AST) node.
|
|
|
|
Args:
|
|
n: The AST node representing an 'if' statement.
|
|
|
|
Returns:
|
|
None or Parsed information from the 'if' statement node.
|
|
"""
|
|
tokens = []
|
|
try:
|
|
if isinstance(n.test, ast.BoolOp):
|
|
tokens = []
|
|
for v in n.test.values:
|
|
tokens.extend(RepoParser._parse_if_compare(v))
|
|
return tokens
|
|
if isinstance(n.test, ast.Compare):
|
|
v = RepoParser._parse_variable(n.test.left)
|
|
if v:
|
|
tokens.append(v)
|
|
if isinstance(n.test, ast.Name):
|
|
v = RepoParser._parse_variable(n.test)
|
|
tokens.append(v)
|
|
if hasattr(n.test, "comparators"):
|
|
for item in n.test.comparators:
|
|
v = RepoParser._parse_variable(item)
|
|
if v:
|
|
tokens.append(v)
|
|
return tokens
|
|
except Exception as e:
|
|
logger.warning(f"Unsupported if: {n}, err:{e}")
|
|
return tokens
|
|
|
|
@staticmethod
|
|
def _parse_if_compare(n):
|
|
"""
|
|
Parses an 'if' condition Abstract Syntax Tree (AST) node.
|
|
|
|
Args:
|
|
n: The AST node representing an 'if' condition.
|
|
|
|
Returns:
|
|
None or Parsed information from the 'if' condition node.
|
|
"""
|
|
if hasattr(n, "left"):
|
|
return RepoParser._parse_variable(n.left)
|
|
else:
|
|
return []
|
|
|
|
@staticmethod
|
|
def _parse_variable(node):
|
|
"""
|
|
Parses a variable Abstract Syntax Tree (AST) node.
|
|
|
|
Args:
|
|
node: The AST node representing a variable.
|
|
|
|
Returns:
|
|
None or Parsed information from the variable node.
|
|
"""
|
|
try:
|
|
funcs = {
|
|
any_to_str(ast.Constant): lambda x: x.value,
|
|
any_to_str(ast.Name): lambda x: x.id,
|
|
any_to_str(ast.Attribute): lambda x: f"{x.value.id}.{x.attr}"
|
|
if hasattr(x.value, "id")
|
|
else f"{x.attr}",
|
|
any_to_str(ast.Call): lambda x: RepoParser._parse_variable(x.func),
|
|
any_to_str(ast.Tuple): lambda x: [d.value for d in x.dims],
|
|
}
|
|
func = funcs.get(any_to_str(node))
|
|
if not func:
|
|
raise NotImplementedError(f"Not implement:{node}")
|
|
return func(node)
|
|
except Exception as e:
|
|
logger.warning(f"Unsupported variable:{node}, err:{e}")
|
|
|
|
@staticmethod
|
|
def _parse_assign(node):
|
|
"""
|
|
Parses an assignment Abstract Syntax Tree (AST) node.
|
|
|
|
Args:
|
|
node: The AST node representing an assignment.
|
|
|
|
Returns:
|
|
None or Parsed information from the assignment node.
|
|
"""
|
|
return [RepoParser._parse_variable(t) for t in node.targets]
|
|
|
|
async def rebuild_class_views(self, path: str | Path = None):
|
|
"""
|
|
Executes `pylint` to reconstruct the dot format class view repository file.
|
|
|
|
Args:
|
|
path (str | Path): The path to the target directory or file. Default is None.
|
|
"""
|
|
if not path:
|
|
path = self.base_directory
|
|
path = Path(path)
|
|
if not path.exists():
|
|
return
|
|
init_file = path / "__init__.py"
|
|
if not init_file.exists():
|
|
raise ValueError("Failed to import module __init__ with error:No module named __init__.")
|
|
command = f"pyreverse {str(path)} -o dot"
|
|
output_dir = path / "__dot__"
|
|
output_dir.mkdir(parents=True, exist_ok=True)
|
|
result = subprocess.run(command, shell=True, check=True, cwd=str(output_dir))
|
|
if result.returncode != 0:
|
|
raise ValueError(f"{result}")
|
|
class_view_pathname = output_dir / "classes.dot"
|
|
class_views = await self._parse_classes(class_view_pathname)
|
|
relationship_views = await self._parse_class_relationships(class_view_pathname)
|
|
packages_pathname = output_dir / "packages.dot"
|
|
class_views, relationship_views, package_root = RepoParser._repair_namespaces(
|
|
class_views=class_views, relationship_views=relationship_views, path=path
|
|
)
|
|
class_view_pathname.unlink(missing_ok=True)
|
|
packages_pathname.unlink(missing_ok=True)
|
|
return class_views, relationship_views, package_root
|
|
|
|
@staticmethod
|
|
async def _parse_classes(class_view_pathname: Path) -> List[DotClassInfo]:
|
|
"""
|
|
Parses a dot format class view repository file.
|
|
|
|
Args:
|
|
class_view_pathname (Path): The path to the dot format class view repository file.
|
|
|
|
Returns:
|
|
List[DotClassInfo]: A list of DotClassInfo objects representing the parsed classes.
|
|
"""
|
|
class_views = []
|
|
if not class_view_pathname.exists():
|
|
return class_views
|
|
data = await aread(filename=class_view_pathname, encoding="utf-8")
|
|
lines = data.split("\n")
|
|
for line in lines:
|
|
package_name, info = RepoParser._split_class_line(line)
|
|
if not package_name:
|
|
continue
|
|
class_name, members, functions = re.split(r"(?<!\\)\|", info)
|
|
class_info = DotClassInfo(name=class_name)
|
|
class_info.package = package_name
|
|
for m in members.split("\n"):
|
|
if not m:
|
|
continue
|
|
attr = DotClassAttribute.parse(m)
|
|
class_info.attributes[attr.name] = attr
|
|
for i in attr.compositions:
|
|
if i not in class_info.compositions:
|
|
class_info.compositions.append(i)
|
|
for f in functions.split("\n"):
|
|
if not f:
|
|
continue
|
|
method = DotClassMethod.parse(f)
|
|
class_info.methods[method.name] = method
|
|
for i in method.aggregations:
|
|
if i not in class_info.compositions and i not in class_info.aggregations:
|
|
class_info.aggregations.append(i)
|
|
class_views.append(class_info)
|
|
return class_views
|
|
|
|
@staticmethod
|
|
async def _parse_class_relationships(class_view_pathname: Path) -> List[DotClassRelationship]:
|
|
"""
|
|
Parses a dot format class view repository file.
|
|
|
|
Args:
|
|
class_view_pathname (Path): The path to the dot format class view repository file.
|
|
|
|
Returns:
|
|
List[DotClassRelationship]: A list of DotClassRelationship objects representing the parsed class relationships.
|
|
"""
|
|
relationship_views = []
|
|
if not class_view_pathname.exists():
|
|
return relationship_views
|
|
data = await aread(filename=class_view_pathname, encoding="utf-8")
|
|
lines = data.split("\n")
|
|
for line in lines:
|
|
relationship = RepoParser._split_relationship_line(line)
|
|
if not relationship:
|
|
continue
|
|
relationship_views.append(relationship)
|
|
return relationship_views
|
|
|
|
@staticmethod
|
|
def _split_class_line(line: str) -> (str, str):
|
|
"""
|
|
Parses a dot format line about class info and returns the class name part and class members part.
|
|
|
|
Args:
|
|
line (str): The dot format line containing class information.
|
|
|
|
Returns:
|
|
Tuple[str, str]: A tuple containing the class name part and class members part.
|
|
"""
|
|
part_splitor = '" ['
|
|
if part_splitor not in line:
|
|
return None, None
|
|
ix = line.find(part_splitor)
|
|
class_name = line[0:ix].replace('"', "")
|
|
left = line[ix:]
|
|
begin_flag = "label=<{"
|
|
end_flag = "}>"
|
|
if begin_flag not in left or end_flag not in left:
|
|
return None, None
|
|
bix = left.find(begin_flag)
|
|
eix = left.rfind(end_flag)
|
|
info = left[bix + len(begin_flag) : eix]
|
|
info = re.sub(r"<br[^>]*>", "\n", info)
|
|
return class_name, info
|
|
|
|
@staticmethod
|
|
def _split_relationship_line(line: str) -> DotClassRelationship:
|
|
"""
|
|
Parses a dot format line about the relationship of two classes and returns 'Generalize', 'Composite',
|
|
or 'Aggregate'.
|
|
|
|
Args:
|
|
line (str): The dot format line containing relationship information.
|
|
|
|
Returns:
|
|
DotClassRelationship: The object of relationship representing either 'Generalize', 'Composite',
|
|
or 'Aggregate' relationship.
|
|
"""
|
|
splitters = [" -> ", " [", "];"]
|
|
idxs = []
|
|
for tag in splitters:
|
|
if tag not in line:
|
|
return None
|
|
idxs.append(line.find(tag))
|
|
ret = DotClassRelationship()
|
|
ret.src = line[0 : idxs[0]].strip('"')
|
|
ret.dest = line[idxs[0] + len(splitters[0]) : idxs[1]].strip('"')
|
|
properties = line[idxs[1] + len(splitters[1]) : idxs[2]].strip(" ")
|
|
mappings = {
|
|
'arrowhead="empty"': GENERALIZATION,
|
|
'arrowhead="diamond"': COMPOSITION,
|
|
'arrowhead="odiamond"': AGGREGATION,
|
|
}
|
|
for k, v in mappings.items():
|
|
if k in properties:
|
|
ret.relationship = v
|
|
if v != GENERALIZATION:
|
|
ret.label = RepoParser._get_label(properties)
|
|
break
|
|
return ret
|
|
|
|
@staticmethod
|
|
def _get_label(line: str) -> str:
|
|
"""
|
|
Parses a dot format line and returns the label information.
|
|
|
|
Args:
|
|
line (str): The dot format line containing label information.
|
|
|
|
Returns:
|
|
str: The label information parsed from the line.
|
|
"""
|
|
tag = 'label="'
|
|
if tag not in line:
|
|
return ""
|
|
ix = line.find(tag)
|
|
eix = line.find('"', ix + len(tag))
|
|
return line[ix + len(tag) : eix]
|
|
|
|
@staticmethod
|
|
def _create_path_mapping(path: str | Path) -> Dict[str, str]:
|
|
"""
|
|
Creates a mapping table between source code files' paths and module names.
|
|
|
|
Args:
|
|
path (str | Path): The path to the source code files or directory.
|
|
|
|
Returns:
|
|
Dict[str, str]: A dictionary mapping source code file paths to their corresponding module names.
|
|
"""
|
|
mappings = {
|
|
str(path).replace("/", "."): str(path),
|
|
}
|
|
files = []
|
|
try:
|
|
directory_path = Path(path)
|
|
if not directory_path.exists():
|
|
return mappings
|
|
for file_path in directory_path.iterdir():
|
|
if file_path.is_file():
|
|
files.append(str(file_path))
|
|
else:
|
|
subfolder_files = RepoParser._create_path_mapping(path=file_path)
|
|
mappings.update(subfolder_files)
|
|
except Exception as e:
|
|
logger.error(f"Error: {e}")
|
|
for f in files:
|
|
mappings[str(Path(f).with_suffix("")).replace("/", ".")] = str(f)
|
|
|
|
return mappings
|
|
|
|
@staticmethod
|
|
def _repair_namespaces(
|
|
class_views: List[DotClassInfo], relationship_views: List[DotClassRelationship], path: str | Path
|
|
) -> (List[DotClassInfo], List[DotClassRelationship], str):
|
|
"""
|
|
Augments namespaces to the path-prefixed classes and relationships.
|
|
|
|
Args:
|
|
class_views (List[DotClassInfo]): List of DotClassInfo objects representing class views.
|
|
relationship_views (List[DotClassRelationship]): List of DotClassRelationship objects representing
|
|
relationships.
|
|
path (str | Path): The path to the source code files or directory.
|
|
|
|
Returns:
|
|
Tuple[List[DotClassInfo], List[DotClassRelationship], str]: A tuple containing the augmented class views,
|
|
relationships, and the root path of the package.
|
|
"""
|
|
if not class_views:
|
|
return [], [], ""
|
|
c = class_views[0]
|
|
full_key = str(path).lstrip("/").replace("/", ".")
|
|
root_namespace = RepoParser._find_root(full_key, c.package)
|
|
root_path = root_namespace.replace(".", "/")
|
|
|
|
mappings = RepoParser._create_path_mapping(path=path)
|
|
new_mappings = {}
|
|
ix_root_namespace = len(root_namespace)
|
|
ix_root_path = len(root_path)
|
|
for k, v in mappings.items():
|
|
nk = k[ix_root_namespace:]
|
|
nv = v[ix_root_path:]
|
|
new_mappings[nk] = nv
|
|
|
|
for c in class_views:
|
|
c.package = RepoParser._repair_ns(c.package, new_mappings)
|
|
for _, v in enumerate(relationship_views):
|
|
v.src = RepoParser._repair_ns(v.src, new_mappings)
|
|
v.dest = RepoParser._repair_ns(v.dest, new_mappings)
|
|
return class_views, relationship_views, str(path)[: len(root_path)]
|
|
|
|
@staticmethod
|
|
def _repair_ns(package: str, mappings: Dict[str, str]) -> str:
|
|
"""
|
|
Replaces the package-prefix with the namespace-prefix.
|
|
|
|
Args:
|
|
package (str): The package to be repaired.
|
|
mappings (Dict[str, str]): A dictionary mapping source code file paths to their corresponding packages.
|
|
|
|
Returns:
|
|
str: The repaired namespace.
|
|
"""
|
|
file_ns = package
|
|
ix = 0
|
|
while file_ns != "":
|
|
if file_ns not in mappings:
|
|
ix = file_ns.rfind(".")
|
|
file_ns = file_ns[0:ix]
|
|
continue
|
|
break
|
|
if file_ns == "":
|
|
return ""
|
|
internal_ns = package[ix + 1 :]
|
|
ns = mappings[file_ns] + ":" + internal_ns.replace(".", ":")
|
|
return ns
|
|
|
|
@staticmethod
|
|
def _find_root(full_key: str, package: str) -> str:
|
|
"""
|
|
Returns the package root path based on the key, which is the full path, and the package information.
|
|
|
|
Args:
|
|
full_key (str): The full key representing the full path.
|
|
package (str): The package information.
|
|
|
|
Returns:
|
|
str: The package root path.
|
|
"""
|
|
left = full_key
|
|
while left != "":
|
|
if left in package:
|
|
break
|
|
if "." not in left:
|
|
break
|
|
ix = left.find(".")
|
|
left = left[ix + 1 :]
|
|
ix = full_key.rfind(left)
|
|
return "." + full_key[0:ix]
|
|
|
|
|
|
def is_func(node) -> bool:
|
|
"""
|
|
Returns True if the given node represents a function.
|
|
|
|
Args:
|
|
node: The Abstract Syntax Tree (AST) node.
|
|
|
|
Returns:
|
|
bool: True if the node represents a function, False otherwise.
|
|
"""
|
|
return isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef))
|