ddp/photoassistant
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Huge change, removed Status class and all "alert" messages are now shown as BS toast() and are via the DB and handled async in the F/E in jscript via Ajax. Fixed BUG-113 where toasts() were repeating. Removed many of the explicit alert messages (other than errors) and hooked {New|Finish}Job to consistently send messages to the F/E. Other messages (F/E without a job, like save settings) now use this model as well. Finally converted most of the older POST responses to formal json

Browse Source
This commit is contained in:
Damien De Paoli
2023-01-11 13:50:05 +11:00
parent 2be2c504b2
commit a29cbb143c
15 changed files with 162 additions and 217 deletions
Download Patch File Download Diff File Expand all files Collapse all files

78
pa_job_manager.py
View File

@@ -494,19 +494,19 @@ class Job(Base):
##############################################################################
# Class describing PA_JobManager_FE_Message and in the DB (via sqlalchemy)
# the job manager (this code) can send a message back to the front end via the
# DB. has to be about a specific job_id and is success/danger, etc. (alert)
# DB. has to be about a specific job_id and is success/danger, etc. (level)
# and a message
################################################################################
class PA_JobManager_FE_Message(Base):
__tablename__ = "pa_job_manager_fe_message"
id = Column(Integer, Sequence('pa_job_manager_fe_message_id_seq'), primary_key=True )
job_id = Column(Integer, ForeignKey('job.id') )
alert = Column(String)
level = Column(String)
message = Column(String)
persistent = Column(Boolean)
cant_close = Column(Boolean)
def __repr__(self):
return "<id: {}, job_id: {}, alert: {}, message: {}".format(self.id, self.job_id, self.alert, self.message)
return "<id: {}, job_id: {}, level: {}, message: {}".format(self.id, self.job_id, self.level, self.message)
class PA_UserState(Base):
@@ -542,7 +542,7 @@ class PA_UserState(Base):
##############################################################################
# NewJob(): convenience function to create a job, appropriately
##############################################################################
def NewJob(name, num_files=0, wait_for=None, jex=None, parent_job=None ):
def NewJob(name, num_files=0, wait_for=None, jex=None, parent_job=None, desc="No description provided" ):
job=Job( name=name, current_file_num=0, current_file='', num_files=num_files,
wait_for=wait_for, state="New", pa_job_state="New", start_time=None,
last_update=datetime.now(pytz.utc) )
@@ -552,6 +552,8 @@ def NewJob(name, num_files=0, wait_for=None, jex=None, parent_job=None ):
session.add(job)
session.commit()
MessageToFE( job_id=job.id, message=f'Created <a class="link-light" href="/job/{job.id}">Job #{job.id}</a> to {desc}',
level="success", persistent=False, cant_close=False )
if parent_job:
str=f"adding <a href='/job/{job.id}'>job id={job.id} {job.name}</a>"
if job.wait_for:
@@ -560,11 +562,11 @@ def NewJob(name, num_files=0, wait_for=None, jex=None, parent_job=None ):
return job
##############################################################################
# MessageToFE(): sends a specific alert/messasge for a given job via the DB to
# MessageToFE(): sends a specific level/messasge for a given job via the DB to
# the front end
##############################################################################
def MessageToFE( job_id, alert, message, persistent, cant_close ):
msg = PA_JobManager_FE_Message( job_id=job_id, alert=alert, message=message, persistent=persistent, cant_close=cant_close)
def MessageToFE( job_id, message, level, persistent, cant_close ):
msg = PA_JobManager_FE_Message( job_id=job_id, message=message, level=level, persistent=persistent, cant_close=cant_close)
session.add(msg)
session.commit()
return msg.id
@@ -686,22 +688,22 @@ def JobsForPath( parent_job, path, ptype ):
jex=[]
jex.append( JobExtra( name="path", value=path ) )
jex.append( JobExtra( name="path_type", value=ptype.id ) )
job1=NewJob( "importdir", cfn, None, jex, parent_job )
job1=NewJob( name="importdir", num_files=cfn, wait_for=None, jex=jex, parent_job=parent_job, desc=f"scan for files from {ptype.name} path" )
# then get file details (hash/thumbs)
jex=[]
jex.append( JobExtra( name="path", value=path ) )
job2=NewJob("getfiledetails", 0, job1.id, jex, parent_job )
job2=NewJob( name="getfiledetails", num_files=0, wait_for=job1.id, jex=jex, parent_job=parent_job, desc=f"get details of files from {ptype.name} path" )
# can start straight after importdir - job1, does not need details (job2)
jex=[]
jex.append( JobExtra( name="person", value="all" ) )
jex.append( JobExtra( name="path_type", value=ptype.id ) )
job3=NewJob("run_ai_on_path", 0, job1.id, jex, parent_job )
job3=NewJob( name="run_ai_on_path", num_files=0, wait_for=job1.id, jex=jex, parent_job=parent_job, desc=f"match faces on files from {ptype.name} path" )
# careful here, wait for getfiledetails (job2), the ai job cannot cause a dup
# but it can fail - in which case the checkdup will be withdrawn
job4=NewJob( "checkdups", 0, job2.id, None, parent_job )
job4=NewJob( name="checkdups", num_files=0, wait_for=job2.id, jex=None, parent_job=parent_job, desc="check for duplicate files" )
# okay, now process all the new jobs
HandleJobs(False)
@@ -832,8 +834,8 @@ def RunJob(job):
# only update start_time if we have never set it - stops restarts resetting start_time
if not job.start_time:
job.start_time=datetime.now(pytz.utc)
if job.name =="scannow":
JobScanNow(job)
if job.name =="scan_ip":
JobScanImportDir(job)
elif job.name =="forcescan":
JobForceScan(job)
elif job.name =="scan_sp":
@@ -874,7 +876,7 @@ def RunJob(job):
##############################################################################
# FinishJob(): finish this job off (if no overrides), its just marked completed
##############################################################################
def FinishJob(job, last_log, state="Completed", pa_job_state="Completed"):
def FinishJob(job, last_log, state="Completed", pa_job_state="Completed", level="success", persistent=False, cant_close=False):
job.state=state
job.pa_job_state=pa_job_state
if not job.start_time:
@@ -884,6 +886,7 @@ def FinishJob(job, last_log, state="Completed", pa_job_state="Completed"):
if job.state=="Failed":
WithdrawDependantJobs( job, job.id, "failed" )
session.commit()
MessageToFE( job_id=job.id, message=last_log, level=level, persistent=persistent, cant_close=cant_close )
if DEBUG:
print( f"DEBUG: {last_log}" )
return
@@ -906,7 +909,7 @@ def HandleJobs(first_run=False):
job.pa_job_state = 'Stale'
session.add(job)
AddLogForJob( job, "ERROR: Job has been marked stale as it did not complete" )
MessageToFE( job.id, "danger", f'Stale job, click&nbsp; <a href="javascript:document.body.innerHTML+=\'<form id=_fm method=GET action=/stale_jobs></form>\'; document.getElementById(\'_fm\').submit();">here</a>&nbsp;to restart or cancel', True, False )
MessageToFE( job_id=job.id, message=f'Stale job, click&nbsp; <a class="link-light" href="javascript:document.body.innerHTML+=\'<form id=_fm method=GET action=/stale_jobs></form>\'; document.getElementById(\'_fm\').submit();">here</a>&nbsp;to restart or cancel', level="danger", persistent=True, cant_close=False )
session.commit()
continue
if job.pa_job_state == 'New':
@@ -932,7 +935,7 @@ def HandleJobs(first_run=False):
# threading.Thread(target=RunJob, args=(job,)).start()
except Exception as e:
try:
MessageToFE( job.id, "danger", "Failed with: {} (try job log for details)".format(e), True, False )
MessageToFE( job_id=job.id, level="danger", message="Failed with: {} (try job log for details)".format(e), persistent=True, cant_close=False )
except Exception as e2:
print("ERROR: Failed to let front-end know, but back-end Failed to run job (id: {}, name: {} -- orig exep was: {}, this exception was: {})".format( job.id, job.name, e, e2) )
print("INFO: PA job manager is waiting for a job")
@@ -950,13 +953,12 @@ def JobProgressState( job, state ):
return
##############################################################################
# JobScanNow(): start and process the job to start scanning now (import paths)
# JobScanImportDir(): start and process the job to start scanning now (import paths)
##############################################################################
def JobScanNow(job):
def JobScanImportDir(job):
JobProgressState( job, "In Progress" )
ProcessImportDirs(job)
FinishJob( job, "Completed (scan for new files)" )
MessageToFE( job.id, "success", "Completed (scan for new files)", False, False )
return
##############################################################################
@@ -966,7 +968,6 @@ def JobScanStorageDir(job):
JobProgressState( job, "In Progress" )
ProcessStorageDirs(job)
FinishJob( job, "Completed (scan for new files)" )
MessageToFE( job.id, "success", "Completed (scan for new files)", False, False )
return
@@ -1080,7 +1081,6 @@ def JobForceScan(job):
ProcessImportDirs(job)
ProcessStorageDirs(job)
FinishJob(job, "Completed (forced remove and recreation of all file data)")
MessageToFE( job.id, "success", "Completed (forced remove and recreation of all file data)", False, False )
return
##############################################################################
@@ -1553,7 +1553,7 @@ def AddJexToDependantJobs(job,name,value):
####################################################################################################################################
def WithdrawDependantJobs( job, id, reason ):
for j in session.query(Job).filter(Job.wait_for==id).all():
FinishJob(j, f"Job (#{j.id}) has been withdrawn -- #{job.id} {reason}", "Withdrawn" )
FinishJob(j, f"Job #{j.id} has been withdrawn -- #{job.id} {reason}", "Withdrawn" )
WithdrawDependantJobs(j, j.id, reason)
return
@@ -1722,17 +1722,18 @@ def JobImportDir(job):
for j in session.query(Job).filter(Job.wait_for==job.id).all():
if j.name == "getfiledetails" and last_file_details > last_scan:
FinishJob(j, f"Job (#{j.id}) has been withdrawn -- #{job.id} (scan job) did not find new files", "Withdrawn" )
FinishJob(j, f"Job #{j.id} has been withdrawn -- #{job.id} (scan job) did not find new files", "Withdrawn" )
# scan found no new files and last ai scan was after the last file scan
if j.name == "run_ai_on_path" and last_ai_scan > last_scan:
newest_refimg = session.query(Refimg).order_by(Refimg.created_on.desc()).limit(1).all()
# IF we also have no new refimgs since last scan, then no need to run any AI again
if newest_refimg and newest_refimg[0].created_on < last_scan:
FinishJob(j, f"Job (#{j.id}) has been withdrawn -- scan did not find new files, and no new reference images since last scan", "Withdrawn" )
FinishJob(j, f"Job #{j.id} has been withdrawn -- scan did not find new files, and no new reference images since last scan", "Withdrawn" )
# IF we also have no new refimgs since last AI scan, then no need to run any AI again
elif newest_refimg and newest_refimg[0].created_on < last_ai_scan:
FinishJob(j, f"Job (#{j.id}) has been withdrawn -- scan did not find new files, and no new reference images since last scan", "Withdrawn" )
FinishJob(job, f"Finished Importing: {path} - Processed {overall_file_cnt} files, Found {found_new_files} new files, Removed {rm_cnt} file(s)")
FinishJob(j, f"Job #{j.id} has been withdrawn -- scan did not find new files, and no new reference images since last scan", "Withdrawn" )
AddLogForJob(job, f"Finished importing: {path} - Processed {overall_file_cnt} files, Found {found_new_files} new files, Removed {rm_cnt} file(s)")
FinishJob(job, f"Finished import of {path_obj.type.name} path")
return
####################################################################################################################################
@@ -2042,10 +2043,9 @@ def JobCheckForDups(job):
for row in res:
if row.count > 0:
AddLogForJob(job, f"Found duplicates, Creating Status message in front-end for attention")
MessageToFE( job.id, "danger", f'Found duplicate(s), click&nbsp; <a href="javascript:document.body.innerHTML+=\'<form id=_fm method=POST action=/fix_dups></form>\'; document.getElementById(\'_fm\').submit();">here</a>&nbsp;to finalise import by removing duplicates', True, True )
FinishJob( job=job, last_log=f'Found duplicate(s), click&nbsp; <a class="link-light" href="javascript:document.body.innerHTML+=\'<form id=_fm method=POST action=/fix_dups></form>\'; document.getElementById(\'_fm\').submit();">here</a>&nbsp;to finalise import by removing duplicates', state="Completed", pa_job_state="Completed", level="danger", persistent=True, cant_close=True )
else:
FinishJob(job, f"No duplicates found")
FinishJob(job, f"Finished looking for duplicates")
return
####################################################################################################################################
@@ -2109,12 +2109,11 @@ def JobRemoveDups(job):
MoveFileToRecycleBin(job,del_me)
dup_cnt += 1
FinishJob(job, f"Finished removing {dup_cnt} duplicate files" )
# Need to put another checkdups job in now to force / validate we have no dups
next_job=NewJob( "checkdups" )
next_job=NewJob( name="checkdups", num_files=0, wait_for=None, jex=None, parent_job=None, desc="check for duplicate files" )
AddLogForJob(job, f"adding <a href='/job/{next_job.id}'>job id={next_job.id} {next_job.name}</a> to confirm there are no more duplicates" )
MessageToFE( job.id, "success", f"Finished Job#{job.id} removing duplicate files", False, False )
FinishJob(job, f"Finished removing {dup_cnt} duplicate files" )
return
####################################################################################################################################
@@ -2143,8 +2142,7 @@ def JobMoveFiles(job):
if 'eid-' in jex.name:
move_me=session.query(Entry).get(jex.value)
MoveEntriesToOtherFolder( job, move_me, dst_storage_path, f"{prefix}{suffix}" )
next_job=NewJob( "checkdups" )
MessageToFE( job.id, "success", "Completed (move of selected files)", False, False )
NewJob( name="checkdups", num_files=0, wait_for=None, jex=None, parent_job=None, desc="check for duplicate files" )
FinishJob(job, f"Finished move selected file(s)")
return
@@ -2158,8 +2156,7 @@ def JobDeleteFiles(job):
if 'eid-' in jex.name:
del_me=session.query(Entry).join(File).filter(Entry.id==jex.value).first()
MoveFileToRecycleBin(job,del_me)
next_job=NewJob( "checkdups" )
MessageToFE( job.id, "success", "Completed (delete of selected files)", False, False )
NewJob( name="checkdups", num_files=0, wait_for=None, jex=None, parent_job=None, desc="check for duplicate files" )
FinishJob(job, f"Finished deleting selected file(s)")
return
@@ -2173,8 +2170,7 @@ def JobRestoreFiles(job):
if 'eid-' in jex.name:
restore_me=session.query(Entry).join(File).filter(Entry.id==jex.value).first()
RestoreFile(job,restore_me)
next_job=NewJob( "checkdups" )
MessageToFE( job.id, "success", "Completed (restore of selected files)", False, False )
NewJob( name="checkdups", num_files=0, wait_for=None, jex=None, parent_job=None, desc="check for duplicate files" )
FinishJob(job, f"Finished restoring selected file(s)")
return
@@ -2339,7 +2335,7 @@ def ReloadMetadata(job):
####################################################################################################################################
def InitialValidationChecks():
now=datetime.now(pytz.utc)
job=NewJob( "init" )
job=NewJob( name="init", num_files=0, wait_for=None, jex=None, parent_job=None, desc="initialise photo assistant" )
job.start_time=datetime.now(pytz.utc)
JobProgressState( job, "In Progress" )
AddLogForJob(job, f"INFO: Starting Initial Validation checks...")
@@ -2640,7 +2636,7 @@ def CheckAndRunBinClean():
now=datetime.now(pytz.utc)
if not j or (now-j.last_update).days >= settings.scheduled_bin_cleanup:
print( f"INFO: Should force clean up bin path, del files older than {settings.bin_cleanup_file_age} days old" )
job=NewJob( "clean_bin" )
NewJob( name="clean_bin", num_files=0, wait_for=None, jex=None, parent_job=None, desc="periodic clean up on Bin path" )
created_jobs=True
return created_jobs
@@ -2660,11 +2656,11 @@ def ScheduledJobs():
now=datetime.now(pytz.utc)
if ndays_since_last_im_scan >= settings.scheduled_import_scan:
print( f"INFO: Time to force an import scan, last scan was {ndays_since_last_im_scan} days ago" )
job=NewJob( "scannow" )
NewJob( name="scan_ip", num_files=0, wait_for=None, jex=None, parent_job=None, desc="periodic clean scan for new files in Import path" )
created_jobs=True
if ndays_since_last_st_scan >= settings.scheduled_storage_scan:
print( f"INFO: Time to force a storage scan, last scan was {ndays_since_last_st_scan}" )
job=NewJob( "scan_sp" )
NewJob( name="scan_sp", num_files=0, wait_for=None, jex=None, parent_job=None, desc="periodic clean scan for new files in Storage path" )
created_jobs=True
if CheckAndRunBinClean():
created_jobs=True