fix(ui,backend): async scan with asyncio.gather, fix CVE back button focus
All checks were successful
Build and Release .deb / build-deb (push) Successful in 21s
All checks were successful
Build and Release .deb / build-deb (push) Successful in 21s
- Replace ThreadPoolExecutor with asyncio.gather + asyncio.to_thread - Use @work async worker for proper Textual integration - Add table.focus() after mount to ensure button events are received - Remove all call_from_thread calls (now in main thread)
This commit is contained in:
parent
8f7cb33a9c
commit
2237d83ddd
2 changed files with 75 additions and 34 deletions
|
|
@ -11,7 +11,7 @@ from full_updater.backend.cache import ensure_cache_dir, read_cache, get_cache_t
|
||||||
from full_updater.backend.scanner import (
|
from full_updater.backend.scanner import (
|
||||||
Target, ScanResult, get_lxc_list, lxc_is_running,
|
Target, ScanResult, get_lxc_list, lxc_is_running,
|
||||||
ensure_debsecan_installed, scan_apt, scan_cve, write_cache,
|
ensure_debsecan_installed, scan_apt, scan_cve, write_cache,
|
||||||
run_full_scan
|
run_full_scan_async
|
||||||
)
|
)
|
||||||
from full_updater.backend.executor import UpgradeExecutor
|
from full_updater.backend.executor import UpgradeExecutor
|
||||||
from full_updater.ui.loader import LoaderScreen
|
from full_updater.ui.loader import LoaderScreen
|
||||||
|
|
@ -57,16 +57,16 @@ class FullUpdaterApp(App):
|
||||||
self.push_screen(self.loader_screen)
|
self.push_screen(self.loader_screen)
|
||||||
self.run_scan_all()
|
self.run_scan_all()
|
||||||
|
|
||||||
@work(thread=True)
|
@work
|
||||||
def run_scan_all(self):
|
async def run_scan_all(self):
|
||||||
total = len(self.targets)
|
total = len(self.targets)
|
||||||
completed = 0
|
completed = 0
|
||||||
|
|
||||||
def progress_cb():
|
async def progress_cb():
|
||||||
nonlocal completed
|
nonlocal completed
|
||||||
completed += 1
|
completed += 1
|
||||||
pct = min(100.0, (completed / (total * 4)) * 100)
|
pct = min(100.0, (completed / (total * 4)) * 100)
|
||||||
self.call_from_thread(self._update_loader, pct)
|
self._update_loader(pct)
|
||||||
|
|
||||||
def on_result(result: ScanResult):
|
def on_result(result: ScanResult):
|
||||||
self.results[result.target.target_id] = result
|
self.results[result.target.target_id] = result
|
||||||
|
|
@ -75,13 +75,13 @@ class FullUpdaterApp(App):
|
||||||
if t.target_id == result.target.target_id:
|
if t.target_id == result.target.target_id:
|
||||||
idx = i
|
idx = i
|
||||||
break
|
break
|
||||||
self.call_from_thread(self._update_loader_status, idx, result.status)
|
self._update_loader_status(idx, result.status)
|
||||||
self.call_from_thread(self._update_sidebar, result.target.target_id, result)
|
self._update_sidebar(result.target.target_id, result)
|
||||||
|
|
||||||
results = run_full_scan(progress_cb)
|
results = await run_full_scan_async(progress_cb)
|
||||||
for r in results:
|
for r in results:
|
||||||
on_result(r)
|
on_result(r)
|
||||||
self.call_from_thread(self._finish_scan)
|
self._finish_scan()
|
||||||
|
|
||||||
def _update_loader(self, pct: float):
|
def _update_loader(self, pct: float):
|
||||||
if self.loader_screen:
|
if self.loader_screen:
|
||||||
|
|
@ -257,6 +257,7 @@ class FullUpdaterApp(App):
|
||||||
self.mount(table)
|
self.mount(table)
|
||||||
self.query_one("#main-layout").display = False
|
self.query_one("#main-layout").display = False
|
||||||
table.load_data(pkgs)
|
table.load_data(pkgs)
|
||||||
|
table.focus()
|
||||||
|
|
||||||
def on_summary_panel_cve_clicked(self, event: SummaryPanel.CveClicked):
|
def on_summary_panel_cve_clicked(self, event: SummaryPanel.CveClicked):
|
||||||
cache_id = "host" if event.target_id == "host" else event.target_id
|
cache_id = "host" if event.target_id == "host" else event.target_id
|
||||||
|
|
@ -268,6 +269,7 @@ class FullUpdaterApp(App):
|
||||||
self.mount(table)
|
self.mount(table)
|
||||||
self.query_one("#main-layout").display = False
|
self.query_one("#main-layout").display = False
|
||||||
table.load_data(cves)
|
table.load_data(cves)
|
||||||
|
table.focus()
|
||||||
|
|
||||||
def on_package_table_back_pressed(self, event: PackageTable.BackPressed):
|
def on_package_table_back_pressed(self, event: PackageTable.BackPressed):
|
||||||
self.query_one("#main-layout").display = True
|
self.query_one("#main-layout").display = True
|
||||||
|
|
|
||||||
|
|
@ -1,10 +1,11 @@
|
||||||
import concurrent.futures
|
import asyncio
|
||||||
import re
|
import re
|
||||||
import subprocess
|
import subprocess
|
||||||
from dataclasses import dataclass, field
|
from dataclasses import dataclass, field
|
||||||
from typing import Any
|
from typing import Any
|
||||||
|
|
||||||
from full_updater.backend.cache import write_cache
|
from full_updater.backend.cache import write_cache
|
||||||
|
from datetime import datetime
|
||||||
|
|
||||||
|
|
||||||
@dataclass
|
@dataclass
|
||||||
|
|
@ -187,7 +188,6 @@ def scan_single_target(target: Target, progress_cb: Any) -> ScanResult:
|
||||||
result.status = "error"
|
result.status = "error"
|
||||||
|
|
||||||
cache_id = "host" if target.is_host else target.target_id
|
cache_id = "host" if target.is_host else target.target_id
|
||||||
from datetime import datetime
|
|
||||||
write_cache(cache_id, {
|
write_cache(cache_id, {
|
||||||
"timestamp": datetime.now().isoformat(),
|
"timestamp": datetime.now().isoformat(),
|
||||||
"apt_count": result.apt_count,
|
"apt_count": result.apt_count,
|
||||||
|
|
@ -200,27 +200,66 @@ def scan_single_target(target: Target, progress_cb: Any) -> ScanResult:
|
||||||
return result
|
return result
|
||||||
|
|
||||||
|
|
||||||
def run_full_scan(progress_cb: Any) -> list[ScanResult]:
|
async def scan_single_target_async(target: Target, progress_cb: Any) -> ScanResult:
|
||||||
|
"""Version async de scan_single_target (pour asyncio.gather)."""
|
||||||
|
result = ScanResult(target=target)
|
||||||
|
result.status = "running"
|
||||||
|
await progress_cb()
|
||||||
|
|
||||||
|
if not target.is_host and not lxc_is_running(target.target_id):
|
||||||
|
result.status = "skipped"
|
||||||
|
await progress_cb()
|
||||||
|
await progress_cb()
|
||||||
|
await progress_cb()
|
||||||
|
await progress_cb()
|
||||||
|
return result
|
||||||
|
|
||||||
|
debsecan_ok, debsecan_err = await asyncio.to_thread(ensure_debsecan_installed, target.is_host, target.target_id)
|
||||||
|
await progress_cb()
|
||||||
|
|
||||||
|
apt_ok, apt_packages, apt_err = await asyncio.to_thread(scan_apt, target)
|
||||||
|
result.apt_ok = apt_ok
|
||||||
|
result.apt_count = len(apt_packages)
|
||||||
|
result.apt_packages = apt_packages
|
||||||
|
await progress_cb()
|
||||||
|
|
||||||
|
if debsecan_ok:
|
||||||
|
cve_ok, cve_list, cve_err = await asyncio.to_thread(scan_cve, target)
|
||||||
|
result.cve_ok = cve_ok
|
||||||
|
result.cve_count = len(cve_list)
|
||||||
|
result.cve_list = cve_list
|
||||||
|
if not cve_ok:
|
||||||
|
result.error = cve_err
|
||||||
|
else:
|
||||||
|
result.cve_ok = False
|
||||||
|
result.error = f"debsecan: {debsecan_err}"
|
||||||
|
await progress_cb()
|
||||||
|
|
||||||
|
result.status = "done" if (result.apt_ok and result.cve_ok) else "error"
|
||||||
|
if result.error:
|
||||||
|
result.status = "error"
|
||||||
|
|
||||||
|
cache_id = "host" if target.is_host else target.target_id
|
||||||
|
write_cache(cache_id, {
|
||||||
|
"timestamp": datetime.now().isoformat(),
|
||||||
|
"apt_count": result.apt_count,
|
||||||
|
"apt_packages": result.apt_packages,
|
||||||
|
"cve_count": result.cve_count,
|
||||||
|
"cve_list": result.cve_list,
|
||||||
|
"error": result.error
|
||||||
|
})
|
||||||
|
await progress_cb()
|
||||||
|
return result
|
||||||
|
|
||||||
|
|
||||||
|
async def run_full_scan_async(progress_cb: Any) -> list[ScanResult]:
|
||||||
targets = [Target(target_id="host", name="hote", is_host=True)] + get_lxc_list()
|
targets = [Target(target_id="host", name="hote", is_host=True)] + get_lxc_list()
|
||||||
results = []
|
coros = [scan_single_target_async(t, progress_cb) for t in targets]
|
||||||
completed = 0
|
results = await asyncio.gather(*coros, return_exceptions=True)
|
||||||
total = len(targets)
|
out = []
|
||||||
|
for r in results:
|
||||||
def _progress():
|
if isinstance(r, Exception):
|
||||||
nonlocal completed
|
out.append(ScanResult(target=Target(target_id="unknown", name="unknown", is_host=False), status="error", error=str(r)))
|
||||||
completed += 1
|
else:
|
||||||
pct = min(100.0, (completed / (total * 4)) * 100)
|
out.append(r)
|
||||||
progress_cb(pct)
|
return out
|
||||||
|
|
||||||
with concurrent.futures.ThreadPoolExecutor(max_workers=8) as executor:
|
|
||||||
futures = {executor.submit(scan_single_target, t, _progress): t for t in targets}
|
|
||||||
for future in concurrent.futures.as_completed(futures):
|
|
||||||
try:
|
|
||||||
result = future.result()
|
|
||||||
results.append(result)
|
|
||||||
except Exception as e:
|
|
||||||
target = futures[future]
|
|
||||||
result = ScanResult(target=target, status="error", error=str(e))
|
|
||||||
results.append(result)
|
|
||||||
|
|
||||||
return results
|
|
||||||
|
|
|
||||||
Loading…
Add table
Add a link
Reference in a new issue