290 lines
11 KiB
Python
290 lines
11 KiB
Python
from flask import Blueprint, request, jsonify
|
|
from flask_jwt_extended import jwt_required, get_jwt_identity
|
|
from models.order_out import OrdersOut
|
|
from models.user import Users
|
|
from models.transporters import Transporters
|
|
from datetime import datetime
|
|
from utils.pdf import generate_order_pdf
|
|
from utils.cancel_order import cancel_order_pdf
|
|
import os
|
|
|
|
from flask import send_from_directory
|
|
from utils.email import send_gmail_with_attachment, send_custom_email_with_attachment
|
|
|
|
orders_bp = Blueprint("orders", __name__, url_prefix="/orders")
|
|
|
|
@orders_bp.route("/", methods=["POST"])
|
|
@jwt_required()
|
|
def create_order_route():
|
|
user_id = get_jwt_identity()
|
|
users = Users()
|
|
user = users.get_user_by_id(user_id)
|
|
if user['user_role'] == 'company_user':
|
|
user_id = user['company_id']
|
|
orders = OrdersOut()
|
|
incoming_data = request.json
|
|
#here we need to first implement the order pdf
|
|
users = Users()
|
|
user = users.get_user_by_id(user_id)
|
|
logo_filename = user.get('logo_filename')
|
|
logo_path = None
|
|
if logo_filename:
|
|
logo_path = os.path.abspath(
|
|
os.path.join(os.path.dirname(__file__), "..", "..", "client", "assets", "images", logo_filename)
|
|
)
|
|
transporters = Transporters()
|
|
transporter = transporters.get_transporter_by_id(incoming_data["transporter_id"])
|
|
generate_order_pdf(order=incoming_data, user_data=user, transporter_data=transporter, logo_path=logo_path)
|
|
#
|
|
#try:
|
|
order_data = {
|
|
'user_id': user_id,
|
|
'client_id': incoming_data["client_id"],
|
|
'transporter_id': incoming_data["transporter_id"],
|
|
'received_price': incoming_data["received_price"],
|
|
'paid_price': incoming_data["paid_price"],
|
|
'order_number': incoming_data["order_number"],
|
|
'created_at': datetime.now(),
|
|
'ldb_quantity': incoming_data["ldb_quantity"],
|
|
'kg_quantity': incoming_data["kg_quantity"],
|
|
'track_reg_number': incoming_data["track_reg_number"],
|
|
'trailer_reg_number': incoming_data["trailer_reg_number"],
|
|
'products_description': incoming_data["products_description"],
|
|
'order_in_number': incoming_data["order_in_number"],
|
|
'currency_received': incoming_data["currency_received"],
|
|
'currency_paid': incoming_data["currency_paid"]
|
|
}
|
|
order_id = orders.create_order(order_data)
|
|
|
|
for address in incoming_data["loading_addresses"]:
|
|
data = {
|
|
"order_id": order_id,
|
|
"destination_id": address['loading_address_id'],
|
|
"informatins": address['loading_informatins'],
|
|
"point_data": address['loading_date'],
|
|
"point_hour": address['loading_hour'],
|
|
"point_type": "loading"
|
|
}
|
|
orders.create_order_point(data)
|
|
|
|
for address in incoming_data["unloading_addresses"]:
|
|
data = {
|
|
"order_id": order_id,
|
|
"destination_id": address['unloading_address_id'],
|
|
"informatins": address['unloading_informatins'],
|
|
"point_data": address['unloading_date'],
|
|
"point_hour": address['unloading_hour'],
|
|
"point_type": "unloading"
|
|
}
|
|
orders.create_order_point(data)
|
|
|
|
|
|
|
|
return jsonify({"message": "Order created", "order_id": order_id}), 201
|
|
#except Exception as e:
|
|
# return jsonify({"error": str(e)}), 400
|
|
|
|
@orders_bp.route("/<int:order_id>", methods=["PUT"])
|
|
@jwt_required()
|
|
def update_order_route(order_id):
|
|
orders = OrdersOut()
|
|
data = request.json
|
|
user_id = get_jwt_identity()
|
|
users = Users()
|
|
user = users.get_user_by_id(user_id)
|
|
if user['user_role'] == 'company_user':
|
|
user_id = user['company_id']
|
|
order = orders.get_order_by_id(order_id)
|
|
if not order:
|
|
return jsonify({"error": "Order not found"}), 404
|
|
if str(order["user_id"]) != str(user_id):
|
|
return jsonify({"error": "Unauthorized"}), 403
|
|
|
|
try:
|
|
orders.update_order({
|
|
"id":data.get("id", order['id']),
|
|
"client_id": data.get("client_id", order["client_id"]),
|
|
"transporter_id": data.get("transporter_id", order["transporter_id"]),
|
|
"received_price": data.get("received_price", order["received_price"]),
|
|
"paid_price": data.get("paid_price", order["paid_price"]),
|
|
"order_number": data.get("order_number", order["order_number"]),
|
|
"ldb_quantity": data.get("ldb_quantity", order["ldb_quantity"]),
|
|
"kg_quantity": data.get("kg_quantity", order["kg_quantity"]),
|
|
"track_reg_number": data.get("track_reg_number", order["track_reg_number"]),
|
|
"trailer_reg_number": data.get("trailer_reg_number", order["trailer_reg_number"]),
|
|
"products_description": data.get("products_description", order["products_description"]),
|
|
"order_in_number": data.get("order_in_number", order["order_in_number"]),
|
|
"currency_received":data.get("currency_received", order["currency_received"]),
|
|
"currency_paid":data.get("currency_paid", order["currency_paid"]),
|
|
"user_id":user_id,
|
|
})
|
|
|
|
orders.delete_points_by_order_id(order_id)
|
|
for address in data["loading_addresses"]:
|
|
loading_data = {
|
|
"order_id": order_id,
|
|
"destination_id": address['loading_address_id'],
|
|
"informatins": address['loading_informatins'],
|
|
"point_data": address['loading_date'],
|
|
"point_hour": address['loading_hour'],
|
|
"point_type": "loading"
|
|
}
|
|
orders.create_order_point(loading_data)
|
|
|
|
for address in data["unloading_addresses"]:
|
|
unloading_data = {
|
|
"order_id": order_id,
|
|
"destination_id": address['unloading_address_id'],
|
|
"informatins": address['unloading_informatins'],
|
|
"point_data": address['unloading_date'],
|
|
"point_hour": address['unloading_hour'],
|
|
"point_type": "unloading"
|
|
}
|
|
orders.create_order_point(unloading_data)
|
|
|
|
#regenerate pdf:
|
|
incoming_data = data
|
|
users = Users()
|
|
user = users.get_user_by_id(user_id)
|
|
transporters = Transporters()
|
|
transporter = transporters.get_transporter_by_id(incoming_data["transporter_id"])
|
|
logo_filename = user.get('logo_filename')
|
|
logo_path = None
|
|
if logo_filename:
|
|
logo_path = os.path.abspath(
|
|
os.path.join(os.path.dirname(__file__), "..", "..", "client", "assets", "images", logo_filename)
|
|
)
|
|
generate_order_pdf(order=incoming_data, user_data=user, transporter_data=transporter, logo_path=logo_path)
|
|
|
|
return jsonify({"message": "Order updated", "order_id": order_id}), 200
|
|
except Exception as e:
|
|
return jsonify({"error": str(e)}), 400
|
|
|
|
@orders_bp.route("/<int:order_id>", methods=["DELETE"])
|
|
@jwt_required()
|
|
def delete_order_route(order_id):
|
|
orders = OrdersOut()
|
|
user_id = get_jwt_identity()
|
|
users = Users()
|
|
user = users.get_user_by_id(user_id)
|
|
if user['user_role'] == 'company_user':
|
|
user_id = user['company_id']
|
|
order = orders.get_order_by_id(order_id)
|
|
if not order:
|
|
return jsonify({"error": "Order not found"}), 404
|
|
if order["user_id"] != user_id:
|
|
return jsonify({"error": "Unauthorized"}), 403
|
|
|
|
try:
|
|
orders.delete_points_by_order_id(order_id)
|
|
orders.delete_order(order_id)
|
|
return jsonify({"message": "Order deleted"}), 200
|
|
except Exception as e:
|
|
return jsonify({"error": str(e)}), 400
|
|
|
|
@orders_bp.route("/list", methods=["GET"])
|
|
@jwt_required()
|
|
def list_orders():
|
|
orders = OrdersOut()
|
|
user_id = get_jwt_identity()
|
|
users = Users()
|
|
user = users.get_user_by_id(user_id)
|
|
if user['user_role'] == 'company_user':
|
|
user_id = user['company_id']
|
|
try:
|
|
user_orders = orders.get_orders_by_user(user_id)
|
|
#result = [{"id": order["id"], "order_number": order["order_number"]} for order in user_orders]
|
|
return jsonify(user_orders), 200
|
|
except Exception as e:
|
|
return jsonify({"error": str(e)}), 400
|
|
|
|
@orders_bp.route("/<int:order_id>", methods=["GET"])
|
|
@jwt_required()
|
|
def get_order(order_id):
|
|
orders = OrdersOut()
|
|
user_id = get_jwt_identity()
|
|
users = Users()
|
|
user = users.get_user_by_id(user_id)
|
|
if user['user_role'] == 'company_user':
|
|
user_id = user['company_id']
|
|
order = orders.get_order_by_id(order_id)
|
|
points = orders.get_order_points_by_order(order['id'])
|
|
loading_points = []
|
|
unloading_points = []
|
|
for point in points:
|
|
if point['point_type'] == 'loading':
|
|
loading_points.append(point)
|
|
else:
|
|
unloading_points.append(point)
|
|
order['loading_points'] = loading_points
|
|
order['unloading_points'] = unloading_points
|
|
if not order:
|
|
return jsonify({"error": "Order not found"}), 404
|
|
print(f'{order["user_id"]} {user_id}')
|
|
print(f'{type(order["user_id"])} {type(user_id)}')
|
|
if order["user_id"] != int(user_id):
|
|
return jsonify({"error": "Unauthorized"}), 403
|
|
return jsonify(order), 200
|
|
|
|
@orders_bp.route("/pdfs/<path:filename>", methods=["GET"])
|
|
#@jwt_required()
|
|
def serve_order_pdf(filename):
|
|
pdf_folder = os.path.abspath(os.path.join(os.path.dirname(__file__), "..", "generated_pdfs"))
|
|
print(pdf_folder)
|
|
print(filename)
|
|
return send_from_directory(pdf_folder, filename, mimetype="application/pdf")
|
|
|
|
@orders_bp.route("/send-email/gmail", methods=["POST"])
|
|
@jwt_required()
|
|
def send_email_with_gmail():
|
|
data = request.json
|
|
try:
|
|
to_email = data["to_email"]
|
|
subject = data["subject"]
|
|
body = data["body"]
|
|
filename = data["filename"]
|
|
attachment_path = os.path.abspath(os.path.join(os.path.dirname(__file__), "..", "uploads", filename))
|
|
|
|
send_gmail_with_attachment(to_email, subject, body, attachment_path)
|
|
return jsonify({"message": "Email sent successfully using Gmail"}), 200
|
|
except Exception as e:
|
|
return jsonify({"error": str(e)}), 400
|
|
|
|
@orders_bp.route("/send-email/custom", methods=["POST"])
|
|
@jwt_required()
|
|
def send_email_with_custom_smtp():
|
|
data = request.json
|
|
try:
|
|
to_email = data["to_email"]
|
|
subject = data["subject"]
|
|
body = data["body"]
|
|
filename = data["filename"]
|
|
smtp_host = data["smtp_host"]
|
|
smtp_port = data["smtp_port"]
|
|
smtp_user = data["smtp_user"]
|
|
smtp_pass = data["smtp_pass"]
|
|
|
|
attachment_path = os.path.abspath(os.path.join(os.path.dirname(__file__), "..", "uploads", filename))
|
|
|
|
send_custom_email_with_attachment(to_email, subject, body, attachment_path, smtp_host, smtp_port, smtp_user, smtp_pass)
|
|
return jsonify({"message": "Email sent successfully using custom SMTP"}), 200
|
|
except Exception as e:
|
|
return jsonify({"error": str(e)}), 400
|
|
|
|
@orders_bp.route("/cancel/<int:order_id>", methods=["DELETE"])
|
|
@jwt_required()
|
|
def cancel_order(order_id):
|
|
try:
|
|
orders = OrdersOut()
|
|
order = orders.get_order_by_id(order_id)
|
|
user_id = get_jwt_identity()
|
|
users = Users()
|
|
user = users.get_user_by_id(user_id)
|
|
if user['user_role'] == 'company_user':
|
|
user_id = user['company_id']
|
|
pdf_name = f'order_{user_id}_{order['order_number']}.pdf'
|
|
cancel_order_pdf(pdf_name)
|
|
orders.cancel_order(order_id)
|
|
return jsonify({"message": "The order was successfully canceled!"}), 200
|
|
except Exception as e:
|
|
return jsonify({"error": str(e)}), 500 |