test_dashboard/server/scanner.py

import os
import re
import threading
import tempfile
import smbclient

from db_py import (
    mark_tests_completed,
    upsert_tests,
    clear_tests,
    extract_measurement_metrics,
    get_config,
    reset_all_results_state,
    update_all_p2p_coe_pairs_sql,
    update_all_p3p_pairs_sql,
)
from parser import (
    parse_result_filename,
    parse_target_filename,
    parse_timestamp,
)


_SMB_SESSIONS = set()
_WIN_DRIVE_PATH_RE = re.compile(r"^[A-Za-z]:[\\/]")
_SCAN_STATE_LOCK = threading.Lock()
_ACTIVE_SCAN_COUNT = 0
_DEFAULT_SCAN_EXCLUSIONS = ""

def _parse_scan_exclusions(value):
    raw = value if value not in (None, "") else _DEFAULT_SCAN_EXCLUSIONS
    if raw == "" or raw is None:
        return None
    parts = re.split(r"[,\n;]", str(raw))
    return {token.strip().upper() for token in parts if token.strip()}


def _build_match_tokens(parsed):
    tokens = set()

    def _add(value):
        if value in (None, ""):
            return
        tokens.add(str(value).upper())

    _add(parsed.get("interference"))
    _add(parsed.get("device"))
    _add(parsed.get("rotation"))
    _add(parsed.get("test_point"))
    _add(parsed.get("rssi"))
    _add(parsed.get("station"))
    _add(parsed.get("band"))
    _add(parsed.get("channel"))
    _add(parsed.get("bandwidth"))
    _add(parsed.get("direction"))
    _add(parsed.get("throttled"))
    _add(parsed.get("test_id"))

    for bw in parsed.get("extra_bandwidths") or []:
        _add(bw)

    # Add numeric alias for devices, e.g. CGW453 -> 453.
    device = (parsed.get("device") or "").upper()
    device_digits = re.sub(r"\D", "", device)
    if device_digits:
        tokens.add(device_digits)

    # SP can appear as flag or specific token variant (e.g., SP40).
    if parsed.get("sp"):
        tokens.add("SP")
        tokens.add(str(parsed.get("sp")).upper())

    return tokens


def _normalize_bandwidth_value(value):
    if value in (None, ""):
        return set()

    raw = str(value).strip().upper()
    if not raw:
        return set()

    normalized = {raw}
    digits = re.sub(r"\D", "", raw)
    if digits:
        normalized.add(digits)
        normalized.add(f"BW{digits}")

    return normalized


def _extract_coe_bandwidth_tokens(ini_path):
    keys = {"Bandwidth_fh2", "Bandwidth_fh5", "Bandwidth_fh6"}
    line_re = re.compile(r"^\s*([A-Za-z0-9_]+)\s*=\s*([^\r\n#;]+)")
    tokens = set()

    try:
        for line in _read_text_lines(ini_path):
            match = line_re.match(line)
            if not match:
                continue

            key = match.group(1).strip()
            if key not in keys:
                continue

            raw_value = match.group(2).strip()
            for part in re.split(r"[,/\s]+", raw_value):
                part = part.strip()
                if not part:
                    continue
                tokens.update(_normalize_bandwidth_value(part))
    except OSError as exc:
        print(f"[scanner] Cannot read ini file {ini_path}: {exc}")

    return tokens


def _rule_matches_target(rule, parsed_tokens):
    # Rule parts are AND-ed: CGW453_P3P_ROT2 => device AND interference AND rotation.
    # Support _, -, or spaces as condition separators.
    parts = [p for p in re.split(r"[_\-\s]+", rule) if p]
    if not parts:
        return False

    def _part_matches(part):
        if part in parsed_tokens:
            return True

        # Support tag variants (e.g., BW80 should match BW80M/BW80+80 in source names).
        # Keep this conservative for very short parts to avoid overmatching.
        if len(part) >= 3:
            for token in parsed_tokens:
                if token.startswith(part) or part in token:
                    return True

        return False

    return all(_part_matches(part) for part in parts)


def _should_exclude_target(parsed, exclusions):
    parsed_tokens = _build_match_tokens(parsed)
    # Any matching rule excludes the testcase.
    return any(_rule_matches_target(rule, parsed_tokens) for rule in exclusions)


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 host paths (Windows or Linux) to the container mount point when running in a container.
    if os.name != "nt":
        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 = resolve_runtime_path(path_value)
    path = str(path).strip()

    # Accept //server/share style and normalize to UNC for smbclient.
    if path.startswith("//"):
        return "\\\\" + path.lstrip("/").replace("/", "\\")

    # Accept /<ipv4>/<share>/... and normalize to UNC for Linux-hosted inputs.
    if re.match(r"^/\d{1,3}(?:\.\d{1,3}){3}/[^/]+", path):
        return "\\\\" + path.lstrip("/").replace("/", "\\")

    return path


def _is_unc_path(path):
    return isinstance(path, str) and path.startswith("\\\\")


def _extract_unc_server(path):
    if not _is_unc_path(path):
        return None
    rest = path[2:]
    return rest.split("\\", 1)[0] if rest else None


def _register_smb_session_if_needed(path):
    if not _is_unc_path(path):
        return

    server = _extract_unc_server(path)
    if not server or server in _SMB_SESSIONS:
        return

    username = os.getenv("SMB_USERNAME", "").strip()
    password = os.getenv("SMB_PASSWORD", "")
    domain = os.getenv("SMB_DOMAIN", "").strip()

    if username and domain and "\\" not in username and "@" not in username:
        username = f"{domain}\\{username}"

    if username:
        smbclient.register_session(server, username=username, password=password)
    else:
        smbclient.register_session(server)

    _SMB_SESSIONS.add(server)


def _iter_dir_entries(path):
    path = _normalize_input_path(path)
    if _is_unc_path(path):
        _register_smb_session_if_needed(path)
        return list(smbclient.scandir(path))
    return list(os.scandir(path))


def _join_path(path, name):
    if _is_unc_path(path):
        base = path.rstrip("\\")
        return f"{base}\\{name}"
    return os.path.join(path, name)


def _read_text_lines(path):
    path = _normalize_input_path(path)
    if _is_unc_path(path):
        _register_smb_session_if_needed(path)
        with smbclient.open_file(path, mode="r", encoding="utf-8", errors="ignore") as fh:
            for line in fh:
                yield line
        return

    with open(path, "r", encoding="utf-8", errors="ignore") as fh:
        for line in fh:
            yield line


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 or not results_dir_ref:
        return

    _scan_started()
    try:
        print(f"[scanner] results-only scan dir     : {results_dir}")
        print(f"[scanner] results-only scan dir ref : {results_dir_ref}")

        # Rebuild completion state from current result folders so deletions are reflected.
        reset_all_results_state()

        scan_results(results_dir)
        scan_results(results_dir_ref)
        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 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)

    _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}")
        print(f"[scanner] results dir ref : {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)
        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):
    target_dir = _normalize_input_path(target_dir)
    exclusions = _parse_scan_exclusions(get_config("scan_exclusions"))
    batch_tests = []

    try:
        parent_entries = [
            entry.name
            for entry in _iter_dir_entries(target_dir)
            if entry.is_dir()
        ]
    except OSError as exc:
        print(f"[scanner] Cannot read target dir: {exc}")
        return False

    print(f"[scanner] subdirectories found: {len(parent_entries)}")

    for parent_name in parent_entries:
        parent_path = _join_path(target_dir, parent_name)
        try:
            files = [
                entry.name
                for entry in _iter_dir_entries(parent_path)
                if entry.is_file()
                and not entry.name.startswith("GLOBAL")
                and entry.name.endswith(".ini")
            ]
        except OSError as exc:
            print(f"[scanner] Cannot read parent dir {parent_name}: {exc}")
            continue

        print(f"[scanner]   {parent_name} -> {len(files)} target test file(s)")

        for filename in files:
            parsed = parse_target_filename(filename, parent_name)
            if not parsed["test_id"]:
                #print(f"[scanner] skip (no test_id): {parent_name}/{filename}")
                continue

            if (parsed.get("interference") or "").upper() == "COE":
                ini_path = _join_path(parent_path, filename)
                parsed["extra_bandwidths"] = sorted(_extract_coe_bandwidth_tokens(ini_path))

            if exclusions and _should_exclude_target(parsed, exclusions):
                continue

            #print( f"[scanner] found: {parent_name}/{filename}  ->  test_id={parsed['test_id']}")
            batch_tests.append(
                {
                    "id": parsed["id"],
                    "test_id": parsed["test_id"],
                    "parent_dir": parent_name,
                    "filename": filename,
                    "interference": parsed["interference"],
                    "device": parsed["device"],
                    "rotation": parsed["rotation"],
                    "test_point": parsed["test_point"],
                    "station": parsed["station"],
                    "band": parsed["band"],
                    "channel": parsed["channel"],
                    "bandwidth": parsed["bandwidth"],
                    "rssi": parsed["rssi"],
                    "direction": parsed["direction"],
                    "throttled": parsed["throttled"],
                }
            )

    upsert_tests(batch_tests)

    return True


def scan_results(results_dir):
    results_dir = _normalize_input_path(results_dir)

    try:
        entries = [
            entry.name
            for entry in _iter_dir_entries(results_dir)
            if entry.is_dir()
        ]
    except OSError as exc:
        print(f"[scanner] Cannot read results dir: {exc}")
        return

    print(f"[scanner] results: {len(entries)} result dir(s) found")
    completed_batch = []
    for dir_name in entries:
        completion = process_result_dir(results_dir, dir_name)
        if completion:
            completed_batch.append(completion)

    mark_tests_completed(completed_batch)


def process_result_dir(results_dir, result_dir_name):
    results_dir = _normalize_input_path(results_dir)

    parsed = parse_result_filename(result_dir_name)
    test_id = parsed["test_id"]
    device = parsed["device"]
    if not test_id or not device:
        return None

    result_dir_path = _join_path(results_dir, result_dir_name)

    try:
        files = [entry.name for entry in _iter_dir_entries(result_dir_path) if entry.is_file()]
    except OSError as exc:
        print(f"[scanner] Cannot read result dir {result_dir_name}: {exc}")
        return None

    measurement_db = next((name for name in files if name.lower() == "measurement.db"), None)
    if not measurement_db:
        print(f"[scanner] completed (no measurement.db yet): {test_id}")
        return {
            "test_id": test_id,
            "device": device,
            "completed_at": None,
            "duration_seconds": None,
            "tput_results": None,
        }

    db_path = _join_path(result_dir_path, measurement_db)
    completed_at = parse_timestamp(result_dir_name)
    data = extract_measurement_data(db_path)

    '''print(
        f"[scanner] completed: {test_id}  device={device}  "
        f"duration={data['duration_seconds']}s  stations={len(data['tputResults'])}  at={completed_at}"
    )'''
    return {
        "test_id": test_id,
        "device": device,
        "completed_at": completed_at,
        "duration_seconds": data["duration_seconds"],
        "tput_results": data["tputResults"],
    }


def parse_deleted_result_dir_name(dir_name):
    segments = re.split(r"[_\-]", dir_name)
    test_id = next((s for s in segments if re.match(r"^R\d+[A-Z0-9]+$", s, re.IGNORECASE)), None)
    device = next((s for s in segments if re.match(r"^CGW\d+$", s, re.IGNORECASE)), None)
    return test_id, device


def extract_measurement_data(db_file_path):
    result = {"duration_seconds": None, "tputResults": []}
    local_db_path = None

    try:
        source_db_path = _normalize_input_path(db_file_path)
        if _is_unc_path(source_db_path):
            _register_smb_session_if_needed(source_db_path)
            with smbclient.open_file(source_db_path, mode="rb") as src_fh:
                with tempfile.NamedTemporaryFile(delete=False, suffix=".db") as tmp_fh:
                    tmp_fh.write(src_fh.read())
                    local_db_path = tmp_fh.name
            source_db_path = local_db_path

        result = extract_measurement_metrics(source_db_path)

    except (OSError, ValueError, Exception) as exc:
        print(f"[scanner] Cannot read measurement db {db_file_path}: {exc}")
    finally:
        if local_db_path:
            try:
                os.remove(local_db_path)
            except OSError:
                pass

    return result