""" services/tool_compressor.py ───────────────────────────────────────────────────────────────────────────── Token-efficient tool output compression and the geocode_address_handler. compress_tool_output(): Converts raw RPC dicts into compact strings for LLM context. geocode_address_handler(): Resolves addresses to lat/lng via Mapbox. """ import csv import io import json import requests from services.observability import setup_logger from config import MAPBOX_ACCESS_TOKEN logger = setup_logger(__name__) _MAX_ROWS = 30 _MAX_RESULT_CHARS = 2000 _LIST_KEYS = ("items", "results", "compared", "rate_history", "ranked_properties", "newly_added", "untracked_or_removed") def _clean_value(val): """Truncates ISO datetime strings to date-only format for token economy.""" if isinstance(val, str) and len(val) >= 19 and "T" in val and (val.endswith("Z") or "+" in val): return val[:10] return val def _extract_raw_data(func_name: str, db_result: dict) -> tuple[str | None, any]: """Validates status and extracts the payload, handling scalars early.""" if db_result.get("status") != "success": msg = db_result.get("message", "unknown error") return f"Tool '{func_name}' error: {msg}", None data = db_result.get("data") if "data" in db_result else {k: v for k, v in db_result.items() if k != "status"} if not data: return f"Tool '{func_name}': no data returned.", None if isinstance(data, (str, int, float, bool)): return f"Tool '{func_name}' result: {data}", None return None, data def _find_list_envelope(data: dict) -> tuple[str | None, list | None, dict]: """Inspects a dict payload for known list envelope keys and extracts metadata.""" for lk in _LIST_KEYS: val = data.get(lk) if isinstance(val, list): outer_meta = {k: v for k, v in data.items() if k != lk and not isinstance(v, (list, dict))} return lk, val, outer_meta return None, None, {} def _unwrap_payload(func_name: str, db_result: dict) -> tuple[str | None, list | None, dict]: """Extracts list data and outer metadata from raw RPC result dictionary.""" early_err, data = _extract_raw_data(func_name, db_result) if early_err is not None: return early_err, None, {} outer_metadata = {} if isinstance(data, dict): found_key, list_val, outer_metadata = _find_list_envelope(data) if not found_key: return f"Tool '{func_name}' result:\n{json.dumps(data, default=str)}", None, {} data = list_val if not isinstance(data, list) or not data: meta_prefix = f"[{', '.join(f'{k}={v}' for k, v in outer_metadata.items())}]\n" if outer_metadata else "" return f"Tool '{func_name}':\n{meta_prefix}0 items returned.", None, {} return None, data, outer_metadata def _filter_and_slice_rows(func_name: str, rows: list) -> tuple[str | None, list[dict], bool]: """Filters out empty rows and slices list to _MAX_ROWS limit.""" data = [row for row in rows if isinstance(row, dict) and any(v is not None for v in row.values())] if not data: return f"Tool '{func_name}': all returned rows were empty.", [], False if len(data) > _MAX_ROWS: return None, data[:_MAX_ROWS], True return None, data, False def _find_candidate_keys(first_row: dict) -> list[str]: """Returns column keys to process, omitting redundant property_id when property_name exists.""" all_keys = list(first_row.keys()) has_name = "property_name" in all_keys if has_name and "property_id" in all_keys: return [k for k in all_keys if k != "property_id"] return all_keys def _hoist_metadata(data: list[dict], outer_metadata: dict) -> tuple[str, list[str]]: """Hoists columns with identical values across all rows into a bracketed header.""" candidate_keys = _find_candidate_keys(data[0]) hoisted = dict(outer_metadata) row_keys = [] for key in candidate_keys: unique_values = {str(row.get(key)) for row in data} if len(unique_values) == 1 and data[0].get(key) is not None: hoisted[key] = data[0].get(key) else: row_keys.append(key) header_parts = [f"{k}={v}" for k, v in hoisted.items()] header_str = f"[{', '.join(header_parts)}]\n" if header_parts else "" active_keys = [k for k in row_keys if any(row.get(k) is not None for row in data)] return header_str, active_keys def _render_csv_table(data: list[dict], active_keys: list[str], header_str: str, truncated: bool) -> str: """Renders data rows as CSV text with optional truncation notice.""" buf = io.StringIO() writer = csv.writer(buf) writer.writerow(active_keys) for row in data: writer.writerow([_clean_value(row.get(k)) for k in active_keys]) truncation_notice = ( f"\n[Truncated to {_MAX_ROWS} rows. Advise user to narrow date range or add filters.]" if truncated else "" ) return f"{header_str}{buf.getvalue().strip()}{truncation_notice}" def _cap_result_string(result_str: str) -> str: """Enforces _MAX_RESULT_CHARS character cap on formatted output.""" if len(result_str) <= _MAX_RESULT_CHARS: return result_str lines = result_str.split("\n") capped = [] total = 0 for line in lines: if total + len(line) > _MAX_RESULT_CHARS: capped.append("[Result capped at 2000 chars. Advise user to narrow filters.]") break capped.append(line) total += len(line) return "\n".join(capped) def compress_tool_output(func_name: str, db_result: dict) -> str: """Converts raw RPC dicts into compact strings for LLM context.""" early_return, data, outer_metadata = _unwrap_payload(func_name, db_result) if early_return is not None: return early_return early_filter_err, valid_rows, truncated = _filter_and_slice_rows(func_name, data) if early_filter_err is not None: return early_filter_err header_str, active_keys = _hoist_metadata(valid_rows, outer_metadata) rendered = _render_csv_table(valid_rows, active_keys, header_str, truncated) return _cap_result_string(rendered) def _parse_geocode_feature(features: list, address: str) -> dict: """Extracts coordinates and formatted address from Mapbox features.""" if not features: return { "status": "error", "message": f"We are unable to locate coordinates for '{address}'. Please try a more specific address.", } feature = features[0] lon, lat = feature["geometry"]["coordinates"] resolved = feature.get("properties", {}).get("full_address", address) return { "status": "success", "latitude": round(lat, 6), "longitude": round(lon, 6), "resolved_address": resolved, } def geocode_address_handler(address: str) -> dict: """Resolves a free-text address to lat/lng via the Mapbox Geocoding API.""" if not MAPBOX_ACCESS_TOKEN: return { "status": "error", "message": "We are unable to geocode addresses at this time - the mapping service is not configured.", } clean_addr = address.strip() if address else "" if not clean_addr: return {"status": "error", "message": "No address was provided to geocode."} params = { "q": clean_addr, "access_token": MAPBOX_ACCESS_TOKEN, "limit": 1, } from config import GEOCODE_COUNTRY_FILTER if GEOCODE_COUNTRY_FILTER: params["country"] = GEOCODE_COUNTRY_FILTER try: response = requests.get("https://api.mapbox.com/search/geocode/v6/forward", params=params, timeout=6) response.raise_for_status() return _parse_geocode_feature(response.json().get("features", []), clean_addr) except requests.exceptions.Timeout: logger.error(f"Mapbox geocode timeout for address: {clean_addr}") return {"status": "error", "message": "Mapping service timed out. Please try again shortly."} except requests.exceptions.HTTPError as e: logger.error(f"Mapbox geocode HTTP error {e.response.status_code} for: {clean_addr}") return {"status": "error", "message": "Mapping service returned an error. Please try again later."} except requests.exceptions.RequestException as e: logger.error(f"Mapbox geocode request failed for '{clean_addr}': {e}") return {"status": "error", "message": "Unable to reach the mapping service. Please try again later."} except (KeyError, IndexError, ValueError) as e: logger.error(f"Mapbox geocode parsing error for '{clean_addr}': {e}") return {"status": "error", "message": f"Unable to parse the location for '{clean_addr}'. Please try a more specific address."}