190 lines
6.3 KiB
Python
190 lines
6.3 KiB
Python
import json
|
|
|
|
from flask import Blueprint, request, jsonify, send_file, Response
|
|
|
|
import utils
|
|
from services import FileService, OrganizationService, UserService, SessionService
|
|
|
|
file_bp = Blueprint("file", __name__)
|
|
upload_service = FileService()
|
|
|
|
|
|
@file_bp.route("/get/<string:file_handle>/content", methods=["GET"])
|
|
def file_get_content(file_handle: str):
|
|
file = FileService.get_file_by_file_handle(file_handle)
|
|
if not file:
|
|
return jsonify({"error": "File not found"}), 404
|
|
|
|
file_content = FileService.get_file_content(file)
|
|
return send_file(file_content, as_attachment=True, download_name=file.name)
|
|
|
|
|
|
@file_bp.route("/get/<string:document_handle>/metadata", methods=["GET"])
|
|
def file_get_metadata(document_handle: str):
|
|
session_token = request.headers.get("Authorization")
|
|
session = SessionService.validate_session(session_token)
|
|
if isinstance(session, tuple):
|
|
return session
|
|
|
|
org = OrganizationService.get_organization(session.org_id)
|
|
if not org:
|
|
return jsonify({"error": "Organization not found"}), 404
|
|
|
|
file = FileService.get_file_by_document_handle(document_handle)
|
|
if not file:
|
|
return jsonify({"error": "File not found"}), 404
|
|
|
|
return jsonify(file.to_dict())
|
|
|
|
|
|
@file_bp.route("/upload/metadata", methods=["POST"])
|
|
def file_upload_metadata():
|
|
session_token = request.headers.get("Authorization")
|
|
print(session_token)
|
|
session = SessionService.validate_session(session_token)
|
|
if isinstance(session, tuple):
|
|
return session
|
|
|
|
data = request.json
|
|
if type(data) is str:
|
|
data = json.loads(data)
|
|
if "document_name" not in data or "key" not in data or "alg" not in data or "nonce" not in data:
|
|
return jsonify({"error": "Missing required fields"}), 400
|
|
|
|
org = OrganizationService.get_organization(session.org_id)
|
|
if not org:
|
|
return jsonify({"error": "Organization not found"}), 404
|
|
|
|
user = UserService.get_user(session.user_id)
|
|
if not user:
|
|
return jsonify({"error": "User not found"}), 404
|
|
|
|
file = upload_service.create_file(session.token, org, user, data["document_name"], data["key"], data["alg"], data["nonce"])
|
|
return jsonify(file.to_dict()), 201
|
|
|
|
|
|
@file_bp.route("/upload/content", methods=["POST"])
|
|
def file_upload_content():
|
|
session_token = request.headers.get("Authorization")
|
|
if not session_token:
|
|
return jsonify({"error": "No session token"}), 400
|
|
|
|
session = SessionService.validate_session(session_token)
|
|
if isinstance(session, tuple):
|
|
return session
|
|
|
|
if "file" not in request.files:
|
|
return jsonify({"error": "No file data"}), 400
|
|
|
|
file = request.files.get("file")
|
|
if file.filename == "":
|
|
return jsonify({"error": "No file selected for uploading"}), 400
|
|
|
|
if not file:
|
|
return jsonify({"error": "Invalid file data"}), 400
|
|
|
|
file_data = utils.get_hex_from_temp_file(file.stream)
|
|
|
|
file_sum = request.headers.get("File-Checksum")
|
|
if not file_sum:
|
|
return jsonify({"error": "No file checksum provided"}), 400
|
|
|
|
if file_sum != str(utils.get_hash(file_data)):
|
|
return jsonify({"error": "File checksum mismatch"}), 400
|
|
|
|
file = upload_service.write_file(session_token, file_sum, file_data)
|
|
if isinstance(file, tuple):
|
|
return file
|
|
|
|
return jsonify(file.to_dict()), 201
|
|
|
|
|
|
@file_bp.route("/list", methods=["GET"])
|
|
def file_list():
|
|
session_token = request.headers.get("Authorization")
|
|
if not session_token:
|
|
return jsonify({"error": "No session token"}), 400
|
|
|
|
session = SessionService.validate_session(session_token)
|
|
if isinstance(session, tuple):
|
|
return session
|
|
|
|
|
|
data = request.json
|
|
if type(data) is str:
|
|
data = json.loads(data)
|
|
|
|
org = OrganizationService.get_organization(session.org_id)
|
|
if not org:
|
|
return jsonify({"error": "Organization not found"}), 404
|
|
|
|
if "datetime" in data:
|
|
try:
|
|
datetime_value = int(data["datetime"]["value"])
|
|
datetime_relation = data["datetime"]["relation"]
|
|
except ValueError:
|
|
return jsonify({"error": "Invalid datetime value"}), 400
|
|
|
|
if "username" in data:
|
|
user = UserService.get_user_by_username(data["username"])
|
|
if not user:
|
|
return jsonify({"error": "User not found"}), 404
|
|
files = FileService.list_files_in_org(org)
|
|
return jsonify([file.to_dict() for file in files if file.creator_id == user.id and (
|
|
utils.check_valid_time(file.created_at, datetime_value, datetime_relation)
|
|
if "datetime" in data else True
|
|
)])
|
|
|
|
files = FileService.list_files_in_org(org)
|
|
return jsonify([file.to_dict() for file in files if (utils.check_valid_time(file.created_at, datetime_value, datetime_relation) if "datetime" in data else True)])
|
|
|
|
|
|
@file_bp.route("/delete/<string:document_handle>", methods=["POST"])
|
|
def file_delete(document_handle: str):
|
|
session_token = request.headers.get("Authorization")
|
|
if not session_token:
|
|
return jsonify({"error": "No session token"}), 400
|
|
|
|
session = SessionService.validate_session(session_token)
|
|
if isinstance(session, tuple):
|
|
return session
|
|
|
|
org = OrganizationService.get_organization(session.org_id)
|
|
if not org:
|
|
return jsonify({"error": "Organization not found"}), 404
|
|
|
|
file = FileService.get_file_by_document_handle(document_handle)
|
|
if not file:
|
|
return jsonify({"error": "File not found"}), 404
|
|
|
|
if file.creator_id != session.user_id:
|
|
return jsonify({"error": "Not authorized to delete file"}), 403
|
|
|
|
file = FileService.delete_file(file)
|
|
return jsonify(file.to_dict())
|
|
|
|
|
|
################################################
|
|
|
|
|
|
@file_bp.route("/create_dummy", methods=["POST"])
|
|
def file_create_dummy():
|
|
session_token = request.headers.get("Authorization")
|
|
if not session_token:
|
|
return jsonify({"error": "No session token"}), 400
|
|
|
|
session = SessionService.validate_session(session_token)
|
|
|
|
if isinstance(session, tuple):
|
|
return session
|
|
|
|
org = OrganizationService.get_organization(session.org_id)
|
|
if not org:
|
|
return jsonify({"error": "Organization not found"}), 404
|
|
|
|
user = UserService.get_user(session.user_id)
|
|
if not user:
|
|
return jsonify({"error": "User not found"}), 404
|
|
|
|
file = FileService.create_dummy_file(org, user)
|
|
return jsonify(file.to_dict()), 201 |