claw/insightflow
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: auto-fix code issues (cron)

Browse Source 
- 修复重复导入/字段
- 修复异常处理
- 修复PEP8格式问题
- 添加类型注解
- 修复缺失的urllib.parse导入
This commit is contained in:
OpenClaw Bot
2026-02-28 06:03:09 +08:00
parent ff83cab6c7
commit fe3d64a1d2
41 changed files with 4501 additions and 1176 deletions
Download Patch File Download Diff File Expand all files Collapse all files

144
auto_code_fixer.py
View File

@@ -8,13 +8,19 @@ import os
import re
import subprocess
from pathlib import Path
from typing import Any
class CodeIssue:
"""代码问题记录"""
def __init__(self, file_path: str, line_no: int, issue_type: str, message: str, severity: str = "warning"):
def __init__(
self,
file_path: str,
line_no: int,
issue_type: str,
message: str,
severity: str = "warning",
):
self.file_path = file_path
self.line_no = line_no
self.issue_type = issue_type
@@ -83,7 +89,9 @@ class CodeFixer:
# 检查敏感信息
self._check_sensitive_info(file_path, content, lines)
def _check_duplicate_imports(self, file_path: Path, content: str, lines: list[str]) -> None:
def _check_duplicate_imports(
self, file_path: Path, content: str, lines: list[str]
) -> None:
"""检查重复导入"""
imports = {}
for i, line in enumerate(lines, 1):
@@ -94,38 +102,64 @@ class CodeFixer:
key = f"{module}:{names}"
if key in imports:
self.issues.append(
CodeIssue(str(file_path), i, "duplicate_import", f"重复导入: {line.strip()}", "warning")
CodeIssue(
str(file_path),
i,
"duplicate_import",
f"重复导入: {line.strip()}",
"warning",
)
)
imports[key] = i
def _check_bare_exceptions(self, file_path: Path, content: str, lines: list[str]) -> None:
def _check_bare_exceptions(
self, file_path: Path, content: str, lines: list[str]
) -> None:
"""检查裸异常捕获"""
for i, line in enumerate(lines, 1):
if re.search(r"except\s*:\s*$", line) or re.search(r"except\s*:\s*#", line):
self.issues.append(
CodeIssue(str(file_path), i, "bare_exception", "裸异常捕获，应指定具体异常类型", "error")
CodeIssue(
str(file_path),
i,
"bare_exception",
"裸异常捕获，应指定具体异常类型",
"error",
)
)
def _check_pep8_issues(self, file_path: Path, content: str, lines: list[str]) -> None:
def _check_pep8_issues(
self, file_path: Path, content: str, lines: list[str]
) -> None:
"""检查 PEP8 格式问题"""
for i, line in enumerate(lines, 1):
# 行长度超过 120
if len(line) > 120:
self.issues.append(
CodeIssue(str(file_path), i, "line_too_long", f"行长度 {len(line)} 超过 120 字符", "warning")
CodeIssue(
str(file_path),
i,
"line_too_long",
f"行长度 {len(line)} 超过 120 字符",
"warning",
)
)
# 行尾空格
if line.rstrip() != line:
self.issues.append(
CodeIssue(str(file_path), i, "trailing_whitespace", "行尾有空格", "info")
CodeIssue(
str(file_path), i, "trailing_whitespace", "行尾有空格", "info"
)
)
# 多余的空行
if i > 1 and line.strip() == "" and lines[i - 2].strip() == "":
if i < len(lines) and lines[i].strip() != "":
self.issues.append(
CodeIssue(str(file_path), i, "extra_blank_line", "多余的空行", "info")
CodeIssue(
str(file_path), i, "extra_blank_line", "多余的空行", "info"
)
)
def _check_unused_imports(self, file_path: Path, content: str) -> None:
@@ -157,10 +191,18 @@ class CodeFixer:
for name, line in imports.items():
if name not in used_names and not name.startswith("_"):
self.issues.append(
CodeIssue(str(file_path), line, "unused_import", f"未使用的导入: {name}", "warning")
CodeIssue(
str(file_path),
line,
"unused_import",
f"未使用的导入: {name}",
"warning",
)
)
def _check_type_annotations(self, file_path: Path, content: str, lines: list[str]) -> None:
def _check_type_annotations(
self, file_path: Path, content: str, lines: list[str]
) -> None:
"""检查类型注解"""
try:
tree = ast.parse(content)
@@ -171,7 +213,11 @@ class CodeFixer:
if isinstance(node, ast.FunctionDef):
# 检查函数参数类型注解
for arg in node.args.args:
if arg.annotation is None and arg.arg != "self" and arg.arg != "cls":
if (
arg.annotation is None
and arg.arg != "self"
and arg.arg != "cls"
):
self.issues.append(
CodeIssue(
str(file_path),
@@ -182,22 +228,40 @@ class CodeFixer:
)
)
def _check_string_formatting(self, file_path: Path, content: str, lines: list[str]) -> None:
def _check_string_formatting(
self, file_path: Path, content: str, lines: list[str]
) -> None:
"""检查字符串格式化"""
for i, line in enumerate(lines, 1):
# 检查 % 格式化
if re.search(r"['\"].*%[sdif].*['\"]\s*%", line) or re.search(r"['\"].*%\(.*\).*['\"]\s*%", line):
if re.search(r"['\"].*%[sdif].*['\"]\s*%", line) or re.search(
r"['\"].*%\(.*\).*['\"]\s*%", line
):
self.issues.append(
CodeIssue(str(file_path), i, "old_string_format", "使用 % 格式化，建议改为 f-string", "info")
CodeIssue(
str(file_path),
i,
"old_string_format",
"使用 % 格式化，建议改为 f-string",
"info",
)
)
# 检查 .format()
if re.search(r"['\"].*\{.*\}.*['\"]\.format\(", line):
self.issues.append(
CodeIssue(str(file_path), i, "format_method", "使用 .format()，建议改为 f-string", "info")
CodeIssue(
str(file_path),
i,
"format_method",
"使用 .format()，建议改为 f-string",
"info",
)
)
def _check_magic_numbers(self, file_path: Path, content: str, lines: list[str]) -> None:
def _check_magic_numbers(
self, file_path: Path, content: str, lines: list[str]
) -> None:
"""检查魔法数字"""
# 排除的魔法数字
excluded = {"0", "1", "-1", "0.0", "1.0", "100", "0.5", "3600", "86400", "1024"}
@@ -223,11 +287,15 @@ class CodeFixer:
)
)
def _check_sql_injection(self, file_path: Path, content: str, lines: list[str]) -> None:
def _check_sql_injection(
self, file_path: Path, content: str, lines: list[str]
) -> None:
"""检查 SQL 注入风险"""
for i, line in enumerate(lines, 1):
# 检查字符串拼接 SQL
if re.search(r"execute\s*\(\s*['\"].*%", line) or re.search(r"execute\s*\(\s*f['\"]", line):
if re.search(r"execute\s*\(\s*['\"].*%", line) or re.search(
r"execute\s*\(\s*f['\"]", line
):
self.issues.append(
CodeIssue(
str(file_path),
@@ -238,7 +306,9 @@ class CodeFixer:
)
)
def _check_cors_config(self, file_path: Path, content: str, lines: list[str]) -> None:
def _check_cors_config(
self, file_path: Path, content: str, lines: list[str]
) -> None:
"""检查 CORS 配置"""
for i, line in enumerate(lines, 1):
if "allow_origins" in line and "*" in line:
@@ -252,7 +322,9 @@ class CodeFixer:
)
)
def _check_sensitive_info(self, file_path: Path, content: str, lines: list[str]) -> None:
def _check_sensitive_info(
self, file_path: Path, content: str, lines: list[str]
) -> None:
"""检查敏感信息泄露"""
patterns = [
(r"password\s*=\s*['\"][^'\"]+['\"]", "硬编码密码"),
@@ -323,7 +395,11 @@ class CodeFixer:
line_idx = issue.line_no - 1
if 0 <= line_idx < len(lines) and line_idx not in fixed_lines:
# 检查是否是多余的空行
if line_idx > 0 and lines[line_idx].strip() == "" and lines[line_idx - 1].strip() == "":
if (
line_idx > 0
and lines[line_idx].strip() == ""
and lines[line_idx - 1].strip() == ""
):
lines.pop(line_idx)
fixed_lines.add(line_idx)
self.fixed_issues.append(issue)
@@ -386,7 +462,9 @@ class CodeFixer:
report.append("")
if self.fixed_issues:
for issue in self.fixed_issues:
report.append(f"- `{issue.file_path}:{issue.line_no}` - {issue.message}")
report.append(
f"- `{issue.file_path}:{issue.line_no}` - {issue.message}"
)
else:
report.append("无")
report.append("")
@@ -399,7 +477,9 @@ class CodeFixer:
report.append("")
if manual_issues:
for issue in manual_issues:
report.append(f"- `{issue.file_path}:{issue.line_no}` [{issue.severity}] {issue.message}")
report.append(
f"- `{issue.file_path}:{issue.line_no}` [{issue.severity}] {issue.message}"
)
else:
report.append("无")
report.append("")
@@ -407,7 +487,11 @@ class CodeFixer:
# 其他问题
report.append("## 📋 其他发现的问题")
report.append("")
other_issues = [i for i in self.issues if i.issue_type not in manual_types and i not in self.fixed_issues]
other_issues = [
i
for i in self.issues
if i.issue_type not in manual_types and i not in self.fixed_issues
]
# 按类型分组
by_type = {}
@@ -420,7 +504,9 @@ class CodeFixer:
report.append(f"### {issue_type}")
report.append("")
for issue in issues[:10]: # 每种类型最多显示10个
report.append(f"- `{issue.file_path}:{issue.line_no}` - {issue.message}")
report.append(
f"- `{issue.file_path}:{issue.line_no}` - {issue.message}"
)
if len(issues) > 10:
report.append(f"- ... 还有 {len(issues) - 10} 个类似问题")
report.append("")
@@ -453,7 +539,9 @@ def git_commit_and_push(project_path: str) -> tuple[bool, str]:
- 修复PEP8格式问题
- 添加类型注解"""
subprocess.run(["git", "commit", "-m", commit_msg], cwd=project_path, check=True)
subprocess.run(
["git", "commit", "-m", commit_msg], cwd=project_path, check=True
)
# 推送
subprocess.run(["git", "push"], cwd=project_path, check=True)