job persitance andupdates
This commit is contained in:
1 parent
40f65be11b
commit
a438493a76
2 files changed
+63
-2
No files matched your search
Binary file not shown.
+63
-2
@@ -3,6 +3,7 @@ import tempfile
|
|||||||
import zipfile
|
import zipfile
|
||||||
import threading
|
import threading
|
||||||
import uuid
|
import uuid
|
||||||
|
import json
|
||||||
from datetime import datetime
|
from datetime import datetime
|
||||||
from functools import wraps
|
from functools import wraps
|
||||||
from flask import Flask, request, render_template_string, redirect, url_for, session, send_from_directory, flash, get_flashed_messages, jsonify
|
from flask import Flask, request, render_template_string, redirect, url_for, session, send_from_directory, flash, get_flashed_messages, jsonify
|
||||||
@@ -54,6 +55,56 @@ except Exception as e:
|
|||||||
print(f"ERROR: Failed to register Microsoft OAuth provider: {e}")
|
print(f"ERROR: Failed to register Microsoft OAuth provider: {e}")
|
||||||
microsoft = None
|
microsoft = None
|
||||||
|
|
||||||
|
# Job persistence
|
||||||
|
JOBS_FILE = os.path.join(os.path.dirname(os.path.abspath(__file__)), 'jobs.json')
|
||||||
|
job_queue = {}
|
||||||
|
job_lock = threading.Lock()
|
||||||
|
|
||||||
|
def save_jobs():
|
||||||
|
"""Save jobs to persistent storage"""
|
||||||
|
try:
|
||||||
|
with job_lock:
|
||||||
|
# Convert datetime objects to strings for JSON serialization
|
||||||
|
jobs_to_save = {}
|
||||||
|
for job_id, job in job_queue.items():
|
||||||
|
jobs_to_save[job_id] = job.copy()
|
||||||
|
# Convert any datetime objects if needed
|
||||||
|
if 'created_at' in jobs_to_save[job_id]:
|
||||||
|
jobs_to_save[job_id]['created_at'] = str(jobs_to_save[job_id]['created_at'])
|
||||||
|
|
||||||
|
with open(JOBS_FILE, 'w') as f:
|
||||||
|
json.dump(jobs_to_save, f, indent=2)
|
||||||
|
except Exception as e:
|
||||||
|
print(f"ERROR: Failed to save jobs: {e}")
|
||||||
|
|
||||||
|
def load_jobs():
|
||||||
|
"""Load jobs from persistent storage"""
|
||||||
|
global job_queue
|
||||||
|
try:
|
||||||
|
if os.path.exists(JOBS_FILE):
|
||||||
|
with open(JOBS_FILE, 'r') as f:
|
||||||
|
loaded_jobs = json.load(f)
|
||||||
|
job_queue = loaded_jobs
|
||||||
|
print(f"DEBUG: Loaded {len(job_queue)} jobs from persistent storage")
|
||||||
|
else:
|
||||||
|
print("DEBUG: No existing jobs file found, starting fresh")
|
||||||
|
except Exception as e:
|
||||||
|
print(f"ERROR: Failed to load jobs: {e}")
|
||||||
|
job_queue = {}
|
||||||
|
|
||||||
|
# Load jobs on startup
|
||||||
|
load_jobs()
|
||||||
|
|
||||||
|
def update_job_status(job_id, status, **kwargs):
|
||||||
|
"""Update job status and persist changes"""
|
||||||
|
with job_lock:
|
||||||
|
if job_id in job_queue:
|
||||||
|
job_queue[job_id]['status'] = status
|
||||||
|
job_queue[job_id].update(kwargs)
|
||||||
|
job_queue[job_id]['updated_at'] = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
|
||||||
|
save_jobs()
|
||||||
|
print(f"DEBUG: Job {job_id} status updated to {status}")
|
||||||
|
|
||||||
def analyze_failed_extractions(log_path):
|
def analyze_failed_extractions(log_path):
|
||||||
"""
|
"""
|
||||||
Analyze failed extraction log to identify new patterns.
|
Analyze failed extraction log to identify new patterns.
|
||||||
@@ -459,6 +510,7 @@ def process_job(job_id, files):
|
|||||||
print("=== DEBUG: process_job STARTED ===")
|
print("=== DEBUG: process_job STARTED ===")
|
||||||
job_queue[job_id]['status'] = 'processing'
|
job_queue[job_id]['status'] = 'processing'
|
||||||
job_queue[job_id]['progress'] = 'Starting job processing...'
|
job_queue[job_id]['progress'] = 'Starting job processing...'
|
||||||
|
save_jobs() # Persist status change
|
||||||
processed_files = []
|
processed_files = []
|
||||||
temp_extract_dirs = []
|
temp_extract_dirs = []
|
||||||
|
|
||||||
@@ -614,6 +666,7 @@ def process_job(job_id, files):
|
|||||||
except Exception as e:
|
except Exception as e:
|
||||||
job_queue[job_id]['progress'] = f'Error creating ZIP: {str(e)}'
|
job_queue[job_id]['progress'] = f'Error creating ZIP: {str(e)}'
|
||||||
job_queue[job_id]['status'] = 'error'
|
job_queue[job_id]['status'] = 'error'
|
||||||
|
save_jobs() # Persist error status
|
||||||
print(f"DEBUG: ZIP creation failed: {str(e)}")
|
print(f"DEBUG: ZIP creation failed: {str(e)}")
|
||||||
else:
|
else:
|
||||||
job_queue[job_id]['progress'] = 'No files to archive - all files were skipped'
|
job_queue[job_id]['progress'] = 'No files to archive - all files were skipped'
|
||||||
@@ -622,11 +675,13 @@ def process_job(job_id, files):
|
|||||||
job_queue[job_id]['status'] = 'completed'
|
job_queue[job_id]['status'] = 'completed'
|
||||||
job_queue[job_id]['processed_files'] = processed_files
|
job_queue[job_id]['processed_files'] = processed_files
|
||||||
job_queue[job_id]['skipped'] = [(p[1], p[2]) for p in processed_files if p[0] == 'skipped']
|
job_queue[job_id]['skipped'] = [(p[1], p[2]) for p in processed_files if p[0] == 'skipped']
|
||||||
|
save_jobs() # Persist completed job
|
||||||
print(f"DEBUG: Job {job_id} completed successfully. Processed: {len(processed_files)}, Skipped: {len(job_queue[job_id]['skipped'])}")
|
print(f"DEBUG: Job {job_id} completed successfully. Processed: {len(processed_files)}, Skipped: {len(job_queue[job_id]['skipped'])}")
|
||||||
|
|
||||||
except Exception as e:
|
except Exception as e:
|
||||||
job_queue[job_id]['status'] = 'error'
|
job_queue[job_id]['status'] = 'error'
|
||||||
job_queue[job_id]['progress'] = f'Job failed with error: {str(e)}'
|
job_queue[job_id]['progress'] = f'Job failed with error: {str(e)}'
|
||||||
|
save_jobs() # Persist error status
|
||||||
print(f"Job {job_id} failed: {str(e)}") # Debug logging
|
print(f"Job {job_id} failed: {str(e)}") # Debug logging
|
||||||
|
|
||||||
@app.route('/login')
|
@app.route('/login')
|
||||||
@@ -776,7 +831,8 @@ def upload_files():
|
|||||||
flash('No files selected.', 'error')
|
flash('No files selected.', 'error')
|
||||||
return redirect(url_for('upload_files'))
|
return redirect(url_for('upload_files'))
|
||||||
|
|
||||||
job_id = str(uuid.uuid4())
|
# Create job ID using timestamp format: YYYY-MM-DD_HH-MM-SS
|
||||||
|
job_id = datetime.now().strftime('%Y-%m-%d_%H-%M-%S')
|
||||||
job_queue[job_id] = {
|
job_queue[job_id] = {
|
||||||
'status': 'queued',
|
'status': 'queued',
|
||||||
'files': [],
|
'files': [],
|
||||||
@@ -784,7 +840,8 @@ def upload_files():
|
|||||||
'zip_name': None,
|
'zip_name': None,
|
||||||
'processed_files': [],
|
'processed_files': [],
|
||||||
'skipped': [],
|
'skipped': [],
|
||||||
'created_at': datetime.now().strftime('%Y-%m-%d %H:%M:%S')
|
'created_at': datetime.now().strftime('%Y-%m-%d %H:%M:%S'),
|
||||||
|
'job_id': job_id # Store the job ID for reference
|
||||||
}
|
}
|
||||||
|
|
||||||
file_list = []
|
file_list = []
|
||||||
@@ -804,6 +861,9 @@ def upload_files():
|
|||||||
job_queue[job_id]['files'] = file_list
|
job_queue[job_id]['files'] = file_list
|
||||||
print(f"Created job {job_id} with {len(file_list)} files") # Debug logging
|
print(f"Created job {job_id} with {len(file_list)} files") # Debug logging
|
||||||
|
|
||||||
|
# Save job to persistent storage
|
||||||
|
save_jobs()
|
||||||
|
|
||||||
# Start processing in background
|
# Start processing in background
|
||||||
def process_with_context():
|
def process_with_context():
|
||||||
with app.app_context():
|
with app.app_context():
|
||||||
@@ -1109,6 +1169,7 @@ def delete_job(job_id):
|
|||||||
os.remove(zip_path)
|
os.remove(zip_path)
|
||||||
# Remove the job from the queue
|
# Remove the job from the queue
|
||||||
del job_queue[job_id]
|
del job_queue[job_id]
|
||||||
|
save_jobs() # Persist job deletion
|
||||||
flash('Job deleted successfully.', 'success')
|
flash('Job deleted successfully.', 'success')
|
||||||
else:
|
else:
|
||||||
flash('Only completed jobs can be deleted.', 'error')
|
flash('Only completed jobs can be deleted.', 'error')
|
||||||
|
|||||||
Reference in new issue
Block a user