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

Merge branch 'dev'

Browse Source
This commit is contained in:
Dang Zerong
2026-03-13 17:37:10 +08:00
parent e46aff2797 97881ee00e
commit 2201f6d696
4 changed files with 138 additions and 246 deletions
Download Patch File Download Diff File Expand all files Collapse all files

281
README.md
View File

@@ -1,224 +1,109 @@
# AI Code Quality Scanner - 飞书通知版
# AI 代码质量扫描系统
一个自动化代码质量扫描系统，在代码提交时自动扫描并发送报告到飞书。
自动化代码质量扫描工具，监听 PR 事件，自动扫描代码缺陷并提供合并决策支持。
## 功能特性
- 🤖 自动监听 Gitea 代码提交事件
- 🔍 多维度代码质量扫描（语法、风格、安全）
- 📊 生成 Markdown 格式扫描报告
- 📱 实时推送飞书机器人通知
## 系统架构
## 工作流程
```
┌─────────────┐ Webhook ┌──────────────────┐
│ Gitea │ ───────────────► │ Webhook Server │
│ 代码仓库 │ │ (Flask) │
└─────────────┘ └────────┬─────────┘
│
┌──────────┐ 1. 创建 PR ┌────────────┐
│ Gitea │ ───────────────► │ Webhook │
└──────────┘ │ Server │
└─────┬──────┘
│ 2. 拉取代码、扫描、存库
▼
┌──────────────────┐
│ Code Scanner │
│ - ESLint │
│ - Pylint │
│ - SonarQube │
└────────┬─────────┘
│
┌────────────┐
│ SQLite │
│ Database │
└────────────┘
│ 3. 前端查询
▼
┌──────────────────┐
│ Report Generator│
│ - Markdown │
└────────┬─────────┘
│
▼
┌──────────────────┐
│ Feishu Bot │
│ - Webhook │
└──────────────────┘
┌────────────┐
│ 前端页面 │
└────────────┘
```
## 三个核心功能
### 1. PR 创建
- Gitea 仓库创建 PR 时自动触发扫描
- 支持事件：`opened`、`reopened`、`synchronize`
### 2. 后端处理
- 拉取 PR 对应的代码
- 执行代码扫描（Python/JavaScript/TypeScript）
- AI 智能审查代码缺陷
- 扫描结果存入 SQLite 数据库
### 3. 前端功能
- 查询所有 PR 及扫描状态
- 查看每个 PR 的缺陷详情
- 一键「拒绝合并」或「同意合并」
## 快速开始
### 1. 安装依赖
```bash
# 安装依赖
pip install -r requirements.txt
```
### 2. 配置飞书机器人
1. 打开飞书群聊 → 设置 → 群机器人
2. 添加机器人 → 选择"自定义机器人"
3. 获取 Webhook 地址
4. 配置 `config.yaml`
### 3. 配置 Gitea Webhook
#### 方式一：Push 时扫描（原有方式）
1. 进入 Gitea 仓库 → 设置 → Webhooks
2. 添加 Webhook：
- 目标 URL: `http://你的服务器IP:5000/webhook/gitea`
- 触发事件: Push
- 密钥: 配置 `config.yaml` 中的 secret
#### 方式二：PR 创建时扫描（推荐）
1. 进入 Gitea 仓库 → 设置 → Webhooks
2. 添加 Webhook：
- 目标 URL: `http://你的服务器IP:5000/webhook/gitea`
- 触发事件: Pull Request
- 密钥: 配置 `config.yaml` 中的 secret
**支持的 PR 事件：**
- `opened` - 创建新 PR
- `reopened` - 重新打开 PR
- `synchronize` - PR 中的提交有更新
- `ready_for_review` - PR 标记为准备好审查
### 4. 运行服务
```bash
# 运行服务
python app.py
```
## 配置说明
访问 http://localhost:5000 查看前端页面。
所有配置在 `config.yaml` 中：
## Docker 部署
### 1. 构建镜像
```bash
docker build -t dcr-by1jwyxk44.71826370.xyz/whlaoding/code-scan:latest .
```
### 2. 登录仓库
```bash
docker login dcr-by1jwyxk44.71826370.xyz
```
### 3. Push 到仓库
```bash
docker push dcr-by1jwyxk44.71826370.xyz/whlaoding/code-scan:latest
```
### 4. 使用 docker compose 启动
```bash
# 启动服务
docker compose up -d
# 查看日志
docker compose logs -f
# 停止服务
docker compose down
```
## 配置
配置文件 `config.yaml`：
```yaml
server:
host: "0.0.0.0"
port: 5000
debug: true
gitea:
base_url: "http://localhost:3000"
# Webhook 签名密钥
webhook_secret: "your_webhook_secret"
base_url: "https://code.deep-pilot.chat"
webhook_secret: "xxx"
api_token: "xxx"
feishu:
# 飞书机器人 Webhook 地址
webhook_url: "https://open.feishu.cn/open-apis/bot/v2/hook/xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx"
# 消息推送 secret（可选，用于签名）
secret: "your_feishu_secret"
scanner:
# 支持的语言
languages:
- python
- javascript
- typescript
# 扫描阈值
max_issues: 10
# 是否启用详细扫描
detailed: true
report:
# 报告保存路径
output_dir: "./reports"
# 是否保留报告文件
keep_files: true
ai:
provider: "api"
model: "qwen3.5-plus"
api_url: "https://dashscope.aliyuncs.com/compatible-mode/v1"
api_key: "sk-xxx"
```
## 项目结构
```
code-scanner/
├── app.py # 主应用入口
├── config.yaml # 配置文件
├── requirements.txt # Python 依赖
├── README.md # 项目说明
├── scanner/
│ ├── __init__.py
│ ├── base.py # 扫描器基类
│ ├── python_scanner.py # Python 代码扫描
│ ├── js_scanner.py # JavaScript/TypeScript 扫描
│ └── security_scanner.py # 安全扫描
├── report/
│ ├── __init__.py
│ └── generator.py # Markdown 报告生成
├── notify/
│ ├── __init__.py
│ └── feishu.py # 飞书通知
├── webhook/
│ ├── __init__.py
│ └── handler.py # Webhook 处理
└── reports/ # 报告输出目录
```
## 支持的扫描工具
### Python
- **Pylint** - 代码风格和错误检查
- **Flake8** - Python 代码检查
- **Bandit** - 安全漏洞扫描
### JavaScript/TypeScript
- **ESLint** - JavaScript/TypeScript 检查
- **Prettier** - 代码格式化
## 飞书消息效果
扫描完成后，将收到类似以下消息：
### Push 扫描消息
```
📊 代码质量扫描报告
仓库: my-project
分支: main
提交: abc1234
提交者: developer@example.com
✅ 扫描通过 (0 issues)
或
⚠️ 发现问题 (5 issues)
```
### PR 扫描消息
```
📊 PR 代码质量扫描报告
仓库: my-project
源分支: feature-xxx → 目标分支: main
PR链接: https://gitea.example.com/user/project/pulls/123
提交: abc1234
提交者: developer@example.com
✅ 扫描通过 (0 issues)
或
⚠️ 发现问题 (5 issues)
```
## Docker 部署
```dockerfile
FROM python:3.11-slim
WORKDIR /app
COPY requirements.txt .
RUN pip install -r requirements.txt
COPY . .
EXPOSE 5000
CMD ["python", "app.py"]
```
## 环境变量
也可以通过环境变量配置：
```bash
export FEISHU_WEBHOOK_URL="https://open.feishu.cn/..."
export GITEA_WEBHOOK_SECRET="secret"
export SCANNER_MAX_ISSUES=10
```
## 许可证
MIT License

21
app.py
View File

@@ -2,6 +2,7 @@
# -*- coding: utf-8 -*-
import os
import time
import logging
import traceback
from typing import Dict, Tuple, Any
@@ -131,21 +132,27 @@ def handle_gitea_webhook():
# Python 扫描
if 'python' in config.get('scanner', {}).get('languages', []):
start_time = time.time()
scan_results['python'] = python_scanner.scan(
clone_url, commit_id, branch
)
logger.info(f"[TIMER] Python 扫描耗时: {time.time() - start_time:.2f}秒")
# JavaScript/TypeScript 扫描
if any(lang in config.get('scanner', {}).get('languages', [])
for lang in ['javascript', 'typescript']):
start_time = time.time()
scan_results['javascript'] = js_scanner.scan(
clone_url, commit_id, branch
)
logger.info(f"[TIMER] JavaScript 扫描耗时: {time.time() - start_time:.2f}秒")
# 安全扫描
start_time = time.time()
scan_results['security'] = security_scanner.scan(
clone_url, commit_id, branch
)
logger.info(f"[TIMER] 安全扫描耗时: {time.time() - start_time:.2f}秒")
# 生成报告
report = report_generator.generate(
@@ -228,27 +235,35 @@ def handle_pull_request(payload: Dict[str, Any]) -> Tuple[Dict, int]:
# Python 扫描
if 'python' in config.get('scanner', {}).get('languages', []):
start_time = time.time()
scan_results['python'] = python_scanner.scan(
clone_url, source_sha, source_branch, changed_files
)
logger.info(f"[TIMER] Python 扫描耗时: {time.time() - start_time:.2f}秒")
# JavaScript/TypeScript 扫描
if any(lang in config.get('scanner', {}).get('languages', [])
for lang in ['javascript', 'typescript']):
start_time = time.time()
scan_results['javascript'] = js_scanner.scan(
clone_url, source_sha, source_branch, changed_files
)
logger.info(f"[TIMER] JavaScript 扫描耗时: {time.time() - start_time:.2f}秒")
# 安全扫描
start_time = time.time()
scan_results['security'] = security_scanner.scan(
clone_url, source_sha, source_branch, changed_files
)
logger.info(f"[TIMER] 安全扫描耗时: {time.time() - start_time:.2f}秒")
# AI 代码审查
if config.get('ai', {}).get('enabled', False):
start_time = time.time()
scan_results['ai'] = ai_reviewer.scan(
clone_url, source_sha, source_branch, changed_files
)
logger.info(f"[TIMER] AI 扫描耗时: {time.time() - start_time:.2f}秒")
# 获取 PR 的代码差异，用于将问题与代码片段关联
pr_diff = None
@@ -326,13 +341,19 @@ def manual_scan():
scan_results = {}
if 'python' in config.get('scanner', {}).get('languages', []):
start_time = time.time()
scan_results['python'] = python_scanner.scan(repo_url, commit_id, branch)
logger.info(f"[TIMER] Python 扫描耗时: {time.time() - start_time:.2f}秒")
if any(lang in config.get('scanner', {}).get('languages', [])
for lang in ['javascript', 'typescript']):
start_time = time.time()
scan_results['javascript'] = js_scanner.scan(repo_url, commit_id, branch)
logger.info(f"[TIMER] JavaScript 扫描耗时: {time.time() - start_time:.2f}秒")
start_time = time.time()
scan_results['security'] = security_scanner.scan(repo_url, commit_id, branch)
logger.info(f"[TIMER] 安全扫描耗时: {time.time() - start_time:.2f}秒")
# 生成报告
report = report_generator.generate(

2
config.yaml
View File

@@ -58,4 +58,4 @@ ai:
# 是否启用 AI 审查
enabled: true
# 每次审查的最大代码行数
max_lines: 200
max_lines: 100

60
scanner/ai_reviewer.py
View File

@@ -30,7 +30,7 @@ class AIReviewer(BaseScanner):
self.config = config
self.enabled = config.get('enabled', True)
self.provider = config.get('provider', 'ollama')
self.provider = config.get('provider', 'api')
self.model = config.get('model', 'llama3')
self.api_url = config.get('api_url', 'http://localhost:11434')
self.api_key = config.get('api_key', '')
@@ -424,13 +424,7 @@ class AIReviewer(BaseScanner):
def _call_ai(self, prompt: str) -> Optional[Dict[str, Any]]:
"""调用 AI 服务"""
try:
if self.provider == 'ollama':
return self._call_ollama(prompt)
elif self.provider == 'api':
return self._call_api(prompt)
else:
logger.warning(f'未知的 AI provider: {self.provider}')
return None
except Exception as e:
print("异常追踪信息:", e.__traceback__)
logger.error(f'AI 调用失败: {str(e)}')
@@ -517,32 +511,6 @@ class AIReviewer(BaseScanner):
logger.debug("_extract_json_obj: 未能提取到有效的 JSON 对象")
return None
def _call_ollama(self, prompt: str) -> Optional[Dict[str, Any]]:
"""调用 Ollama 本地模型"""
import requests
url = f"{self.api_url}/api/generate"
payload = {
"model": self.model,
"prompt": prompt,
"stream": False,
"format": "json"
}
logger.info(f"调用 Ollama: {url}, model={self.model}")
response = requests.post(url, json=payload, timeout=120)
if response.status_code == 200:
result = response.json()
content = result.get('response', '')
logger.info(f"Ollama 返回内容长度: {len(content) if content else 0}")
logger.debug(f"Ollama 返回内容预览: {content[:200] if content else 'empty'}")
parsed = self._extract_json_obj(content)
return parsed
logger.warning(f'Ollama 返回错误: {response.status_code}')
return None
def _call_api(self, prompt: str) -> Optional[Dict[str, Any]]:
"""调用在线 API"""
import requests
@@ -560,7 +528,7 @@ class AIReviewer(BaseScanner):
payload = {
"model": self.model,
"messages": [{"role": "user", "content": prompt}],
"max_tokens": 1024*5,
"max_tokens": 1024,
"temperature": 0.7
}
elif 'deepseek' in self.api_url:
@@ -568,25 +536,43 @@ class AIReviewer(BaseScanner):
payload = {
"model": self.model,
"messages": [{"role": "user", "content": prompt}],
"max_tokens": 1024*5,
"max_tokens": 1024,
"temperature": 0.7
}
elif 'dashscope' in self.api_url:
# 阿里云 dashscope 专用端点
url = f"{self.api_url}/chat/completions"
payload = {
"model": self.model,
"messages": [{"role": "user", "content": prompt}],
"max_tokens": 1024,
"temperature": 0.7,
"stream": False # 显式关闭流式
}
else:
url = f"{self.api_url}/chat/completions"
payload = {
"model": self.model,
"messages": [{"role": "user", "content": prompt}],
"max_tokens": 1024*5,
"max_tokens": 1024,
"temperature": 0.7
}
logger.info(f"调用 API: {url}, model={self.model}")
try:
response = requests.post(url, json=payload, headers=headers, timeout=120)
if response.status_code == 200:
result = response.json()
content = result['choices'][0]['message']['content']
logger.info(f"API 返回内容长度: {len(content) if content else 0}")
parsed = self._extract_json_obj(content)
return parsed
logger.warning(f'API 返回错误: {response.status_code}')
logger.warning(f'API 返回错误: {response.status_code}, {response.text[:200]}')
return None
except Exception as e:
logger.warning(f'API 调用失败: {e}')
return None