jobrunner => registers + queues. misc

back/constants.py

@@ -3,3 +3,4 @@ BASE_PATH="/back" #changes based on webserver
 
 RAW_PATH='/home/david/git/youtube-uploader/web/front/raw/'
 CUT_OUTPUT_PATH='/home/david/git/youtube-uploader/web/front/videos/'
+INCOMING_PATH='/home/david/git/youtube-uploader/watched/'

back/tasks.py

@@ -2,10 +2,12 @@ import os.path
 import datetime
 import json
 import thumb
+import time
 
 from constants import CUT_OUTPUT_PATH
 from subprocess import Popen,PIPE
 from rq import get_current_job,Queue
+from rq.registry import StartedJobRegistry,FinishedJobRegistry 
 from rq.job import Job
 from rq.decorators import job
 from worker import conn,q
@@ -13,17 +15,8 @@ from dateutil import tz
 from db import db
 
 d = db()
-job_key_list = []
 date_fmt='%d/%m/%Y, %H:%M'
 
-
-def jobrunner(fn,args):
-    print(args)
-    job = fn.delay(*args)
-    print(job.get_id())
-    job_key_list.append(job.get_id())
-    return job
-
 def utctolocal(date):
     date=date.replace(tzinfo=tz.tzutc())
     return date.astimezone(tz.tzlocal())
@@ -38,23 +31,32 @@ def result(job_key):
     else:
         return ("Process: %s" % json.dumps(job.meta), 202)
 
-def jobs():
+def l_from_ids(ids,status=None):
     ret = []
-    for j in job_key_list:#q.jobs:
-        try:
-            job = Job.fetch(j, connection=conn)
-        except Exception as e:
-            job_key_list.remove(j)
-            print("%s deleted" % j)
-            continue
+    for job_id in ids:
+        job = Job.fetch(job_id, connection=conn)
         out = job.meta
-        out["id"]=j
         out["eq"]=utctolocal(job.enqueued_at).strftime(date_fmt)
-        out["finished"]=job.is_finished
+        out["working"]=True
         out["result"]=job.result
+        if status:
+            out["progress"]=status
         ret.append(out)
     return ret
 
+def jobs():
+    ret = []
+    ids = StartedJobRegistry(name='default', connection=conn).get_job_ids()
+    ret.extend(l_from_ids(ids))
+    ids = FinishedJobRegistry(name='default', connection=conn).get_job_ids()
+    ret.extend(l_from_ids(ids,'Finished'))
+
+    for q_name in q.all(connection=conn):
+        ids = q_name.job_ids
+        ret.extend(l_from_ids(ids,'Waiting'))
+
+    return ret
+
 def upload_video(_id,prof,texto1,texto2):
     vid=d.processed(_id)
     ul_file = vid["out_fname"]
@@ -65,8 +67,6 @@ def upload_video(_id,prof,texto1,texto2):
     #job = jobrunner(upload (titulo, thumb_file, ul_file)) 
     #FIXME
     job = upload.delay(titulo, thumb_file, ul_file)
-    job_key_list.append(job.get_id())
-    print(job.get_id())
 
 
 @job('default',connection=conn, timeout=3600*2, result_ttl=3600*2)
@@ -79,13 +79,18 @@ def upload(title,thumb_path,video_path):
     d.insert_uploaded({'title': title,'yt': v_id, 'video':video_path, 'enqueued_at': job_start, 'finished_at': job_end })
     return v_id
 
-@job('default',connection=conn, timeout=30)
+@job('default',connection=conn, timeout=30, result_ttl=135)
 def long_job(arg):
+    job = get_current_job()
+    job.meta['type'] = "Test"
+    job.meta['target'] = 20
+    job.save()
     for i in range(20):
         time.sleep(1)
         job = get_current_job()
         job.meta['progress'] = i
         job.save()
+        print("working")
     #push to db, return id if ok. return error if not ok
     return "37"
 
@@ -119,7 +124,7 @@ def crop_video(in_fname,start_time,end_time,out_fname):
 
     out = run_process(process)
     if out == "":
-        out = { "status": "Proceso finalizado" }
+        out = "Proceso finalizado"
 
 
     job_end = datetime.datetime.now()
@@ -146,5 +151,7 @@ def run_process(process = None):
             job.meta['progress'] = line.rstrip()
             job.save()
         else:
+            job.meta['progress'] = "Finalizado"
+            job.save()
             break
     return ""

back/thumb.py

@@ -8,7 +8,7 @@ from wand.color import Color
 draw = Drawing()
 
 def create_thumb(prof,texto1,texto2):
-    with tempfile.NamedTemporaryFile(suffix='.png',delete=None) as fp:
+    with tempfile.NamedTemporaryFile(suffix='.png',delete=False) as fp:
         with Image(filename='especificosos-nuevo.png') as img:
             draw.font = "Helvetica"
             draw.gravity = "center"

back/web.py

@@ -65,8 +65,7 @@ def crop():
                                 today.strftime('%Y%m%d'),
                                 data["desc"])
 
-    #job = tasks.crop_video.delay(video["file"],
-    job = tasks.jobrunner(tasks.crop_video, 
+    job = tasks.crop_video.delay(
                             (video["file"],
                             str(data["inicio"]),
                             str(data["fin"]),
@@ -88,8 +87,7 @@ def upload():
 
 @app.route(BASE_PATH + '/job')
 def job():
-    job = tasks.jobrunner(tasks.long_job, ("test",))
-    #job = tasks.long_job.delay("test")
+    job = tasks.long_job.delay("test")
 
     ret = {"id":job.get_id()}
 

back/youtube_upload/main.py

@@ -45,7 +45,7 @@ def get_progress_info():
     def _callback(total_size, completed):
         if JOB is not None:
             JOB.meta['status'] = 'uploading'
-            JOB.meta['progress'] = (completed/total_size)*100
+            JOB.meta['progress'] = "%.2f%%" % ((completed/total_size)*100)
             JOB.save()
     def _finish():
         if JOB is not None:
@@ -140,6 +140,8 @@ def run_main(title,thumb,path,output=sys.stdout):
                     'default_audio_language': None, 'publish_at': None, 'auth_browser': None
                     })
 
+    JOB.meta["type"]="Youtube"
+    JOB.meta["target"]=title
     youtube = get_youtube_handler(options)
     video_path = path
     video_id=None

front/controllers/jobsCtrl.js

@@ -1,10 +1,10 @@
-app.controller('jobsCtrl', ['$scope', '$http', function ($scope, $http) {
+app.controller('jobsCtrl', function ($scope, $http,$interval) {
 	$scope.update = function() {
 		$http.get('back/list/jobs').success(function(data) {
-	   		$scope.finished = data.finished;
 	   		$scope.working  = data.working;
 		});
 	};
 
 	$scope.update();
-}]);
+	$interval($scope.update, 1500);
+});

front/views/jobs.html

@@ -5,10 +5,11 @@
 		<th>Tipo</th>
 		<th>Target</th>
 		<th>Estado</th>
+		<th>Resultado</th>
 	</tr>
-	<tr ng-repeat="j in joblist">
+	<tr ng-repeat="j in working">
 		<td>
-			{{j.start}}
+			{{j.eq}}
 		</td>
 		<td>
 			{{j.type}}
@@ -17,7 +18,10 @@
 			{{j.target}}
 		</td>
 		<td>
-			{{j.status}}
+			{{j.progress}}
+		</td>
+		<td>
+			{{j.result}}
 		</td>
 	</tr>
 </table>

front/views/uploadList.html

@@ -3,6 +3,7 @@
 	<tr>
 		<th>Inicio</th>
 		<th>Fin</th>
+		<th>Titulo</th>
 		<th>Archivo</th>
 		<th>Link</th>
 	</tr>
@@ -13,6 +14,9 @@
 		<td>
 			{{v.finished_at}}
 		</td>
+		<td>
+			{{v.title}}
+		</td>
 		<td>
 			{{v.video}}
 		</td>