diff --git a/lib/address_book.py b/lib/address_book.py deleted file mode 100644 index f9827f6..0000000 --- a/lib/address_book.py +++ /dev/null @@ -1,160 +0,0 @@ -""" -RECON Address Book — YAML-backed saved-location lookup. - -Provides named locations (home, work, etc.) that short-circuit Photon -geocoding when an exact alias match is found. - -Config: /opt/recon/config/address_book.yaml -""" - -import os -import re -import threading - -import yaml - -from .utils import setup_logging - -logger = setup_logging('recon.address_book') - -_CONFIG_PATH = os.path.join( - os.path.dirname(os.path.dirname(os.path.abspath(__file__))), - 'config', 'address_book.yaml', -) - -_lock = threading.Lock() -_entries: list[dict] = [] -_mtime: float = 0.0 - - -def _reload_if_changed(): - """Reload the YAML file if its mtime has changed.""" - global _entries, _mtime - try: - st = os.stat(_CONFIG_PATH) - except FileNotFoundError: - logger.warning("Address book not found: %s", _CONFIG_PATH) - _entries = [] - _mtime = 0.0 - return - - if st.st_mtime == _mtime: - return - - with _lock: - # Double-check after acquiring lock - try: - st = os.stat(_CONFIG_PATH) - except FileNotFoundError: - _entries = [] - _mtime = 0.0 - return - if st.st_mtime == _mtime: - return - - with open(_CONFIG_PATH, 'r') as f: - data = yaml.safe_load(f) or {} - - raw = data.get('entries', []) - loaded = [] - for entry in raw: - # Normalise aliases to lowercase for matching - aliases = [a.lower() for a in entry.get('aliases', [])] - loaded.append({ - 'id': entry.get('id', ''), - 'name': entry.get('name', ''), - 'aliases': aliases, - 'address': entry.get('address', ''), - 'lat': entry.get('lat'), - 'lon': entry.get('lon'), - 'tags': entry.get('tags', []), - }) - _entries = loaded - _mtime = st.st_mtime - logger.info("Address book loaded: %d entries from %s", len(_entries), _CONFIG_PATH) - - -def load(): - """Ensure the address book is loaded (and refreshed if the file changed).""" - _reload_if_changed() - return _entries - - -def _normalize(text: str) -> str: - """Lowercase, strip, remove commas, collapse whitespace.""" - t = text.strip().lower() - t = t.replace(',', ' ') - return ' '.join(t.split()) - - -def lookup(query: str): - """ - Look up a query against name and aliases. - - Returns dict with the matching entry plus a 'confidence' field: - - "exact": full name/alias match, OR query starts with alias + word boundary - - "partial": alias starts with query + word boundary, or alias appears - as a contiguous token sequence inside the query - - None if no match - - Matching order (first exact wins, else first partial): - 1. normalized(query) == normalized(name or alias) → exact - 2. normalized(query) starts with normalized(alias) + " " → exact - 3. normalized(alias) starts with normalized(query) + " " → partial - 4. normalized(alias) is a contiguous token sub-sequence → partial - """ - _reload_if_changed() - q = _normalize(query) - if not q: - return None - - first_exact = None - first_partial = None - - for entry in _entries: - norm_name = _normalize(entry['name']) - check_aliases = [_normalize(a) for a in entry.get('aliases', [])] - all_forms = [norm_name] + check_aliases - - for form in all_forms: - if not form: - continue - - # Rule 1: exact match - if q == form: - return {**entry, 'confidence': 'exact'} - - # Rule 2: query starts with alias + word boundary - if q.startswith(form + ' '): - if first_exact is None: - first_exact = entry - continue - - # Rule 3: alias starts with query (user still typing) - if form.startswith(q) and len(q) < len(form): - if first_partial is None: - first_partial = entry - continue - - # Rule 4: alias is contiguous token sub-sequence in query - # Build regex: token1\s+token2\s+...tokenN - tokens = form.split() - if len(tokens) >= 1: - pattern = r'(?:^|\s)' + r'\s+'.join(re.escape(t) for t in tokens) + r'(?:\s|$)' - if re.search(pattern, q): - if first_partial is None: - first_partial = entry - - if first_exact is not None: - return {**first_exact, 'confidence': 'exact'} - - if first_partial is not None: - return {**first_partial, 'confidence': 'partial'} - - return None - - -def list_all(): - """Return all address book entries.""" - _reload_if_changed() - return list(_entries) diff --git a/lib/address_book_test.py b/lib/address_book_test.py deleted file mode 100644 index 75905f0..0000000 --- a/lib/address_book_test.py +++ /dev/null @@ -1,91 +0,0 @@ -#!/usr/bin/env python3 -"""Tests for RECON address book module.""" -import sys -import os - -# Add project root to path -sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) - -from lib import address_book - -TESTS = [ - # ── Existing tests ── - ("lookup('home') → exact", - lambda: address_book.lookup("home"), - lambda r: r is not None and r['confidence'] == 'exact' and r['id'] == 'home'), - - ("lookup('Home') → exact (case-insensitive)", - lambda: address_book.lookup("Home"), - lambda r: r is not None and r['confidence'] == 'exact' and r['id'] == 'home'), - - ("lookup('214 north st') → exact via alias", - lambda: address_book.lookup("214 north st"), - lambda r: r is not None and r['confidence'] == 'exact' and r['id'] == 'home'), - - ("lookup('214 North Street') → exact via alias", - lambda: address_book.lookup("214 North Street"), - lambda r: r is not None and r['confidence'] == 'exact' and r['id'] == 'home'), - - ("lookup('nonexistent place') → None", - lambda: address_book.lookup("nonexistent place"), - lambda r: r is None), - - ("list_all() → 1 entry", - lambda: address_book.list_all(), - lambda r: isinstance(r, list) and len(r) == 1 and r[0]['id'] == 'home'), - - # ── New prefix+boundary tests ── - ("lookup('214 north st filer') → exact (query starts with alias)", - lambda: address_book.lookup("214 north st filer"), - lambda r: r is not None and r['confidence'] == 'exact' and r['id'] == 'home'), - - ("lookup('214 North St Filer ID') → exact (case + trailing state)", - lambda: address_book.lookup("214 North St Filer ID"), - lambda r: r is not None and r['confidence'] == 'exact' and r['id'] == 'home'), - - ("lookup('214 north st, filer, id') → exact (commas stripped)", - lambda: address_book.lookup("214 north st, filer, id"), - lambda r: r is not None and r['confidence'] == 'exact' and r['id'] == 'home'), - - ("lookup('home today') → exact (short alias + trailing text)", - lambda: address_book.lookup("home today"), - lambda r: r is not None and r['confidence'] == 'exact' and r['id'] == 'home'), - - ("lookup('214') → partial (query is prefix of alias)", - lambda: address_book.lookup("214"), - lambda r: r is not None and r['confidence'] == 'partial'), - - ("lookup('214 n') → partial (partial prefix of alias)", - lambda: address_book.lookup("214 n"), - lambda r: r is not None and r['confidence'] == 'partial'), - - ("lookup('completely unrelated query') → None", - lambda: address_book.lookup("completely unrelated query"), - lambda r: r is None), - - ("lookup('214 north streets of filer') → None (no word boundary after st)", - lambda: address_book.lookup("214 north streets of filer"), - lambda r: r is None), -] - -passed = 0 -failed = 0 -for name, fn, check in TESTS: - try: - result = fn() - ok = check(result) - except Exception as e: - ok = False - result = f"EXCEPTION: {e}" - - status = "PASS" if ok else "FAIL" - if ok: - passed += 1 - else: - failed += 1 - print(f" [{status}] {name}") - if not ok: - print(f" got: {result}") - -print(f"\n{passed} passed, {failed} failed") -sys.exit(0 if failed == 0 else 1) diff --git a/lib/netsyms_api.py b/lib/netsyms_api.py index 2caf47c..dbae24e 100644 --- a/lib/netsyms_api.py +++ b/lib/netsyms_api.py @@ -8,7 +8,6 @@ GET /api/netsyms/health from flask import Blueprint, request, jsonify from . import netsyms -from . import address_book from .utils import setup_logging logger = setup_logging('recon.netsyms_api')