diff --git a/app/importer.py b/app/importer.py new file mode 100644 index 0000000..742f184 --- /dev/null +++ b/app/importer.py @@ -0,0 +1,305 @@ +"""Bike-Tinder importer — read the hunt's Kijiji state JSONs, gate them, and +upsert into the app DB (new / changed / gone / dup states), scoring every row +with the ported rubric. + +Run verbatim from the repo root: + python3 -m app.importer --state-dir /var/home/admin/.hermes/scripts/motorcycle-hunt.d --db data/bike-tinder.sqlite3 + +Idempotent: re-running changes no listing rows (a new import_runs row is +expected). Gates fail loud and leave the deck last-known-good. +""" +import argparse +import datetime +import json +import os +import re +import sys + +from . import db, rubric + +# Phase 1 consumes only the two Kijiji state files. +REGIONS = {'kijiji-ns': 'ns', 'kijiji-nb': 'nb'} + +GATE_MAX_AGE_HOURS = 26 + + +def now_iso(): + return datetime.datetime.now(datetime.timezone.utc).strftime('%Y-%m-%dT%H:%M:%SZ') + + +def parse_iso(s): + if not s: + return None + s = str(s).strip() + if s.endswith('Z'): + s = s[:-1] + '+00:00' + try: + return datetime.datetime.fromisoformat(s) + except ValueError: + return None + + +def norm_title(t): + """Normalize a title for dedupe: lowercase, non-alnum -> space, collapse.""" + t = (t or '').lower() + t = re.sub(r'[^a-z0-9]+', ' ', t) + t = re.sub(r'\s+', ' ', t).strip() + return t + + +def num_from_price(price): + if not price or '$' not in str(price): + return None + m = re.search(r'\d[\d,]*', str(price)) + return int(m.group(0).replace(',', '')) if m else None + + +def gate_region(st): + """Return (ok, error). Mirrors docs/DATA-MODEL.md import gates.""" + if not st.get('ok'): + return False, 'ok != true' + if st.get('partial'): + return False, 'partial == true' + fetched = parse_iso(st.get('fetched_at')) + if fetched is None: + return False, 'fetched_at missing/unparseable' + age = datetime.datetime.now(datetime.timezone.utc) - fetched + if age.total_seconds() > GATE_MAX_AGE_HOURS * 3600: + return False, 'fetched_at older than %dh' % GATE_MAX_AGE_HOURS + if not st.get('items'): + return False, 'items empty' + return True, None + + +def iter_items(st): + """Yield (src_id, item_dict) for a state file. Handles dict-keyed or list + items defensively; src_id comes from the ad id (dict key / url id).""" + items = st.get('items') or {} + if isinstance(items, dict): + for src_id, it in items.items(): + yield str(src_id), dict(it) + elif isinstance(items, list): + for it in items: + it = dict(it) + src_id = str(it.get('id') or '') + if not src_id: + m = re.search(r'/(\d+)(?:/|$)', it.get('url') or '') + src_id = m.group(1) if m else '' + yield src_id, it + + +def normalize(region, src_id, it, now): + """Normalize one state item into a row dict (strings->ints robustly, + missing fields -> NULL).""" + title = (it.get('title') or '').strip() + price_display = it.get('price') or None + price_num = it.get('price_num') + if not price_num: + price_num = num_from_price(price_display) + km = rubric.km_num(it.get('km')) + if km == 0: + km = None # 0 km is a placeholder, not delivery mileage + year = rubric.year_from_title(title) + score, family = rubric.score_item(title, it.get('desc') or '', price_num, km, year) + return { + 'region': region, + 'src_id': src_id, + 'title': title, + 'price_display': price_display, + 'price_num': price_num, + 'km': km, + 'location': (it.get('loc') or it.get('city') or None), + 'url': it.get('url') or '', + 'img_url': it.get('img') or None, + 'desc': it.get('desc') or None, + 'family': family, + 'score': score, + 'first_seen': it.get('first_seen') or now, + 'last_seen': now, + } + + +def upsert(conn, row, now): + """Insert or update one listing. Returns (db_id, is_new). Present rows are + reset to 'active' (reappearance flips gone back; dedupe re-marks dups).""" + cur = conn.execute( + 'SELECT id FROM listings WHERE region=? AND src_id=?', + (row['region'], row['src_id'])) + existing = cur.fetchone() + if existing: + conn.execute( + '''UPDATE listings SET title=?, price_display=?, price_num=?, km=?, + location=?, url=?, img_url=?, desc=?, family=?, score=?, + status='active', dup_of=NULL, last_seen=?, updated_at=? + WHERE id=?''', + (row['title'], row['price_display'], row['price_num'], row['km'], + row['location'], row['url'], row['img_url'], row['desc'], + row['family'], row['score'], now, now, existing['id'])) + return existing['id'], False + cur = conn.execute( + '''INSERT INTO listings (region, src_id, title, price_display, price_num, + km, location, url, img_url, desc, family, score, status, dup_of, + first_seen, last_seen, created_at, updated_at) + VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)''', + (row['region'], row['src_id'], row['title'], row['price_display'], + row['price_num'], row['km'], row['location'], row['url'], + row['img_url'], row['desc'], row['family'], row['score'], 'active', + None, row['first_seen'], row['last_seen'], now, now)) + return cur.lastrowid, True + + +def dedupe(conn, present): + """Collapse cross-province twins. `present` maps (region, src_id) -> db_id. + Groups by (normalized_title, price_num); a group spanning both provinces + keeps the most-complete row canonical (img > km > older) and marks the rest + 'dup'. The twin's URL is preserved on the canonical row's desc.""" + groups = {} + for (region, src_id), db_id in present.items(): + row = conn.execute( + 'SELECT title, price_num, img_url, km, first_seen, url, desc FROM listings WHERE id=?', + (db_id,)).fetchone() + key = (norm_title(row['title']), row['price_num']) + groups.setdefault(key, []).append((region, db_id, row)) + + dup_count = 0 + for key, members in groups.items(): + if len({m[0] for m in members}) < 2: + continue # not cross-province + # most complete first: img present, then km present, then older row + members.sort(key=lambda m: ( + 0 if m[2]['img_url'] else 1, + 0 if m[2]['km'] is not None else 1, + m[2]['first_seen'] or '')) + canonical = members[0] + twin_urls = [] + for region, db_id, row in members[1:]: + conn.execute( + "UPDATE listings SET status='dup', dup_of=?, updated_at=? WHERE id=?", + (canonical[1], now_iso(), db_id)) + dup_count += 1 + if row['url']: + twin_urls.append(row['url']) + if twin_urls: + # preserve twin URLs on the canonical row (desc suffix), after scoring + cur = conn.execute('SELECT desc FROM listings WHERE id=?', (canonical[1],)) + base_desc = cur.fetchone()['desc'] or '' + suffix = ' [also listed: ' + ' | '.join(twin_urls) + ']' + conn.execute('UPDATE listings SET desc=? WHERE id=?', + (base_desc + suffix, canonical[1])) + return dup_count + + +def mark_gone(conn, present, prev_good): + """Mark active listings absent from >=2 consecutive good imports as gone. + `prev_good` is the previous good import's started_at (ISO); a listing whose + last_seen is older than that was also absent last run. Single misses and + failed/partial runs never mark gone (prev_good only advances on good runs).""" + gone = 0 + rows = conn.execute( + "SELECT id, region, src_id, last_seen FROM listings WHERE status='active'").fetchall() + for r in rows: + if (r['region'], r['src_id']) in present: + continue + if prev_good and (r['last_seen'] or '') < prev_good: + conn.execute( + "UPDATE listings SET status='gone', updated_at=? WHERE id=?", + (now_iso(), r['id'])) + gone += 1 + return gone + + +def run_import(state_dir, db_path): + conn = db.connect(db_path) + db.init_db(conn) + started = now_iso() + + # Load + gate each region. + regions, errors = {}, {} + for fname, region in REGIONS.items(): + path = os.path.join(state_dir, fname + '.json') + try: + with open(path, encoding='utf-8') as fh: + st = json.load(fh) + except Exception as e: + errors[region] = 'read error: %s' % e + continue + ok, err = gate_region(st) + if ok: + regions[region] = st + else: + errors[region] = err + + if errors: + # Fail loud, keep last good: log the run, leave listings untouched. + conn.execute( + '''INSERT INTO import_runs (started_at, ok, partial, src_fetched_ns, + src_fetched_nb, total, new, gone, dup, error) + VALUES (?,?,?,?,?,?,?,?,?,?)''', + (started, 0, 0, + 1 if 'ns' not in errors else 0, + 1 if 'nb' not in errors else 0, + None, None, None, None, + '; '.join('%s: %s' % (r, e) for r, e in sorted(errors.items())))) + conn.commit() + conn.close() + return {'ok': False, 'errors': errors} + + # Previous good import time (for gone-marking). + prev = conn.execute( + "SELECT started_at FROM import_runs WHERE ok=1 ORDER BY id DESC LIMIT 1").fetchone() + prev_good = prev['started_at'] if prev else None + + # Upsert all present items. + present = {} + new_count = 0 + for region, st in regions.items(): + for src_id, it in iter_items(st): + row = normalize(region, src_id, it, started) + db_id, is_new = upsert(conn, row, started) + present[(region, src_id)] = db_id + if is_new: + new_count += 1 + + # Dedupe cross-province twins. + dup_count = dedupe(conn, present) + + # Gone-marking for active rows absent from this import. + gone_count = mark_gone(conn, present, prev_good) + + # Advance the last-good-import marker (only on a good run). + conn.execute( + "INSERT INTO prefs (key, value) VALUES ('last_good_import', ?) " + "ON CONFLICT(key) DO UPDATE SET value=excluded.value", (started,)) + + total = conn.execute('SELECT COUNT(*) AS n FROM listings').fetchone()['n'] + conn.execute( + '''INSERT INTO import_runs (started_at, ok, partial, src_fetched_ns, + src_fetched_nb, total, new, gone, dup, error) + VALUES (?,?,?,?,?,?,?,?,?,?)''', + (started, 1, 0, 1, 1, total, new_count, gone_count, dup_count, None)) + conn.commit() + conn.close() + return {'ok': True, 'total': total, 'new': new_count, 'gone': gone_count, + 'dup': dup_count} + + +def main(argv=None): + ap = argparse.ArgumentParser(description='Bike-Tinder importer (Phase 1)') + ap.add_argument('--state-dir', required=True, + help='hunt state dir (read-only)') + ap.add_argument('--db', default='data/bike-tinder.sqlite3', + help='SQLite DB path') + args = ap.parse_args(argv) + + result = run_import(args.state_dir, args.db) + if result['ok']: + print('import: ok=1 total=%d new=%d gone=%d dup=%d active=%d' % ( + result['total'], result['new'], result['gone'], result['dup'], + result['total'] - result['gone'] - result['dup'])) + return 0 + print('import: ok=0 errors=%s' % result['errors'], file=sys.stderr) + return 1 + + +if __name__ == '__main__': + sys.exit(main())