Swagger UI added + merge

Author: MonsefRH

backend/app.py

@@ -17,6 +17,7 @@
 from routes.full_scan_routes import full_scan_bp
 from routes.t5_base import t5_base_bp
 from routes.admin import admin_bp
+from routes.report_routes import report_bp
 
 from flasgger import Swagger
 from utils.api_spec import swagger_template, swagger_config
@@ -55,6 +56,7 @@
 app.register_blueprint(full_scan_bp, url_prefix="/")
 app.register_blueprint(t5_base_bp, url_prefix="/")
 app.register_blueprint(admin_bp, url_prefix="/")
+app.register_blueprint(report_bp, url_prefix="/")
 
 if __name__ == "__main__" :
     app.run(debug=True, port=5000)

backend/models/notification.py

@@ -0,0 +1,22 @@
+from utils.db import db
+from datetime import datetime, timezone
+
+class Notification(db.Model):
+    __tablename__ = "notifications"
+
+    id = db.Column(db.Integer, primary_key=True)
+    user_id = db.Column(db.Integer, nullable=False, index=True)
+    scan_id = db.Column(db.Integer, nullable=False, index=True)
+    repo = db.Column(db.String(255), nullable=False)
+    status = db.Column(db.String(32), nullable=False)  # SUCCESS / FAILED
+    findings_count = db.Column(db.Integer, nullable=False, default=0)
+    email = db.Column(db.String(255), nullable=False)
+    subject = db.Column(db.String(255), nullable=False)
+    sent = db.Column(db.Boolean, default=False)
+    error_text = db.Column(db.Text, nullable=True)
+    created_at = db.Column(db.DateTime, default=lambda: datetime.now(timezone.utc))
+    sent_at = db.Column(db.DateTime, nullable=True)
+
+    __table_args__ = (
+        db.UniqueConstraint('scan_id', 'user_id', name='uq_notifications_scan_user'),
+    )

backend/models/user_preference.py

@@ -0,0 +1,8 @@
+from utils.db import db
+
+class UserPreference(db.Model):
+    __tablename__ = "user_preferences"
+
+    id = db.Column(db.Integer, primary_key=True)
+    user_id = db.Column(db.Integer, unique=True, index=True, nullable=False)
+    email_notifications_enabled = db.Column(db.Boolean, default=True)

backend/requirements.txt

@@ -18,5 +18,7 @@ google-generativeai
 transformers
 torch
 flask_sqlalchemy
+semgrep
+check
 flasgger
 apispec-pydantic-plugin

backend/routes/notifications.py

@@ -0,0 +1,39 @@
+from flask import Blueprint, request, jsonify
+from flask_jwt_extended import jwt_required, get_jwt_identity
+from utils.db import db
+from models.notification import Notification
+from models.user_preference import UserPreference
+
+bp_notifications = Blueprint("notifications", __name__, url_prefix="/api/notifications")
+
+@bp_notifications.get("/")
+@jwt_required()
+def list_notifications():
+    user_id = int(get_jwt_identity())
+    rows = (Notification.query
+            .filter_by(user_id=user_id)
+            .order_by(Notification.created_at.desc())
+            .limit(100).all())
+    data = [{
+        "id": r.id, "scan_id": r.scan_id, "repo": r.repo, "status": r.status,
+        "findings_count": r.findings_count, "email": r.email, "subject": r.subject,
+        "sent": r.sent, "error_text": r.error_text,
+        "created_at": r.created_at.isoformat() if r.created_at else None,
+        "sent_at": r.sent_at.isoformat() if r.sent_at else None,
+    } for r in rows]
+    return jsonify(data)
+
+@bp_notifications.patch("/preferences/email")
+@jwt_required()
+def toggle_email_pref():
+    user_id = int(get_jwt_identity())
+    body = request.get_json(force=True) or {}
+    enabled = bool(body.get("enabled", True))
+    pref = UserPreference.query.filter_by(user_id=user_id).first()
+    if not pref:
+        pref = UserPreference(user_id=user_id, email_notifications_enabled=enabled)
+        db.session.add(pref)
+    else:
+        pref.email_notifications_enabled = enabled
+    db.session.commit()
+    return jsonify({"email_notifications_enabled": pref.email_notifications_enabled})

backend/routes/report_routes.py

@@ -0,0 +1,33 @@
+from flask import Blueprint, send_file, request, jsonify
+from flask_jwt_extended import jwt_required, get_jwt_identity
+from utils.db import db
+from models.scan_history import ScanHistory
+from models.user import User
+from services.report_service import generate_csv_for_scan, send_csv_report_email
+
+report_bp = Blueprint("reports", __name__)
+
+@report_bp.route("/reports/<int:scan_id>/csv", methods=["GET"])
+@jwt_required()
+def download_report_csv(scan_id: int):
+    user_id = int(get_jwt_identity())
+    scan: ScanHistory = db.session.get(ScanHistory, scan_id)
+    if not scan:
+        return jsonify({"error": "scan not found"}), 404
+
+    user = db.session.get(User, user_id)
+    if (scan.user_id != user_id) and (user.role != "admin"):
+        return jsonify({"error": "forbidden"}), 403
+
+    try:
+        csv_path, csv_filename = generate_csv_for_scan(scan_id)
+
+        if request.args.get("email", "false").lower() == "true":
+            subject = f"SafeOps — Rapport CSV ({scan.scan_type})"
+            executed_at = ((scan.scan_result or {}).get("meta", {}) or {}).get("executed_at", scan.created_at.isoformat() if getattr(scan, "created_at", None) else "")
+            body = f"Bonjour {user.name},\n\nVeuillez trouver ci-joint le rapport CSV de votre scan.\nExécuté le : {executed_at}\n"
+            send_csv_report_email(user.email, subject, body, csv_path, csv_filename)
+
+        return send_file(csv_path, mimetype="text/csv", as_attachment=True, download_name=csv_filename)
+    except Exception as e:
+        return jsonify({"error": str(e)}), 500