190 lines
8.5 KiB
Python
190 lines
8.5 KiB
Python
# app.py
|
|
import os
|
|
import logging
|
|
from flask import Flask, render_template, request, jsonify, session, redirect, url_for
|
|
|
|
# Import functions from api/endpoints.py
|
|
# check_link is assumed to handle VT internally and return combined result for 'link' key
|
|
from api.endpoints import check_message, check_phone, check_link
|
|
# Import Validators from api/validators.py - Needed for type detection
|
|
from api.validators import Validators
|
|
|
|
# Konfiguracja logowania
|
|
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
|
|
logger = logging.getLogger(__name__)
|
|
|
|
app = Flask(__name__)
|
|
# Setting secret_key is crucial for session security. Change this in production!
|
|
app.secret_key = os.getenv("FLASK_SECRET_KEY", "default_key_that_should_be_changed")
|
|
|
|
# Global statistics (since server start)
|
|
GLOBAL_STATS = {
|
|
"messages_checked": 0,
|
|
"phones_checked": 0,
|
|
"links_checked": 0
|
|
}
|
|
|
|
def increment_stat(key):
|
|
"""Increments session and global stats."""
|
|
# Session
|
|
session[key] = session.get(key, 0) + 1
|
|
# Global
|
|
GLOBAL_STATS[key] += 1
|
|
|
|
# ZASTĄP CAŁĄ FUNKCJĘ index PONIŻSZYM KODEM
|
|
@app.route("/", methods=["GET", "POST"])
|
|
def index():
|
|
"""
|
|
Handles the main page, processing form submissions (POST)
|
|
and displaying results after redirect (GET).
|
|
Implements Post/Redirect/Get (PRG) pattern.
|
|
Processes a single input field, detecting data type (Link, Phone, Message).
|
|
"""
|
|
user_ip = request.remote_addr
|
|
|
|
# Initialize session stats if they don't exist
|
|
if "messages_checked" not in session:
|
|
session["messages_checked"] = 0
|
|
session["phones_checked"] = 0
|
|
session["links_checked"] = 0
|
|
|
|
logger.info(f"Request to index ({request.method}) from IP: %s", user_ip)
|
|
|
|
# --- Handle POST Request (Process & Store, then Redirect) ---
|
|
if request.method == "POST":
|
|
# Clear previous results from session before processing new ones
|
|
session.pop('check_results', None)
|
|
|
|
# Get data from the single input field named "input_data"
|
|
input_data = request.form.get("input_data", "").strip() # Get and strip whitespace
|
|
|
|
current_check_results = {}
|
|
|
|
if input_data:
|
|
logger.info(f"Received input data: '{input_data}'")
|
|
|
|
# --- Type Detection Logic ---
|
|
# Implement a simple prioritization: Link -> Phone -> Message
|
|
# This order can be adjusted based on expected input types
|
|
|
|
# 1. Check if it's a Link (URL) first
|
|
# Use Validators.is_valid_url from api/validators.py
|
|
if Validators.is_valid_url(input_data):
|
|
logger.info(f"Input detected as URL: {input_data}")
|
|
# Call check_link. Based on your endpoints.py, it returns the combined result dictionary for 'link'
|
|
link_result = check_link(input_data)
|
|
# Store the result dictionary under the 'link' key for the HTML template to read
|
|
current_check_results["link"] = link_result
|
|
increment_stat("links_checked") # Increment link stat
|
|
logger.info(f"Processed link check. Result: {link_result.get('is_suspicious', 'N/A')}")
|
|
|
|
# 2. Else, check if it's a Phone Number
|
|
# Use Validators.is_valid_phone from api/validators.py
|
|
elif Validators.is_valid_phone(input_data):
|
|
logger.info(f"Input detected as Phone: {input_data}")
|
|
phone_result = check_phone(input_data)
|
|
# Store the result dictionary under the 'phone' key for the HTML template
|
|
current_check_results["phone"] = phone_result
|
|
increment_stat("phones_checked") # Increment phone stat
|
|
logger.info(f"Processed phone check. Result: {phone_result.get('is_suspicious', 'N/A')}")
|
|
|
|
# 3. Else, if it's neither a valid URL nor a valid phone number, assume it's a Message
|
|
else:
|
|
logger.info(f"Input treated as Message: {input_data}")
|
|
message_result = check_message(input_data)
|
|
# Store the result dictionary under the 'message' key for the HTML template
|
|
current_check_results["message"] = message_result
|
|
increment_stat("messages_checked") # Increment message stat
|
|
logger.info(f"Processed message check. Result: {message_result.get('is_suspicious', 'N/A')}")
|
|
|
|
else:
|
|
logger.info("Received empty input data.")
|
|
# If input is empty, current_check_results will be empty.
|
|
# The HTML template is designed to handle this gracefully (no results displayed).
|
|
|
|
# Store the results from this single check in the session to be retrieved by the subsequent GET request
|
|
session['check_results'] = current_check_results
|
|
|
|
# Redirect to the same URL with GET method to display results and prevent form re-submission on refresh
|
|
logger.info("Redirecting after POST.")
|
|
return redirect(url_for('index'))
|
|
|
|
# --- Handle GET Request (Retrieve & Display) ---
|
|
else: # request.method == "GET"
|
|
# Retrieve the results from session if they exist (after a POST redirect)
|
|
# Use session.pop to get the value and remove it in one step. Defaults to empty dict if no results in session.
|
|
result = session.pop('check_results', {})
|
|
logger.info(f"Handling GET request. Results retrieved from session: {bool(result)}")
|
|
|
|
# Prepare session stats to pass to the template
|
|
session_stats = {
|
|
"messages_checked": session.get("messages_checked", 0),
|
|
"phones_checked": session.get("phones_checked", 0),
|
|
"links_checked": session.get("links_checked", 0)
|
|
}
|
|
|
|
# Render the template with the retrieved results (or empty result dict for a normal initial GET)
|
|
# The 'result' dictionary will contain 'message', 'phone', or 'link' key based on input type detected in POST
|
|
return render_template("index.html",
|
|
result=result, # Pass the result dictionary (can be empty or contain one check result)
|
|
global_stats=GLOBAL_STATS,
|
|
session_stats=session_stats)
|
|
|
|
# --- API ENDPOINTS ---
|
|
# These endpoints are designed for specific input types and typically return JSON.
|
|
# They remain unchanged as they are separate from the main single-input form.
|
|
|
|
@app.route("/api/check_message", methods=["POST"])
|
|
def api_check_message():
|
|
"""API endpoint for message check."""
|
|
data = request.get_json()
|
|
message = data.get("message", "")
|
|
if not message:
|
|
logger.warning("API check_message: Missing 'message' parameter.")
|
|
return jsonify({"error": "Missing 'message' parameter."}), 400
|
|
|
|
message_result = check_message(message)
|
|
increment_stat("messages_checked") # Increment message stat for API
|
|
logger.info(f"API check_message: Result: {message_result.get('is_suspicious', 'N/A')}")
|
|
return jsonify(message_result)
|
|
|
|
@app.route("/api/check_phone", methods=["POST"])
|
|
def api_check_phone():
|
|
"""API endpoint for phone check."""
|
|
data = request.get_json()
|
|
phone = data.get("phone_number", "")
|
|
if not phone:
|
|
logger.warning("API check_phone: Missing 'phone_number' parameter.")
|
|
return jsonify({"error": "Missing 'phone_number' parameter."}), 400
|
|
|
|
phone_result = check_phone(phone)
|
|
increment_stat("phones_checked") # Increment phone stat for API
|
|
logger.info(f"API check_phone: Result: {phone_result.get('is_suspicious', 'N/A')}")
|
|
return jsonify(phone_result)
|
|
|
|
@app.route("/api/check_link", methods=["POST"])
|
|
def api_check_link():
|
|
"""API endpoint for link check."""
|
|
data = request.get_json()
|
|
url = data.get("url", "")
|
|
if not url:
|
|
logger.warning("API check_link: Missing 'url' parameter.")
|
|
return jsonify({"error": "Missing 'url' parameter."}), 400
|
|
|
|
# Note: check_link in your endpoints.py handles VT internally and returns the combined dict
|
|
link_result = check_link(url)
|
|
|
|
increment_stat("links_checked") # Increment link stat for API
|
|
logger.info(f"API check_link: Result: {link_result.get('is_suspicious', 'N/A')}, Source: {link_result.get('source', 'N/A')}")
|
|
# Return the combined result dictionary from check_link
|
|
return jsonify(link_result)
|
|
|
|
|
|
# --- Application Entry Point ---
|
|
if __name__ == "__main__":
|
|
# Port from environment variable or default 5000
|
|
port = int(os.environ.get("PORT", 5000))
|
|
# host='0.0.0.0' allows external access (e.g., in Docker)
|
|
# debug=True is useful during development
|
|
app.run(host="0.0.0.0", port=port, debug=True)
|