25 Commits

Author SHA1 Message Date
279a01b897 Update test_demo/demo_flaws.py 2026-03-15 13:29:58 +08:00
77fd09e6d2 Update test_demo/demo_flaws.py 2026-03-15 12:27:43 +08:00
dangzerong
91c16cbc88 Merge pull request '测试的扫描文件' (#20) from dev into main 2026-03-13 21:04:31 +08:00
Dang Zerong
c8c0ef1620 测试的扫描文件 2026-03-13 21:00:53 +08:00
dangzerong
95831d5190 Merge pull request 'dev' (#19) from dev into main 2026-03-13 18:09:32 +08:00
Dang Zerong
9a14c0b219 测试的扫描文件 2026-03-13 18:00:27 +08:00
Dang Zerong
87b2dacf65 测试的扫描文件 2026-03-13 18:00:22 +08:00
dangzerong
453414efb2 Merge pull request 'dev' (#17) from dev into main 2026-03-13 17:57:36 +08:00
Dang Zerong
04518812f4 Merge branch 'dev' of https://code.deep-pilot.chat/Bosch_Demo/code_scan into dev 2026-03-13 17:42:54 +08:00
Dang Zerong
6c4ee107f9 测试的扫描文件 2026-03-13 17:42:27 +08:00
dangzerong
d11b349d5e Merge pull request '测试的扫描文件' (#15) from dev into main 2026-03-13 17:41:51 +08:00
dangzerong
2a2ff1ad5f Merge branch 'main' into dev 2026-03-13 17:40:33 +08:00
Dang Zerong
bc5a19fffc 测试的扫描文件 2026-03-13 17:39:20 +08:00
Dang Zerong
78655ce5dc 测试的扫描文件 2026-03-13 17:37:46 +08:00
Dang Zerong
2201f6d696 Merge branch 'dev' 2026-03-13 17:37:10 +08:00
Dang Zerong
97881ee00e 测试的扫描文件 2026-03-13 17:32:23 +08:00
dangzerong
e46aff2797 Merge pull request 'dev' (#13) from dev into main 2026-03-13 17:25:28 +08:00
Dang Zerong
887c8ae154 测试的扫描文件 2026-03-13 16:49:13 +08:00
Dang Zerong
ecc39402d5 测试的扫描文件 2026-03-13 16:27:32 +08:00
dangzerong
dc9b921091 Merge pull request '测试的扫描文件' (#11) from dev into main
Reviewed-on: #11
2026-03-13 16:26:54 +08:00
Dang Zerong
a928b79d6d 测试的扫描文件 2026-03-13 16:25:37 +08:00
dangzerong
0991b3de26 Merge pull request 'dev' (#10) from dev into main 2026-03-13 16:24:21 +08:00
Dang Zerong
1876be1777 测试的扫描文件 2026-03-13 16:22:23 +08:00
Dang Zerong
51fc1a6aae 先删除测试代码,后面再提交 2026-03-13 16:21:39 +08:00
Dang Zerong
726c21feac 可演示 2026-03-13 16:04:20 +08:00
12 changed files with 242 additions and 836 deletions

34
Dockerfile Normal file
View File

@@ -0,0 +1,34 @@
# 使用中科大镜像源的 Python 基础镜像
FROM python:3.11.15-slim
# 设置工作目录
WORKDIR /app
# 设置环境变量
ENV PYTHONDONTWRITEBYTECODE=1
ENV PYTHONUNBUFFERED=1
ENV FLASK_RUN_HOST=0.0.0.0
# 安装系统依赖(用于 git 等工具)
RUN apt-get update && apt-get install -y --no-install-recommends \
git \
&& rm -rf /var/lib/apt/lists/*
# 复制依赖文件
COPY requirements.txt .
# 安装 Python 依赖(显式安装,避免缓存导致遗漏)
RUN pip install --no-cache-dir -r requirements.txt \
&& pip install --no-cache-dir "GitPython>=3.1.0" "gitdb>=4.0.1" "smmap>=3.0.1"
# 复制应用代码
COPY . .
# 创建报告目录
RUN mkdir -p reports
# 暴露端口
EXPOSE 5000
# 启动应用
CMD ["python", "app.py"]

285
README.md
View File

@@ -1,224 +1,109 @@
# AI Code Quality Scanner - 飞书通知版
# AI 代码质量扫描系统
一个自动化代码质量扫描系统,在代码提交时自动扫描并发送报告到飞书
自动化代码质量扫描工具,监听 PR 事件,自动扫描代码缺陷并提供合并决策支持
## 功能特性
- 🤖 自动监听 Gitea 代码提交事件
- 🔍 多维度代码质量扫描(语法、风格、安全)
- 📊 生成 Markdown 格式扫描报告
- 📱 实时推送飞书机器人通知
## 系统架构
## 工作流程
```
┌───────────── Webhook ┌──────────────────┐
Gitea │ ───────────────► │ Webhook Server
│ 代码仓库 │ (Flask)
└─────────────┘ └────────┬─────────┘
┌──────────────────┐
│ Code Scanner
│ - ESLint
│ - Pylint │
- SonarQube │
└────────┬─────────┘
┌──────────────────
│ Report Generator│
│ - Markdown │
└────────┬─────────┘
┌──────────────────┐
│ Feishu Bot │
│ - Webhook │
└──────────────────┘
┌──────────┐ 1. 创建 PR ┌────────────┐
│ Gitea │ ───────────────► │ Webhook │
└──────────┘Server
└───────────┘
│ 2. 拉取代码、扫描、存库
────────────┐
│ SQLite
│ Database
└────────────┘
│ 3. 前端查询
┌────────────┐
│ 前端页面
────────────
```
## 三个核心功能
### 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 buildx build --load --push -t dcr-by1jwyxk44.71826370.xyz/whlaoding/code-scan:latest .
```
### 2. 登录仓库
```bash
docker login dcr-by1jwyxk44.71826370.xyz
```
### 3. Push 到仓库
```bash
docker run -d --name code-scan -p 5000:5000 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(

View File

@@ -49,14 +49,13 @@ ai:
# AI 审查器配置
# 支持: "ollama" (本地) 或 "api" (在线API)
provider: "api"
# 模型名称(硅基流动可用模型)- Qwen 最强语言模型
model: "deepseek-ai/DeepSeek-V3.2"
# 模型名称(阿里云通义千问)
model: "qwen3.5-plus"
# API 地址
# 硅基流动: https://api.siliconflow.cn/v1
api_url: "https://api.siliconflow.cn/v1"
api_url: "https://dashscope.aliyuncs.com/compatible-mode/v1"
# API 密钥
api_key: "sk-cqxhnsxdxaalxlykfkjksyinjftdyejnblmgkfxmhwmmvdyu"
api_key: "sk-616332b2afa94699b4572d0fe6ac370a"
# 是否启用 AI 审查
enabled: true
# 每次审查的最大代码行数
max_lines: 200
max_lines: 100

9
docker-compose.yml Normal file
View File

@@ -0,0 +1,9 @@
version: "3.8"
services:
code-scan:
image: dcr-by1jwyxk44.71826370.xyz/whlaoding/code-scan:latest
container_name: code-scan
ports:
- "5000:5000"
restart: unless-stopped

View File

@@ -109,7 +109,7 @@ class GiteaClient:
timeout=30
)
if response.status_code == 200:
if response.status_code in (200, 201):
logger.info(f"成功关闭 PR #{pr_number}")
return True
else:

24
pyproject.toml Normal file
View File

@@ -0,0 +1,24 @@
[build-system]
requires = ["setuptools>=45", "wheel"]
build-backend = "setuptools.build_meta"
[project]
name = "code-scan"
version = "1.0.0"
description = "代码扫描工具"
readme = "README.md"
requires-python = ">=3.8"
dependencies = [
"flask>=2.0.0",
"pyyaml>=5.0",
"requests>=2.25.0",
"python-dotenv>=0.19.0",
"GitPython>=3.1.0",
]
[project.scripts]
code-scan = "app:main"
[tool.setuptools.packages.find]
where = ["."]
include = ["scanner*"]

View File

@@ -2,3 +2,9 @@ flask>=2.0.0
pyyaml>=5.0
requests>=2.25.0
python-dotenv>=0.19.0
GitPython>=3.1.0
gitdb>=4.0.1
smmap>=3.0.1
pylint>=2.17.0
flake8>=6.0.0
bandit>=1.7.0

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
return self._call_api(prompt)
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
}
response = requests.post(url, json=payload, headers=headers, timeout=120)
logger.info(f"调用 API: {url}, model={self.model}")
if response.status_code == 200:
result = response.json()
content = result['choices'][0]['message']['content']
parsed = self._extract_json_obj(content)
return parsed
try:
response = requests.post(url, json=payload, headers=headers, timeout=120)
logger.warning(f'API 返回错误: {response.status_code}')
return None
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}, {response.text[:200]}')
return None
except Exception as e:
logger.warning(f'API 调用失败: {e}')
return None

218
test.py
View File

@@ -1,218 +0,0 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
import os
import logging
os.environ.setdefault('FLASK_RUN_HOST', '0.0.0.0')
from flask import Flask, request, jsonify
import yaml
from webhook.handler import GiteaWebhookHandler
from scanner.python_scanner import PythonScanner
from scanner.js_scanner import JavaScriptScanner
from scanner.security_scanner import SecurityScanner
from report.generator import ReportGenerator
from notify.feishu import FeishuNotifier
# 配置日志
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)
# 加载配置
def load_config():
"""加载配置文件"""
config_path = os.path.join(os.path.dirname(__file__), 'config.yaml')
with open(config_path, 'r', encoding='utf-8') as f:
return yaml.safe_load(f)
# 全局配置
config = load_config()
# 初始化应用
app = Flask(__name__)
app.config['SECRET_KEY'] = config.get('server', {}).get('secret_key', 'dev-secret-key')
# 初始化组件
webhook_handler = GiteaWebhookHandler(config['gitea'])
python_scanner = PythonScanner(config.get('scanner', {}))
js_scanner = JavaScriptScanner(config.get('scanner', {}))
security_scanner = SecurityScanner(config.get('scanner', {}))
report_generator = ReportGenerator(config.get('report', {}))
feishu_notifier = FeishuNotifier(config['feishu'])
@app.route('/')
def index():
"""健康检查接口"""
return jsonify({
'status': 'ok',
'service': 'AI Code Quality Scanner',
'version': '1.0.0'
})
@app.route('/webhook/gitea', methods=['POST'])
def handle_gitea_webhook():
"""处理 Gitea Webhook 请求"""
try:
# 验证签名
signature = request.headers.get('X-Gitea-Signature')
if signature:
if not webhook_handler.verify_signature(
request.data,
signature,
config['gitea']['webhook_secret']
):
logger.warning('Webhook 签名验证失败')
return jsonify({'error': 'Invalid signature'}), 401
# 解析 Webhook payload
payload = request.json
if not payload:
return jsonify({'error': 'No payload'}), 400
event_type = request.headers.get('X-Gitea-Event', 'push')
logger.info(f'收到 Gitea Webhook 事件: {event_type}')
# 只处理 push 事件
if event_type != 'push':
return jsonify({'message': 'Event ignored'}), 200
# 提取提交信息
commits = payload.get('commits', [])
if not commits:
return jsonify({'message': 'No commits'}), 200
repo = payload.get('repository', {})
repo_name = repo.get('full_name', 'unknown')
branch = payload.get('ref', '').replace('refs/heads/', '')
pusher = payload.get('pusher', {}).get('name', 'unknown')
logger.info(f'处理仓库 {repo_name}{len(commits)} 个提交')
# 处理每个提交
for commit in commits:
commit_id = commit.get('id', '')[:8]
commit_message = commit.get('message', '')
author = commit.get('author', {}).get('name', 'unknown')
logger.info(f'扫描提交 {commit_id}: {commit_message}')
try:
# 获取仓库 URL
clone_url = repo.get('clone_url')
if not clone_url:
# 尝试从 web_url 构建
web_url = repo.get('web_url', '')
if web_url:
clone_url = web_url.replace('http://', 'http://').replace('https://', 'https://')
clone_url = clone_url.rstrip('/') + '.git'
# 执行代码扫描
scan_results = {}
# Python 扫描
if 'python' in config.get('scanner', {}).get('languages', []):
scan_results['python'] = python_scanner.scan(
clone_url, commit_id, branch
)
# JavaScript/TypeScript 扫描
if any(lang in config.get('scanner', {}).get('languages', [])
for lang in ['javascript', 'typescript']):
scan_results['javascript'] = js_scanner.scan(
clone_url, commit_id, branch
)
# 安全扫描
scan_results['security'] = security_scanner.scan(
clone_url, commit_id, branch
)
# 生成报告
report = report_generator.generate(
repo_name=repo_name,
branch=branch,
commit_id=commit_id,
commit_message=commit_message,
author=author,
scan_results=scan_results
)
# 发送飞书通知
feishu_notifier.send_report(report)
logger.info(f'提交 {commit_id} 扫描完成')
except Exception as e:
logger.error(f'扫描提交 {commit_id} 失败: {str(e)}')
# 继续处理其他提交
continue
return jsonify({'status': 'ok', 'message': 'Scan completed'}), 200
except Exception as e:
logger.error(f'处理 Webhook 失败: {str(e)}', exc_info=True)
return jsonify({'error': str(e)}), 500
@app.route('/scan/manual', methods=['POST'])
def manual_scan():
"""手动触发扫描接口"""
try:
data = request.json
repo_url = data.get('repo_url')
branch = data.get('branch', 'main')
commit_id = data.get('commit_id')
if not repo_url:
return jsonify({'error': 'repo_url is required'}), 400
# 执行扫描
scan_results = {}
if 'python' in config.get('scanner', {}).get('languages', []):
scan_results['python'] = python_scanner.scan(repo_url, commit_id, branch)
if any(lang in config.get('scanner', {}).get('languages', [])
for lang in ['javascript', 'typescript']):
scan_results['javascript'] = js_scanner.scan(repo_url, commit_id, branch)
scan_results['security'] = security_scanner.scan(repo_url, commit_id, branch)
# 生成报告
report = report_generator.generate(
repo_name=repo_url.split('/')[-1].replace('.git', ''),
branch=branch,
commit_id=commit_id or 'manual',
commit_message='Manual scan',
author='manual',
scan_results=scan_results
)
# 发送飞书通知
feishu_notifier.send_report(report)
return jsonify({
'status': 'ok',
'report': report
}), 200
except Exception as e:
logger.error(f'手动扫描失败: {str(e)}', exc_info=True)
return jsonify({'error': str(e)}), 500
if __name__ == '__main__':
# 强制监听所有网络接口
host = "0.0.0.0"
port = config.get('server', {}).get('port', 5000)
debug = config.get('server', {}).get('debug', True)
logger.info(f'启动服务: {host}:{port}')
app.run(host=host, port=port, debug=debug)

View File

@@ -25,6 +25,32 @@ def unused_variable_demo():
print("Function executed")
# 缺陷8: 行太长(风格问题)
def long_line():
"""这是一行非常非常非常非常非常非常非常非常非常非常非常非常长的代码超过了 120 个字符的限制"""
# 缺陷9: 缺少空格
def missing_spaces():
"""缺少必要空格"""
x=1+2
y=3*99
if x==1:
print(x)
# 缺陷1: 未使用的导入
import unused_module # 未使用
import collections as col # 使用了 col 但 flake8 可能检测
# 缺陷2: 未使用的变量
# def unused_variable_demo():
# """演示未使用的变量"""
# result = calculate() # result 未被使用
# print("Function executed")
def calculate():
"""计算并返回结果"""
return 42
@@ -41,227 +67,3 @@ def use_before_define():
"""在定义前使用变量"""
print(before_var) # before_var 在下面才定义
before_var = 100
# 缺陷5: 硬编码密码(安全问题)
def connect_database():
"""连接数据库"""
password = "admin123" # 硬编码密码
username = "root"
return f"Connecting with {username}:{password}"
# 缺陷6: 使用 eval安全问题
def unsafe_eval():
"""危险使用 eval"""
user_input = "os.system('ls')"
result = eval(user_input) # 危险!
return result
# 缺陷7: 使用 pickle 反序列化(安全问题)
def unsafe_pickle():
"""不安全的 pickle 反序列化"""
data = b"..." # 模拟恶意数据
obj = pickle.loads(data) # 危险!
# 缺陷8: 行太长(风格问题)
def long_line():
"""这是一行非常非常非常非常非常非常非常非常非常非常非常非常长的代码超过了 120 个字符的限制"""
# 缺陷9: 缺少空格
def missing_spaces():
"""缺少必要空格"""
x=1+2
y=3*4
if x==1:
print(x)
# 缺陷10: 多余空格
def extra_spaces():
"""多余空格"""
x = 1
y = 2
# 缺陷11: 未捕获的异常
def unhandled_exception():
"""捕获异常后未处理"""
try:
result = 10 / 0
except ZeroDivisionError:
pass # 捕获但未处理
# 缺陷12: 过于宽泛的异常
def broad_exception():
"""捕获所有异常"""
try:
data = json.loads('{"key": "value"}')
except Exception:
pass
# 缺陷13: 裸 except 子句
def bare_except():
"""使用裸 except"""
try:
x = int("abc")
except:
pass
# 缺陷14: 重复代码
def duplicate_code():
"""重复代码示例"""
a = 1
b = 2
c = a + b
print(c)
a = 3
b = 4
c = a + b
print(c)
# 缺陷15: 变量名与内置函数冲突
def shadow_builtin():
"""变量名覆盖内置函数"""
list = [1, 2, 3] # 覆盖内置 list
dict = {} # 覆盖内置 dict
str = "hello" # 覆盖内置 str
return list, dict, str
# 缺陷16: 不必要的 pass
def unnecessary_pass():
"""不必要的 pass"""
if True:
pass # 可以直接删除
# 缺陷17: 使用 + 进行字符串拼接(推荐用 join
def string_concat():
"""低效字符串拼接"""
result = ""
for i in range(100):
result = result + str(i)
return result
# 缺陷18: 在循环中修改集合
def modify_during_iteration():
"""在迭代时修改列表"""
items = [1, 2, 3, 4, 5]
for item in items:
if item % 2 == 0:
items.remove(item) # 在迭代时修改
# 缺陷19: 全局变量
global_counter = 0 # 全局变量
def increment():
global global_counter # 依赖全局变量
global_counter += 1
# 缺陷20: 魔法数字
def calculate_price():
"""使用魔法数字"""
price = 100
tax = price * 1.1 # 1.1 是什么?
discount = price * 0.9
return tax, discount
# 缺陷21: 函数参数过多
def bad_function(a, b, c, d, e, f, g, h):
"""参数过多的函数"""
return a + b + c + d + e + f + g + h
# 缺陷22: 空函数体
def empty_function():
"""空函数应该使用 pass 或文档字符串"""
pass
# 缺陷23: 使用 time.sleep 测试
def bad_sleep():
"""生产代码中使用 time.sleep"""
import time
time.sleep(5) # 阻塞
# 缺陷24: 注释掉的代码
def commented_code():
# print("This is commented out")
pass
# 缺陷25: TODO/FIXME 注释
def todo_comment():
# TODO: Implement this
# FIXME: This is broken
pass
# 缺陷26: 导入顺序错误(应先标准库,再第三方,本地)
import sys # 标准库
import flask # 第三方
from . import local # 本地
# 缺陷27: 不必要的列表推导式
def unnecessary_list_comp():
"""不必要的列表推导式"""
result = [x for x in range(10)] # 可简化为 list(range(10))
return result
# 缺陷28: 条件表达式中的赋值
def assignment_in_condition():
"""在条件中赋值(不推荐)"""
if (x := get_value()) > 0: # 海象运算符但可能难以阅读
print(x)
def get_value():
return 5
# 缺陷29: 比较布尔值
def compare_bool():
"""与布尔值比较"""
flag = True
if flag == True: # 应直接用 if flag:
print("yes")
# 缺陷30: 使用 hasattr/getattr 而非异常处理
def use_hasattr():
"""滥用 hasattr"""
class Foo:
pass
obj = Foo()
if hasattr(obj, 'bar'): # 可直接用 try/except
print(obj.bar)
# 主函数入口
def main():
"""主函数"""
connect_database()
unsafe_eval()
unsafe_pickle()
print("Demo executed")
if __name__ == "__main__":
main()

View File

@@ -97,30 +97,9 @@
</li>
</ul>
<h6 class="mt-4 mb-2 text-uppercase" style="color: rgba(255,255,255,0.5); font-size: 11px;">AI 智能分析</h6>
<ul class="nav flex-column">
<li class="nav-item">
<a class="nav-link" href="#" onclick="showPage('ai-quality')">
<i class="bi bi-stars me-2"></i>AI 质量评分
</a>
</li>
<li class="nav-item">
<a class="nav-link" href="#" onclick="showPage('ai-insights')">
<i class="bi bi-lightbulb me-2"></i>智能洞察
</a>
</li>
</ul>
<ul class="nav flex-column mt-3">
</ul>
<div class="mt-4 p-3" style="background: linear-gradient(135deg, #667eea 0%, #764ba2 100%); border-radius: 8px;">
<small><i class="bi bi-cpu me-1"></i> AI 引擎</small>
<div class="mt-2 small">基于大模型智能分析代码质量</div>
<div class="mt-2">
<span class="badge bg-success">在线</span>
</div>
</div>
</div>
<!-- 主内容区 -->
@@ -289,126 +268,6 @@
</div>
<!-- 设置页面 -->
<!-- AI 质量评分页面 -->
<div id="page-ai-quality" style="display:none;">
<h2 class="mb-4"><i class="bi bi-stars text-primary"></i> AI 质量评分</h2>
<div class="alert alert-info py-2 mb-3">
<i class="bi bi-info-circle"></i> 基于 AI 大模型对 PR 代码进行多维度质量评估
</div>
<div class="row mb-3">
<div class="col-md-3">
<div class="card text-center bg-gradient-primary text-white">
<div class="card-body">
<div class="display-4 fw-bold" id="aiq-total">--</div>
<small>综合评分</small>
</div>
</div>
</div>
<div class="col-md-3">
<div class="card text-center">
<div class="card-body">
<div class="h3 fw-bold text-success" id="aiq-security">--</div>
<small class="text-muted">安全性</small>
</div>
</div>
</div>
<div class="col-md-3">
<div class="card text-center">
<div class="card-body">
<div class="h3 fw-bold text-warning" id="aiq-maintain">--</div>
<small class="text-muted">可维护性</small>
</div>
</div>
</div>
<div class="col-md-3">
<div class="card text-center">
<div class="card-body">
<div class="h3 fw-bold text-info" id="aiq-readability">--</div>
<small class="text-muted">可读性</small>
</div>
</div>
</div>
</div>
<div class="card">
<div class="card-header">
<h5 class="mb-0">评分说明</h5>
</div>
<div class="card-body">
<ul class="list-group list-group-flush">
<li class="list-group-item">
<strong class="text-success">安全性 (35%)</strong> - 检测 SQL 注入、XSS、密码泄露等安全风险
</li>
<li class="list-group-item">
<strong class="text-warning">可维护性 (30%)</strong> - 代码复杂度、重复代码、硬编码等问题
</li>
<li class="list-group-item">
<strong class="text-info">可读性 (15%)</strong> - 命名规范、注释、代码风格
</li>
<li class="list-group-item">
<strong class="text-primary">最佳实践 (20%)</strong> - 遵循语言最佳实践和设计模式
</li>
</ul>
</div>
</div>
</div>
<!-- AI 智能洞察页面 -->
<div id="page-ai-insights" style="display:none;">
<h2 class="mb-4"><i class="bi bi-lightbulb text-warning"></i> AI 智能洞察</h2>
<div class="row mb-4">
<div class="col-md-4">
<div class="card border-primary">
<div class="card-body text-center">
<i class="bi bi-robot display-4 text-primary"></i>
<h5 class="mt-3">智能分析</h5>
<p class="text-muted small">基于 AI 大模型深度分析代码问题,提供精准修复建议</p>
</div>
</div>
</div>
<div class="col-md-4">
<div class="card border-success">
<div class="card-body text-center">
<i class="bi bi-lightning display-4 text-success"></i>
<h5 class="mt-3">自动修复</h5>
<p class="text-muted small">一键生成修复代码,直接应用到项目中</p>
</div>
</div>
</div>
<div class="col-md-4">
<div class="card border-info">
<div class="card-body text-center">
<i class="bi bi-graph-up display-4 text-info"></i>
<h5 class="mt-3">趋势预测</h5>
<p class="text-muted small">分析历史数据,预测代码质量变化趋势</p>
</div>
</div>
</div>
</div>
<div class="card">
<div class="card-header">AI 能力展示</div>
<div class="card-body">
<div class="row">
<div class="col-md-6">
<h6><i class="bi bi-check2-all text-success"></i> 已支持功能</h6>
<ul class="list-unstyled ms-3">
<li><i class="bi bi-check text-success me-2"></i>多维度代码质量评分</li>
<li><i class="bi bi-check text-success me-2"></i>问题根因分析</li>
<li><i class="bi bi-check text-success me-2"></i>智能修复建议生成</li>
<li><i class="bi bi-check text-success me-2"></i>历史趋势分析</li>
</ul>
</div>
<div class="col-md-6">
<h6><i class="bi bi-gear text-primary"></i> 扫描器类型</h6>
<ul class="list-unstyled ms-3">
<li><i class="bi bi-code-slash me-2"></i>Python 代码分析</li>
<li><i class="bi bi-code-slash me-2"></i>JavaScript/TypeScript 分析</li>
<li><i class="bi bi-shield-check me-2"></i>安全漏洞检测</li>
<li><i class="bi bi-stars me-2"></i>AI 智能审查</li>
</ul>
</div>
</div>
</div>
</div>
</div>
</div>
@@ -660,7 +519,6 @@
if (page === 'dashboard') loadDashboard();
if (page === 'prs') loadPRs();
if (page === 'ai-quality') loadAIQualityOverview();
}
// 加载概览数据