|
from flask import Flask, render_template, request, redirect, url_for, session, jsonify |
|
from sklearn.metrics.pairwise import cosine_similarity |
|
import numpy as np |
|
from bson.objectid import ObjectId |
|
import bcrypt |
|
from passlib.hash import sha256_crypt |
|
from flask_pymongo import PyMongo |
|
from models import model_ir |
|
from sklearn.feature_extraction.text import TfidfVectorizer |
|
from config import DATABASE_URI, SECRET_KEY |
|
|
|
app = Flask(__name__) |
|
|
|
|
|
app.config["MONGO_URI"] = DATABASE_URI |
|
app.config["SECRET_KEY"] = SECRET_KEY |
|
mongo = PyMongo(app) |
|
|
|
|
|
|
|
def test_db_connection(): |
|
try: |
|
|
|
mongo.db.users.find_one() |
|
|
|
return True |
|
except Exception as e: |
|
|
|
return False |
|
|
|
|
|
|
|
|
|
|
|
|
|
@app.route("/", methods=["GET", "POST"]) |
|
def login(): |
|
if request.method == "POST": |
|
username = request.form.get("email") |
|
password = request.form.get("password") |
|
|
|
user = mongo.db.users.find_one({"email": username}) |
|
if user: |
|
|
|
if user["role"] == "admin": |
|
if bcrypt.checkpw( |
|
password.encode("utf-8"), user["password"].encode("utf-8") |
|
): |
|
|
|
session["user_id"] = str(user["_id"]) |
|
session["user_name"] = user["name"] |
|
return redirect(url_for("admin")) |
|
else: |
|
if bcrypt.checkpw( |
|
password.encode("utf-8"), user["password"].encode("utf-8") |
|
): |
|
|
|
session["user_id"] = str(user["_id"]) |
|
session["user_name"] = user["name"] |
|
return redirect(url_for("regular_user")) |
|
|
|
|
|
return "Invalid credentials or Account not Registered" |
|
else: |
|
return render_template( |
|
"authentication-login.html", user_name=session.get("user_name") |
|
) |
|
|
|
|
|
@app.route("/register", methods=["GET", "POST"]) |
|
def register(): |
|
if request.method == "POST": |
|
name = request.form.get("name") |
|
email = request.form.get("email") |
|
password = request.form.get("password") |
|
|
|
|
|
if email == "admin@example.com": |
|
hashed_password = sha256_crypt.hash(password) |
|
else: |
|
hashed_password = bcrypt.hashpw( |
|
password.encode("utf-8"), bcrypt.gensalt() |
|
).decode( |
|
"utf-8" |
|
) |
|
|
|
role = "regular_user" |
|
if email == "admin@arazim.com": |
|
role = "admin" |
|
|
|
|
|
user = { |
|
"name": name, |
|
"email": email, |
|
"password": hashed_password, |
|
"role": role |
|
|
|
} |
|
|
|
mongo.db.users.insert_one(user) |
|
|
|
return redirect(url_for("login")) |
|
else: |
|
return render_template("authentication-register.html") |
|
|
|
|
|
@app.route("/logout") |
|
def logout(): |
|
session.pop("user_id", None) |
|
session.pop("user_name", None) |
|
return redirect(url_for("login")) |
|
|
|
|
|
@app.route("/admin") |
|
def admin(): |
|
return render_template("admin-dashboard.html", user_name=session.get("user_name")) |
|
|
|
|
|
@app.route("/admin/Results", methods=["GET", "POST"]) |
|
def AdminResults(): |
|
if request.method == "POST": |
|
selected_rows = request.form.getlist("selected-rows") |
|
results = session.get("results") |
|
|
|
for row_index in selected_rows: |
|
row = results[int(row_index) - 1] |
|
mongo.db.customers.insert_one({"post": row[0], "label": row[1]}) |
|
|
|
return render_template("success.html") |
|
else: |
|
results = model_ir.lr_BS() |
|
session["results"] = results |
|
return render_template( |
|
"admin-dashboard.html", results=results, user_name=session.get("user_name") |
|
) |
|
|
|
|
|
@app.route("/regular_user") |
|
def regular_user(): |
|
return render_template("regular_user.html", user_name=session.get("user_name")) |
|
|
|
|
|
|
|
|
|
|
|
@app.route("/buyer.html") |
|
def buyer(): |
|
results = get_buyer_posts() |
|
return render_template( |
|
"buyer.html", user_name=session.get("user_name"), results=results |
|
) |
|
|
|
|
|
def get_buyer_posts(): |
|
|
|
results = mongo.db.customers.find({"label": "Buyer"}) |
|
|
|
posts_with_ids = [ |
|
{"_id": str(post["_id"]), "post": post["post"]} for post in results |
|
] |
|
return posts_with_ids |
|
|
|
|
|
@app.route("/similar_seller_posts/<buyer_post_id>") |
|
def similar_seller_posts(buyer_post_id): |
|
similar_posts = get_similar_seller_posts(buyer_post_id) |
|
buyer_post = mongo.db.customers.find_one({"_id": ObjectId(buyer_post_id)}) |
|
return render_template( |
|
"post-display.html", |
|
buyer_post=buyer_post, |
|
user_name=session.get("user_name"), |
|
similar_posts=similar_posts, |
|
) |
|
|
|
|
|
def get_similar_seller_posts(buyer_post_id): |
|
if buyer_post_id: |
|
buyer_post = mongo.db.customers.find_one({"_id": ObjectId(buyer_post_id)}) |
|
if buyer_post: |
|
buyer_attributes = extract_attributes(buyer_post["post"]) |
|
similar_seller_posts = content_based_filtering(buyer_attributes) |
|
return similar_seller_posts |
|
|
|
return [] |
|
|
|
|
|
def extract_attributes(post): |
|
|
|
attributes = { |
|
"content": post, |
|
} |
|
return attributes |
|
|
|
|
|
def content_based_filtering(buyer_attributes): |
|
|
|
seller_posts = mongo.db.customers.find({"label": "Seller"}) |
|
|
|
|
|
most_similar_post = None |
|
highest_similarity_score = -1 |
|
|
|
for post in seller_posts: |
|
seller_attributes = extract_attributes( |
|
post["post"] |
|
) |
|
similarity_score = compute_similarity(buyer_attributes, seller_attributes) |
|
|
|
|
|
if similarity_score > highest_similarity_score: |
|
highest_similarity_score = similarity_score |
|
most_similar_post = post |
|
|
|
if most_similar_post: |
|
return [most_similar_post] |
|
else: |
|
return [] |
|
|
|
|
|
def compute_similarity(attributes1, attributes2): |
|
|
|
vector1 = convert_attributes_to_vector(attributes1).reshape(1, -1) |
|
vector2 = convert_attributes_to_vector(attributes2).reshape(1, -1) |
|
|
|
|
|
similarity_score = cosine_similarity(vector1, vector2)[0][0] |
|
return similarity_score |
|
|
|
|
|
def convert_attributes_to_vector(attributes): |
|
|
|
feature_indices = { |
|
"content": 0, |
|
} |
|
|
|
|
|
vector_length = len(feature_indices) |
|
vector = np.zeros(vector_length) |
|
|
|
|
|
if "content" in attributes: |
|
content = attributes["content"] |
|
tfidf_vectorizer = TfidfVectorizer() |
|
tfidf_matrix = tfidf_vectorizer.fit_transform([content]) |
|
content_vector = tfidf_matrix.toarray() |
|
|
|
|
|
content_vector = content_vector.flatten() |
|
|
|
|
|
if "content" in feature_indices: |
|
feature_index = feature_indices["content"] |
|
vector[feature_index] = content_vector[ |
|
0 |
|
] |
|
|
|
return vector |
|
|
|
|
|
|
|
|
|
|
|
@app.route("/seller.html") |
|
def seller(): |
|
results = get_seller_posts() |
|
return render_template( |
|
"seller.html", user_name=session.get("user_name"), results=results |
|
) |
|
|
|
|
|
def get_seller_posts(): |
|
results = mongo.db.customers.find({"label": "Seller"}) |
|
|
|
posts_with_ids = [ |
|
{"_id": str(post["_id"]), "post": post["post"]} for post in results |
|
] |
|
return posts_with_ids |
|
|
|
|
|
@app.route("/similar_buyer_posts/<seller_post_id>") |
|
def similar_buyer_posts(seller_post_id): |
|
similar_posts = get_similar_buyer_posts(seller_post_id) |
|
seller_post = mongo.db.customers.find_one({"_id": ObjectId(seller_post_id)}) |
|
return render_template( |
|
"post-display-seller.html", |
|
seller_post=seller_post, |
|
user_name=session.get("user_name"), |
|
similar_posts=similar_posts, |
|
) |
|
|
|
|
|
def get_similar_buyer_posts(seller_post_id): |
|
if seller_post_id: |
|
seller_post = mongo.db.customers.find_one({"_id": ObjectId(seller_post_id)}) |
|
if seller_post: |
|
seller_attributes = extract_attributes(seller_post["post"]) |
|
similar_buyer_posts = content_based_filtering_seller(seller_attributes) |
|
return similar_buyer_posts |
|
|
|
return [] |
|
|
|
|
|
def content_based_filtering_seller(seller_attributes): |
|
|
|
buyer_posts = mongo.db.customers.find({"label": "Buyer"}) |
|
|
|
|
|
most_similar_post = None |
|
highest_similarity_score = -1 |
|
|
|
for post in buyer_posts: |
|
buyer_attributes = extract_attributes( |
|
post["post"] |
|
) |
|
similarity_score = compute_similarity(seller_attributes, buyer_attributes) |
|
|
|
|
|
if similarity_score > highest_similarity_score: |
|
highest_similarity_score = similarity_score |
|
most_similar_post = post |
|
|
|
if most_similar_post: |
|
return [most_similar_post] |
|
else: |
|
return [] |
|
|
|
|
|
|
|
|
|
|
|
|
|
@app.route("/post_counts", methods=["GET"]) |
|
def post_counts(): |
|
seller_count = mongo.db.customers.count_documents({"label": "Seller"}) |
|
buyer_count = mongo.db.customers.count_documents({"label": "Buyer"}) |
|
return jsonify({"seller_posts": seller_count, "buyer_posts": buyer_count}) |
|
|
|
|
|
if __name__ == "__main__": |
|
try: |
|
with app.app_context(): |
|
|
|
if test_db_connection(): |
|
app.run() |
|
else: |
|
print("Database connection failed.") |
|
except Exception as e: |
|
print(e) |