Bike-Tinder/app/importer.py

"""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 imgs_from_item(it):
    """Gallery images from a state item: deduplicated, non-empty URL strings
    in page order, or None. The hunt guarantees imgs[0] equals `img` (the
    established primary)."""
    raw = it.get('imgs')
    if not isinstance(raw, (list, tuple)):
        return None
    out, seen = [], set()
    for u in raw:
        if not isinstance(u, str):
            continue
        u = u.strip()
        if u and u not in seen:
            seen.add(u)
            out.append(u)
    return out or None


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)
    extras = rubric.extras_from_text(title, it.get('desc') or '')
    # B1: gallery (page order, entry 0 = the primary). img_url keeps the
    # established single `img` signal and falls back to imgs[0].
    imgs = imgs_from_item(it)
    img = it.get('img') or (imgs[0] if imgs else None)
    # seller_type only from a REAL signal in the state item (the hunt image
    # backfill / enrichment passes) — never inferred from titles.
    seller = it.get('seller_type') or it.get('seller')
    if seller not in ('dealer', 'private'):
        seller = None
    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': img,
        'imgs': json.dumps(imgs) if imgs else None,
        'desc': it.get('desc') or None,
        'family': family,
        'score': score,
        'year': year,
        'seller_type': seller,
        'extras': json.dumps(extras) if extras else None,
        '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=?, imgs=?, desc=?, family=?, score=?,
               year=?, seller_type=?, extras=?,
               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['imgs'],
             row['desc'], row['family'], row['score'], row['year'],
             row['seller_type'], row['extras'], 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, imgs, desc, family, score, year,
           seller_type, extras, 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['imgs'], row['desc'], row['family'], row['score'],
         row['year'], row['seller_type'], row['extras'], 'active',
         None, row['first_seen'], row['last_seen'], now, now))
    return cur.lastrowid, True


def dedupe(conn, present):
    """Collapse cross-province twins with min(ns,nb) completeness pairing.

    `present` maps (region, src_id) -> db_id. Groups by
    (normalized_title, price_num). For a group spanning both provinces the
    two region lists are sorted best-first (img present > km present > older)
    and paired one-to-one up to min(ns_count, nb_count): each pair keeps the
    more-complete row canonical and marks the twin 'dup'. Surplus siblings
    (multi-unit dealer ads) stay active instead of being false-merged.
    The twin's URL is preserved on the canonical row's desc. (DATA-MODEL P2
    addendum: pairing replaces collapse-to-one — dup ≈ 77 vs 80 on today's
    snapshot.) Reversible: dubs carry dup_of; flipping status back to
    'active' and re-importing re-derives the pairing."""
    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))

    def completeness(m):
        """Best first: img present, then km present, then older row."""
        return (0 if m[2]['img_url'] else 1,
                0 if m[2]['km'] is not None else 1,
                m[2]['first_seen'] or '')

    dup_count = 0
    for key, members in groups.items():
        if len({m[0] for m in members}) < 2:
            continue  # not cross-province
        ns = sorted((m for m in members if m[0] == 'ns'), key=completeness)
        nb = sorted((m for m in members if m[0] == 'nb'), key=completeness)
        for a, b in zip(ns, nb):
            canonical, twin = (a, b) if completeness(a) <= completeness(b) else (b, a)
            conn.execute(
                "UPDATE listings SET status='dup', dup_of=?, updated_at=? WHERE id=?",
                (canonical[1], now_iso(), twin[1]))
            dup_count += 1
            if twin[2]['url']:
                # preserve the twin's URL on the canonical row (desc suffix)
                cur = conn.execute('SELECT desc FROM listings WHERE id=?',
                                   (canonical[1],))
                base_desc = cur.fetchone()['desc'] or ''
                suffix = ' [also listed: ' + twin[2]['url'] + ']'
                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())