#!/usr/bin/env python3 """ GeoLynx stocklist importer (v2). Launched detached by www/fn/stocklist_import_manager.php, which writes our stdout and stderr to a per-job log and reads the tail back when something goes wrong. Connection settings come from the libpq service file at /etc/geolynx/pg_service.conf, service [geolynx] — the same entry www/fn/db.php, psql, QGIS and GeoServer read, so the site has one definition of how to reach the database rather than a copy per tool. The script used to carry its own credentials in an out-of-repo config.py. The manager passes PGSERVICE when db.php resolved from the service file, and explicit PGHOST/PGPORT/PGDATABASE/PGUSER/PGPASSWORD when it fell back to its own literals, so the importer always lands in the same database as the job row it is writing to. See _resolve_db(). Run by hand — no environment needed while the service file is in place: /opt/netplanner_env/bin/python3 python/stocklist_import_v2.py --check ... -stocklist 42 -file 0123abcd....xlsx -user 1 Override with PGSERVICE= (plus PGSERVICEFILE if the file is elsewhere), or with the explicit PG* variables. """ from __future__ import annotations import argparse import logging import os import re import sys import traceback import uuid from datetime import datetime from pathlib import Path from typing import Dict, List, Tuple, Optional, Any import psycopg2 import psycopg2.extras from psycopg2.extras import execute_batch logging.basicConfig( level=logging.INFO, stream=sys.stderr, format='%(asctime)s %(levelname)s %(message)s' ) logger = logging.getLogger(__name__) FILE_STORE_ROOT = os.environ.get('GEOLYNX_FILE_STORE') or '/var/www/netplanner-files' # Heavy third-party modules are bound here by load_runtime_dependencies() rather # than imported at module scope. A missing or broken pandas/openpyxl install would # otherwise kill the process before it could reach the jobs table, which is exactly # how an import failure used to present as a job frozen on 'Started'. psycopg2 is # the one exception — without it nothing can be reported to the database at all, # so it stays at the top and its failure is caught by the manager's log reading. # `from __future__ import annotations` above keeps the pd.DataFrame annotations # from being evaluated while pd is still None. pd = None openpyxl = None get_column_letter = None SequenceMatcher = None REQUIRED_ENV = ('PGHOST', 'PGDATABASE', 'PGUSER') # How one row of the import file is described back to the user, rendered in this # order — e.g. "5 Added, 3 In Another Stocklist". # # A single verdict cannot describe a row, because a row is not one premise. A UPRN # is one-to-one, but an address resolves to as many UPRNs as it covers: eight flats # where three are held by another stocklist is an addition AND an exclusion at once. # The feedback used to keep whichever UPRN sorted first and drop the rest, so that # row read "added_ok" and the exclusion vanished from the file (Dave, 2026-09-18). IMPORT_ACTION_LABELS = ( ('added_ok', 'Added'), ('address_already_in_stocklist', 'Already In This Stocklist'), ('address_in_another_stocklist', 'In Another Stocklist'), ('unable_to_match_address', 'Unmatched'), ('validation_error', 'Not In AddressBase'), ('parse_error', 'Parse Error'), ('skipped_no_uprn', 'No UPRN'), ) def summarise_import_actions(counts: Dict[str, int], resolved_uprns: int) -> str: """ One line describing what happened to one row of the import file. Counts are shown when the row resolved to at least one UPRN, since that is what is being counted. A row that resolved to none — an address that matched nothing, a line that would not parse — gets the bare label, because "1 Unmatched" would be counting something that does not exist. Unknown actions are passed through under their own name rather than dropped: a silently missing outcome is how the previous version lost exclusions. """ labels = dict(IMPORT_ACTION_LABELS) ordered = [a for a, _ in IMPORT_ACTION_LABELS if counts.get(a)] ordered += [a for a in counts if a not in labels and counts[a]] parts = [ f"{counts[a]} {labels.get(a, a)}" if resolved_uprns else labels.get(a, a) for a in ordered ] return ', '.join(parts) or 'No result' # The libpq service file is the site's one definition of how to reach the # database — db.php reads the same entry, as do psql, QGIS and GeoServer. # These defaults are what let the script run on the server with no environment # set at all, which matters when the thing being debugged is the environment. DEFAULT_SERVICE_FILE = '/etc/geolynx/pg_service.conf' DEFAULT_SERVICE_NAME = 'geolynx' def load_runtime_dependencies() -> None: """Import the processing libraries, binding them to the module globals.""" global pd, openpyxl, get_column_letter, SequenceMatcher import pandas as _pd import openpyxl as _openpyxl from openpyxl.utils import get_column_letter as _get_column_letter try: from cdifflib import CSequenceMatcher as _SequenceMatcher except ImportError: from difflib import SequenceMatcher as _SequenceMatcher pd = _pd openpyxl = _openpyxl get_column_letter = _get_column_letter SequenceMatcher = _SequenceMatcher def _point_libpq_at_service_file() -> None: """ Make sure libpq can find the service file. libpq searches ~/.pg_service.conf and $PGSYSCONFDIR/pg_service.conf and nowhere else, so a file at /etc/geolynx needs PGSERVICEFILE naming it. Setting PGSERVICEFILE without also naming a service does nothing at all — libpq reads a service file only when a service is asked for. """ if not os.environ.get('PGSERVICEFILE') and os.path.isfile(DEFAULT_SERVICE_FILE): os.environ['PGSERVICEFILE'] = DEFAULT_SERVICE_FILE def _resolve_db() -> Tuple[Dict[str, Any], str]: """ Work out how to connect, and say where that came from. Order matters: 1. PGSERVICE — someone naming a service means it. 2. A complete set of explicit PG* variables. This is what stocklist_import_manager.php passes when db.php could *not* read the service file and fell back to its own literals. Taking it ahead of the default below is what stops the importer resolving to one database while the job row it is writing to lives in another. 3. The default service, when its file is on disk — the no-environment case. Returned settings go straight to psycopg2.connect(**settings); libpq accepts 'service' as a connection keyword and resolves the rest itself, password included, so nothing secret need ever be in the environment. """ service = os.environ.get('PGSERVICE') if service: _point_libpq_at_service_file() return {'service': service}, _service_description(service) if all(os.environ.get(k) for k in REQUIRED_ENV): return { 'host': os.environ['PGHOST'], 'port': int(os.environ.get('PGPORT') or 5432), 'dbname': os.environ['PGDATABASE'], 'user': os.environ['PGUSER'], 'password': os.environ.get('PGPASSWORD') or None, }, 'PGHOST/PGPORT/PGDATABASE/PGUSER environment' if os.path.isfile(DEFAULT_SERVICE_FILE): _point_libpq_at_service_file() return ({'service': DEFAULT_SERVICE_NAME}, _service_description(DEFAULT_SERVICE_NAME) + ' (default)') raise RuntimeError( "No database settings. Set PGSERVICE (the service file is normally " + DEFAULT_SERVICE_FILE + ", named by PGSERVICEFILE if it is elsewhere), or " "set PGHOST, PGDATABASE and PGUSER explicitly. " "stocklist_import_manager.php passes whichever of the two www/fn/db.php used." ) def _service_description(service: str) -> str: return ('service ' + service + ' from ' + (os.environ.get('PGSERVICEFILE') or 'the libpq default service file path')) def db_settings() -> Dict[str, Any]: """Connection settings for psycopg2.connect(). See _resolve_db().""" return _resolve_db()[0] def db_settings_source() -> str: """Where the settings came from, for logs and --check. Never the password.""" return _resolve_db()[1] def report_job_failure(job_id: Optional[str], message: str) -> bool: """ Last-resort write of a failure to the jobs table, on its own connection. Used by the bootstrap for anything that goes wrong outside the processor — a dependency that will not import, an unreadable file, a connection that was never made. Returns False when even this could not be recorded, in which case the manager's log reading is the only channel left. """ if not job_id: return False try: conn = psycopg2.connect(**db_settings()) conn.autocommit = True with conn, conn.cursor() as cur: cur.execute(""" UPDATE stocklists.stocklist_import_jobs SET status = 'Error', progress = 0, message = 'Import failed', error_message = %s, completed_at = NOW() WHERE job_id = %s """, (message[:20000], job_id)) conn.close() return True except Exception: logger.error("Could not record the failure on the job row:\n%s", traceback.format_exc()) return False UK_POSTCODE_RE = re.compile( r'\b([A-Z]{1,2}[0-9][0-9A-Z]?\s*[0-9][A-Z]{2})\b', re.IGNORECASE ) SAO_PREFIX_RE = re.compile( r'^(flat|flats|apartment|apartments|unit|units|room|rooms|suite|suites|' r'maisonette|maisonettes|penthouse|studio|' r'ground\s+floor|first\s+floor|second\s+floor|third\s+floor|top\s+floor|' r'lower\s+ground|upper\s+ground|basement)(?:\s+flat)?(?:\s+\S+)?$', re.IGNORECASE ) # Lettered SAO range: "FLATS A-G", "APARTMENTS B-D" SAO_LETTER_RANGE_RE = re.compile( r'^(flat|flats|apartment|apartments|unit|units)s?\s+([A-Z])\s*[-\u2013]\s*([A-Z])$', re.IGNORECASE ) PAO_NAMED_RE = re.compile( r'^(block|tower|court|house|lodge|wing|hall|manor|grange|mill|barn|cottage)\s+\S+', re.IGNORECASE ) PAO_RANGE_RE = re.compile(r'^(\d+)[A-Za-z]?[-\u2013](\d+)[A-Za-z]?$') PAO_NUMBER_RE = re.compile(r'^(\d+)([A-Za-z]?)$') NUMBER_WORD_RE = re.compile(r'^(\d+[A-Za-z]?)\s+(.+)$') # Trailing number in a building name e.g. "WILLIAMSON COURT 163A" → name="WILLIAMSON COURT", num=163, suf="A" PAO_TRAILING_NUMBER_RE = re.compile(r'^(.+?)\s+(\d+)([A-Za-z]?)$') # Trailing range in a building name e.g. "RUTLAND LODGE 81-85" → name="RUTLAND LODGE", range="81-85" PAO_TRAILING_RANGE_RE = re.compile(r'^(.+?)\s+(\d+[-\u2013]\d+[A-Za-z]?)$') # Building-name suffixes — used to detect "1 BARNHILL HOUSE" as SAO+PAO rather than PAO+street. # Deliberately excludes ambiguous street-type words (CLOSE, ROAD, AVENUE, MEWS, PLACE, etc.) BUILDING_SUFFIX_RE = re.compile( r'\b(house|court|courts|apartments?|towers?|lodge|lodges|hall|halls|' r'mansions?|heights|blocks?|buildings|chambers|villas?)\b', re.IGNORECASE ) ABP_FILTERS = "AND (building_classification_code IS NULL OR building_classification_code != 'PP')" # Exclude property shells class GeoLynxExcelProcessor: """Stocklist importer — supports UPRN direct, TOID direct, and address matching.""" def __init__(self, stocklist_id: int, user_id: int = 0, job_id: str = None): self.stocklist_id = int(stocklist_id) self.user_id = int(user_id) self.job_id = job_id # UUID from stocklist_import_jobs self.connection = None self.cursor = None self.job_connection = None # separate autocommit conn for progress self.batch_id = None self.file_upload_id = None self.primary_identifier = None # 'UPRN' | 'TOID' | 'ADDRESS' self.uprn_pattern = re.compile(r'^\d+$') self.toid_pattern = re.compile(r'^osgb\d{12,20}$') # ------------------------------------------------------------------ # # Infrastructure # ------------------------------------------------------------------ # def update_job_progress(self, progress: int, status: str, message: str = "", error: str = "", **kwargs): """Write progress to the jobs table. Falls back to stderr logging if no job_id.""" log_line = f"[{status}] {progress}% — {message}" + (f" ERROR: {error}" if error else "") logger.info(log_line) if not self.job_id: return try: if not self.job_connection or self.job_connection.closed: self._connect_job_connection() with self.job_connection.cursor() as cur: cur.execute(""" UPDATE stocklists.stocklist_import_jobs SET status = %s, progress = %s, message = %s, error_message = %s, processed_rows = COALESCE(%s, processed_rows), matched_rows = COALESCE(%s, matched_rows), unmatched_rows = COALESCE(%s, unmatched_rows), total_rows = COALESCE(%s, total_rows), rows_inserted = COALESCE(%s, rows_inserted), batch_id = COALESCE(%s, batch_id), completed_at = CASE WHEN %s IN ('Complete','Error') THEN NOW() ELSE completed_at END WHERE job_id = %s """, ( status, progress, message, error or None, kwargs.get('processed_rows'), kwargs.get('matched_rows'), kwargs.get('unmatched_rows'), kwargs.get('total_rows'), kwargs.get('rows_inserted'), kwargs.get('batch_id'), status, self.job_id, )) except Exception: # Never silently. A progress write that fails is how a job ends up # frozen on its last status while the run carries on or dies unseen, # so the traceback goes to the log the manager reads. logger.error("Failed to update job progress (status=%s progress=%s):\n%s", status, progress, traceback.format_exc()) def _connect_job_connection(self) -> None: self.job_connection = psycopg2.connect(**db_settings()) self.job_connection.autocommit = True def connect_to_database(self) -> None: try: self.connection = psycopg2.connect(**db_settings()) self.cursor = self.connection.cursor( cursor_factory=psycopg2.extras.RealDictCursor ) # Log what libpq actually resolved, not what we asked for — with a # service file those differ, and the resolved values are the ones # worth having in the log when a database turns out to be the wrong one. info = self.connection.get_dsn_parameters() logger.info("Database connection established: %s@%s:%s/%s (settings from %s)", info.get('user'), info.get('host'), info.get('port'), info.get('dbname'), db_settings_source()) except Exception as e: # The progress write below shares these settings, so it will usually # fail too. report_job_failure() in the bootstrap is the backstop. self.update_job_progress(0, "Error", error=f"Database connection failed: {e}") raise def close_connection(self) -> None: if self.cursor: self.cursor.close() if self.connection: self.connection.close() if self.job_connection and not self.job_connection.closed: self.job_connection.close() def generate_batch_id(self) -> str: return f"{datetime.now().strftime('%Y%m%d_%H%M%S')}_{str(uuid.uuid4())[:8]}" def ensure_schema_migrations(self) -> None: """Idempotently apply required schema changes.""" self.cursor.execute(""" ALTER TABLE public.geolynx_identifiers ADD COLUMN IF NOT EXISTS file_upload_id BIGINT """) self.cursor.execute(""" CREATE TABLE IF NOT EXISTS public.geolynx_address_staging ( id SERIAL PRIMARY KEY, file_upload_id BIGINT NOT NULL, stocklist_id INTEGER NOT NULL, batch_id VARCHAR(50) NOT NULL, source_row INTEGER, address_raw TEXT NOT NULL, postcode VARCHAR(8), address_type VARCHAR(20), match_method VARCHAR(30), match_confidence VARCHAR(10), uprn BIGINT, import_action VARCHAR(50) NOT NULL, notes TEXT, upload_timestamp TIMESTAMPTZ DEFAULT NOW() ) """) self.cursor.execute(""" CREATE INDEX IF NOT EXISTS geolynx_address_staging_file_upload_idx ON public.geolynx_address_staging (file_upload_id) """) self.cursor.execute(""" CREATE INDEX IF NOT EXISTS geolynx_address_staging_stocklist_idx ON public.geolynx_address_staging (stocklist_id) """) self.connection.commit() def resolve_file_upload(self, safe_name: str) -> None: """Fetch the file_uploads record already created by PHP.""" self.cursor.execute(""" SELECT id, file_upload_user FROM public.file_uploads WHERE file_id = %s AND stocklist_id = %s AND file_is_deleted = false """, (safe_name, self.stocklist_id)) row = self.cursor.fetchone() if not row: raise ValueError( f"No file_uploads record for '{safe_name}' / stocklist {self.stocklist_id}" ) self.file_upload_id = row['id'] if self.user_id == 0 and row['file_upload_user']: self.user_id = row['file_upload_user'] logger.info(f"file_upload_id={self.file_upload_id}") # ------------------------------------------------------------------ # # File reading # ------------------------------------------------------------------ # def read_file(self, file_path: str) -> pd.DataFrame: ext = Path(file_path).suffix.lower() try: if ext == '.csv': df = pd.read_csv(file_path, encoding='utf-8-sig') else: df = pd.read_excel(file_path, engine='openpyxl') df.attrs['file_path'] = file_path logger.info(f"Read {df.shape[0]} rows from {file_path}") return df except Exception as e: logger.error(f"Failed to read file: {e}") raise # ------------------------------------------------------------------ # # Column detection & validation # ------------------------------------------------------------------ # def find_identifier_columns(self, df: pd.DataFrame) -> Dict[str, str]: """ Map identifier kind to column name, returning every kind present. Collecting all of them matters: a file can carry a UPRN column holding some other system's numbering, and the address column beside it is then the only usable identifier in the sheet. Which one to *use* is validate_file's decision, made once it knows which actually yielded values. This used to stop looking for an address as soon as a UPRN column was seen, which discarded the fallback and made the result depend on column order — ADDRESS before UPRN found both, UPRN before ADDRESS found only the UPRN. """ found: Dict[str, str] = {} # Feedback columns appended by a previous run (e.g. uprn_matched, match_method) # must not be mistaken for source identifier columns. Use exact or prefix # matches. matched_address is here because 'address' in cl would otherwise # pick a previous run's output as this run's address source. feedback_cols = {'import_action', 'uprns_matched', 'uprn_matched', 'match_method', 'match_confidence', 'matched_address', 'notes'} for col in df.columns: cl = str(col).lower().strip() if cl in feedback_cols: continue if cl == 'uprn' or cl.startswith('uprn_id') or cl.startswith('uprn '): found.setdefault('uprn', col) elif cl == 'toid': found.setdefault('toid', col) elif cl == 'address': found['address'] = col # an exact header beats a partial one elif 'address' in cl: found.setdefault('address', col) return found def validate_uprn(self, value: Any) -> bool: """ Whether a value is *shaped* like a UPRN. It cannot tell you whether it is one. A UPRN has no checksum and no enforced length, so this can only check the value is all digits — which means a supplier's own reference '1' passes and is then indistinguishable from Ordnance Survey UPRN 1, a different property somewhere else entirely. That failure is silent and wrong rather than empty: the import succeeds and assigns premises nobody uploaded. A non-numeric supplier reference is the safe case — it fails validation and the file falls through to address matching. The numeric case has no defence here. assign_premises() warns when a whole batch matches nothing in AddressBase, which catches wholly foreign numbering but says nothing about a partial collision. See the Stocklist Import section of CLAUDE.md. """ if pd.isna(value): return False s = str(value).strip() if '.' in s: s = s.split('.')[0] return bool(self.uprn_pattern.match(s)) def validate_toid(self, value: Any) -> bool: if pd.isna(value): return False return bool(self.toid_pattern.match(str(value).strip().lower())) @staticmethod def _sample_values(rows: List[dict], limit: int = 3) -> str: sample = ', '.join(f"'{r['value']}'" for r in rows[:limit]) extra = f" and {len(rows) - limit} more" if len(rows) > limit else '' return sample + extra def _describe_no_identifier(self, details: Dict[str, Any]) -> str: """ Say which columns were read and why each gave nothing. 'No valid UPRN, TOID, or address values found' sends people to look at the wrong thing — usually the whole file, when the answer is that one named column held the wrong kind of value and is quotable back at them. """ parts = [] if details['has_uprn']: bad = details['invalid_uprns'] if bad: parts.append( f"column '{details['uprn_column']}' held {len(bad)} value(s), none of " f"them a UPRN (a UPRN is all digits) — e.g. {self._sample_values(bad)}" ) else: parts.append(f"column '{details['uprn_column']}' was empty") if details['has_toid']: bad = details['invalid_toids'] if bad: parts.append( f"column '{details['toid_column']}' held {len(bad)} value(s), none of " f"them a TOID — e.g. {self._sample_values(bad)}" ) else: parts.append(f"column '{details['toid_column']}' was empty") if details['has_address']: parts.append( f"column '{details['address_column']}' had no non-blank addresses" ) else: parts.append("no address column was found to fall back to") return ("Nothing in this file could be used as an identifier: " + '; '.join(parts) + '.') def validate_file(self, df: pd.DataFrame) -> Tuple[bool, str, Dict[str, Any]]: self.update_job_progress(25, "validating", "Analysing file structure...") details: Dict[str, Any] = { 'total_rows': len(df), 'has_uprn': False, 'uprn_column': None, 'valid_uprns': 0, 'invalid_uprns': [], 'has_toid': False, 'toid_column': None, 'valid_toids': 0, 'invalid_toids': [], 'has_address': False, 'address_column': None, 'address_rows': 0, 'parse_errors': [], 'primary_identifier': None } if df.empty: return False, "File is empty", details id_cols = self.find_identifier_columns(df) self.update_job_progress(26, "validating", f"Detected columns: {list(df.columns)}", identifier_columns=id_cols) if not id_cols: return False, ( "No identifier column found. File must contain a column whose header " "includes 'uprn', 'toid', or 'address'." ), details if 'uprn' in id_cols: details['has_uprn'] = True details['uprn_column'] = id_cols['uprn'] for idx, val in df[id_cols['uprn']].items(): if not pd.isna(val): if self.validate_uprn(val): details['valid_uprns'] += 1 else: details['invalid_uprns'].append({'row': idx + 2, 'value': val}) if 'toid' in id_cols: details['has_toid'] = True details['toid_column'] = id_cols['toid'] for idx, val in df[id_cols['toid']].items(): if not pd.isna(val): if self.validate_toid(val): details['valid_toids'] += 1 else: details['invalid_toids'].append({'row': idx + 2, 'value': val}) # Addresses are the fallback, analysed only when no direct identifier # yielded anything — a UPRN column full of another system's numbering # should not stop the addresses beside it being used. Gating on values # rather than on the column being absent also keeps this off the common # path: parse_address_string() runs per row and is not free. if 'address' in id_cols and not details['valid_uprns'] and not details['valid_toids']: details['has_address'] = True details['address_column'] = id_cols['address'] for idx, val in df[id_cols['address']].items(): if not pd.isna(val) and str(val).strip(): details['address_rows'] += 1 parsed = self.parse_address_string(str(val)) if parsed['address_type'] == 'parse_error': details['parse_errors'].append( {'row': idx + 2, 'value': val, 'reason': parsed['error']} ) if details['valid_uprns'] > 0: details['primary_identifier'] = 'UPRN' elif details['valid_toids'] > 0: details['primary_identifier'] = 'TOID' elif details['has_address'] and details['address_rows'] > 0: details['primary_identifier'] = 'ADDRESS' else: return False, self._describe_no_identifier(details), details self.primary_identifier = details['primary_identifier'] parts = [] if details['valid_uprns']: parts.append(f"{details['valid_uprns']} valid UPRNs") if details['valid_toids']: parts.append(f"{details['valid_toids']} valid TOIDs") if details['address_rows']: parts.append(f"{details['address_rows']} address rows") if details['parse_errors']: parts.append(f"{len(details['parse_errors'])} parse errors") msg = f"Validation complete: {', '.join(parts)}. Using {details['primary_identifier']}." # Say so when a direct identifier was present but unusable. Otherwise a # run that fell back to address matching is indistinguishable from one # that read the UPRN column, and the two carry very different confidence. if details['primary_identifier'] == 'ADDRESS' and details['has_uprn']: msg += (f" The UPRN column '{details['uprn_column']}' held no Ordnance Survey " f"UPRN, so addresses were matched instead.") self.update_job_progress(30, "validated", msg) return True, msg, details # ------------------------------------------------------------------ # # Address parsing # ------------------------------------------------------------------ # @staticmethod def parse_address_string(raw: str) -> dict: result = { 'raw': raw, 'postcode': None, 'pao_number': None, 'pao_start_suffix': None, 'range_start': None, 'range_end': None, 'pao_text': None, 'sao_text': None, 'sao_start_number': None, 'sao_start_suffix': None, 'sao_letter_start': None, 'sao_letter_end': None, 'street': None, 'town': None, 'address_type': None, 'error': None } elements = [e.strip() for e in raw.split(',') if e.strip()] # Extract postcode (required) postcode, postcode_idx = None, None for i, el in enumerate(elements): m = UK_POSTCODE_RE.search(el) if m: pc = re.sub(r'\s+', '', m.group(1).upper()) postcode = pc[:-3] + ' ' + pc[-3:] postcode_idx = i break if postcode is None: result['address_type'] = 'parse_error' result['error'] = ( 'missing postcode — address must include a UK postcode ' '(e.g. ME14 1AE) as a comma-separated element' ) return result result['postcode'] = postcode remaining = [el for i, el in enumerate(elements) if i != postcode_idx] if len(remaining) < 2: result['address_type'] = 'parse_error' result['error'] = 'too few address elements after postcode removed (minimum: number/name + street)' return result result['town'] = remaining[-1] components = remaining[:-1] # If an SAO element exists but isn't first (e.g. "2 HILLSIDE, FLAT 1, LONDON, PC"), # move it to the front so downstream logic always sees SAO → PAO → street order. sao_idx = next( (i for i, c in enumerate(components) if SAO_LETTER_RANGE_RE.match(c) or SAO_PREFIX_RE.match(c)), None ) if sao_idx is not None and sao_idx > 0: components = ( [components[sao_idx]] + [c for i, c in enumerate(components) if i != sao_idx] ) first = components[0] street_parts = components[1:] # Lettered SAO range — "FLATS A-G", "APARTMENTS B-D" m = SAO_LETTER_RANGE_RE.match(first) if m: result['sao_text'] = first result['sao_letter_start'] = m.group(2).upper() result['sao_letter_end'] = m.group(3).upper() # Parse remaining elements as PAO + street using same logic as SAO branch below if street_parts: next_el = street_parts[0] rest = street_parts[1:] if len(street_parts) == 1: mv = NUMBER_WORD_RE.match(next_el) if mv: nm = PAO_NUMBER_RE.match(mv.group(1)) if nm: result['pao_number'] = int(nm.group(1)) result['pao_start_suffix'] = nm.group(2) or None result['street'] = mv.group(2) elif PAO_NUMBER_RE.match(next_el): nm = PAO_NUMBER_RE.match(next_el) result['pao_number'] = int(nm.group(1)) result['pao_start_suffix'] = nm.group(2) or None else: result['street'] = next_el else: if PAO_NUMBER_RE.match(next_el): nm = PAO_NUMBER_RE.match(next_el) result['pao_number'] = int(nm.group(1)) result['pao_start_suffix'] = nm.group(2) or None else: result['pao_text'] = next_el result['street'] = ', '.join(rest) result['address_type'] = 'sao_letter_range' return result # SAO prefix — flat/unit/floor etc. if SAO_PREFIX_RE.match(first): result['sao_text'] = first m = re.search(r'\d+', first) if m: result['sao_start_number'] = int(m.group()) if street_parts: next_el = street_parts[0] rest = street_parts[1:] if len(street_parts) == 1: # Single remaining element — may be "2 HILLSIDE" (number fused with street) m = NUMBER_WORD_RE.match(next_el) if m: nm = PAO_NUMBER_RE.match(m.group(1)) if nm: result['pao_number'] = int(nm.group(1)) result['pao_start_suffix'] = nm.group(2) or None result['street'] = m.group(2) elif PAO_NUMBER_RE.match(next_el): # Bare number, no street name (unusual) nm = PAO_NUMBER_RE.match(next_el) result['pao_number'] = int(nm.group(1)) result['pao_start_suffix'] = nm.group(2) or None result['street'] = None else: result['street'] = next_el else: # 2+ remaining elements — first is the PAO (number or building name), # the rest is the street. e.g. ["OCTAVE HOUSE", "EMPIRE WAY"] if PAO_NUMBER_RE.match(next_el): nm = PAO_NUMBER_RE.match(next_el) result['pao_number'] = int(nm.group(1)) result['pao_start_suffix'] = nm.group(2) or None else: # Detect trailing number in building name e.g. "WILLIAMSON COURT 163A" # ABP stores this as pao='WILLIAMSON COURT, 163A', pao_start_number=163 mt = PAO_TRAILING_NUMBER_RE.match(next_el) if mt and BUILDING_SUFFIX_RE.search(mt.group(1)): result['pao_text'] = mt.group(1) result['pao_number'] = int(mt.group(2)) result['pao_start_suffix'] = mt.group(3) or None else: # Detect trailing range in building name e.g. "RUTLAND LODGE 81-85" # ABP stores building name separately from range pao mr = PAO_TRAILING_RANGE_RE.match(next_el) if mr and BUILDING_SUFFIX_RE.search(mr.group(1)): result['pao_text'] = mr.group(1) rm = PAO_RANGE_RE.match(mr.group(2)) if rm: result['range_start'] = int(rm.group(1)) result['range_end'] = int(rm.group(2)) else: result['pao_text'] = next_el # If there is exactly one street element and it starts with a number, # extract it as pao_number — ABP may combine building name + number # into a single pao value e.g. "LUCERNE APARTMENTS, 91" with # pao_start_number=91 and street_description="HILLTOP AVENUE". if len(rest) == 1 and result.get('pao_text'): m = NUMBER_WORD_RE.match(rest[0]) if m: nm = PAO_NUMBER_RE.match(m.group(1)) if nm: result['pao_number'] = int(nm.group(1)) result['pao_start_suffix'] = nm.group(2) or None result['street'] = m.group(2) else: result['street'] = rest[0] else: result['street'] = ', '.join(rest) else: result['street'] = None result['address_type'] = 'unit' return result # PAO range — 1-10 m = PAO_RANGE_RE.match(first) if m: result['range_start'] = int(m.group(1)) result['range_end'] = int(m.group(2)) result['street'] = ', '.join(street_parts) if street_parts else None result['address_type'] = 'range' return result # PAO number only — 42 or 42A m = PAO_NUMBER_RE.match(first) if m: pao_num = int(m.group(1)) pao_suf = m.group(2) or None # Detect when the bare number is actually an SAO — happens when the next element # is a pure-text named building, e.g.: # "2, JOHN WILSON HOUSE, 167 WILLESDEN LANE" → sao=2, pao='JOHN WILSON HOUSE, 167', street='WILLESDEN LANE' # "1, BARNHILL HOUSE, ST DAVIDS CLOSE" → sao=1, pao='BARNHILL HOUSE', street='ST DAVIDS CLOSE' if (len(street_parts) >= 2 and not PAO_NUMBER_RE.match(street_parts[0]) and not NUMBER_WORD_RE.match(street_parts[0])): result['sao_start_number'] = pao_num result['sao_start_suffix'] = pao_suf result['sao_text'] = str(pao_num) + (pao_suf or '') result['pao_text'] = street_parts[0] if NUMBER_WORD_RE.match(street_parts[1]): # Next element fuses a pao number with the street name # e.g. "167 WILLESDEN LANE" → pao_number=167, street='WILLESDEN LANE' mv = NUMBER_WORD_RE.match(street_parts[1]) nm = PAO_NUMBER_RE.match(mv.group(1)) if nm: result['pao_number'] = int(nm.group(1)) result['pao_start_suffix'] = nm.group(2) or None # Use only the extracted street name — any further elements are # locality/area names (e.g. "KILBURN") not in ABP street_description result['street'] = mv.group(2) else: # Next element is the plain street name — e.g. "ST DAVIDS CLOSE" # Any further elements are locality names, discard them result['street'] = street_parts[1] result['address_type'] = 'unit' return result result['pao_number'] = pao_num result['pao_start_suffix'] = pao_suf result['street'] = ', '.join(street_parts) if street_parts else None result['address_type'] = 'single' return result # Named building keyword if PAO_NAMED_RE.match(first): result['pao_text'] = first result['street'] = ', '.join(street_parts) if street_parts else None result['address_type'] = 'named_building' return result # Number + word in same element — "42 High Street" or "1 BARNHILL HOUSE" m = NUMBER_WORD_RE.match(first) if m: bldg_text = m.group(2) # text portion after the number nm = PAO_NUMBER_RE.match(m.group(1)) # If the text after the number contains a clear building-name keyword AND # there are further address elements, treat number as SAO and text as PAO. # e.g. "1 BARNHILL HOUSE, ST. DAVIDS CLOSE" → sao=1, pao=BARNHILL HOUSE, street=ST DAVIDS CLOSE if BUILDING_SUFFIX_RE.search(bldg_text) and street_parts: result['sao_start_number'] = int(nm.group(1)) if nm else None result['sao_text'] = m.group(1) result['pao_text'] = bldg_text # Use only the first remaining element as street — further elements # are locality names (e.g. "FOREST HILL") not in ABP street_description result['street'] = street_parts[0] result['address_type'] = 'unit' return result # Standard: number is the PAO house number, remaining text + street_parts form the street if nm: result['pao_number'] = int(nm.group(1)) result['pao_start_suffix'] = nm.group(2) or None street_els = [bldg_text] + street_parts result['street'] = ', '.join(street_els) result['address_type'] = 'single' return result # Fallback — if elements remain after first, treat first as a named building (PAO) # and the rest as the street. Catches anything not matched above: # "HILLSIDE COURT, EMPIRE WAY, ...", "THE OLD BARN, HIGH STREET, ..." if street_parts: result['pao_text'] = first result['street'] = ', '.join(street_parts) result['address_type'] = 'named_building' else: result['street'] = first result['address_type'] = 'single' return result # ------------------------------------------------------------------ # # ABP matching — phases 1-6 # ------------------------------------------------------------------ # def _abp_query(self, sql: str, params: tuple) -> list: self.cursor.execute(sql, params) return self.cursor.fetchall() def match_address_in_abp(self, parsed: dict) -> List[dict]: atype = parsed.get('address_type') if atype == 'sao_letter_range': res = self._phase_sao_letter_range(parsed) if res: return res if atype == 'unit': res = self._phase1_unit(parsed) if res: return res if atype == 'single' and parsed.get('pao_number') and parsed.get('street'): res = self._phase2_exact(parsed) if res: return res # Number may be an SAO within a range PAO e.g. '8 PLASSY ROAD' → pao='6-28', sao='8' res = self._phase2b_range_sao(parsed) if res: return res if atype == 'range': res = self._phase3_range(parsed) if res: return res if atype == 'named_building' and parsed.get('pao_text'): res = self._phase4_named_building(parsed) if res: return res # Phase 5 fuzzy fallback for single/unit that found nothing above if atype in ('single', 'unit'): res = self._phase5_fuzzy(parsed) if res: return res # Phase 6 — no match; run diagnostics to show where matching broke down diag = self._diagnose_no_match(parsed) return [{ 'uprn': None, 'address_raw': parsed['raw'], 'postcode': parsed.get('postcode'), 'address_type': atype, 'match_method': 'unmatched', 'match_confidence': None, 'import_action': 'unable_to_match_address', 'notes': diag }] def _diagnose_no_match(self, p: dict) -> str: """Run stepped diagnostic counts to show where matching breaks down.""" lines = [] # Always show what was parsed parsed_summary = ( f"parsed: type={p.get('address_type')} " f"postcode={p.get('postcode')!r} " f"street={p.get('street')!r} " f"pao_number={p.get('pao_number')} " f"pao_text={p.get('pao_text')!r} " f"sao_text={p.get('sao_text')!r} " f"sao_number={p.get('sao_start_number')} " f"range={p.get('range_start')}-{p.get('range_end')}" ) lines.append(parsed_summary) try: # Step 1 — postcode exists in ABP at all? self.cursor.execute( f"SELECT COUNT(*) AS n FROM basedata.abp_main " f"WHERE postcode = %s {ABP_FILTERS}", (p.get('postcode'),) ) n_pc = (self.cursor.fetchone() or {}).get('n', 0) lines.append(f"postcode {p.get('postcode')!r}: {n_pc} ABP rows (filtered)") if n_pc == 0: # Check without filters — reveals if postcode exists but is filtered out, # or if the postcode format itself doesn't match self.cursor.execute( "SELECT COUNT(*) AS n FROM basedata.abp_main WHERE postcode = %s", (p.get('postcode'),) ) n_unfiltered = (self.cursor.fetchone() or {}).get('n', 0) lines.append(f"postcode unfiltered: {n_unfiltered} rows") if n_unfiltered == 0: # Show a sample of nearby postcodes to reveal the stored format self.cursor.execute( "SELECT DISTINCT postcode FROM basedata.abp_main " "WHERE postcode LIKE %s LIMIT 3", (p.get('postcode', '')[:3] + '%',) ) samples = [r['postcode'] for r in self.cursor.fetchall()] lines.append(f"ABP postcode format samples ({p.get('postcode', '')[:3]}...): {samples}") if samples: lines.append("STOP: postcode not found, though the area is present " "— likely a format mismatch") else: # No postcode in this area at all, which is a different problem # entirely. Saying 'format mismatch' here sent a real # investigation the wrong way (2026-09-13) when the table was # simply not loaded. reltuples rather than COUNT(*): this runs # once per unmatched row, and an estimate answers 'empty or not'. self.cursor.execute( "SELECT reltuples::bigint AS n FROM pg_class " "WHERE oid = 'basedata.abp_main'::regclass" ) n_all = (self.cursor.fetchone() or {}).get('n', 0) lines.append(f"basedata.abp_main holds roughly {n_all} rows") lines.append( "STOP: no postcode in this area at all — basedata.abp_main is " "empty or does not cover it. That is a data-load problem, not " "an address problem." ) else: lines.append("STOP: postcode exists but every row is excluded by " "ABP_FILTERS (building_classification_code = 'PP')") return ' | '.join(lines) # Step 2 — postcode + street if p.get('street'): self.cursor.execute( f"SELECT COUNT(*) AS n FROM basedata.abp_main " f"WHERE postcode = %s AND street_description = %s {ABP_FILTERS}", (p['postcode'], p['street']) ) n_st = (self.cursor.fetchone() or {}).get('n', 0) lines.append(f"postcode+street {p['street']!r}: {n_st} rows") if n_st == 0: # Show what street values actually exist for this postcode self.cursor.execute( f"SELECT DISTINCT street_description FROM basedata.abp_main " f"WHERE postcode = %s {ABP_FILTERS} LIMIT 5", (p['postcode'],) ) streets = [r['street_description'] for r in self.cursor.fetchall()] lines.append(f"streets in ABP for this postcode: {streets}") lines.append("STOP: street not matched") return ' | '.join(lines) # Step 3 — postcode + street + pao_number if p.get('pao_number') is not None and p.get('street'): self.cursor.execute( f"SELECT COUNT(*) AS n FROM basedata.abp_main " f"WHERE postcode = %s AND street_description = %s " f"AND pao_start_number = %s {ABP_FILTERS}", (p['postcode'], p['street'], p['pao_number']) ) n_pao = (self.cursor.fetchone() or {}).get('n', 0) lines.append(f"postcode+street+pao_number={p['pao_number']}: {n_pao} rows") # Step 4 — postcode + street + pao if p.get('pao_text') and p.get('street'): self.cursor.execute( f"SELECT COUNT(*) AS n FROM basedata.abp_main " f"WHERE postcode = %s AND street_description = %s " f"AND pao ILIKE %s {ABP_FILTERS}", (p['postcode'], p['street'], f"%{p['pao_text']}%") ) n_pt = (self.cursor.fetchone() or {}).get('n', 0) lines.append(f"postcode+street+pao LIKE {p['pao_text']!r}: {n_pt} rows") # Step 5 — postcode + sao if p.get('sao_text'): self.cursor.execute( f"SELECT COUNT(*) AS n FROM basedata.abp_main " f"WHERE postcode = %s AND sao = %s {ABP_FILTERS}", (p['postcode'], p['sao_text']) ) n_sao = (self.cursor.fetchone() or {}).get('n', 0) lines.append(f"postcode+sao={p['sao_text']!r}: {n_sao} rows") except Exception as e: lines.append(f"diagnostic error: {e}") return ' | '.join(lines) def _phase_sao_letter_range(self, p: dict) -> List[dict]: """Match a lettered SAO range e.g. FLATS A-G. Tries two strategies: 1. Match individual records where sao_text falls within the letter range. 2. Find a parent block record and expand its children. """ if not p.get('sao_letter_start') or not p.get('sao_letter_end'): return [] prefix = p['sao_text'].split()[0].rstrip('s').title() # "FLATS" → "Flat" letter_start = p['sao_letter_start'] letter_end = p['sao_letter_end'] params_base = [p['postcode']] pao_clause, pao_params = '', [] if p.get('pao_number') is not None: pao_clause = 'AND pao_start_number = %s' pao_params = [p['pao_number']] elif p.get('pao_text'): pao_clause = 'AND pao ILIKE %s' pao_params = [p['pao_text']] street_clause = 'AND street_description = %s' if p.get('street') else '' street_params = [p['street']] if p.get('street') else [] # Strategy 1: individual records with sao_text like "FLAT A" .. "FLAT G" sql = f""" SELECT uprn FROM basedata.abp_main WHERE postcode = %s AND sao BETWEEN %s AND %s {pao_clause} {street_clause} {ABP_FILTERS} ORDER BY sao """ rows = self._abp_query(sql, tuple( params_base + [f'{prefix} {letter_start}', f'{prefix} {letter_end}'] + pao_params + street_params )) if rows: uprns = [r['uprn'] for r in rows] return self._build_results(uprns, p, 'sao_letter_range', 'high') # Strategy 2: find parent block record and return all children sql2 = f""" SELECT uprn, multi_occ_count FROM basedata.abp_main WHERE postcode = %s {pao_clause} {street_clause} AND sao IS NULL {ABP_FILTERS} LIMIT 1 """ parent_rows = self._abp_query(sql2, tuple(params_base + pao_params + street_params)) if parent_rows and (parent_rows[0].get('multi_occ_count') or 0) > 0: child_sql = f""" SELECT uprn FROM basedata.abp_main WHERE parent_uprn = %s {ABP_FILTERS} """ children = self._abp_query(child_sql, (parent_rows[0]['uprn'],)) if children: uprns = [r['uprn'] for r in children] return self._build_results(uprns, p, 'sao_letter_range', 'medium') return [] def _phase1_unit(self, p: dict) -> List[dict]: if not p.get('street'): return [] sao_conds, params = [], [p['postcode']] if p.get('sao_text'): # Match bare sao value OR prefixed form e.g. '1' matches 'FLAT 1', 'UNIT 1' etc. # The trailing-space anchor ('% 1') prevents '1' matching '11' or '21'. sao_conds.append("(sao = %s OR sao ILIKE '%%' || ' ' || %s)") params.extend([p['sao_text'], p['sao_text']]) if p.get('sao_start_number') is not None: sao_suf = p.get('sao_start_suffix') or None if sao_suf: sao_conds.append("(sao_start_number = %s AND sao_start_suffix = %s)") params.extend([p['sao_start_number'], sao_suf]) else: sao_conds.append("(sao_start_number = %s AND (sao_start_suffix IS NULL OR sao_start_suffix = ''))") params.append(p['sao_start_number']) if not sao_conds: return [] pao_conds = [] if p.get('pao_number') is not None: suffix = p.get('pao_start_suffix') or None pao_str = str(p['pao_number']) + (suffix or '') if suffix: pao_conds.append("(pao = %s OR (pao_start_number = %s AND pao_start_suffix = %s))") params.extend([pao_str, p['pao_number'], suffix]) else: pao_conds.append("(pao = %s OR (pao_start_number = %s AND (pao_start_suffix IS NULL OR pao_start_suffix = '')))") params.extend([pao_str, p['pao_number']]) if p.get('pao_text'): # Exact match OR "NAME, number" pattern (e.g. "LUCERNE APARTMENTS, 91") pao_conds.append("(pao = %s OR pao ILIKE %s || ', %%')") params.extend([p['pao_text'], p['pao_text']]) # Range PAO e.g. pao_start_number=81, pao_end_number=85 from "RUTLAND LODGE 81-85" # Building name is not verified in ABP so confidence is medium has_range_pao = p.get('range_start') is not None and p.get('range_end') is not None if has_range_pao: pao_conds.append("(pao_start_number = %s AND pao_end_number = %s)") params.extend([p['range_start'], p['range_end']]) params.append(p['street']) pao_clause = f"AND ({' OR '.join(pao_conds)})" if pao_conds else "" sql = f""" SELECT uprn FROM basedata.abp_main WHERE postcode = %s AND ({' OR '.join(sao_conds)}) {pao_clause} AND street_description = %s {ABP_FILTERS} """ rows = self._abp_query(sql, tuple(params)) if not rows: return [] uprns = [r['uprn'] for r in rows] if len(uprns) == 1: method = 'exact_unit' confidence = 'medium' if has_range_pao else 'high' return self._build_results(uprns, p, method, confidence) # Multiple matches — let phase 5 rank by similarity rather than returning low confidence return [] def _phase2_exact(self, p: dict) -> List[dict]: suffix = p.get('pao_start_suffix') or None pao_str = str(p['pao_number']) + (suffix or '') if suffix: pao_cond = "(pao = %s OR (pao_start_number = %s AND pao_end_number IS NULL AND pao_start_suffix = %s))" pao_params = (pao_str, p['pao_number'], suffix) else: pao_cond = "(pao = %s OR (pao_start_number = %s AND pao_end_number IS NULL AND (pao_start_suffix IS NULL OR pao_start_suffix = '')))" pao_params = (pao_str, p['pao_number']) sql = f""" SELECT uprn FROM basedata.abp_main WHERE postcode = %s AND {pao_cond} AND street_description = %s {ABP_FILTERS} """ rows = self._abp_query(sql, (p['postcode'],) + pao_params + (p['street'],)) if not rows: return [] uprns = [r['uprn'] for r in rows] method = 'exact' if len(uprns) == 1 else 'exact_multi' return self._build_results(uprns, p, method, 'high') def _phase2b_range_sao(self, p: dict) -> List[dict]: """Match input number as an SAO within a range PAO. e.g. '8 PLASSY ROAD' → ABP pao='6-28', pao_start_number=6, pao_end_number=28, sao='8' """ if not p.get('pao_number') or not p.get('street'): return [] n = p['pao_number'] n_str = str(n) + (p.get('pao_start_suffix') or '') sql = f""" SELECT uprn FROM basedata.abp_main WHERE postcode = %s AND street_description = %s AND pao_start_number <= %s AND pao_end_number >= %s AND (sao = %s OR sao ILIKE '%%' || ' ' || %s OR sao_start_number = %s) {ABP_FILTERS} """ rows = self._abp_query(sql, ( p['postcode'], p['street'], n, n, n_str, n_str, n )) if not rows: return [] uprns = [r['uprn'] for r in rows] method = 'exact' if len(uprns) == 1 else 'exact_multi' return self._build_results(uprns, p, method, 'high') def _phase3_range(self, p: dict) -> List[dict]: if p.get('range_start') is None or not p.get('street'): return [] sql = f""" SELECT uprn, multi_occ_count FROM basedata.abp_main WHERE postcode = %s AND ( pao_start_number BETWEEN %s AND %s OR (pao_start_number <= %s AND pao_end_number >= %s) ) AND street_description = %s {ABP_FILTERS} ORDER BY pao_start_number """ rows = self._abp_query(sql, ( p['postcode'], p['range_start'], p['range_end'], p['range_start'], p['range_end'], p['street'] )) if not rows: return [] # If we matched a single block record (pao_end_number set), expand to child UPRNs if len(rows) == 1 and (rows[0].get('multi_occ_count') or 0) > 0: child_sql = f""" SELECT uprn FROM basedata.abp_main WHERE parent_uprn = %s {ABP_FILTERS} """ children = self._abp_query(child_sql, (rows[0]['uprn'],)) if children: rows = children uprns = [r['uprn'] for r in rows] return self._build_results(uprns, p, 'range_expansion', 'high') def _phase4_named_building(self, p: dict) -> List[dict]: if not p.get('street'): return [] sql = f""" SELECT uprn, multi_occ_count FROM basedata.abp_main WHERE postcode = %s AND (pao ILIKE %s || '%%' OR sao ILIKE %s || '%%') AND street_description = %s {ABP_FILTERS} """ rows = self._abp_query(sql, (p['postcode'], p['pao_text'], p['pao_text'], p['street'])) if not rows: return [] # Phase 4b — expand parent block record to individual child UPRNs if len(rows) == 1 and (rows[0].get('multi_occ_count') or 0) > 0: child_sql = f""" SELECT uprn FROM basedata.abp_main WHERE parent_uprn = %s {ABP_FILTERS} """ children = self._abp_query(child_sql, (rows[0]['uprn'],)) if children: rows = children uprns = [r['uprn'] for r in rows] return self._build_results(uprns, p, 'named_building', 'high') def _phase5_fuzzy(self, p: dict) -> List[dict]: conds, params = [], [p['postcode']] if p.get('pao_number') is not None: conds.append("pao_start_number = %s") params.append(p['pao_number']) if p.get('pao_text'): conds.append("(pao ILIKE '%%' || %s || '%%' OR sao ILIKE '%%' || %s || '%%')") params.extend([p['pao_text'], p['pao_text']]) if not conds: return [] sql = f""" SELECT uprn, sao, pao, street_description FROM basedata.abp_main WHERE postcode = %s AND ({' OR '.join(conds)}) {ABP_FILTERS} """ rows = self._abp_query(sql, tuple(params)) if not rows: return [] if len(rows) == 1: return self._build_results([rows[0]['uprn']], p, 'fuzzy', 'medium') # Build a normalised input string to score candidates against input_parts = [ p.get('sao_text') or '', p.get('pao_text') or '', str(p['pao_number']) if p.get('pao_number') is not None else '', p.get('street') or '', ] input_str = ' '.join(x for x in input_parts if x).upper() # Score each candidate by similarity of its reconstructed address to input scored = [] for row in rows: abp_parts = [row['sao'] or '', row['pao'] or '', row['street_description'] or ''] abp_str = ' '.join(x for x in abp_parts if x).upper() score = SequenceMatcher(None, input_str, abp_str).ratio() scored.append((score, row['uprn'])) scored.sort(key=lambda x: x[0], reverse=True) best_score, best_uprn = scored[0] second_score = scored[1][0] if len(scored) > 1 else 0.0 # Only return a match if the best score clears the threshold and # is meaningfully ahead of the next candidate if best_score >= 0.65 and (best_score - second_score) >= 0.10: return self._build_results([best_uprn], p, 'fuzzy', 'medium') return [] @staticmethod def _build_results(uprns: List[int], parsed: dict, method: str, confidence: str) -> List[dict]: n = len(uprns) uprn_list = ', '.join(str(u) for u in uprns) if method == 'fuzzy_ambiguous': action = 'low_confidence_match' notes = f"Ambiguous — {n} candidates found, review required ({uprn_list})" else: action = 'pending' # resolved to added_ok / duplicate after check_duplicates notes = f"{n} UPRN{'s' if n != 1 else ''} matched" + (f" ({uprn_list})" if n > 1 else "") return [ { 'uprn': u, 'address_raw': parsed['raw'], 'postcode': parsed.get('postcode'), 'address_type': parsed.get('address_type'), 'match_method': method, 'match_confidence': confidence, 'import_action': action, 'notes': notes, } for u in uprns ] # ------------------------------------------------------------------ # # Address resolution pipeline # ------------------------------------------------------------------ # def resolve_addresses(self, df: pd.DataFrame, address_col: str) -> List[dict]: self.update_job_progress(25, "parsing_addresses", "Parsing address strings...") rows = [ (idx, str(val)) for idx, val in df[address_col].items() if not pd.isna(val) and str(val).strip() ] total = len(rows) staging: List[dict] = [] for i, (idx, raw) in enumerate(rows): source_row = idx + 2 parsed = self.parse_address_string(raw) # Normalise text fields to uppercase once — ABP data is all uppercase # and doing this in Python lets PostgreSQL use indexes on the columns. # Also strip periods: abbreviations like "ST." appear in input but ABP # stores them without the period (e.g. "ST DAVIDS CLOSE"). for _f in ('street', 'sao_text', 'pao_text', 'town'): if parsed[_f]: parsed[_f] = parsed[_f].upper().replace('.', '') if parsed['address_type'] == 'parse_error': staging.append({ 'file_upload_id': self.file_upload_id, 'stocklist_id': self.stocklist_id, 'batch_id': self.batch_id, 'source_row': source_row, 'address_raw': raw, 'postcode': None, 'address_type': 'parse_error', 'match_method': 'parse_error', 'match_confidence': None, 'uprn': None, 'import_action': 'parse_error', 'notes': f"Could not parse address — {parsed['error']}" }) continue if i % 10 == 0: progress = 30 + int((i / total) * 40) self.update_job_progress( progress, "matching_addresses", f"Matching addresses ({i}/{total})", processed_rows=i, total_rows=total ) try: matches = self.match_address_in_abp(parsed) except Exception as e: logger.error(f"ABP match failed for row {source_row} '{raw}': {e}") matches = [{ 'uprn': None, 'match_method': 'unmatched', 'match_confidence': None, 'import_action': 'unable_to_match_address', 'notes': f"Match error: {e}" }] # Reconnect if the DB connection was lost try: self.connection.rollback() except Exception: self.connect_to_database() for m in matches: staging.append({ 'file_upload_id': self.file_upload_id, 'stocklist_id': self.stocklist_id, 'batch_id': self.batch_id, 'source_row': source_row, 'address_raw': raw, 'postcode': m.get('postcode') or parsed.get('postcode'), 'address_type': parsed.get('address_type'), 'match_method': m['match_method'], 'match_confidence': m.get('match_confidence'), 'uprn': m.get('uprn'), 'import_action': m['import_action'], 'notes': m.get('notes', '') }) by_method: Dict[str, int] = {} for r in staging: by_method[r['match_method']] = by_method.get(r['match_method'], 0) + 1 n_matched = sum(1 for r in staging if r['uprn'] is not None) n_unmatched = sum(1 for r in staging if r['uprn'] is None) self.update_job_progress( 70, "addresses_resolved", f"Matching complete — {n_matched} UPRNs resolved, {n_unmatched} unmatched", total_rows=total, processed_rows=total, matched_rows=n_matched, unmatched_rows=n_unmatched, ) return staging # ------------------------------------------------------------------ # # Duplicate checking # ------------------------------------------------------------------ # def check_duplicates(self, staging: List[dict]) -> List[dict]: """Update import_action for resolved UPRNs based on existing stocklist membership.""" pending_uprns = [r['uprn'] for r in staging if r['uprn'] is not None and r['import_action'] == 'pending'] if not pending_uprns: return staging self.cursor.execute(""" SELECT uprn FROM stocklists.stocklist_premises WHERE stocklist_id = %s AND uprn = ANY(%s) """, (self.stocklist_id, pending_uprns)) in_this = {r['uprn'] for r in self.cursor.fetchall()} self.cursor.execute(""" SELECT uprn, STRING_AGG(stocklist_id::text, ', ' ORDER BY stocklist_id::text) AS other_ids FROM stocklists.stocklist_premises WHERE stocklist_id != %s AND uprn = ANY(%s) GROUP BY uprn """, (self.stocklist_id, pending_uprns)) in_other = {r['uprn']: r['other_ids'] for r in self.cursor.fetchall()} for r in staging: if r['uprn'] is None or r['import_action'] != 'pending': continue u = r['uprn'] if u in in_this: r['import_action'] = 'address_already_in_stocklist' r['notes'] = f"UPRN {u} already exists in stocklist {self.stocklist_id}" elif u in in_other: r['import_action'] = 'address_in_another_stocklist' r['notes'] = f"UPRN {u} already assigned to stocklist(s): {in_other[u]}" else: r['import_action'] = 'added_ok' r['notes'] = f"Added to stocklist {self.stocklist_id} — {r['notes']}" return staging # ------------------------------------------------------------------ # # Database insertion # ------------------------------------------------------------------ # def insert_address_staging(self, staging: List[dict]) -> int: if not staging: return 0 cols = ['file_upload_id', 'stocklist_id', 'batch_id', 'source_row', 'address_raw', 'postcode', 'address_type', 'match_method', 'match_confidence', 'uprn', 'import_action', 'notes'] sql = ( "INSERT INTO public.geolynx_address_staging " f"({', '.join(cols)}) " f"VALUES ({', '.join(['%s'] * len(cols))})" ) execute_batch(self.cursor, sql, [tuple(r.get(c) for c in cols) for r in staging], page_size=500) self.connection.commit() logger.info(f"Inserted {len(staging)} rows into geolynx_address_staging") return len(staging) def insert_resolved_uprns(self, staging: List[dict]) -> int: """Copy added_ok UPRNs into geolynx_identifiers for assign_premises().""" resolved = [r for r in staging if r['uprn'] is not None and r['import_action'] == 'added_ok'] if not resolved: return 0 sql = """ INSERT INTO public.geolynx_identifiers (uprn, toid, stocklist_id, batch_id, source_row, source_file, upload_timestamp, file_upload_id) VALUES (%s, NULL, %s, %s, %s, %s, NOW(), %s) ON CONFLICT DO NOTHING """ execute_batch(self.cursor, sql, [(r['uprn'], self.stocklist_id, self.batch_id, r['source_row'], 'address_import', self.file_upload_id) for r in resolved], page_size=500) self.connection.commit() logger.info(f"Inserted {len(resolved)} resolved UPRNs into geolynx_identifiers") return len(resolved) def prepare_data_for_insert(self, df: pd.DataFrame, details: Dict[str, Any]) -> List[Dict[str, Any]]: """Build insert records for UPRN/TOID direct imports.""" data = [] pt = details['primary_identifier'] source_file = os.path.basename(df.attrs.get('file_path', 'unknown')) if pt == 'UPRN': col = details['uprn_column'] for idx, row in df.iterrows(): val = row[col] if not pd.isna(val) and self.validate_uprn(val): s = str(val).strip() if '.' in s: s = s.split('.')[0] data.append({ 'uprn': int(s), 'toid': None, 'stocklist_id': self.stocklist_id, 'batch_id': self.batch_id, 'source_row': idx + 2, 'source_file': source_file, 'upload_timestamp': datetime.now(), 'file_upload_id': self.file_upload_id }) elif pt == 'TOID': col = details['toid_column'] for idx, row in df.iterrows(): val = row[col] if not pd.isna(val) and self.validate_toid(val): data.append({ 'uprn': None, 'toid': str(val).strip().lower(), 'stocklist_id': self.stocklist_id, 'batch_id': self.batch_id, 'source_row': idx + 2, 'source_file': source_file, 'upload_timestamp': datetime.now(), 'file_upload_id': self.file_upload_id }) return data def insert_to_database(self, data: List[Dict[str, Any]], batch_size: int = 1000) -> int: if not data: return 0 sql = """ INSERT INTO public.geolynx_identifiers (uprn, toid, stocklist_id, batch_id, source_row, source_file, upload_timestamp, file_upload_id) VALUES (%s, %s, %s, %s, %s, %s, %s, %s) """ values = [ (r['uprn'], r['toid'], r['stocklist_id'], r['batch_id'], r['source_row'], r['source_file'], r['upload_timestamp'], r['file_upload_id']) for r in data ] total, n_batches = 0, (len(values) + batch_size - 1) // batch_size for i in range(0, len(values), batch_size): batch = values[i:i + batch_size] current = (i // batch_size) + 1 self.update_job_progress( 30 + int((current / n_batches) * 50), "inserting", f"Inserting batch {current}/{n_batches}", rows_processed=total + len(batch), total_rows=len(data) ) execute_batch(self.cursor, sql, batch, page_size=batch_size) self.connection.commit() total += len(batch) self.update_job_progress(80, "inserted", f"Inserted {total} rows", rows_inserted=total) return total # ------------------------------------------------------------------ # # Premises assignment # ------------------------------------------------------------------ # def assign_premises(self) -> Dict[str, Any]: self.update_job_progress(85, "assigning", "Assigning premises to stocklist...") try: self.cursor.execute(""" SELECT COUNT(*) AS cnt FROM public.geolynx_identifiers WHERE batch_id = %s AND uprn IS NOT NULL """, (self.batch_id,)) total_in_batch = self.cursor.fetchone()['cnt'] # COUNT(DISTINCT) rather than COUNT(*): these are premise counts, a file # can name the same UPRN twice, and the insert below is SELECT DISTINCT. self.cursor.execute(""" SELECT COUNT(DISTINCT gi.uprn) AS cnt FROM public.geolynx_identifiers gi WHERE gi.batch_id = %s AND gi.uprn IS NOT NULL AND EXISTS ( SELECT 1 FROM stocklists.stocklist_premises sp WHERE sp.uprn = gi.uprn AND sp.stocklist_id = %s ) """, (self.batch_id, self.stocklist_id)) already_exists = self.cursor.fetchone()['cnt'] # Held by a different stocklist, so not added — the count that explains # a premise going missing from an import that otherwise looks clean. self.cursor.execute(""" SELECT COUNT(DISTINCT gi.uprn) AS cnt FROM public.geolynx_identifiers gi WHERE gi.batch_id = %s AND gi.uprn IS NOT NULL AND NOT EXISTS ( SELECT 1 FROM stocklists.stocklist_premises sp WHERE sp.uprn = gi.uprn AND sp.stocklist_id = %s ) AND EXISTS ( SELECT 1 FROM stocklists.stocklist_premises sp WHERE sp.uprn = gi.uprn AND sp.stocklist_id <> %s ) """, (self.batch_id, self.stocklist_id, self.stocklist_id)) in_other_stocklist = self.cursor.fetchone()['cnt'] # A premise belongs to ONE stocklist (Dave, 2026-09-18). The NOT EXISTS is # therefore unscoped — a UPRN held by ANY stocklist is not added here. # # It used to carry `AND sp.stocklist_id = %s`, which only skipped premises # already in this stocklist, so the UPRN and TOID paths added premises that # belonged to another one while the address path — which filters them out # upstream in check_duplicates() — did not. The same properties uploaded as # UPRNs and as addresses gave different answers, and the identifier fallback # meant one file could switch between the two policies depending on whether # its UPRN column happened to be numeric. # # Enforcing it here rather than alongside check_duplicates() is deliberate: # this is the one statement both paths pass through, so they cannot drift # apart again. On the address path it is a harmless second guard. self.cursor.execute(""" INSERT INTO stocklists.stocklist_premises (uprn, stocklist_id, added_datetime, upload_batch_id) SELECT DISTINCT gi.uprn, %s, NOW(), %s FROM public.geolynx_identifiers gi WHERE gi.batch_id = %s AND gi.uprn IS NOT NULL AND EXISTS (SELECT 1 FROM basedata.abp WHERE abp.uprn = gi.uprn) AND NOT EXISTS ( SELECT 1 FROM stocklists.stocklist_premises sp WHERE sp.uprn = gi.uprn ) """, (self.stocklist_id, self.batch_id, self.batch_id)) rows_inserted = self.cursor.rowcount self.connection.commit() # A batch that resolved identifiers but matched no AddressBase premise # is the signature of numbering from some other system. Reported # plainly because it otherwise completes with a count of zero, which # reads as a successful import of nothing. msg = f"Assigned {rows_inserted} premises" if already_exists: msg += f", {already_exists} already in this stocklist" if in_other_stocklist: msg += f", {in_other_stocklist} held by another stocklist" # Only claim nothing matched AddressBase when nothing was held back for # another reason. A file whose premises all belong to another stocklist # also inserts zero rows, and blaming AddressBase for that would send the # next person looking at the address data instead of the stocklists. if (total_in_batch and not rows_inserted and not already_exists and not in_other_stocklist): msg = (f"No premises assigned: none of the {total_in_batch} identifier(s) " f"resolved from this file matched a premise in Ordnance Survey " f"AddressBase") self.update_job_progress(90, "assigned", msg, rows_assigned=rows_inserted, already_existed=already_exists) return { 'success': True, 'total_in_batch': total_in_batch, 'already_existed': already_exists, 'rows_inserted': rows_inserted, 'in_other_stocklist': in_other_stocklist } except Exception as e: self.connection.rollback() self.update_job_progress(0, "Error", error=f"assign_premises failed: {e}") raise # ------------------------------------------------------------------ # # Feedback — write columns back to the uploaded file # ------------------------------------------------------------------ # def _get_address_feedback(self) -> Dict[int, dict]: self.cursor.execute(""" SELECT s.source_row, s.import_action, s.match_method, s.match_confidence, s.uprn, s.notes, COUNT(s.uprn) FILTER (WHERE s.uprn IS NOT NULL) OVER (PARTITION BY s.source_row) AS uprns_matched, TRIM(CONCAT_WS(', ', NULLIF(a.sao, ''), NULLIF(a.pao, ''), NULLIF(a.street_description, ''), NULLIF(a.postcode, '') )) AS matched_address FROM public.geolynx_address_staging s LEFT JOIN basedata.abp_main a ON a.uprn = s.uprn WHERE s.file_upload_id = %s ORDER BY s.source_row, s.uprn NULLS LAST """, (self.file_upload_id,)) # One staging row per matched UPRN, so a source row arrives as several rows # and every one of them has to be counted. match_method and match_confidence # are taken from the first: all UPRNs for a row come out of the same matching # phase together, so they are uniform across it. feedback: Dict[int, dict] = {} for r in self.cursor.fetchall(): sr = r['source_row'] if sr not in feedback: feedback[sr] = { 'import_action': None, 'uprns_matched': r['uprns_matched'], 'uprn_matched': None, 'match_method': r['match_method'], 'match_confidence': r['match_confidence'], 'matched_address': r['matched_address'] or '', 'notes': None, '_counts': {}, '_notes': [], '_uprns': [], } fb = feedback[sr] fb['_counts'][r['import_action']] = fb['_counts'].get(r['import_action'], 0) + 1 if r['notes'] and r['notes'] not in fb['_notes']: fb['_notes'].append(r['notes']) if r['uprn'] is not None: fb['_uprns'].append(r['uprn']) for fb in feedback.values(): uprns = fb.pop('_uprns') counts = fb.pop('_counts') notes = fb.pop('_notes') fb['uprn_matched'] = uprns[0] if len(uprns) == 1 else None fb['import_action'] = summarise_import_actions(counts, len(uprns)) # Capped: eight flats each naming their holding stocklist runs long in a # spreadsheet cell, and the count in import_action already carries the shape. fb['notes'] = '; '.join(notes[:5]) + ( f" (+{len(notes) - 5} more)" if len(notes) > 5 else '') return feedback def _get_uprn_toid_feedback(self) -> Dict[int, dict]: self.cursor.execute(""" WITH other_sl AS ( SELECT uprn, STRING_AGG(stocklist_id::text, ', ' ORDER BY stocklist_id::text) AS other_ids FROM stocklists.stocklist_premises WHERE stocklist_id != %s GROUP BY uprn ) SELECT gi.source_row, gi.uprn, gi.toid, -- Ordered most-specific first. A premise held by another stocklist is -- no longer added (see assign_premises), so it reports the reason -- rather than 'added_ok' with an aside — the wording matches -- check_duplicates() on the address path so a reader of the feedback -- file sees the same words whichever way the file was matched. CASE WHEN gi.uprn IS NULL THEN 'skipped_no_uprn' WHEN sp.uprn IS NOT NULL THEN 'address_already_in_stocklist' WHEN os.uprn IS NOT NULL THEN 'address_in_another_stocklist' WHEN abp.uprn IS NOT NULL THEN 'added_ok' ELSE 'validation_error' END AS import_action, -- Matched, not added. The address path's uprns_matched is a window -- count of resolved UPRNs regardless of what became of them, and -- import_action now carries added-versus-excluded on both paths, so -- this counts the same thing the column name says. CASE WHEN gi.uprn IS NOT NULL AND abp.uprn IS NOT NULL THEN 1 ELSE 0 END AS uprns_matched, CASE WHEN abp.uprn IS NULL AND gi.uprn IS NOT NULL THEN 'UPRN not found in address base' WHEN sp.uprn IS NOT NULL THEN 'UPRN ' || gi.uprn || ' already in stocklist ' || %s WHEN os.uprn IS NOT NULL THEN 'UPRN ' || gi.uprn || ' already assigned to stocklist(s): ' || os.other_ids ELSE NULL END AS notes FROM public.geolynx_identifiers gi LEFT JOIN basedata.abp abp ON gi.uprn = abp.uprn LEFT JOIN stocklists.stocklist_premises sp ON gi.uprn = sp.uprn AND sp.stocklist_id = %s LEFT JOIN other_sl os ON gi.uprn = os.uprn WHERE gi.batch_id = %s ORDER BY gi.source_row """, (self.stocklist_id, self.stocklist_id, self.stocklist_id, self.batch_id)) # One row per source row here — a UPRN is one-to-one — but rendered through the # same summariser as the address path, so the column reads the same whichever # way the file was matched. feedback: Dict[int, dict] = {} for r in self.cursor.fetchall(): resolved = 1 if r['uprn'] is not None else 0 feedback[r['source_row']] = { 'import_action': summarise_import_actions({r['import_action']: 1}, resolved), 'uprns_matched': r['uprns_matched'], 'uprn_matched': r['uprn'] if r['uprns_matched'] == 1 else None, 'match_method': 'uprn_direct' if r['uprn'] else 'toid_direct', 'match_confidence': 'high', 'matched_address': '', 'notes': r['notes'] or '' } return feedback def write_feedback_columns_to_upload(self, file_path: str) -> None: self.update_job_progress(92, "generating_feedback", "Writing feedback columns to upload file...") feedback = (self._get_address_feedback() if self.primary_identifier == 'ADDRESS' else self._get_uprn_toid_feedback()) ext = Path(file_path).suffix.lower() if ext == '.csv': self._write_feedback_csv(file_path, feedback) else: self._write_feedback_xlsx(file_path, feedback) self.update_job_progress(95, "feedback_written", "Feedback columns written to upload file") def _write_feedback_xlsx(self, file_path: str, feedback: Dict[int, dict]) -> None: wb = openpyxl.load_workbook(file_path) ws = wb.active last_col = ws.max_column headers = ['import_action', 'uprns_matched', 'uprn_matched', 'match_method', 'match_confidence', 'matched_address', 'notes'] for i, h in enumerate(headers, start=last_col + 1): ws.cell(row=1, column=i, value=h) for row_idx in range(2, ws.max_row + 1): fb = feedback.get(row_idx) if not fb: continue base = last_col + 1 ws.cell(row=row_idx, column=base, value=fb['import_action']) ws.cell(row=row_idx, column=base + 1, value=fb['uprns_matched']) ws.cell(row=row_idx, column=base + 2, value=fb['uprn_matched']) ws.cell(row=row_idx, column=base + 3, value=fb['match_method']) ws.cell(row=row_idx, column=base + 4, value=fb['match_confidence']) ws.cell(row=row_idx, column=base + 5, value=fb['matched_address']) ws.cell(row=row_idx, column=base + 6, value=fb['notes']) for i, h in enumerate(headers, start=last_col + 1): ws.column_dimensions[get_column_letter(i)].width = max(len(h) + 2, 15) wb.save(file_path) logger.info(f"Feedback written to {file_path}") def _write_feedback_csv(self, file_path: str, feedback: Dict[int, dict]) -> None: df = pd.read_csv(file_path, encoding='utf-8-sig') for col in ['import_action', 'uprns_matched', 'uprn_matched', 'match_method', 'match_confidence', 'matched_address', 'notes']: df[col] = pd.NA for idx in df.index: fb = feedback.get(idx + 2) if fb: df.at[idx, 'import_action'] = fb['import_action'] df.at[idx, 'uprns_matched'] = fb['uprns_matched'] df.at[idx, 'uprn_matched'] = fb['uprn_matched'] df.at[idx, 'match_method'] = fb['match_method'] df.at[idx, 'match_confidence'] = fb['match_confidence'] df.at[idx, 'matched_address'] = fb['matched_address'] df.at[idx, 'notes'] = fb['notes'] df.to_csv(file_path, index=False, encoding='utf-8-sig') logger.info(f"Feedback written to {file_path}") # ------------------------------------------------------------------ # # Main orchestration # ------------------------------------------------------------------ # def process_file(self, file_path: str) -> Dict[str, Any]: self.update_job_progress(20, "starting", "Initialising processing...") self.batch_id = self.generate_batch_id() results: Dict[str, Any] = { 'success': False, 'file_path': file_path, 'stocklist_id': self.stocklist_id, 'batch_id': self.batch_id, 'rows_inserted': 0, 'error': None } try: if not self.connection: self.connect_to_database() self.ensure_schema_migrations() safe_name = os.path.basename(file_path) self.resolve_file_upload(safe_name) self.update_job_progress(22, "reading", "Reading file...") df = self.read_file(file_path) is_valid, msg, details = self.validate_file(df) if not is_valid: self.update_job_progress(0, "Error", error=f"Validation failed: {msg}") results['error'] = msg return results results['primary_identifier_used'] = details['primary_identifier'] if details['primary_identifier'] == 'ADDRESS': staging = self.resolve_addresses(df, details['address_column']) staging = self.check_duplicates(staging) self.insert_address_staging(staging) self.insert_resolved_uprns(staging) assign = self.assign_premises() results['rows_inserted'] = assign['rows_inserted'] else: self.update_job_progress(35, "preparing", "Preparing data for insertion...") insert_data = self.prepare_data_for_insert(df, details) self.insert_to_database(insert_data) assign = self.assign_premises() results['rows_inserted'] = assign['rows_inserted'] self.write_feedback_columns_to_upload(file_path) results['success'] = True except Exception as e: results['error'] = str(e) logger.error(f"Processing error: {e}", exc_info=True) self.update_job_progress(0, "Error", error=str(e)) return results # --------------------------------------------------------------------------- # Entry point # --------------------------------------------------------------------------- def upload_path(stocklist_id: int, file_name: str) -> str: """Where file_upload.php puts a stocklist attachment.""" return os.path.join(FILE_STORE_ROOT, 'stocklist', str(stocklist_id), file_name) def build_arg_parser() -> argparse.ArgumentParser: parser = argparse.ArgumentParser(description="GeoLynx Stocklist Importer v2") parser.add_argument("-stocklist", type=int, help="Stocklist ID to add premises to") parser.add_argument("-file", type=str, help="Safe filename of the uploaded file (e.g. abc123.xlsx)") parser.add_argument("-user", type=int, default=0, help="User ID performing the import") parser.add_argument("-job", type=str, default=None, help="Job UUID from stocklist_import_jobs, for progress tracking") parser.add_argument("--check", action="store_true", help="Report on everything the import needs — interpreter, " "dependencies, environment, database, tables, file store " "— and exit. Add -stocklist and -file to check one upload.") return parser def run_check(args) -> int: """ Diagnostic report, one PASS/FAIL line per thing an import depends on. This exists because the app runs on a server with no interactive session to hand: `--check` answers 'why did the import not start' in one command. """ ok = True def line(label: str, good: bool, detail: str = "") -> None: nonlocal ok if not good: ok = False print(f"[{'PASS' if good else 'FAIL'}] {label}" + (f" — {detail}" if detail else "")) print(f"script : {os.path.abspath(__file__)}") print(f"interpreter : {sys.executable}") print(f"python : {sys.version.split()[0]}") print(f"user : uid={os.getuid()} gid={os.getgid()}") print("") service_file_ok = os.path.isfile(DEFAULT_SERVICE_FILE) line(f"service file {DEFAULT_SERVICE_FILE}", service_file_ok, "readable" if service_file_ok and os.access(DEFAULT_SERVICE_FILE, os.R_OK) else ("present but not readable by this user" if service_file_ok else "not found — explicit PG* settings will be needed")) try: line("database settings", True, db_settings_source()) except Exception as e: line("database settings", False, f"{type(e).__name__}: {e}") try: load_runtime_dependencies() line("dependencies", True, f"pandas {pd.__version__}, openpyxl {openpyxl.__version__}, " f"psycopg2 {psycopg2.__version__.split()[0]}, " f"{SequenceMatcher.__module__}.SequenceMatcher") except Exception as e: line("dependencies", False, f"{type(e).__name__}: {e}") conn = None try: conn = psycopg2.connect(**db_settings()) info = conn.get_dsn_parameters() with conn.cursor() as cur: cur.execute("SELECT current_database(), current_user, version()") dbname, whoami, ver = cur.fetchone() line("database connection", True, f"{dbname} as {whoami} on {info.get('host')}:{info.get('port')} " f"— {ver.split(' on ')[0]}") except Exception as e: line("database connection", False, f"{type(e).__name__}: {e}") if conn is not None: for table in ('stocklists.stocklist_import_jobs', 'stocklists.stocklist_premises', 'public.file_uploads', 'public.geolynx_identifiers', 'public.geolynx_address_staging', 'basedata.abp', 'basedata.abp_main'): try: with conn.cursor() as cur: cur.execute("SELECT to_regclass(%s)", (table,)) found = cur.fetchone()[0] is not None line(f"table {table}", found, "" if found else "does not exist") except Exception as e: line(f"table {table}", False, f"{type(e).__name__}: {e}") # Existence is not enough for this one. An empty basedata.abp_main matches # nothing and reports every address as unmatched, which reads as a matching # bug rather than a missing load — the failure that prompted migration 065. try: with conn.cursor() as cur: cur.execute("SELECT reltuples::bigint FROM pg_class " "WHERE oid = 'basedata.abp_main'::regclass") row = cur.fetchone() approx = row[0] if row else 0 line("basedata.abp_main is populated", approx > 0, f"roughly {approx} rows" if approx > 0 else "empty — load it from basedata.abp_import, or every address " "will come back unmatched") except Exception as e: line("basedata.abp_main is populated", False, f"{type(e).__name__}: {e}") conn.close() store_ok = os.path.isdir(FILE_STORE_ROOT) line(f"file store {FILE_STORE_ROOT}", store_ok, "" if store_ok else "directory not found") if args.stocklist and args.file: path = upload_path(args.stocklist, args.file) found = os.path.isfile(path) line(f"upload {path}", found, f"{os.path.getsize(path)} bytes" if found else "not found") print("") print("CHECK " + ("PASSED" if ok else "FAILED")) return 0 if ok else 1 def main(argv: Optional[List[str]] = None) -> int: args = build_arg_parser().parse_args(argv) if args.check: return run_check(args) missing = [name for name, value in (('-stocklist', args.stocklist), ('-file', args.file)) if not value] if missing: build_arg_parser().error("missing required argument(s): " + ", ".join(missing)) processor = GeoLynxExcelProcessor( stocklist_id=args.stocklist, user_id=args.user, job_id=args.job ) # Announce ourselves before anything that can fail. A job still sitting on # 'Started' after this point never reached here at all — a wrong interpreter # path, a missing script, or psycopg2 failing to import — and none of those # can reach the jobs table. That distinction is the whole diagnostic. try: processor.update_job_progress(5, "Running", "Import process started") load_runtime_dependencies() file_path = upload_path(args.stocklist, args.file) if not os.path.isfile(file_path): raise FileNotFoundError(f"Uploaded file is not on disk: {file_path}") processor.connect_to_database() results = processor.process_file(file_path) logger.info("Success: %s", results['success']) logger.info("Rows inserted: %s", results['rows_inserted']) if results.get('success'): processor.update_job_progress( 100, "Complete", "Processing complete", batch_id=results['batch_id'], rows_inserted=results['rows_inserted'] ) return 0 processor.update_job_progress( 0, "Error", error=results.get('error', 'Unknown error') ) return 1 finally: processor.close_connection() def _job_id_from_argv() -> Optional[str]: """Read -job straight off argv, so the bootstrap has it even if parsing failed.""" for i, arg in enumerate(sys.argv): if arg == '-job' and i + 1 < len(sys.argv): return sys.argv[i + 1] return None if __name__ == "__main__": try: sys.exit(main()) except SystemExit: raise except BaseException: # Anything at all that escapes main() lands here. The traceback goes to # the log the manager captures, and — where a connection is possible — # onto the job row so the user sees it without anyone reading a log. detail = traceback.format_exc() logger.error("Import failed:\n%s", detail) lines = [l for l in detail.strip().splitlines() if l.strip()] summary = lines[-1] if lines else 'Unknown error' job_id = _job_id_from_argv() if not report_job_failure(job_id, f"{summary}\n\n{detail}") and job_id: logger.error("Job %s was not updated; the manager will reconcile it " "from this log.", job_id) sys.exit(1)