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) | |