#!/usr/bin/env python3 # -*- coding: utf-8 -*- from flask import Flask, render_template_string, request, redirect, url_for, session, jsonify import sqlite3 import hashlib import os from datetime import datetime app = Flask(__name__) app.secret_key = os.urandom(24) # --- Configuration --- DB_NAME = "nexapy_chat.db" HOST = "0.0.0.0" PORT = 4000 # --- Initialisation de la base de données --- def init_db(): conn = sqlite3.connect(DB_NAME) cursor = conn.cursor() # Table des utilisateurs cursor.execute(''' CREATE TABLE IF NOT EXISTS users ( user_id INTEGER PRIMARY KEY AUTOINCREMENT, username TEXT UNIQUE NOT NULL, password TEXT NOT NULL, email TEXT UNIQUE, bio TEXT DEFAULT '', profile_pic TEXT DEFAULT '/static/default_profile.png', is_admin BOOLEAN DEFAULT FALSE, created_at DATETIME DEFAULT CURRENT_TIMESTAMP ) ''') # Table des amitiés cursor.execute(''' CREATE TABLE IF NOT EXISTS friendships ( friendship_id INTEGER PRIMARY KEY AUTOINCREMENT, user1_id INTEGER NOT NULL, user2_id INTEGER NOT NULL, status TEXT DEFAULT 'pending', created_at DATETIME DEFAULT CURRENT_TIMESTAMP, FOREIGN KEY (user1_id) REFERENCES users(user_id), FOREIGN KEY (user2_id) REFERENCES users(user_id), UNIQUE(user1_id, user2_id) ) ''') # Table des likes de profil cursor.execute(''' CREATE TABLE IF NOT EXISTS profile_likes ( like_id INTEGER PRIMARY KEY AUTOINCREMENT, liker_id INTEGER NOT NULL, liked_id INTEGER NOT NULL, created_at DATETIME DEFAULT CURRENT_TIMESTAMP, FOREIGN KEY (liker_id) REFERENCES users(user_id), FOREIGN KEY (liked_id) REFERENCES users(user_id), UNIQUE(liker_id, liked_id) ) ''') # Table des conversations privées cursor.execute(''' CREATE TABLE IF NOT EXISTS conversations ( conversation_id INTEGER PRIMARY KEY AUTOINCREMENT, user1_id INTEGER NOT NULL, user2_id INTEGER NOT NULL, created_at DATETIME DEFAULT CURRENT_TIMESTAMP, FOREIGN KEY (user1_id) REFERENCES users(user_id), FOREIGN KEY (user2_id) REFERENCES users(user_id), UNIQUE(user1_id, user2_id) ) ''') # Table des messages cursor.execute(''' CREATE TABLE IF NOT EXISTS messages ( message_id INTEGER PRIMARY KEY AUTOINCREMENT, conversation_id INTEGER, group_id INTEGER, sender_id INTEGER NOT NULL, content TEXT NOT NULL, timestamp DATETIME DEFAULT CURRENT_TIMESTAMP, is_read BOOLEAN DEFAULT FALSE, FOREIGN KEY (conversation_id) REFERENCES conversations(conversation_id), FOREIGN KEY (group_id) REFERENCES groups(group_id), FOREIGN KEY (sender_id) REFERENCES users(user_id) ) ''') # Table des groupes cursor.execute(''' CREATE TABLE IF NOT EXISTS groups ( group_id INTEGER PRIMARY KEY AUTOINCREMENT, group_name TEXT NOT NULL, access_code TEXT UNIQUE NOT NULL, created_by INTEGER NOT NULL, created_at DATETIME DEFAULT CURRENT_TIMESTAMP, FOREIGN KEY (created_by) REFERENCES users(user_id) ) ''') # Table des membres de groupes cursor.execute(''' CREATE TABLE IF NOT EXISTS group_members ( group_id INTEGER NOT NULL, user_id INTEGER NOT NULL, PRIMARY KEY (group_id, user_id), FOREIGN KEY (group_id) REFERENCES groups(group_id), FOREIGN KEY (user_id) REFERENCES users(user_id) ) ''') conn.commit() conn.close() # --- Fonctions utilitaires --- def hash_password(password): return hashlib.sha256(password.encode()).hexdigest() def get_db_connection(): conn = sqlite3.connect(DB_NAME) conn.row_factory = sqlite3.Row return conn # --- Routes --- @app.route("/") def index(): if "user_id" in session: return redirect(url_for("dashboard")) return redirect(url_for("login")) @app.route("/login", methods=["GET", "POST"]) def login(): if request.method == "POST": username = request.form["username"] password = request.form["password"] conn = get_db_connection() cursor = conn.cursor() cursor.execute("SELECT * FROM users WHERE username = ? AND password = ?", (username, hash_password(password))) user = cursor.fetchone() conn.close() if user: session["user_id"] = user["user_id"] session["username"] = user["username"] session["is_admin"] = bool(user["is_admin"]) session["profile_pic"] = user["profile_pic"] return redirect(url_for("dashboard")) else: return render_template_string(""" Connexion - NexaPy

Nom d'utilisateur ou mot de passe incorrect.

ou

Pas de compte ? S'inscrire

""") return render_template_string(""" Connexion - NexaPy
ou

Pas de compte ? S'inscrire

""") @app.route("/register", methods=["GET", "POST"]) def register(): if request.method == "POST": username = request.form["username"] password = request.form["password"] email = request.form.get("email", "") bio = request.form.get("bio", "") conn = get_db_connection() cursor = conn.cursor() try: cursor.execute( "INSERT INTO users (username, password, email, bio) VALUES (?, ?, ?, ?)", (username, hash_password(password), email, bio) ) conn.commit() return redirect(url_for("login")) except sqlite3.IntegrityError as e: error = "Nom d'utilisateur déjà utilisé." if "email" in str(e): error = "Email déjà utilisé." conn.close() return render_template_string(""" Inscription - NexaPy

{{ error }}

Déjà un compte ? Se connecter

""", error=error) finally: conn.close() return render_template_string(""" Inscription - NexaPy

Déjà un compte ? Se connecter

""") @app.route("/dashboard") def dashboard(): if "user_id" not in session: return redirect(url_for("login")) conn = get_db_connection() cursor = conn.cursor() # Récupérer les conversations privées cursor.execute(''' SELECT c.conversation_id, u.user_id, u.username, u.profile_pic, u.bio, (SELECT content FROM messages WHERE conversation_id = c.conversation_id ORDER BY timestamp DESC LIMIT 1) as last_message, (SELECT timestamp FROM messages WHERE conversation_id = c.conversation_id ORDER BY timestamp DESC LIMIT 1) as last_message_time FROM conversations c JOIN users u ON (u.user_id = c.user1_id OR u.user_id = c.user2_id) AND u.user_id != ? WHERE c.user1_id = ? OR c.user2_id = ? GROUP BY c.conversation_id ORDER BY last_message_time DESC ''', (session["user_id"], session["user_id"], session["user_id"])) conversations = cursor.fetchall() # Récupérer les groupes cursor.execute(''' SELECT g.group_id, g.group_name, g.access_code, (SELECT content FROM messages WHERE group_id = g.group_id ORDER BY timestamp DESC LIMIT 1) as last_message, (SELECT timestamp FROM messages WHERE group_id = g.group_id ORDER BY timestamp DESC LIMIT 1) as last_message_time FROM groups g JOIN group_members gm ON g.group_id = gm.group_id WHERE gm.user_id = ? GROUP BY g.group_id ORDER BY last_message_time DESC ''', (session["user_id"],)) groups = cursor.fetchall() # Récupérer les suggestions d'amis cursor.execute(''' SELECT u.user_id, u.username, u.profile_pic, u.bio FROM users u WHERE u.user_id != ? AND u.user_id NOT IN ( SELECT user2_id FROM friendships WHERE user1_id = ? AND status = 'accepted' UNION SELECT user1_id FROM friendships WHERE user2_id = ? AND status = 'accepted' ) AND u.user_id NOT IN ( SELECT user2_id FROM friendships WHERE user1_id = ? AND status = 'pending' UNION SELECT user1_id FROM friendships WHERE user2_id = ? AND status = 'pending' ) LIMIT 5 ''', (session["user_id"], session["user_id"], session["user_id"], session["user_id"], session["user_id"])) friend_suggestions = cursor.fetchall() # Récupérer les demandes d'ami en attente cursor.execute(''' SELECT u.user_id, u.username, u.profile_pic, u.bio FROM friendships f JOIN users u ON f.user1_id = u.user_id WHERE f.user2_id = ? AND f.status = 'pending' ''', (session["user_id"],)) pending_requests = cursor.fetchall() # Récupérer les amis cursor.execute(''' SELECT u.user_id, u.username, u.profile_pic, u.bio FROM friendships f JOIN users u ON ( (f.user1_id = ? AND f.user2_id = u.user_id AND f.status = 'accepted') OR (f.user2_id = ? AND f.user1_id = u.user_id AND f.status = 'accepted') ) ''', (session["user_id"], session["user_id"])) friends = cursor.fetchall() conn.close() return render_template_string(""" Tableau de bord - NexaPy

Tableau de bord

Bienvenue sur NexaPy

Sélectionnez une conversation ou un ami pour commencer à discuter.

""", conversations=conversations, groups=groups, friend_suggestions=friend_suggestions, pending_requests=pending_requests, friends=friends) @app.route("/search_users") def search_users(): query = request.args.get("q", "") if len(query) < 2: return jsonify([]) conn = get_db_connection() cursor = conn.cursor() cursor.execute(''' SELECT user_id, username, profile_pic, bio FROM users WHERE username LIKE ? AND user_id != ? LIMIT 5 ''', (f"%{query}%", session["user_id"])) users = cursor.fetchall() conn.close() return jsonify([{ "user_id": user["user_id"], "username": user["username"], "profile_pic": user["profile_pic"], "bio": user["bio"] } for user in users]) @app.route("/send_friend_request", methods=["POST"]) def send_friend_request(): if "user_id" not in session: return jsonify({"error": "Non autorisé"}), 401 data = request.get_json() user2_id = data.get("user_id") conn = get_db_connection() cursor = conn.cursor() cursor.execute(''' SELECT * FROM friendships WHERE (user1_id = ? AND user2_id = ?) OR (user1_id = ? AND user2_id = ?) ''', (session["user_id"], user2_id, user2_id, session["user_id"])) existing = cursor.fetchone() if existing: conn.close() return jsonify({"error": "Demande déjà envoyée ou ami existant"}), 400 cursor.execute("INSERT INTO friendships (user1_id, user2_id, status) VALUES (?, ?, 'pending')", (session["user_id"], user2_id)) conn.commit() conn.close() return jsonify({"success": True}) @app.route("/accept_friend_request", methods=["POST"]) def accept_friend_request(): if "user_id" not in session: return jsonify({"error": "Non autorisé"}), 401 data = request.get_json() user1_id = data.get("user_id") conn = get_db_connection() cursor = conn.cursor() cursor.execute(''' UPDATE friendships SET status = 'accepted' WHERE user1_id = ? AND user2_id = ? AND status = 'pending' ''', (user1_id, session["user_id"])) cursor.execute("INSERT OR IGNORE INTO conversations (user1_id, user2_id) VALUES (?, ?)", (session["user_id"], user1_id)) conn.commit() conn.close() return jsonify({"success": True}) @app.route("/reject_friend_request", methods=["POST"]) def reject_friend_request(): if "user_id" not in session: return jsonify({"error": "Non autorisé"}), 401 data = request.get_json() user1_id = data.get("user_id") conn = get_db_connection() cursor = conn.cursor() cursor.execute(''' DELETE FROM friendships WHERE user1_id = ? AND user2_id = ? AND status = 'pending' ''', (user1_id, session["user_id"])) conn.commit() conn.close() return jsonify({"success": True}) @app.route("/like_profile", methods=["POST"]) def like_profile(): if "user_id" not in session: return jsonify({"error": "Non autorisé"}), 401 data = request.get_json() liked_id = data.get("user_id") conn = get_db_connection() cursor = conn.cursor() try: cursor.execute("INSERT INTO profile_likes (liker_id, liked_id) VALUES (?, ?)", (session["user_id"], liked_id)) conn.commit() except sqlite3.IntegrityError: conn.close() return jsonify({"error": "Profil déjà liké"}), 400 conn.close() return jsonify({"success": True}) @app.route("/join_group", methods=["POST"]) def join_group(): if "user_id" not in session: return jsonify({"error": "Non autorisé"}), 401 data = request.get_json() access_code = data.get("access_code") if not access_code: return jsonify({"error": "Code d'accès requis"}), 400 conn = get_db_connection() cursor = conn.cursor() cursor.execute("SELECT group_id FROM groups WHERE access_code = ?", (access_code,)) group = cursor.fetchone() if not group: conn.close() return jsonify({"error": "Code d'accès invalide"}), 404 cursor.execute("SELECT * FROM group_members WHERE group_id = ? AND user_id = ?", (group["group_id"], session["user_id"])) existing_member = cursor.fetchone() if existing_member: conn.close() return jsonify({"error": "Vous êtes déjà membre de ce groupe"}), 400 cursor.execute("INSERT INTO group_members (group_id, user_id) VALUES (?, ?)", (group["group_id"], session["user_id"])) conn.commit() conn.close() return jsonify({"success": True}) @app.route("/chat/user/") def chat_user(user_id): if "user_id" not in session: return redirect(url_for("login")) conn = get_db_connection() cursor = conn.cursor() cursor.execute("SELECT * FROM users WHERE user_id = ?", (user_id,)) user = cursor.fetchone() if not user: conn.close() return redirect(url_for("dashboard")) cursor.execute(''' SELECT c.conversation_id FROM conversations c WHERE (c.user1_id = ? AND c.user2_id = ?) OR (c.user1_id = ? AND c.user2_id = ?) ''', (session["user_id"], user_id, user_id, session["user_id"])) conversation = cursor.fetchone() if not conversation: cursor.execute("INSERT INTO conversations (user1_id, user2_id) VALUES (?, ?)", (session["user_id"], user_id)) conversation_id = cursor.lastrowid else: conversation_id = conversation["conversation_id"] cursor.execute(''' SELECT m.message_id, m.sender_id, m.content, m.timestamp, u.username, u.profile_pic FROM messages m JOIN users u ON m.sender_id = u.user_id WHERE m.conversation_id = ? ORDER BY m.timestamp ASC ''', (conversation_id,)) messages = cursor.fetchall() cursor.execute(''' UPDATE messages SET is_read = TRUE WHERE conversation_id = ? AND sender_id != ? ''', (conversation_id, session["user_id"])) conn.commit() conn.close() return render_template_string(""" Chat avec {{ user['username'] }} - NexaPy
{% if user['profile_pic'] %} Photo de profil {% else %} {% endif %}

{{ user['username'] }}

En ligne

{% for message in messages %}
{{ message['username'] }} {{ message['timestamp'] }}

{{ message['content'] }}

{% endfor %}
""", user=user, messages=messages, conversation_id=conversation_id) @app.route("/send_message", methods=["POST"]) def send_message(): if "user_id" not in session: return jsonify({"error": "Non autorisé"}), 401 data = request.get_json() conversation_id = data.get("conversation_id") content = data.get("content") conn = get_db_connection() cursor = conn.cursor() cursor.execute("INSERT INTO messages (conversation_id, sender_id, content) VALUES (?, ?, ?)", (conversation_id, session["user_id"], content)) conn.commit() cursor.execute(''' SELECT m.message_id, m.sender_id, m.content, m.timestamp, u.username FROM messages m JOIN users u ON m.sender_id = u.user_id WHERE m.message_id = ? ''', (cursor.lastrowid,)) message = cursor.fetchone() conn.close() return jsonify({ "message_id": message["message_id"], "sender_id": message["sender_id"], "content": message["content"], "timestamp": message["timestamp"], "username": message["username"] }) @app.route("/get_messages") def get_messages(): conversation_id = request.args.get("conversation_id") if not conversation_id or "user_id" not in session: return jsonify([]) conn = get_db_connection() cursor = conn.cursor() cursor.execute(''' SELECT m.message_id, m.sender_id, m.content, m.timestamp, u.username FROM messages m JOIN users u ON m.sender_id = u.user_id WHERE m.conversation_id = ? ORDER BY m.timestamp ASC ''', (conversation_id,)) messages = cursor.fetchall() conn.close() return jsonify([{ "message_id": message["message_id"], "sender_id": message["sender_id"], "content": message["content"], "timestamp": message["timestamp"], "username": message["username"] } for message in messages]) @app.route("/create_group", methods=["GET", "POST"]) def create_group(): if "user_id" not in session: return redirect(url_for("login")) if request.method == "POST": group_name = request.form["group_name"] access_code = request.form["access_code"] conn = get_db_connection() cursor = conn.cursor() try: cursor.execute("INSERT INTO groups (group_name, access_code, created_by) VALUES (?, ?, ?)", (group_name, access_code, session["user_id"])) group_id = cursor.lastrowid cursor.execute("INSERT INTO group_members (group_id, user_id) VALUES (?, ?)", (group_id, session["user_id"])) conn.commit() return redirect(url_for("dashboard")) except sqlite3.IntegrityError: return "Code d'accès déjà utilisé." finally: conn.close() return render_template_string(""" Créer un groupe - NexaPy

Retour au tableau de bord

""") @app.route("/chat/group/") def chat_group(group_id): if "user_id" not in session: return redirect(url_for("login")) conn = get_db_connection() cursor = conn.cursor() cursor.execute("SELECT * FROM groups WHERE group_id = ?", (group_id,)) group = cursor.fetchone() if not group: conn.close() return redirect(url_for("dashboard")) cursor.execute("SELECT * FROM group_members WHERE group_id = ? AND user_id = ?", (group_id, session["user_id"])) if not cursor.fetchone(): conn.close() return redirect(url_for("dashboard")) cursor.execute(''' SELECT m.message_id, m.sender_id, m.content, m.timestamp, u.username, u.profile_pic FROM messages m JOIN users u ON m.sender_id = u.user_id WHERE m.group_id = ? ORDER BY m.timestamp ASC ''', (group_id,)) messages = cursor.fetchall() cursor.execute(''' UPDATE messages SET is_read = TRUE WHERE group_id = ? AND sender_id != ? ''', (group_id, session["user_id"])) conn.commit() conn.close() return render_template_string(""" Chat - {{ group['group_name'] }} - NexaPy

{{ group['group_name'] }}

Groupe • Code: {{ group['access_code'] }}

{% for message in messages %}
{{ message['username'] }} {{ message['timestamp'] }}

{{ message['content'] }}

{% endfor %}
""", group=group, messages=messages) @app.route("/send_group_message", methods=["POST"]) def send_group_message(): if "user_id" not in session: return jsonify({"error": "Non autorisé"}), 401 data = request.get_json() group_id = data.get("group_id") content = data.get("content") conn = get_db_connection() cursor = conn.cursor() cursor.execute("SELECT * FROM group_members WHERE group_id = ? AND user_id = ?", (group_id, session["user_id"])) if not cursor.fetchone(): conn.close() return jsonify({"error": "Non autorisé"}), 401 cursor.execute("INSERT INTO messages (group_id, sender_id, content) VALUES (?, ?, ?)", (group_id, session["user_id"], content)) conn.commit() cursor.execute(''' SELECT m.message_id, m.sender_id, m.content, m.timestamp, u.username FROM messages m JOIN users u ON m.sender_id = u.user_id WHERE m.message_id = ? ''', (cursor.lastrowid,)) message = cursor.fetchone() conn.close() return jsonify({ "message_id": message["message_id"], "sender_id": message["sender_id"], "content": message["content"], "timestamp": message["timestamp"], "username": message["username"] }) @app.route("/get_group_messages") def get_group_messages(): group_id = request.args.get("group_id") if not group_id or "user_id" not in session: return jsonify([]) conn = get_db_connection() cursor = conn.cursor() cursor.execute("SELECT * FROM group_members WHERE group_id = ? AND user_id = ?", (group_id, session["user_id"])) if not cursor.fetchone(): conn.close() return jsonify([]) cursor.execute(''' SELECT m.message_id, m.sender_id, m.content, m.timestamp, u.username FROM messages m JOIN users u ON m.sender_id = u.user_id WHERE m.group_id = ? ORDER BY m.timestamp ASC ''', (group_id,)) messages = cursor.fetchall() conn.close() return jsonify([{ "message_id": message["message_id"], "sender_id": message["sender_id"], "content": message["content"], "timestamp": message["timestamp"], "username": message["username"] } for message in messages]) @app.route("/logout") def logout(): session.clear() return redirect(url_for("login")) # --- Lancement du serveur --- if __name__ == "__main__": init_db() app.run(host=HOST, port=PORT, debug=True)