#!/usr/bin/env python3 """Build NJ FedEx shipping points from ATP's FedEx locator spider.""" import json import copy import math import os import sys import tempfile import urllib.request from urllib.error import URLError from collections import Counter from datetime import datetime, timezone from pathlib import Path OUT = (Path(__file__).resolve().parent / 'shipping_fedex.geojson') URL = 'https://data.alltheplaces.xyz/runs/latest/output/fedex.geojson' STAFFED = { 'FedEx OnSite', 'FedEx Authorized ShipCenter', 'FedEx ShipSite', 'FedEx Office Print & Ship Center', 'FedEx Office Ship Center', 'FedEx Ship Center', } def build(features): out, seen = [], set() for f in features: p, g = f.get('properties') or {}, f.get('geometry') or {} if str(p.get('addr:state') or '').strip().upper() not in ('NJ', 'NEW JERSEY'): continue service = p.get('name') or '' if service not in STAFFED | {'FedEx Drop Box', 'FedEx Express Drop Box'}: continue # Freight facilities and Kohl's returns aren't general drop-offs. c = g.get('coordinates') or [] if g.get('type') != 'Point' or len(c) != 2 or not all(isinstance(v, (int, float)) for v in c): continue hours = p.get('opening_hours') or '' if hours.strip().lower() in ('closed', 'off', 'no'): continue ref = str(p.get('ref') or f.get('id') or '') key = ref or (tuple(c), p.get('addr:street_address'), service) if key in seen: continue seen.add(key) host = p.get('located_in') or '' street = p.get('addr:street_address') or ' '.join(filter(None, [p.get('addr:housenumber'), p.get('addr:street')])) out.append({'type': 'Feature', 'geometry': g, 'properties': { 'carrier': 'FedEx', 'name': f'{service} at {host}' if host else service, 'service_type': service, 'host': host, 'location_type': 'staffed' if service in STAFFED else 'drop_box', 'address': ' '.join(filter(None, [street, p.get('addr:city'), p.get('addr:state'), p.get('addr:postcode')])), 'address_parts': {'street':street,'city':p.get('addr:city') or '', 'state':'NJ','postcode':p.get('addr:postcode') or ''}, 'hours': hours, 'phone': p.get('phone') or '', 'ref': ref, 'website': p.get('website') or '', 'osm': '', 'source': 'FedEx locator via All The Places', 'source_url': p.get('@source_uri') or URL, }}) return out class IncompleteData(ValueError): pass def check_counts(features): counts = Counter(f['properties']['location_type'] for f in features) if counts['staffed'] < 100 or counts['drop_box'] < 100: raise IncompleteData(f'Too few FedEx locations: {dict(counts)}; refusing incomplete data') return counts def source_features(data): if not isinstance(data, dict) or data.get('type') != 'FeatureCollection' or not isinstance(data.get('features'), list): raise IncompleteData('ATP FedEx response is not a FeatureCollection') features = build(data['features']) check_counts(features) return features def retain_previous(previous, reason, attempted_at): """Keep the entire validated prior snapshot, never blend a partial refresh.""" if not isinstance(previous, dict) or previous.get('type') != 'FeatureCollection': raise IncompleteData('No valid previous FedEx FeatureCollection') features = previous.get('features') if not isinstance(features, list): raise IncompleteData('No previous FedEx features') refs = set() for f in features: p, g = f.get('properties') or {}, f.get('geometry') or {} c = g.get('coordinates') or [] if (g.get('type') != 'Point' or len(c) != 2 or not all(isinstance(v, (int, float)) and not isinstance(v, bool) and math.isfinite(v) for v in c) or not (-75.57 <= c[0] <= -73.88 and 38.88 <= c[1] <= 41.36) or p.get('carrier') != 'FedEx' or p.get('source') != 'FedEx locator via All The Places' or p.get('location_type') not in ('staffed', 'drop_box') or not p.get('ref') or p['ref'] in refs): raise IncompleteData('Invalid or duplicate record in previous FedEx snapshot') refs.add(p['ref']) check_counts(features) result = copy.deepcopy(previous) prior = previous.get('source_refresh') or {} result['source_refresh'] = { 'status': 'retained_previous', 'attempted_at': attempted_at, 'last_successful_at': prior.get('last_successful_at'), 'source_url': URL, 'reason': reason, } for f in result['features']: f['properties']['source_refresh_status'] = ( f'ATP FedEx refresh unavailable; retained previous dataset (checked {attempted_at})') return result def download(): req = urllib.request.Request(URL, headers={'User-Agent': 'Personal NJ shipping map (github.com/mbligh/mapping)'}) with urllib.request.urlopen(req, timeout=120) as response: return json.load(response) def write_atomic(path, data): path.parent.mkdir(parents=True, exist_ok=True) temporary = None try: with tempfile.NamedTemporaryFile(mode='w', dir=path.parent, prefix='.fedex-', delete=False) as output: temporary = Path(output.name) json.dump(data, output, separators=(',', ':')) os.replace(temporary, path) finally: if temporary is not None: temporary.unlink(missing_ok=True) def main(argv=None): args = sys.argv[1:] if argv is None else argv attempted_at = datetime.now(timezone.utc).isoformat() try: data = json.loads(Path(args[0]).read_text()) if args else download() features = source_features(data) except (URLError, TimeoutError, OSError, ValueError) as error: # Explicit file imports remain strict and reproducible. Automatic # refreshes can use tracked history when the remote source is unusable. if args: raise SystemExit(str(error)) from error try: previous = json.loads(OUT.read_text()) result = retain_previous(previous, str(error), attempted_at) except (OSError, ValueError, TypeError, AttributeError) as history_error: raise SystemExit(f'{error}; cannot retain FedEx history: {history_error}') from history_error print(f'::warning::ATP FedEx refresh unavailable: {error}; retained {len(result["features"])} previous locations', flush=True) else: for f in features: f['properties']['source_refresh_status'] = f'Refreshed from ATP on {attempted_at}' result = {'type': 'FeatureCollection', 'features': features, 'source_refresh': { 'status': 'success', 'attempted_at': attempted_at, 'last_successful_at': attempted_at, 'source_url': URL, }} write_atomic(OUT, result) features = result['features'] counts = check_counts(features) print(f'FedEx NJ: {dict(counts)}', flush=True) print(f'FedEx staffed hosts: {dict(Counter(f["properties"]["host"] or f["properties"]["service_type"] for f in features if f["properties"]["location_type"] == "staffed"))}', flush=True) if __name__ == '__main__': main()