mirror of
https://github.com/zvx-echo6/recon.git
synced 2026-05-20 14:44:54 +02:00
feat(navi): address book with geocoding integration
- YAML-backed saved locations (config/address_book.yaml) - Exact/partial alias matching with case-insensitive lookup - Flask blueprint: /api/address_book/lookup, /api/address_book/list - Geocoder short-circuits Photon when address book has exact match - Test suite for lookup behavior Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
parent
3243f2f252
commit
23483e8198
6 changed files with 263 additions and 1 deletions
132
lib/address_book.py
Normal file
132
lib/address_book.py
Normal file
|
|
@ -0,0 +1,132 @@
|
|||
"""
|
||||
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 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 lookup(query: str):
|
||||
"""
|
||||
Look up a query against name and aliases.
|
||||
|
||||
Returns dict with the matching entry plus a 'confidence' field:
|
||||
- "exact": full name or alias match
|
||||
- "partial": query is a substring of an alias or name (or vice versa)
|
||||
- None if no match
|
||||
"""
|
||||
_reload_if_changed()
|
||||
q = query.strip().lower()
|
||||
if not q:
|
||||
return None
|
||||
|
||||
best = None
|
||||
best_confidence = None
|
||||
|
||||
for entry in _entries:
|
||||
# Exact match on name
|
||||
if q == entry['name'].lower():
|
||||
return {**entry, 'confidence': 'exact'}
|
||||
|
||||
# Exact match on any alias
|
||||
if q in entry['aliases']:
|
||||
return {**entry, 'confidence': 'exact'}
|
||||
|
||||
# Partial: query is substring of name/alias, or name/alias is substring of query
|
||||
name_lower = entry['name'].lower()
|
||||
if q in name_lower or name_lower in q:
|
||||
if best is None:
|
||||
best = entry
|
||||
best_confidence = 'partial'
|
||||
continue
|
||||
|
||||
for alias in entry['aliases']:
|
||||
if q in alias or alias in q:
|
||||
if best is None:
|
||||
best = entry
|
||||
best_confidence = 'partial'
|
||||
break
|
||||
|
||||
if best is not None:
|
||||
return {**best, 'confidence': best_confidence}
|
||||
|
||||
return None
|
||||
|
||||
|
||||
def list_all():
|
||||
"""Return all address book entries."""
|
||||
_reload_if_changed()
|
||||
return list(_entries)
|
||||
31
lib/address_book_api.py
Normal file
31
lib/address_book_api.py
Normal file
|
|
@ -0,0 +1,31 @@
|
|||
"""
|
||||
RECON Address Book API — Flask Blueprint.
|
||||
|
||||
GET /api/address_book/lookup?q=<query> — best match or 404
|
||||
GET /api/address_book/list — all entries
|
||||
"""
|
||||
|
||||
from flask import Blueprint, request, jsonify
|
||||
|
||||
from . import address_book
|
||||
|
||||
address_book_bp = Blueprint('address_book', __name__)
|
||||
|
||||
|
||||
@address_book_bp.route('/api/address_book/lookup')
|
||||
def api_address_book_lookup():
|
||||
q = request.args.get('q', '').strip()
|
||||
if not q:
|
||||
return jsonify({'error': 'Missing q parameter'}), 400
|
||||
|
||||
result = address_book.lookup(q)
|
||||
if result is None:
|
||||
return '', 404
|
||||
|
||||
return jsonify(result)
|
||||
|
||||
|
||||
@address_book_bp.route('/api/address_book/list')
|
||||
def api_address_book_list():
|
||||
entries = address_book.list_all()
|
||||
return jsonify(entries)
|
||||
57
lib/address_book_test.py
Normal file
57
lib/address_book_test.py
Normal file
|
|
@ -0,0 +1,57 @@
|
|||
#!/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 = [
|
||||
("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'),
|
||||
]
|
||||
|
||||
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)
|
||||
|
|
@ -44,6 +44,11 @@ app = Flask(__name__,
|
|||
|
||||
app.config['MAX_CONTENT_LENGTH'] = None # ZIM files can be multi-GB
|
||||
|
||||
# ── Address Book Blueprint ──
|
||||
from .address_book_api import address_book_bp
|
||||
app.register_blueprint(address_book_bp)
|
||||
|
||||
|
||||
# ── Navigation Constants ──
|
||||
|
||||
KNOWLEDGE_SUBNAV = [
|
||||
|
|
|
|||
|
|
@ -3,6 +3,10 @@
|
|||
import re
|
||||
import requests
|
||||
|
||||
from .utils import setup_logging
|
||||
|
||||
logger = setup_logging('recon.nav_tools')
|
||||
|
||||
PHOTON_URL = "http://localhost:2322"
|
||||
VALHALLA_URL = "http://localhost:8002"
|
||||
|
||||
|
|
@ -20,11 +24,26 @@ def _parse_coords(text: str):
|
|||
|
||||
|
||||
def _geocode(query: str):
|
||||
"""Geocode a place name via Photon. Returns (lat, lon, display_name) or raises."""
|
||||
"""Geocode a place name via address book then Photon. Returns (lat, lon, display_name) or raises."""
|
||||
coords = _parse_coords(query)
|
||||
if coords:
|
||||
return coords[0], coords[1], query
|
||||
|
||||
# ── Address book lookup (before Photon) ──
|
||||
try:
|
||||
from . import address_book
|
||||
match = address_book.lookup(query)
|
||||
if match and match['confidence'] == 'exact' and match.get('lat') and match.get('lon'):
|
||||
logger.info("Address book exact match: %r → %s (%s, %s)",
|
||||
query, match['name'], match['lat'], match['lon'])
|
||||
return match['lat'], match['lon'], match.get('address') or match['name']
|
||||
elif match and match['confidence'] == 'partial':
|
||||
logger.info("Address book partial match: %r → %s (falling through to Photon)",
|
||||
query, match['name'])
|
||||
except Exception as e:
|
||||
logger.debug("Address book lookup failed: %s", e)
|
||||
|
||||
# ── Photon geocoding ──
|
||||
try:
|
||||
resp = requests.get(
|
||||
f"{PHOTON_URL}/api",
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue