90 lines
2.5 KiB
Python
90 lines
2.5 KiB
Python
"""
|
|
检测结果路由
|
|
"""
|
|
from flask import Blueprint, request, jsonify
|
|
from models.database import db, Detection
|
|
|
|
detection_bp = Blueprint('detections', __name__)
|
|
|
|
|
|
@detection_bp.route('/', methods=['GET'])
|
|
def get_detections():
|
|
"""获取检测结果"""
|
|
website_id = request.args.get('website_id')
|
|
|
|
if website_id:
|
|
detections = Detection.query.filter_by(website_id=website_id).order_by(Detection.created_at.desc()).all()
|
|
else:
|
|
detections = Detection.query.order_by(Detection.created_at.desc()).all()
|
|
|
|
return jsonify([detection.to_dict() for detection in detections])
|
|
|
|
|
|
@detection_bp.route('/', methods=['POST'])
|
|
def create_detection():
|
|
"""保存检测结果"""
|
|
data = request.json
|
|
|
|
detection = Detection(
|
|
website_id=data.get('website_id'),
|
|
status=data.get('status'),
|
|
response_time=data.get('response_time'),
|
|
http_status=data.get('http_status'),
|
|
message=data.get('message')
|
|
)
|
|
|
|
db.session.add(detection)
|
|
db.session.commit()
|
|
|
|
return jsonify(detection.to_dict()), 201
|
|
|
|
|
|
@detection_bp.route('/latest', methods=['GET'])
|
|
def get_latest_detections():
|
|
"""获取每个网站的最新检测结果"""
|
|
from sqlalchemy import func
|
|
subquery = db.session.query(
|
|
Detection.website_id,
|
|
func.max(Detection.id).label('max_id')
|
|
).group_by(Detection.website_id).subquery()
|
|
|
|
detections = db.session.query(Detection).join(
|
|
subquery,
|
|
(Detection.website_id == subquery.c.website_id) &
|
|
(Detection.id == subquery.c.max_id)
|
|
).all()
|
|
|
|
result = {}
|
|
for d in detections:
|
|
result[d.website_id] = d.to_dict()
|
|
|
|
return jsonify(result)
|
|
|
|
|
|
@detection_bp.route('/<int:detection_id>', methods=['GET'])
|
|
def get_detection(detection_id):
|
|
"""获取单个检测结果"""
|
|
detection = Detection.query.get_or_404(detection_id)
|
|
return jsonify(detection.to_dict())
|
|
|
|
|
|
# 为Detection模型添加to_dict方法
|
|
def detection_to_dict(self):
|
|
created_at = self.created_at
|
|
if created_at:
|
|
if created_at.tzinfo is None:
|
|
from datetime import timezone
|
|
created_at = created_at.replace(tzinfo=timezone.utc)
|
|
created_at = created_at.astimezone()
|
|
return {
|
|
'id': self.id,
|
|
'website_id': self.website_id,
|
|
'status': self.status,
|
|
'response_time': self.response_time,
|
|
'http_status': self.http_status,
|
|
'message': self.message,
|
|
'created_at': created_at.isoformat() if created_at else None
|
|
}
|
|
|
|
Detection.to_dict = detection_to_dict
|