Spaces:
Runtime error
Runtime error
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__) | |
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) | |