| import functools |
| import json |
| import os |
| import uuid |
| from typing import Any, Dict |
|
|
| from dotenv import load_dotenv |
| from flask import Flask, abort, flash, jsonify, redirect, render_template, request, session, url_for |
| from flask import Response |
| from werkzeug.utils import secure_filename |
|
|
| from core.observability import init_langsmith_observability |
| from core.storage import ( |
| create_password_reset_token, |
| change_password, |
| create_analysis, |
| create_user, |
| get_admin_stats, |
| get_analysis_by_id, |
| get_dashboard_stats, |
| get_user_by_id, |
| get_user_by_email, |
| init_db, |
| list_all_analyses, |
| list_users, |
| latest_analysis_for_user, |
| list_user_analyses, |
| seed_first_admin, |
| set_user_admin, |
| get_password_reset_token, |
| use_password_reset_token, |
| update_user_profile, |
| verify_user, |
| ) |
| from core.workflow import run_aispy_pipeline |
| from utils import downloader |
|
|
| load_dotenv() |
|
|
| app = Flask(__name__) |
| app.config["SECRET_KEY"] = os.getenv("SECRET_KEY", os.getenv("FLASK_SECRET_KEY", "aispy-dev-secret")) |
| app.config["MAX_CONTENT_LENGTH"] = 50 * 1024 * 1024 |
| app.config["SESSION_COOKIE_HTTPONLY"] = True |
| app.config["SESSION_COOKIE_SAMESITE"] = "Lax" |
|
|
| ALLOWED_EXTENSIONS = {"png", "jpg", "jpeg", "webp", "mp4", "mov", "avi"} |
|
|
|
|
| def allowed_file(filename: str) -> bool: |
| return "." in filename and filename.rsplit(".", 1)[1].lower() in ALLOWED_EXTENSIONS |
|
|
|
|
| def to_jsonable(value: Any): |
| if value is None: |
| return None |
| if isinstance(value, (str, int, float, bool)): |
| return value |
| if isinstance(value, dict): |
| return {str(k): to_jsonable(v) for k, v in value.items()} |
| if isinstance(value, (list, tuple, set)): |
| return [to_jsonable(v) for v in value] |
| if hasattr(value, "model_dump"): |
| return to_jsonable(value.model_dump()) |
| if hasattr(value, "dict"): |
| try: |
| return to_jsonable(value.dict()) |
| except Exception: |
| pass |
| if hasattr(value, "__dict__"): |
| return to_jsonable({k: v for k, v in value.__dict__.items() if not k.startswith("_")}) |
| return str(value) |
|
|
|
|
| def extract_value(obj: Any, key: str, default=None): |
| if obj is None: |
| return default |
| if hasattr(obj, key): |
| return getattr(obj, key) |
| if isinstance(obj, dict): |
| return obj.get(key, default) |
| return default |
|
|
|
|
| def current_user(): |
| user_id = session.get("user_id") |
| if not user_id: |
| return None |
| return { |
| "id": user_id, |
| "name": session.get("user_name"), |
| "email": session.get("user_email"), |
| "is_admin": bool(session.get("user_is_admin")), |
| } |
|
|
|
|
| def login_required(view): |
| @functools.wraps(view) |
| def wrapped(*args, **kwargs): |
| if not session.get("user_id"): |
| if request.path == "/analyze" or request.path.startswith("/api/"): |
| return jsonify({"error": "Authentication required."}), 401 |
| flash("Please sign in to continue.", "warning") |
| return redirect(url_for("login")) |
| return view(*args, **kwargs) |
|
|
| return wrapped |
|
|
|
|
| def admin_required(view): |
| @functools.wraps(view) |
| def wrapped(*args, **kwargs): |
| if not session.get("user_id"): |
| if request.path == "/analyze" or request.path.startswith("/api/"): |
| return jsonify({"error": "Authentication required."}), 401 |
| flash("Please sign in to continue.", "warning") |
| return redirect(url_for("login")) |
| if not session.get("user_is_admin"): |
| abort(403) |
| return view(*args, **kwargs) |
|
|
| return wrapped |
|
|
|
|
| @app.context_processor |
| def inject_globals(): |
| return {"current_user": current_user()} |
|
|
|
|
| @app.before_request |
| def initialize(): |
| init_db() |
| seed_first_admin() |
|
|
|
|
| init_langsmith_observability() |
|
|
|
|
| @app.route("/") |
| def index(): |
| if session.get("user_id"): |
| return redirect(url_for("dashboard")) |
| return redirect(url_for("login")) |
|
|
|
|
| @app.route("/register", methods=["GET", "POST"]) |
| def register(): |
| if session.get("user_id"): |
| return redirect(url_for("dashboard")) |
|
|
| if request.method == "POST": |
| name = request.form.get("name", "").strip() |
| email = request.form.get("email", "").strip().lower() |
| password = request.form.get("password", "") |
| confirm = request.form.get("confirm_password", "") |
|
|
| if not name or not email or not password: |
| flash("All fields are required.", "danger") |
| elif password != confirm: |
| flash("Passwords do not match.", "danger") |
| elif get_user_by_email(email): |
| flash("An account already exists for that email.", "danger") |
| else: |
| user = create_user(name, email, password) |
| session["user_id"] = user["id"] |
| session["user_name"] = user["name"] |
| session["user_email"] = user["email"] |
| session["user_is_admin"] = bool(user.get("is_admin")) |
| flash("Account created successfully.", "success") |
| return redirect(url_for("dashboard")) |
|
|
| return render_template("register.html") |
|
|
|
|
| @app.route("/login", methods=["GET", "POST"]) |
| def login(): |
| if session.get("user_id"): |
| return redirect(url_for("dashboard")) |
|
|
| if request.method == "POST": |
| email = request.form.get("email", "").strip().lower() |
| password = request.form.get("password", "") |
| user = verify_user(email, password) |
|
|
| if not user: |
| flash("Invalid email or password.", "danger") |
| else: |
| session["user_id"] = user["id"] |
| session["user_name"] = user["name"] |
| session["user_email"] = user["email"] |
| session["user_is_admin"] = bool(user.get("is_admin")) |
| flash("Welcome back.", "success") |
| return redirect(url_for("dashboard")) |
|
|
| return render_template("login.html") |
|
|
|
|
| @app.route("/logout") |
| def logout(): |
| session.clear() |
| flash("You have been signed out.", "info") |
| return redirect(url_for("login")) |
|
|
|
|
| @app.route("/forgot-password", methods=["GET", "POST"]) |
| def forgot_password(): |
| if request.method == "POST": |
| email = request.form.get("email", "").strip().lower() |
| token_info = create_password_reset_token(email) |
| if token_info: |
| reset_link = url_for("reset_password", token=token_info["token"], _external=True) |
| flash(f"Reset link created: {reset_link}", "success") |
| return render_template("forgot_password.html", reset_link=reset_link, email=email) |
| flash("If the account exists, a reset link can be generated.", "info") |
| return render_template("forgot_password.html") |
|
|
|
|
| @app.route("/reset-password/<token>", methods=["GET", "POST"]) |
| def reset_password(token: str): |
| record = get_password_reset_token(token) |
| if not record: |
| flash("Invalid or expired reset token.", "danger") |
| return redirect(url_for("forgot_password")) |
|
|
| if request.method == "POST": |
| new_password = request.form.get("new_password", "") |
| confirm_password = request.form.get("confirm_password", "") |
| if not new_password: |
| flash("Please enter a new password.", "danger") |
| elif new_password != confirm_password: |
| flash("Passwords do not match.", "danger") |
| else: |
| used = use_password_reset_token(token) |
| if not used: |
| flash("This reset link has already been used or expired.", "danger") |
| return redirect(url_for("forgot_password")) |
| change_password(record["user_id"], new_password) |
| flash("Password updated. Please sign in.", "success") |
| return redirect(url_for("login")) |
|
|
| return render_template("reset_password.html", token=token, email=record.get("email")) |
|
|
|
|
| @app.route("/dashboard") |
| @login_required |
| def dashboard(): |
| user = get_user_by_email(session.get("user_email", "")) |
| analyses = list_user_analyses(session["user_id"], limit=8) |
| stats = get_dashboard_stats(session["user_id"]) |
| return render_template( |
| "dashboard.html", |
| user=user, |
| stats=stats, |
| analyses=analyses, |
| ) |
|
|
|
|
| @app.route("/settings", methods=["GET", "POST"]) |
| @login_required |
| def settings(): |
| user = get_user_by_email(session.get("user_email", "")) |
| if request.method == "POST": |
| name = request.form.get("name", "").strip() |
| email = request.form.get("email", "").strip().lower() |
| current_password = request.form.get("current_password", "") |
| new_password = request.form.get("new_password", "") |
|
|
| if not user: |
| abort(404) |
|
|
| if name and email: |
| try: |
| updated = update_user_profile(user["id"], name=name, email=email) |
| session["user_name"] = updated["name"] |
| session["user_email"] = updated["email"] |
| session["user_is_admin"] = bool(updated.get("is_admin")) |
| flash("Profile updated.", "success") |
| except Exception: |
| flash("Email is already in use.", "danger") |
|
|
| if new_password: |
| if not verify_user(user["email"], current_password): |
| flash("Current password is incorrect.", "danger") |
| else: |
| change_password(user["id"], new_password) |
| flash("Password updated.", "success") |
|
|
| return redirect(url_for("settings")) |
|
|
| return render_template("settings.html", user=user) |
|
|
|
|
| @app.route("/admin") |
| @admin_required |
| def admin_dashboard(): |
| return render_template( |
| "admin.html", |
| admin_stats=get_admin_stats(), |
| users=list_users(limit=200), |
| analyses=list_all_analyses(limit=100), |
| ) |
|
|
|
|
| @app.route("/admin/users/<int:user_id>/toggle-admin", methods=["POST"]) |
| @admin_required |
| def admin_toggle(user_id: int): |
| current = get_user_by_email(session.get("user_email", "")) |
| target = get_user_by_id(user_id) |
| if not target: |
| abort(404) |
| if current and current.get("id") == target.get("id") and target.get("is_admin"): |
| flash("Cannot remove your own admin role.", "warning") |
| return redirect(url_for("admin_dashboard")) |
| if target.get("is_admin") and get_admin_stats().get("admins", 0) <= 1: |
| flash("At least one admin account must remain active.", "warning") |
| return redirect(url_for("admin_dashboard")) |
| set_user_admin(user_id, not bool(target.get("is_admin"))) |
| flash("Admin role updated.", "success") |
| return redirect(url_for("admin_dashboard")) |
|
|
|
|
| @app.route("/analyze", methods=["POST"]) |
| @login_required |
| def analyze(): |
| file_obj = request.files.get("file") |
| user_input_url = request.form.get("input", "").strip() |
| text_claim = request.form.get("text_claim", "").strip() |
|
|
| if not file_obj and not user_input_url and not text_claim: |
| return jsonify({"error": "Please provide a media file, URL, or text claim."}), 400 |
|
|
| media_path = None |
| input_type = "text" |
| request_id = str(uuid.uuid4()) |
| file_name = None |
|
|
| try: |
| if file_obj and file_obj.filename: |
| if not allowed_file(file_obj.filename): |
| return jsonify({"error": "Invalid file type."}), 400 |
| file_name = secure_filename(file_obj.filename) |
| file_prefix = f"{request_id[:8]}_{file_name}" |
| media_path = os.path.join(downloader.DOWNLOAD_FOLDER, file_prefix) |
| file_obj.save(media_path) |
| input_type = "media" |
| elif user_input_url and user_input_url.startswith("http"): |
| media_path, _ = downloader.download_media(user_input_url) |
| if not media_path: |
| return jsonify({"error": "Could not download content from link."}), 400 |
| file_name = os.path.basename(media_path) |
| input_type = "media" |
|
|
| final_state = run_aispy_pipeline( |
| input_type=input_type, |
| media_path=media_path, |
| text_claim=text_claim if text_claim else None, |
| request_id=request_id, |
| ) |
|
|
| if final_state.get("errors") and not final_state.get("final_result"): |
| return jsonify({"error": "Analysis failed: " + " | ".join(final_state["errors"])}), 500 |
|
|
| verdict_data = final_state.get("final_result") |
| forensics_data = final_state.get("forensics_data") |
| osint_data = final_state.get("osint_data") |
|
|
| if not verdict_data: |
| return jsonify({"error": "The AI auditor failed to generate a verdict."}), 500 |
|
|
| response_data = { |
| "verdict": extract_value(verdict_data, "verdict", "No verdict"), |
| "confidence": extract_value(verdict_data, "confidence", 0.0), |
| "reasoning": f"{extract_value(verdict_data, 'reasoning', '')} {extract_value(verdict_data, 'xai_breakdown', '')}".strip(), |
| "final_report": final_state.get("final_report", ""), |
| "extracted_context": final_state.get("media_context", "") or extract_value(forensics_data, "extracted_caption", "No vision context."), |
| "visual_evidence": extract_value(forensics_data, "visual_evidence", "No visual forensics run."), |
| "context_sources": [{"url": url, "title": "Verified Source"} for url in (extract_value(osint_data, "sources_used", []) or [])], |
| "request_id": request_id, |
| } |
|
|
| analysis = create_analysis( |
| { |
| "user_id": session["user_id"], |
| "request_id": request_id, |
| "input_type": input_type, |
| "file_name": file_name, |
| "source_url": user_input_url or None, |
| "text_claim": text_claim or None, |
| "verdict": response_data["verdict"], |
| "confidence": response_data["confidence"], |
| "reasoning": response_data["reasoning"], |
| "final_report": response_data["final_report"], |
| "payload_json": { |
| "response": to_jsonable(response_data), |
| "final_state": to_jsonable(final_state), |
| }, |
| } |
| ) |
|
|
| return jsonify( |
| { |
| **response_data, |
| "analysis_id": analysis["id"], |
| "report_url": url_for("report_view", analysis_id=analysis["id"]), |
| } |
| ) |
|
|
| except Exception as exc: |
| return jsonify({"error": str(exc)}), 500 |
| finally: |
| if media_path: |
| downloader.cleanup_file(media_path) |
|
|
|
|
| @app.route("/reports") |
| @login_required |
| def reports(): |
| analyses = list_user_analyses(session["user_id"], limit=50) |
| return render_template("reports.html", analyses=analyses) |
|
|
|
|
| @app.route("/report") |
| @login_required |
| def report_latest(): |
| latest = latest_analysis_for_user(session["user_id"]) |
| if not latest: |
| flash("No reports found yet.", "info") |
| return redirect(url_for("dashboard")) |
| return redirect(url_for("report_view", analysis_id=latest["id"])) |
|
|
|
|
| @app.route("/report/<int:analysis_id>") |
| @login_required |
| def report_view(analysis_id: int): |
| analysis = get_analysis_by_id(analysis_id, user_id=session["user_id"]) |
| if not analysis: |
| abort(404) |
|
|
| payload = analysis.get("payload_json") or {} |
| response = payload.get("response", {}) if isinstance(payload, dict) else {} |
| final_state = payload.get("final_state", {}) if isinstance(payload, dict) else {} |
|
|
| return render_template( |
| "report.html", |
| analysis=analysis, |
| response=response, |
| final_state=final_state, |
| ) |
|
|
|
|
| @app.route("/report/<int:analysis_id>/download.json") |
| @login_required |
| def download_report_json(analysis_id: int): |
| analysis = get_analysis_by_id(analysis_id, user_id=session["user_id"]) |
| if not analysis: |
| abort(404) |
| payload = analysis.get("payload_json") or {} |
| return jsonify({"analysis": analysis, "payload": payload}) |
|
|
|
|
| @app.route("/report/<int:analysis_id>/download.md") |
| @login_required |
| def download_report_markdown(analysis_id: int): |
| analysis = get_analysis_by_id(analysis_id, user_id=session["user_id"]) |
| if not analysis: |
| abort(404) |
| markdown = analysis.get("final_report") or "" |
| filename = f"aispy-report-{analysis_id}.md" |
| return Response( |
| markdown, |
| mimetype="text/markdown; charset=utf-8", |
| headers={"Content-Disposition": f'attachment; filename="{filename}"'}, |
| ) |
|
|
|
|
| @app.route("/api/reports/<int:analysis_id>") |
| @login_required |
| def api_report(analysis_id: int): |
| analysis = get_analysis_by_id(analysis_id, user_id=session["user_id"]) |
| if not analysis: |
| return jsonify({"error": "Report not found."}), 404 |
| return jsonify(analysis) |
|
|
|
|
| if __name__ == "__main__": |
| app.run(debug=True, port=int(os.getenv("PORT", "5000"))) |
