القائمة
当进行Python代码审查、性能优化、类型安全检查或测试策略规划时，分析Python代码质量和最佳实践。
التثبيت باستخدام Codex أو Claude انسخ هذا Prompt والصقه في Codex أو Claude أو مساعد آخر ليراجع صفحة Skill ويثبّتها لك.
استنادا إلى تصنيف SOC المهني
以聚合根为边界，包含多个相关Entity和ValueObject的集合。保证数据一致性和事务边界。
在DDD中具有唯一身份标识和生命周期的对象，通过身份而非属性值相等判断。
封装复杂对象和聚合的创建过程，将创建职责从领域对象中剥离，保证聚合创建时的不变量满足。
没有身份标识，通过属性值判断相等的对象。不可变，通常代表领域中的度量或描述。
命令查询责任分离，将数据的写入操作和读取操作分别用不同的模型处理，优化各自的性能。
将DDD战略设计应用于微服务架构，限界上下文指导服务拆分，领域事件实现服务间通信。
| name | Python分析器 |
| description | 当进行Python代码审查、性能优化、类型安全检查或测试策略规划时，分析Python代码质量和最佳实践。 |
| license | MIT |
Python让编写糟糕代码变得容易。分析代码质量以防止技术债务。
核心原则: Python让编写糟糕代码变得容易。分析代码质量以防止技术债务。
始终:
触发短语:
问题:
不符合PEP 8编码规范
错误示例:
- 变量名使用驼峰命名
- 行长度超过79字符
- 缺少空行分隔
- 导入语句不规范
解决方案:
1. 使用black自动格式化
2. 配置flake8检查
3. 使用isort整理导入
4. 遵循PEP 8指南
问题:
Python代码性能低下
错误示例:
- 在循环中使用+拼接字符串
- 不必要的列表推导
- 全局变量访问
- 缺少缓存机制
解决方案:
1. 使用join拼接字符串
2. 优化数据结构选择
3. 使用局部变量
4. 实现缓存策略
问题:
缺少类型注解导致运行时错误
错误示例:
- 函数参数无类型提示
- 返回值类型不明确
- 可选参数未标注
- 泛型使用不当
解决方案:
1. 添加类型注解
2. 使用Optional标注可选值
3. 使用TypeVar定义泛型
4. 运行mypy检查
import ast
import os
import re
from collections import defaultdict
from typing import List, Dict, Any, Optional
from dataclasses import dataclass
from pathlib import Path
@dataclass
class CodeIssue:
"""代码问题"""
file_path: str
line_number: int
column: int
severity: str # error, warning, info
message: str
rule_id: str
suggestion: Optional[str] = None
@dataclass
class FunctionMetrics:
"""函数指标"""
name: str
line_start: int
line_end: int
complexity: int
arguments: int
returns: int
docstring: bool
type_annotations: bool
@dataclass
class ClassMetrics:
"""类指标"""
name: str
line_start: int
line_end: int
methods: int
attributes: int
inheritance_depth: int
docstring: bool
class PythonAnalyzer:
def __init__(self, config: Optional[Dict] = None):
self.config = config or self.get_default_config()
self.issues: List[CodeIssue] = []
self.function_metrics: List[FunctionMetrics] = []
self.class_metrics: List[ClassMetrics] = []
self.imports: Dict[str, List[str]] = defaultdict(list)
self.complexity_threshold = self.config.get('complexity_threshold', 10)
def get_default_config(self) -> Dict:
"""获取默认配置"""
return {
'max_line_length': 79,
'max_function_length': 50,
'max_class_length': 200,
'complexity_threshold': 10,
'require_type_annotations': True,
'require_docstrings': True,
'check_imports': True,
'check_naming': True
}
def analyze_file(self, file_path: str) -> Dict[str, Any]:
"""分析单个Python文件"""
try:
with open(file_path, 'r', encoding='utf-8') as f:
content = f.read()
# 解析AST
tree = ast.parse(content, filename=file_path)
# 重置状态
self.issues = []
self.function_metrics = []
self.class_metrics = []
self.imports = defaultdict(list)
# 执行各种分析
self.check_style(file_path, content)
self.analyze_ast(tree, file_path)
self.check_complexity(tree, file_path)
self.check_type_safety(tree, file_path)
self.check_documentation(tree, file_path)
return {
'file_path': file_path,
'issues': self.issues,
'function_metrics': self.function_metrics,
'class_metrics': self.class_metrics,
'imports': dict(self.imports),
'summary': self.generate_summary()
}
except SyntaxError as e:
return {
'file_path': file_path,
'error': f'语法错误: {e}',
'issues': [CodeIssue(
file_path=file_path,
line_number=e.lineno or 0,
column=e.offset or 0,
severity='error',
message=f'语法错误: {e.msg}',
rule_id='syntax_error'
)]
}
except Exception as e:
return {
'file_path': file_path,
'error': f'分析失败: {e}',
'issues': []
}
def analyze_directory(self, directory: str) -> Dict[str, Any]:
"""分析整个目录"""
results = []
all_issues = []
for root, dirs, files in os.walk(directory):
# 跳过虚拟环境和缓存目录
dirs[:] = [d for d in dirs if not d.startswith('.') and d not in ['__pycache__', 'venv', 'env']]
for file in files:
if file.endswith('.py'):
file_path = os.path.join(root, file)
result = self.analyze_file(file_path)
results.append(result)
all_issues.extend(result.get('issues', []))
return {
'directory': directory,
'files': results,
'total_issues': len(all_issues),
'issue_summary': self.categorize_issues(all_issues),
'recommendations': self.generate_recommendations(all_issues)
}
def check_style(self, file_path: str, content: str) -> None:
"""检查代码风格"""
lines = content.split('\n')
for line_num, line in enumerate(lines, 1):
# 检查行长度
if len(line) > self.config['max_line_length']:
self.issues.append(CodeIssue(
file_path=file_path,
line_number=line_num,
column=self.config['max_line_length'],
severity='warning',
message=f'行长度超过{self.config["max_line_length"]}字符',
rule_id='line_too_long',
suggestion='考虑拆分长行或使用括号换行'
))
# 检查尾随空格
if line.endswith(' '):
self.issues.append(CodeIssue(
file_path=file_path,
line_number=line_num,
column=len(line.rstrip()),
severity='info',
message='行末有多余空格',
rule_id='trailing_whitespace',
suggestion='删除行末空格'
))
# 检查Tab字符
if '\t' in line:
self.issues.append(CodeIssue(
file_path=file_path,
line_number=line_num,
column=line.find('\t'),
severity='warning',
message='使用了Tab字符，应该使用空格',
rule_id='tab_character',
suggestion='将Tab替换为4个空格'
))
def analyze_ast(self, tree: ast.AST, file_path: str) -> None:
"""分析AST结构"""
# 分析导入
for node in ast.walk(tree):
if isinstance(node, ast.Import):
for alias in node.names:
self.imports['import'].append(alias.name)
elif isinstance(node, ast.ImportFrom):
module = node.module or ''
for alias in node.names:
self.imports['from'].append(f'{module}.{alias.name}')
# 分析函数和类
for node in ast.walk(tree):
if isinstance(node, ast.FunctionDef):
self.analyze_function(node, file_path)
elif isinstance(node, ast.ClassDef):
self.analyze_class(node, file_path)
def analyze_function(self, node: ast.FunctionDef, file_path: str) -> None:
"""分析函数"""
# 计算复杂度
complexity = self.calculate_complexity(node)
# 统计参数
args = len(node.args.args) + len(node.args.kwonlyargs)
if node.args.vararg:
args += 1
if node.args.kwarg:
args += 1
# 检查返回语句
returns = len([n for n in ast.walk(node) if isinstance(n, ast.Return)])
# 检查文档字符串
has_docstring = (ast.get_docstring(node) is not None)
# 检查类型注解
has_type_annotations = (
all(arg.annotation for arg in node.args.args) and
(node.returns is not None)
)
metrics = FunctionMetrics(
name=node.name,
line_start=node.lineno,
line_end=node.end_lineno or node.lineno,
complexity=complexity,
arguments=args,
returns=returns,
docstring=has_docstring,
type_annotations=has_type_annotations
)
self.function_metrics.append(metrics)
# 检查问题
if complexity > self.complexity_threshold:
self.issues.append(CodeIssue(
file_path=file_path,
line_number=node.lineno,
column=0,
severity='warning',
message=f'函数{node.name}复杂度过高({complexity})',
rule_id='high_complexity',
suggestion='考虑拆分函数或简化逻辑'
))
if not has_docstring and self.config['require_docstrings']:
self.issues.append(CodeIssue(
file_path=file_path,
line_number=node.lineno,
column=0,
severity='info',
message=f'函数{node.name}缺少文档字符串',
rule_id='missing_docstring',
suggestion='添加函数文档说明'
))
if not has_type_annotations and self.config['require_type_annotations']:
self.issues.append(CodeIssue(
file_path=file_path,
line_number=node.lineno,
column=0,
severity='info',
message=f'函数{node.name}缺少类型注解',
rule_id='missing_type_annotations',
suggestion='添加参数和返回值类型注解'
))
def analyze_class(self, node: ast.ClassDef, file_path: str) -> None:
"""分析类"""
# 统计方法和属性
methods = 0
attributes = 0
for item in node.body:
if isinstance(item, ast.FunctionDef):
methods += 1
elif isinstance(item, ast.Assign):
for target in item.targets:
if isinstance(target, ast.Name):
attributes += 1
# 计算继承深度
inheritance_depth = self.calculate_inheritance_depth(node)
# 检查文档字符串
has_docstring = (ast.get_docstring(node) is not None)
metrics = ClassMetrics(
name=node.name,
line_start=node.lineno,
line_end=node.end_lineno or node.lineno,
methods=methods,
attributes=attributes,
inheritance_depth=inheritance_depth,
docstring=has_docstring
)
self.class_metrics.append(metrics)
# 检查问题
if inheritance_depth > 3:
self.issues.append(CodeIssue(
file_path=file_path,
line_number=node.lineno,
column=0,
severity='warning',
message=f'类{node.name}继承层次过深({inheritance_depth})',
rule_id='deep_inheritance',
suggestion='考虑使用组合替代继承'
))
if not has_docstring and self.config['require_docstrings']:
self.issues.append(CodeIssue(
file_path=file_path,
line_number=node.lineno,
column=0,
severity='info',
message=f'类{node.name}缺少文档字符串',
rule_id='missing_docstring',
suggestion='添加类文档说明'
))
def calculate_complexity(self, node: ast.AST) -> int:
"""计算圈复杂度"""
complexity = 1 # 基础复杂度
for child in ast.walk(node):
if isinstance(child, (ast.If, ast.While, ast.For, ast.AsyncFor)):
complexity += 1
elif isinstance(child, ast.ExceptHandler):
complexity += 1
elif isinstance(child, ast.With, ast.AsyncWith):
complexity += 1
elif isinstance(child, ast.BoolOp):
complexity += len(child.values) - 1
return complexity
def calculate_inheritance_depth(self, node: ast.ClassDef) -> int:
"""计算继承深度"""
if not node.bases:
return 0
max_depth = 0
for base in node.bases:
if isinstance(base, ast.Name):
# 简化实现，实际需要解析整个模块
max_depth = max(max_depth, 1)
return max_depth
def check_complexity(self, tree: ast.AST, file_path: str) -> None:
"""检查复杂度相关问题"""
# 检查嵌套深度
for node in ast.walk(tree):
if isinstance(node, (ast.For, ast.While, ast.If)):
depth = self.calculate_nesting_depth(node)
if depth > 3:
self.issues.append(CodeIssue(
file_path=file_path,
line_number=node.lineno,
column=0,
severity='warning',
message=f'嵌套层次过深({depth})',
rule_id='deep_nesting',
suggestion='考虑提取函数或使用早期返回'
))
def calculate_nesting_depth(self, node: ast.AST, current_depth: int = 0) -> int:
"""计算嵌套深度"""
max_depth = current_depth
for child in ast.iter_child_nodes(node):
if isinstance(child, (ast.For, ast.While, ast.If, ast.With, ast.Try)):
child_depth = self.calculate_nesting_depth(child, current_depth + 1)
max_depth = max(max_depth, child_depth)
return max_depth
def check_type_safety(self, tree: ast.AST, file_path: str) -> None:
"""检查类型安全"""
for node in ast.walk(tree):
# 检查函数参数类型
if isinstance(node, ast.FunctionDef):
self.check_function_types(node, file_path)
# 检查可能的None值
elif isinstance(node, ast.Attribute):
self.check_none_safety(node, file_path)
def check_function_types(self, node: ast.FunctionDef, file_path: str) -> None:
"""检查函数类型安全"""
if self.config['require_type_annotations']:
# 检查参数类型注解
for arg in node.args.args:
if not arg.annotation:
self.issues.append(CodeIssue(
file_path=file_path,
line_number=node.lineno,
column=0,
severity='info',
message=f'参数{arg.arg}缺少类型注解',
rule_id='missing_parameter_type',
suggestion=f'添加类型注解: {arg.arg}: type'
))
# 检查返回值类型注解
if not node.returns:
self.issues.append(CodeIssue(
file_path=file_path,
line_number=node.lineno,
column=0,
severity='info',
message='函数缺少返回值类型注解',
rule_id='missing_return_type',
suggestion='添加返回值类型注解'
))
def check_none_safety(self, node: ast.Attribute, file_path: str) -> None:
"""检查None值安全"""
# 简化实现，检查可能的None访问
if isinstance(node.value, ast.Name) and node.value.id == 'None':
self.issues.append(CodeIssue(
file_path=file_path,
line_number=node.lineno,
column=node.col_offset,
severity='error',
message='尝试访问None的属性',
rule_id='none_attribute_access',
suggestion='检查对象是否为None'
))
def check_documentation(self, tree: ast.AST, file_path: str) -> None:
"""检查文档"""
# 检查模块文档
if not ast.get_docstring(tree) and self.config['require_docstrings']:
self.issues.append(CodeIssue(
file_path=file_path,
line_number=1,
column=0,
severity='info',
message='模块缺少文档字符串',
rule_id='missing_module_docstring',
suggestion='添加模块级文档说明'
))
def generate_summary(self) -> Dict[str, Any]:
"""生成分析摘要"""
total_issues = len(self.issues)
error_count = len([i for i in self.issues if i.severity == 'error'])
warning_count = len([i for i in self.issues if i.severity == 'warning'])
info_count = len([i for i in self.issues if i.severity == 'info'])
return {
'total_issues': total_issues,
'errors': error_count,
'warnings': warning_count,
'info': info_count,
'functions': len(self.function_metrics),
'classes': len(self.class_metrics),
'average_complexity': self.calculate_average_complexity()
}
def calculate_average_complexity(self) -> float:
"""计算平均复杂度"""
if not self.function_metrics:
return 0.0
total_complexity = sum(f.complexity for f in self.function_metrics)
return total_complexity / len(self.function_metrics)
def categorize_issues(self, issues: List[CodeIssue]) -> Dict[str, List[CodeIssue]]:
"""分类问题"""
categorized = {
'errors': [],
'warnings': [],
'info': []
}
for issue in issues:
categorized[issue.severity].append(issue)
return categorized
def generate_recommendations(self, issues: List[CodeIssue]) -> List[Dict[str, str]]:
"""生成改进建议"""
recommendations = []
# 统计问题类型
issue_counts = defaultdict(int)
for issue in issues:
issue_counts[issue.rule_id] += 1
# 基于问题类型生成建议
if issue_counts['line_too_long'] > 5:
recommendations.append({
'type': 'style',
'priority': 'medium',
'message': '发现多个长行，建议使用black自动格式化',
'action': 'pip install black && black .'
})
if issue_counts['missing_docstring'] > 3:
recommendations.append({
'type': 'documentation',
'priority': 'low',
'message': '多个函数缺少文档字符串，建议完善文档',
'action': '为所有公共函数和类添加文档字符串'
})
if issue_counts['high_complexity'] > 2:
recommendations.append({
'type': 'complexity',
'priority': 'high',
'message': '发现多个高复杂度函数，建议重构',
'action': '拆分复杂函数，提取辅助函数'
})
if issue_counts['missing_type_annotations'] > 5:
recommendations.append({
'type': 'type_safety',
'priority': 'medium',
'message': '缺少类型注解，建议添加类型提示',
'action': '使用mypy检查并添加类型注解'
})
return recommendations
# 使用示例
def main():
analyzer = PythonAnalyzer()
# 分析单个文件
result = analyzer.analyze_file('./example.py')
print(f"文件: {result['file_path']}")
print(f"问题数: {len(result['issues'])}")
# 分析整个目录
directory_result = analyzer.analyze_directory('./src')
print(f"目录: {directory_result['directory']}")
print(f"总问题数: {directory_result['total_issues']}")
# 打印建议
for rec in directory_result['recommendations']:
print(f"- {rec['message']}")
if __name__ == '__main__':
main()
import time
import cProfile
import pstats
import io
from functools import wraps
from typing import Callable, Dict, Any, List
from dataclasses import dataclass
@dataclass
class PerformanceMetric:
"""性能指标"""
function_name: str
execution_time: float
call_count: int
memory_usage: int
cpu_usage: float
class PythonPerformanceAnalyzer:
def __init__(self):
self.metrics: List[PerformanceMetric] = []
self.profiles: Dict[str, cProfile.Profile] = {}
def profile_function(self, func: Callable) -> Callable:
"""函数性能分析装饰器"""
@wraps(func)
def wrapper(*args, **kwargs):
# 创建性能分析器
profiler = cProfile.Profile()
# 开始分析
start_time = time.time()
profiler.enable()
try:
result = func(*args, **kwargs)
return result
finally:
# 结束分析
profiler.disable()
end_time = time.time()
# 记录指标
execution_time = end_time - start_time
self.record_metric(func.__name__, execution_time, profiler)
# 保存分析器
self.profiles[func.__name__] = profiler
return result
return wrapper
def record_metric(self, func_name: str, execution_time: float, profiler: cProfile.Profile) -> None:
"""记录性能指标"""
# 获取调用统计
stats = pstats.Stats(profiler)
# 获取调用次数
call_count = 0
for func_info, (calls, _, _, _, _) in stats.stats.items():
if func_info[2] == func_name:
call_count = calls
break
metric = PerformanceMetric(
function_name=func_name,
execution_time=execution_time,
call_count=call_count,
memory_usage=0, # 需要memory_profiler
cpu_usage=0.0 # 需要psutil
)
self.metrics.append(metric)
def get_performance_report(self, func_name: str) -> Dict[str, Any]:
"""获取性能报告"""
if func_name not in self.profiles:
return {'error': f'函数{func_name}没有性能数据'}
profiler = self.profiles[func_name]
stats = pstats.Stats(profiler)
# 获取热点函数
hotspots = []
for func_info, (calls, total_time, cum_time, _, _) in stats.stats.items():
hotspots.append({
'function': func_info[2],
'calls': calls,
'total_time': total_time,
'cumulative_time': cum_time
})
# 按累计时间排序
hotspots.sort(key=lambda x: x['cumulative_time'], reverse=True)
return {
'function_name': func_name,
'hotspots': hotspots[:10], # 前10个热点
'total_calls': sum(h['calls'] for h in hotspots),
'total_time': sum(h['cumulative_time'] for h in hotspots)
}
def optimize_suggestions(self, func_name: str) -> List[str]:
"""生成优化建议"""
report = self.get_performance_report(func_name)
suggestions = []
if 'error' in report:
return suggestions
hotspots = report['hotspots']
# 分析热点函数
for hotspot in hotspots[:5]:
if hotspot['cumulative_time'] > 0.1: # 超过100ms
suggestions.append(
f"函数{hotspot['function']}耗时较多({hotspot['cumulative_time']:.3f}s)，"
f"考虑优化算法或使用缓存"
)
if hotspot['calls'] > 1000:
suggestions.append(
f"函数{hotspot['function']}调用频繁({hotspot['calls']}次)，"
f"考虑使用缓存或批量处理"
)
return suggestions
# 使用示例
@PythonPerformanceAnalyzer().profile_function
def slow_function():
"""模拟慢函数"""
time.sleep(0.1)
result = sum(i * i for i in range(1000))
return result
def main():
# 运行函数多次
for _ in range(5):
slow_function()
# 获取性能报告
analyzer = PythonPerformanceAnalyzer()
report = analyzer.get_performance_report('slow_function')
print("性能报告:", report)
# 获取优化建议
suggestions = analyzer.optimize_suggestions('slow_function')
print("优化建议:", suggestions)
if __name__ == '__main__':
main()