Spaces:
Runtime error
Runtime error
File size: 2,961 Bytes
975d1b2 4ad299d e4dba65 975d1b2 e4dba65 975d1b2 4ad299d e4dba65 975d1b2 e4dba65 4ad299d 975d1b2 e4dba65 975d1b2 e4dba65 975d1b2 e4dba65 4ad299d e4dba65 975d1b2 e4dba65 975d1b2 e4dba65 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 |
import os
import zipfile
import shutil
import datetime
from flask import Blueprint, render_template, request, send_file, jsonify, current_app
from .utils.zip_handler import handle_zip_upload
from .utils.vector_db import process_files_to_vectors
v_bp = Blueprint('routes', __name__)
@v_bp.route('/', methods=['GET', 'POST'])
def home():
if request.method == 'POST':
# 1. Validate the uploaded file
v_uploaded_file = request.files.get('file')
if not v_uploaded_file or not v_uploaded_file.filename.endswith('.zip'):
return jsonify({'error': 'Please upload a valid zip file (.zip).'}), 400
# 2. Get the uploads folder from Flask config
v_uploads_folder = current_app.config['UPLOAD_FOLDER']
# Example: "app/uploads"
# 3. Clean up the existing contents in uploads folder
cleanup_uploads_folder(v_uploads_folder)
os.makedirs(v_uploads_folder, exist_ok=True)
# 4. Save the new ZIP file into the uploads folder
v_upload_path = os.path.join(v_uploads_folder, v_uploaded_file.filename)
v_uploaded_file.save(v_upload_path)
# 5. Extract the ZIP file (directly into app/uploads)
handle_zip_upload(v_upload_path)
# 6. Process all extracted files to create/update vector DB in app/uploads/vectors
process_files_to_vectors(v_uploads_folder)
# 7. Build a timestamped filename for the final zip
v_timestamp = datetime.datetime.now().strftime("%Y%m%d_%H%M%S")
v_zip_filename = f"uploads_all_{v_timestamp}.zip"
# Final zip will be inside app/uploads
v_final_zip_path = os.path.join(v_uploads_folder, v_zip_filename)
# 8. Zip the entire app/uploads folder (skip the final zip to prevent recursion)
with zipfile.ZipFile(v_final_zip_path, 'w', zipfile.ZIP_DEFLATED) as obj_zip:
for v_root, _, v_files in os.walk(v_uploads_folder):
for v_file in v_files:
# Exclude the new zip itself from being re-zipped
if v_file == v_zip_filename:
continue
v_full_path = os.path.join(v_root, v_file)
v_arcname = os.path.relpath(v_full_path, start=v_uploads_folder)
obj_zip.write(v_full_path, arcname=v_arcname)
# 9. Send the final zip file to the user
return send_file(v_final_zip_path, as_attachment=True)
# If GET, render the upload form
return render_template('index.html')
def cleanup_uploads_folder(v_folder_path):
"""
Deletes all files/folders inside v_folder_path.
"""
if os.path.exists(v_folder_path):
for v_item in os.listdir(v_folder_path):
v_item_path = os.path.join(v_folder_path, v_item)
if os.path.isfile(v_item_path):
os.remove(v_item_path)
elif os.path.isdir(v_item_path):
shutil.rmtree(v_item_path)
|