danvics/ytptube
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions 1

fix: tasks handler summary should reflect correct status.

Browse source
This commit is contained in:
arabcoders 2026-05-13 06:58:43 +03:00
parent 56dbf6fc2d
commit 34e99951fc
1 changed files with 26 additions and 3 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

29
app/features/tasks/definitions/service.py
View file

@ -70,6 +70,7 @@ class TaskHandle:
s: dict[str, list[str]] = {"h": [], "d": [], "u": [], "f": []} s: dict[str, list[str]] = {"h": [], "d": [], "u": [], "f": []}
handler_groups: dict[str, list[tuple[HandleTask, type]]] = {} handler_groups: dict[str, list[tuple[HandleTask, type]]] = {}
dispatches: list[tuple[HandleTask, type, asyncio.Task[TaskResult | TaskFailure | None]]] = []
tasks: list[TaskModel] = await self._repo.list() tasks: list[TaskModel] = await self._repo.list()
@ -95,7 +96,6 @@ class TaskHandle:
if handler_name not in handler_groups: if handler_name not in handler_groups:
handler_groups[handler_name] = [] handler_groups[handler_name] = []
handler_groups[handler_name].append((task, handler)) handler_groups[handler_name].append((task, handler))
s["h"].append(task.name)
except Exception as e: except Exception as e:
LOG.error(f"Failed to handle task '{task.name}'. '{e!s}'.") LOG.error(f"Failed to handle task '{task.name}'. '{e!s}'.")
s["f"].append(task.name) s["f"].append(task.name)
@ -112,8 +112,27 @@ class TaskHandle:
name=f"taskHandler-{task.id}", name=f"taskHandler-{task.id}",
) )
t.add_done_callback(lambda fut, t=task: self._handle_exception(fut, t)) t.add_done_callback(lambda fut, t=task: self._handle_exception(fut, t))
dispatches.append((task, handler, t))
except Exception as e: except Exception as e:
LOG.error(f"Failed to dispatch task '{task.name}'. '{e!s}'.") LOG.error(f"Failed to dispatch task '{task.name}'. '{e!s}'.")
s["f"].append(task.name)
if dispatches:
results = await asyncio.gather(*(t for _, _, t in dispatches), return_exceptions=True)
for (task, handler, _), result in zip(dispatches, results, strict=True):
if isinstance(result, TaskResult):
s["h"].append(task.name)
continue
if isinstance(result, TaskFailure):
s["f"].append(task.name)
continue
if result is None:
LOG.error(f"Handler '{handler.__name__}' returned no result for task '{task.name}'.")
s["f"].append(task.name)
if len(tasks) > 0: if len(tasks) > 0:
LOG.info( LOG.info(
@ -193,7 +212,11 @@ class TaskHandle:
raise raise
if isinstance(extraction, TaskFailure): if isinstance(extraction, TaskFailure):
LOG.error(f"Handler '{handler.__name__}' failed to extract items: {extraction.message}") msg: str = extraction.message
if extraction.error and extraction.error != extraction.message:
msg = f"{msg} {extraction.error}"
LOG.error(f"Handler '{handler.__name__}' failed to extract items for task '{task.name}': {msg}")
return extraction return extraction
if not isinstance(extraction, TaskResult): if not isinstance(extraction, TaskResult):
@ -226,7 +249,7 @@ class TaskHandle:
for item in raw_items: for item in raw_items:
if not isinstance(item, TaskItem): if not isinstance(item, TaskItem):
LOG.warning("Handler '{handler.__name__}' produced non-TaskItem entry: {item!r}") LOG.warning(f"Handler '{handler.__name__}' returned unexpected result: {item!r}")
continue continue
url: str = item.url url: str = item.url