Bosch_Demo/code_scan
4
0
Fork 0
Code Issues Pull Requests 2 Actions Packages Projects Releases Wiki Activity

add web

Browse Source
This commit is contained in:
Dang Zerong
2026-03-12 14:42:23 +08:00
parent 9ae55407fc
commit 027cf50759
8 changed files with 225 additions and 52 deletions
Download Patch File Download Diff File Expand all files Collapse all files

25
scanner/python_scanner.py
View File

@@ -20,7 +20,7 @@ class PythonScanner(BaseScanner):
super().__init__(config)
self.extensions = ['.py']
def scan(self, repo_url: str, commit_id: Optional[str], branch: str) -> Dict[str, Any]:
def scan(self, repo_url: str, commit_id: Optional[str], branch: str, changed_files: Optional[List[str]] = None) -> Dict[str, Any]:
"""
执行 Python 代码扫描
@@ -28,6 +28,7 @@ class PythonScanner(BaseScanner):
repo_url: 仓库 URL
commit_id: 提交 ID
branch: 分支名
changed_files: 可选的变更文件列表（来自 PR）
Returns:
扫描结果
@@ -51,8 +52,8 @@ class PythonScanner(BaseScanner):
# 克隆仓库
clone_dir = self.clone_repo(repo_url, commit_id, branch)
# 获取 Python 文件
py_files = self.get_changed_files(clone_dir, self.extensions)
# 获取 Python 文件（只扫描变更的文件）
py_files = self.get_changed_files(clone_dir, self.extensions, changed_files)
result['files_scanned'] = len(py_files)
if not py_files:
@@ -84,7 +85,7 @@ class PythonScanner(BaseScanner):
return result
def _run_pylint(self, cwd: str, files: List[str]) -> Dict[str, Any]:
def _run_pylint(self, clone_dir: str, files: List[str]) -> Dict[str, Any]:
"""运行 Pylint 扫描"""
result = {
'tool': 'pylint',
@@ -95,7 +96,7 @@ class PythonScanner(BaseScanner):
# 只扫描变更的文件
try:
cmd = ['python', '-m', 'pylint', '--output-format=json'] + files
output = self.run_command(cmd, cwd, timeout=120)
output = self.run_command(cmd, clone_dir, timeout=120)
result['raw_output'] = output.get('stdout', '')
@@ -104,12 +105,15 @@ class PythonScanner(BaseScanner):
try:
issues = json.loads(output['stdout'])
for issue in issues:
# 使用相对于 clone_dir 的路径
full_path = issue.get('path', '')
rel_path = os.path.relpath(full_path, clone_dir) if full_path else ''
result['issues'].append({
'tool': 'pylint',
'type': issue.get('type', 'info'),
'severity': issue.get('severity', 'Info'),
'message': issue.get('message', ''),
'file': os.path.basename(issue.get('path', '')),
'file': rel_path,
'line': issue.get('line', 0),
'column': issue.get('column', 0),
'symbol': issue.get('symbol', '')
@@ -122,7 +126,7 @@ class PythonScanner(BaseScanner):
return result
def _run_flake8(self, cwd: str, files: List[str]) -> Dict[str, Any]:
def _run_flake8(self, clone_dir: str, files: List[str]) -> Dict[str, Any]:
"""运行 Flake8 扫描"""
result = {
'tool': 'flake8',
@@ -132,7 +136,7 @@ class PythonScanner(BaseScanner):
try:
cmd = ['python', '-m', 'flake8', '--format=json'] + files
output = self.run_command(cmd, cwd, timeout=120)
output = self.run_command(cmd, clone_dir, timeout=120)
result['raw_output'] = output.get('stdout', '')
@@ -141,12 +145,15 @@ class PythonScanner(BaseScanner):
try:
issues = json.loads(output['stdout'])
for issue in issues:
# 使用相对于 clone_dir 的路径
full_path = issue.get('filename', '')
rel_path = os.path.relpath(full_path, clone_dir) if full_path else ''
result['issues'].append({
'tool': 'flake8',
'type': self._map_flake8_code(issue.get('code', '')),
'severity': 'Warning',
'message': issue.get('text', ''),
'file': os.path.basename(issue.get('filename', '')),
'file': rel_path,
'line': issue.get('line_number', 0),
'column': issue.get('column_number', 0),
'symbol': issue.get('code', '')