215 lines
8.4 KiB
Python
215 lines
8.4 KiB
Python
from flask import Blueprint, request, jsonify, abort
|
|
from flask_jwt_extended import jwt_required, get_jwt_identity
|
|
from models.order_in import OrdersIn
|
|
from models.transporters import Transporters
|
|
from models.user import Users
|
|
from datetime import datetime
|
|
import os
|
|
from flask import send_from_directory
|
|
import mimetypes
|
|
|
|
orders_in_bp = Blueprint("orders_in", __name__, url_prefix="/orders_in")
|
|
|
|
@orders_in_bp.route("/", methods=["POST"])
|
|
@jwt_required()
|
|
def create_order_in_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 = OrdersIn()
|
|
incoming_data = request.json
|
|
try:
|
|
order_data = {
|
|
'user_id': user_id,
|
|
'client_id': incoming_data["client_id"],
|
|
'received_price': incoming_data["received_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"],
|
|
'file': incoming_data['file'],
|
|
'expenses': incoming_data['expenses'],
|
|
'currency': incoming_data['currency']
|
|
}
|
|
#print(order_data)
|
|
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 in created", "order_id": order_id}), 201
|
|
except Exception as e:
|
|
return jsonify({"error": str(e)}), 400
|
|
|
|
@orders_in_bp.route("/<int:order_id>", methods=["PUT"])
|
|
@jwt_required()
|
|
def update_order_route(order_id):
|
|
orders = OrdersIn()
|
|
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 in 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"]),
|
|
"received_price": data.get("received_price", order["received_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"]),
|
|
"file": data.get("file", order["file"]),
|
|
"expenses": data.get("expenses", order["expenses"]),
|
|
"currency": data.get("currency", order["currency"]),
|
|
"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)
|
|
|
|
return jsonify({"message": "Order updated"}), 200
|
|
except Exception as e:
|
|
return jsonify({"error": str(e)}), 400
|
|
|
|
@orders_in_bp.route("/<int:order_id>", methods=["DELETE"])
|
|
@jwt_required()
|
|
def delete_order_route(order_id):
|
|
orders = OrdersIn()
|
|
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 in not found"}), 404
|
|
if str(order["user_id"]) != str(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 in deleted"}), 200
|
|
except Exception as e:
|
|
return jsonify({"error": str(e)}), 400
|
|
|
|
@orders_in_bp.route("/list", methods=["GET"])
|
|
@jwt_required()
|
|
def list_orders():
|
|
orders = OrdersIn()
|
|
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_in_bp.route("/<int:order_id>", methods=["GET"])
|
|
@jwt_required()
|
|
def get_order(order_id):
|
|
orders = OrdersIn()
|
|
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_in_bp.route("/files/<path:filename>", methods=["GET"])
|
|
#@jwt_required()
|
|
def serve_order_pdf(filename):
|
|
try:
|
|
# Directory containing uploaded client files
|
|
uploads_dir = os.path.abspath(
|
|
os.path.join(os.path.dirname(__file__), "..", "..", "client", "uploads")
|
|
)
|
|
# Security: prevent path traversal and ensure file exists
|
|
abs_file_path = os.path.abspath(os.path.join(uploads_dir, filename))
|
|
if not abs_file_path.startswith(uploads_dir + os.sep):
|
|
abort(404)
|
|
if not os.path.isfile(abs_file_path):
|
|
abort(404)
|
|
|
|
guessed_type = mimetypes.guess_type(filename)[0] or "application/octet-stream"
|
|
# send_from_directory expects the directory and the filename relative to it
|
|
return send_from_directory(uploads_dir, filename, mimetype=guessed_type, as_attachment=False)
|
|
except Exception as e:
|
|
print(e)
|
|
return jsonify({"error": "File not found"}), 404 |