()
| 1031 | _BOLD_DIVIDER = "─" * 62 |
| 1032 | |
| 1033 | # ── Shared mutable state (accessed from background timer threads) ───── |
| 1034 | cache = get_cache(path) |
| 1035 | # Use containers so nested functions can mutate without `nonlocal` |
| 1036 | _state: Dict[str, Any] = {"prev_fps": {}} # fingerprint -> issue |
| 1037 | _changed: set = set() |
| 1038 | _lock = threading.Lock() |
| 1039 | _pending: List = [None] # [Optional[threading.Timer]] |
| 1040 | |
| 1041 | # ── Initial full scan ───────────────────────────────────────────────── |
| 1042 | click.echo("[~] Running initial scan...") |
| 1043 | try: |
| 1044 | initial = _scan_to_issues( |
| 1045 | path, config_path, severity_level, ai_scan, debug=debug, cache=cache |
| 1046 | ) |
| 1047 | except Exception as exc: |
| 1048 | click.echo(f"[!] Initial scan failed: {exc}") |
| 1049 | initial = [] |
| 1050 | |
| 1051 | _state["prev_fps"] = {iss.get_fingerprint(): iss for iss in initial} |
| 1052 | |
| 1053 | if initial: |
| 1054 | click.echo(f"\n[!] Initial scan: {len(initial)} issue(s) found:") |
| 1055 | click.echo(Reporter(initial, "console").generate()) |
| 1056 | else: |
| 1057 | click.echo("[+] Initial scan: clean — no issues found.") |
| 1058 | |
| 1059 | click.echo("\n[*] Watching for changes...\n") |
| 1060 | |
| 1061 | # ── Re-scan + diff ──────────────────────────────────────────────────── |
| 1062 | def _do_rescan() -> None: |
| 1063 | with _lock: |
| 1064 | changed_snapshot = set(_changed) |
| 1065 | _changed.clear() |
| 1066 | _pending[0] = None |
| 1067 | |
| 1068 | ts = time.strftime("%H:%M:%S") |
| 1069 | if changed_snapshot: |
| 1070 | names = sorted(Path(p).name for p in changed_snapshot) |
| 1071 | label = ", ".join(names[:3]) + ( |
| 1072 | f" (+{len(names) - 3} more)" if len(names) > 3 else "" |
| 1073 | ) |
| 1074 | else: |
| 1075 | label = "re-scan" |
| 1076 | |
| 1077 | click.echo(_BOLD_DIVIDER) |
| 1078 | click.echo(f"[{ts}] " + click.style(label, bold=True)) |
| 1079 | click.echo(_BOLD_DIVIDER) |
| 1080 | |
| 1081 | try: |
| 1082 | new_issues = _scan_to_issues( |
| 1083 | path, config_path, severity_level, ai_scan, debug=debug, cache=cache |
| 1084 | ) |
| 1085 | except Exception as exc: |
| 1086 | click.echo(f" [!] Scan error: {exc}") |
| 1087 | click.echo(_DIVIDER + "\n") |
| 1088 | return |
nothing calls this directly
no test coverage detected