Mia.Wu/test_dashboard
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix scan

Browse Source
This commit is contained in:
Mia Wu
2026-05-28 12:34:01 -04:00
parent d239de257d
commit 49310e6d59
9 changed files with 240 additions and 164 deletions
Download Patch File Download Diff File Expand all files Collapse all files
server/scanner.py
+90 -118
View File
@@ -1,13 +1,13 @@
import os
import re
import threading
import smbclient
from db_py import (
clear_tests,
get_all_tests,
mark_completed,
set_coe_pair,
set_p3p_pair,
update_all_p2p_coe_pairs_sql,
update_all_p3p_pairs_sql,
upsert_test,
)
from parser import (
@@ -20,13 +20,67 @@ from parser import (
_SMB_SESSIONS = set()
_WIN_DRIVE_PATH_RE = re.compile(r"^[A-Za-z]:[\\/]")
_SCAN_STATE_LOCK = threading.Lock()
_ACTIVE_SCAN_COUNT = 0
def _scan_started():
global _ACTIVE_SCAN_COUNT
with _SCAN_STATE_LOCK:
_ACTIVE_SCAN_COUNT += 1
def _scan_finished():
global _ACTIVE_SCAN_COUNT
with _SCAN_STATE_LOCK:
_ACTIVE_SCAN_COUNT = max(0, _ACTIVE_SCAN_COUNT - 1)
def is_scan_in_progress():
with _SCAN_STATE_LOCK:
return _ACTIVE_SCAN_COUNT > 0
def resolve_runtime_path(path_value):
if not path_value:
return path_value
raw_path = str(path_value).strip()
if not raw_path:
return raw_path
# If the path is already valid in the current runtime, keep it.
if os.path.exists(raw_path):
return raw_path
# UNC/network paths are handled separately via smbclient.
if raw_path.startswith("\\\\") or raw_path.startswith("//"):
return raw_path
# Map Windows host paths when backend runs in Linux container.
if os.name != "nt" and _WIN_DRIVE_PATH_RE.match(raw_path):
mount_root = os.getenv("HOST_MOUNT_ROOT", "/host").strip() or "/host"
host_root = os.getenv("HOST_BROWSE_ROOT", "").strip()
raw_norm = raw_path.replace("\\", "/")
if host_root:
host_norm = host_root.replace("\\", "/").rstrip("/")
if raw_norm.lower() == host_norm.lower() or raw_norm.lower().startswith(host_norm.lower() + "/"):
relative = raw_norm[len(host_norm):].lstrip("/")
if relative:
return os.path.join(mount_root, *relative.split("/"))
return mount_root
return raw_path
def _normalize_input_path(path_value):
if not path_value:
return path_value
path = str(path_value).strip()
path = resolve_runtime_path(path_value)
path = str(path).strip()
# Accept //server/share style and normalize to UNC for smbclient.
if path.startswith("//"):
@@ -102,138 +156,54 @@ def _read_text_lines(path):
yield line
def scan_results_only(results_dir, results_dir_ref=None):
def scan_results_only(results_dir, results_dir_ref):
"""Re-scan only the results directories without clearing or re-scanning targets."""
results_dir = _normalize_input_path(results_dir)
results_dir_ref = _normalize_input_path(results_dir_ref)
if not results_dir:
if not results_dir or not results_dir_ref:
return
print(f"[scanner] results-only scan dir : {results_dir}")
if results_dir_ref:
_scan_started()
try:
print(f"[scanner] results-only scan dir : {results_dir}")
print(f"[scanner] results-only scan dir ref : {results_dir_ref}")
scan_results(results_dir)
if results_dir_ref:
scan_results(results_dir)
scan_results(results_dir_ref)
update_p2p_coe_pairs()
update_p3p_throttle_pairs()
update_p2p_coe_pairs()
update_p3p_throttle_pairs()
finally:
_scan_finished()
def full_scan(target_dir, results_dir, results_dir_ref=None):
def full_scan(target_dir, results_dir, results_dir_ref):
target_dir = _normalize_input_path(target_dir)
results_dir = _normalize_input_path(results_dir)
results_dir_ref = _normalize_input_path(results_dir_ref)
clear_tests()
if not target_dir or not results_dir:
return
_scan_started()
try:
clear_tests()
if not target_dir or not results_dir or not results_dir_ref:
return
print(f"[scanner] target dir : {target_dir}")
print(f"[scanner] results dir : {results_dir}")
if results_dir_ref:
print(f"[scanner] target dir : {target_dir}")
print(f"[scanner] results dir : {results_dir}")
print(f"[scanner] results dir ref : {results_dir_ref}")
scan_targets(target_dir)
scan_results(results_dir)
if results_dir_ref:
if not scan_targets(target_dir):
print("[scanner] Skipping results scan because target scan failed")
return
scan_results(results_dir)
scan_results(results_dir_ref)
update_p2p_coe_pairs()
update_p3p_throttle_pairs()
def update_p2p_coe_pairs():
pair_fields = [
"device",
"rotation",
"test_point",
"rssi",
"station",
"band",
"channel",
"bandwidth",
"direction",
]
tests = get_all_tests()
coe_by_key = {}
for test in tests:
if test.get("interference") != "COE":
continue
key = tuple(test.get(field) for field in pair_fields)
device = test.get("device")
test_id = test.get("test_id")
if not device or not test_id:
continue
coe_by_key.setdefault(key, []).append(f"{device}_{test_id}")
updated = 0
for test in tests:
if test.get("interference") != "P2P":
continue
key = tuple(test.get(field) for field in pair_fields)
pairs = sorted(set(coe_by_key.get(key, [])))
set_coe_pair(test.get("id"), pairs)
updated += 1
print(f"[scanner] coe_pair updated for {updated} P2P test(s)")
def update_p3p_throttle_pairs():
tests = get_all_tests()
p3p_lookup = {}
for test in tests:
if test.get("interference") != "P3P":
continue
device = test.get("device")
test_id = test.get("test_id")
if not device or not test_id:
continue
p3p_lookup.setdefault(f"{str(device).upper()}_{str(test_id).upper()}", []).append(test)
def _pair_test_id(test_id, throttled):
if not test_id or not throttled:
return None
upper_id = str(test_id).upper()
upper_throttled = str(throttled).upper()
if upper_throttled == "TH":
return re.sub("TH", "UT", upper_id, count=1)
if upper_throttled == "UT":
return re.sub("UT", "TH", upper_id, count=1)
return None
updated = 0
for test in tests:
if test.get("interference") != "P3P":
continue
device = test.get("device")
test_id = test.get("test_id")
pair_test_id = _pair_test_id(test_id, test.get("throttled"))
pairs = []
if device and pair_test_id:
lookup_key = f"{str(device).upper()}_{str(pair_test_id).upper()}"
matches = p3p_lookup.get(lookup_key, [])
for match in matches:
match_device = match.get("device")
match_test_id = match.get("test_id")
if match_device and match_test_id:
pairs.append(f"{match_device}_{match_test_id}")
set_p3p_pair(test.get("id"), sorted(set(pairs)))
updated += 1
print(f"[scanner] p3p_pair updated for {updated} P3P test(s)")
coe_pairs = update_all_p2p_coe_pairs_sql()
print(f"[scanner] coe_pair updated for {coe_pairs} P2P test(s)")
p3p_pairs = update_all_p3p_pairs_sql()
print(f"[scanner] p3p_pair updated for {p3p_pairs} P3P test(s)")
finally:
_scan_finished()
def scan_targets(target_dir):
@@ -247,7 +217,7 @@ def scan_targets(target_dir):
]
except OSError as exc:
print(f"[scanner] Cannot read target dir: {exc}")
return
return False
print(f"[scanner] subdirectories found: {len(parent_entries)}")
@@ -296,6 +266,8 @@ def scan_targets(target_dir):
}
)
return True
def scan_results(results_dir):
results_dir = _normalize_input_path(results_dir)