rickub
hlassiege/darylpublic Fork
⑂ 2886ebe
Branches
main View all branches → View all tags →
Commits
⬇ Clone ▾
git clone https://git.rickub.com/hlassiege/daryl.git
git clone ssh://git@rickub.com/hlassiege/daryl.git

Host key fingerprint (ed25519): SHA256:iycHnxEyq0Q7uyVpB7JlznP0G7JrTPXLYRcAU5CSLhc — verify it before your first connect.

Download ZIP

chore(worker): timestamp job log lines

Each worker line starts with the date and time, and a closing line gives each job's outcome and duration.
hlassiege committed 2026-10-11T21:35:11+02:00 Browse files
2886ebe parent: 3723761
modified backend/studio/core/worker.py +15 -5
@@ -28,14 +28,23 @@ def _load_tasks() -> dict[str, Task]:
2828 return tasks
2929
3030
31-def _execute(job: jobs.Job, task: Task) -> None:
31+def _log(message: str) -> None:
32+ """Une ligne de log, à l'heure : pour savoir quand un job a démarré, fini ou planté."""
33+ print(f"{time.strftime('%Y-%m-%d %H:%M:%S')} {message}", flush=True)
34+
35+
36+def _execute(job: jobs.Job, task: Task, queue: str) -> None:
3237 ctx = TaskContext(job)
38+ start = time.time()
39+ end = lambda status: _log(f"[{queue}] {job.kind} {job.id} : {status} en {time.time() - start:.0f} s")
3340 try:
3441 task.run(ctx)
3542 except Cancelled:
3643 ctx.save(force=True)
3744 jobs.update(job.id, status="cancelled", finished_at=time.time())
45+ end("annulé")
3846 except Exception as e: # noqa: BLE001 l'erreur est affichee dans l'UI
47+ end("erreur")
3948 traceback.print_exc()
4049 ctx.save(force=True)
4150 jobs.update(job.id, status="error", error=str(e), finished_at=time.time())
@@ -43,6 +52,7 @@ def _execute(job: jobs.Job, task: Task) -> None:
4352 ctx.progress = 1.0
4453 ctx.save(force=True)
4554 jobs.update(job.id, status="done", finished_at=time.time())
55+ end("terminé")
4656
4757
4858 def _consume(queue: str, tasks: dict[str, Task]) -> None:
@@ -56,8 +66,8 @@ def _consume(queue: str, tasks: dict[str, Task]) -> None:
5666 if job is None:
5767 time.sleep(IDLE_POLL)
5868 continue
59- print(f"[{queue}] {job.kind} {job.id}")
60- _execute(job, tasks[job.kind])
69+ _log(f"[{queue}] {job.kind} {job.id}")
70+ _execute(job, tasks[job.kind], queue)
6171
6272
6373 def _heartbeat(queues: list[str]) -> None:
@@ -74,13 +84,13 @@ def main() -> None:
7484 jobs.init()
7585 requeued = jobs.requeue_interrupted()
7686 if requeued:
77- print(f"{requeued} job(s) interrompu(s) remis en file.")
87+ _log(f"{requeued} job(s) interrompu(s) remis en file.")
7888 by_queue: dict[str, dict[str, Task]] = defaultdict(dict)
7989 for kind, task in _load_tasks().items():
8090 by_queue[task.queue][kind] = task
8191 for queue, tasks in by_queue.items():
8292 threading.Thread(target=_consume, args=(queue, tasks), daemon=True, name=queue).start()
83- print(f"Worker pret · files : {', '.join(by_queue)}")
93+ _log(f"Worker pret · files : {', '.join(by_queue)}")
8494 _heartbeat(list(by_queue))
8595
8696
@@ -28,14 +28,23 @@ def _load_tasks() -> dict[str, Task]:
28 return tasks28 return tasks
29 29
30 30
31-def _execute(job: jobs.Job, task: Task) -> None:31+def _log(message: str) -> None:
32+ """Une ligne de log, à l'heure : pour savoir quand un job a démarré, fini ou planté."""
33+ print(f"{time.strftime('%Y-%m-%d %H:%M:%S')} {message}", flush=True)
34+
35+
36+def _execute(job: jobs.Job, task: Task, queue: str) -> None:
32 ctx = TaskContext(job)37 ctx = TaskContext(job)
38+ start = time.time()
39+ end = lambda status: _log(f"[{queue}] {job.kind} {job.id} : {status} en {time.time() - start:.0f} s")
33 try:40 try:
34 task.run(ctx)41 task.run(ctx)
35 except Cancelled:42 except Cancelled:
36 ctx.save(force=True)43 ctx.save(force=True)
37 jobs.update(job.id, status="cancelled", finished_at=time.time())44 jobs.update(job.id, status="cancelled", finished_at=time.time())
45+ end("annulé")
38 except Exception as e: # noqa: BLE001 l'erreur est affichee dans l'UI46 except Exception as e: # noqa: BLE001 l'erreur est affichee dans l'UI
47+ end("erreur")
39 traceback.print_exc()48 traceback.print_exc()
40 ctx.save(force=True)49 ctx.save(force=True)
41 jobs.update(job.id, status="error", error=str(e), finished_at=time.time())50 jobs.update(job.id, status="error", error=str(e), finished_at=time.time())
@@ -43,6 +52,7 @@ def _execute(job: jobs.Job, task: Task) -> None:
43 ctx.progress = 1.052 ctx.progress = 1.0
44 ctx.save(force=True)53 ctx.save(force=True)
45 jobs.update(job.id, status="done", finished_at=time.time())54 jobs.update(job.id, status="done", finished_at=time.time())
55+ end("terminé")
46 56
47 57
48 def _consume(queue: str, tasks: dict[str, Task]) -> None:58 def _consume(queue: str, tasks: dict[str, Task]) -> None:
@@ -56,8 +66,8 @@ def _consume(queue: str, tasks: dict[str, Task]) -> None:
56 if job is None:66 if job is None:
57 time.sleep(IDLE_POLL)67 time.sleep(IDLE_POLL)
58 continue68 continue
59- print(f"[{queue}] {job.kind} {job.id}")69+ _log(f"[{queue}] {job.kind} {job.id}")
60- _execute(job, tasks[job.kind])70+ _execute(job, tasks[job.kind], queue)
61 71
62 72
63 def _heartbeat(queues: list[str]) -> None:73 def _heartbeat(queues: list[str]) -> None:
@@ -74,13 +84,13 @@ def main() -> None:
74 jobs.init()84 jobs.init()
75 requeued = jobs.requeue_interrupted()85 requeued = jobs.requeue_interrupted()
76 if requeued:86 if requeued:
77- print(f"{requeued} job(s) interrompu(s) remis en file.")87+ _log(f"{requeued} job(s) interrompu(s) remis en file.")
78 by_queue: dict[str, dict[str, Task]] = defaultdict(dict)88 by_queue: dict[str, dict[str, Task]] = defaultdict(dict)
79 for kind, task in _load_tasks().items():89 for kind, task in _load_tasks().items():
80 by_queue[task.queue][kind] = task90 by_queue[task.queue][kind] = task
81 for queue, tasks in by_queue.items():91 for queue, tasks in by_queue.items():
82 threading.Thread(target=_consume, args=(queue, tasks), daemon=True, name=queue).start()92 threading.Thread(target=_consume, args=(queue, tasks), daemon=True, name=queue).start()
83- print(f"Worker pret · files : {', '.join(by_queue)}")93+ _log(f"Worker pret · files : {', '.join(by_queue)}")
84 _heartbeat(list(by_queue))94 _heartbeat(list(by_queue))
85 95
86 96