merge: Phase 1 data spine (verified): schema, importer, rubric, deck CLI, selfcheck

.gitignore

@@ -0,0 +1,12 @@
+# runtime state
+data/
+*.sqlite3
+*.sqlite3-wal
+*.sqlite3-shm
+
+# python
+__pycache__/
+*.pyc
+
+# scratch
+scratch/

app/__init__.py

@@ -0,0 +1 @@
+"""Bike-Tinder application package (Phase 1: data spine)."""

app/db.py

@@ -0,0 +1,29 @@
+"""SQLite connection + schema init for Bike-Tinder.
+
+Stdlib only. The sqlite3 CLI is not installed on this host, so everything
+goes through the sqlite3 module.
+"""
+import os
+import sqlite3
+
+SCHEMA_PATH = os.path.join(os.path.dirname(__file__), 'schema.sql')
+
+
+def connect(db_path):
+    """Open a connection with WAL mode and a busy_timeout, then ensure schema."""
+    parent = os.path.dirname(os.path.abspath(db_path))
+    if parent:
+        os.makedirs(parent, exist_ok=True)
+    conn = sqlite3.connect(db_path)
+    conn.row_factory = sqlite3.Row
+    conn.execute('PRAGMA journal_mode=WAL')
+    conn.execute('PRAGMA busy_timeout=5000')
+    conn.execute('PRAGMA foreign_keys=ON')
+    return conn
+
+
+def init_db(conn):
+    """Apply schema.sql (idempotent)."""
+    with open(SCHEMA_PATH, encoding='utf-8') as fh:
+        conn.executescript(fh.read())
+    conn.commit()

app/deck.py

@@ -0,0 +1,65 @@
+"""Read-only proof CLI: print the ranked deck and DB counts.
+
+  python3 -m app.deck --db data/bike-tinder.sqlite3 --top 15
+"""
+import argparse
+import sys
+
+from . import db
+
+DECK_QUERY = """
+SELECT * FROM listings
+WHERE status = 'active'
+  AND id NOT IN (SELECT listing_id FROM swipes WHERE action = 'nope')
+  AND id NOT IN (SELECT listing_id FROM swipes WHERE action IN ('like','maybe')
+                 AND (action = 'like' OR updated_at > datetime('now','-7 days')))
+ORDER BY score DESC, first_seen DESC
+LIMIT ?
+"""
+
+
+def fmt_price(p):
+    return p if p else '-'
+
+
+def fmt_km(k):
+    return ('%d km' % k) if k is not None else '-'
+
+
+def main(argv=None):
+    ap = argparse.ArgumentParser(description='Bike-Tinder deck (read-only)')
+    ap.add_argument('--db', default='data/bike-tinder.sqlite3')
+    ap.add_argument('--top', type=int, default=15)
+    args = ap.parse_args(argv)
+
+    conn = db.connect(args.db)
+    db.init_db(conn)
+
+    rows = conn.execute(DECK_QUERY, (args.top,)).fetchall()
+    print('=== Bike-Tinder deck (top %d) ===' % len(rows))
+    for i, r in enumerate(rows, 1):
+        print('%3d. %s · %s · %s · %s · %d' % (
+            i, r['title'], fmt_price(r['price_display']), fmt_km(r['km']),
+            r['region'], r['score']))
+
+    counts = {}
+    for status in ('active', 'gone', 'dup'):
+        counts[status] = conn.execute(
+            'SELECT COUNT(*) AS n FROM listings WHERE status=?', (status,)).fetchone()['n']
+    total = conn.execute('SELECT COUNT(*) AS n FROM listings').fetchone()['n']
+    last = conn.execute(
+        'SELECT new FROM import_runs ORDER BY id DESC LIMIT 1').fetchone()
+    new = last['new'] if last else 0
+
+    print('=== counts ===')
+    print('total:  %d' % total)
+    print('active: %d' % counts['active'])
+    print('gone:   %d' % counts['gone'])
+    print('dup:    %d' % counts['dup'])
+    print('new:    %d' % new)
+    conn.close()
+    return 0
+
+
+if __name__ == '__main__':
+    sys.exit(main())

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())

app/init.py

@@ -0,0 +1,20 @@
+"""Explicit DB init path: python3 -m app.init --db <path>."""
+import argparse
+import sys
+
+from . import db
+
+
+def main(argv=None):
+    ap = argparse.ArgumentParser(description='Initialize the Bike-Tinder DB')
+    ap.add_argument('--db', default='data/bike-tinder.sqlite3')
+    args = ap.parse_args(argv)
+    conn = db.connect(args.db)
+    db.init_db(conn)
+    conn.close()
+    print('init: schema applied to %s' % args.db)
+    return 0
+
+
+if __name__ == '__main__':
+    sys.exit(main())

app/rubric.py

@@ -0,0 +1,152 @@
+"""Rubric scoring for Bike-Tinder, ported verbatim from the hunt's
+reference/motorcycle-hunt-rank.py (family tiers + extras + veto + price fit).
+
+The hunt's ranker drops rows with no family match (returns None). Bike-Tinder
+shows everything, so an unmatched/vetoed row scores 0 and sorts below scored
+rows (docs/DATA-MODEL.md: "Unscored rows sort below scored ones; nothing is
+hidden").
+"""
+import re
+
+# (pattern, weight, tier, label) — copied from motorcycle-hunt-rank.py.
+FAMILIES = [
+    (r'\bscout\b', 92, 'S', 'Indian Scout — top priority (sport×cruise)'),
+    (r'vulcan\s*900\s*custom', 88, 'S', 'Vulcan 900 Custom — the dream spec'),
+    (r'vulcan\s*900', 62, 'T1', 'Vulcan 900'),
+    (r'vulcan\s*(classic|lt|nomad)', 58, 'T1', 'Vulcan Classic/LT'),
+    (r'vulcan\s*s\b', 50, 'T1', 'Vulcan S'),
+    (r'vulcan\s*500', 40, 'T2', 'Vulcan 500'),
+    (r'\bvulcan\b', 45, 'T1', 'Vulcan'),
+    (r'\bvolusia\b', 44, 'T1', 'Volusia (C50 predecessor)'),
+    (r'rebel\s*500', 50, 'T1', 'Rebel 500'),
+    (r'rebel\s*300', 44, 'T1', 'Rebel 300'),
+    (r'\brebel\b', 46, 'T1', 'Rebel'),
+    (r'boulevard\s*c50', 50, 'T1', 'Boulevard C50'),
+    (r'\bc50t?\b', 48, 'T1', 'Boulevard C50/C50T'),
+    (r'boulevard\s*(m50|s50)', 44, 'T1', 'Boulevard M50/S50'),
+    (r'boulevard\s*(c90|m90|m109)', 42, 'T2', 'Boulevard big twin'),
+    (r'm109r', 42, 'T2', 'Boulevard M109R'),
+    (r'\bboulevard\b', 40, 'T1', 'Boulevard'),
+    (r'\bintruder\b', 36, 'T2', 'Intruder'),
+    (r'shadow\s*(aero|phantom|spirit)', 48, 'T1', 'Shadow Aero/Phantom/Spirit'),
+    (r'shadow\s*750', 44, 'T1', 'Shadow 750'),
+    (r'\bshadow\b', 42, 'T1', 'Shadow'),
+    (r'\bsabre\b', 34, 'T2', 'Shadow Sabre'),
+    (r'v-?star\s*(650|950)', 46, 'T1', 'V-Star 650/950'),
+    (r'v-?star\s*1100', 40, 'T2', 'V-Star 1100'),
+    (r'v-?star', 42, 'T1', 'V-Star'),
+    (r'iron\s*883', 46, 'T1', 'Iron 883'),
+    (r'\bsportster\b', 40, 'T1', 'Sportster'),
+    (r'super\s*meteor', 48, 'T1', 'Super Meteor 650'),
+    (r'meteor\s*350', 42, 'T1', 'Meteor 350'),
+    (r'\bmeteor\b', 42, 'T1', 'Meteor'),
+    (r'hunter\s*350', 38, 'T2', 'Hunter 350'),
+    (r'classic\s*(350|650)', 36, 'T2', 'RE Classic'),
+    (r'royal\s*enfield', 34, 'T2', 'Royal Enfield'),
+    (r'\beliminator\b', 38, 'T2', 'Eliminator'),
+    (r'\bvirago\b', 34, 'T2', 'Virago'),
+    (r'\bmagna\b', 32, 'T2', 'Magna'),
+    (r'\bnightster\b', 32, 'T2', 'Nightster'),
+    (r'forty-?eight', 32, 'T2', 'Forty-Eight'),
+    (r'\bbonneville\b', 30, 'style', 'Bonneville (style-adjacent)'),
+    (r'\bw800\b', 28, 'style', 'W800'),
+    (r'z650rs', 28, 'style', 'Z650RS'),
+    (r'moto\s*guzzi|\bv7\b', 30, 'style', 'V7'),
+    (r'\bscrambler\b', 24, 'style', 'Scrambler'),
+    (r'\bcontinental\s*gt\b', 26, 'style', 'Continental GT'),
+]
+
+# category vetoes — not a small cruiser (dirt/sport/touring/powersports/gear)
+VETO = [
+    (r'\b(atv|quad|side[\s-]?by[\s-]?side|utv|ranger|outlander|renegade|ryker|spyder|ski-doo|seadoo|sea-doo|snowmobile|snow\s*blower|lawn|generator|pump|watercraft|boat)\b', 'powersports'),
+    (r'\b(dirt\s*bike|motocross|enduro)\b', 'dirt'),
+    (r'\b(crf|kx\d|yz\d|rm-?z|klx|exc|xc-w|xc-f|fe\s*\d|fc\s*\d|mc\s*\d|sx\b|dr-?z|ttr|kdx)\b', 'dirt'),
+    (r'\b(ninja|zx-?\d|gsx-?[rs]|cbr|yzf|hayabusa|panigale|fireblade|gixxer|katana|r\d\s)\b', 'sport'),
+    (r'\b(v-?strom|versys|africa\s*twin|multistrada|t[eé]n[eé]r[eé]|transalp|super\s*adventure)\b', 'adv'),
+    (r'\b(gold\s*wing|voyager|electra\s*glide|road\s*king|ultra\s*classic|f6b)\b', 'tourer'),
+    (r'\b(helmet|half-?shell|jacket|glove|boots?\b|seat\b|exhaust|muffler|tire|tyre|rim\b|brake|lever|mirror|cover\b|battery|sprocket|for\s*parts|parts\b|project\s*bike|salvage|wreck)\b', 'gear'),
+]
+
+EXTRAS = [
+    (r'\b(mvi|safety|inspected|inspection|certified|e-?test)\b', 6, 'MVI/cert'),
+    (r'white\s*wall', 5, 'whitewalls'),
+    (r'\bspokes?\b|spoked', 3, 'spokes'),
+    (r'wind\s*screen|windshield', 3, 'windshield'),
+    (r'saddle\s*bags?|panniers?|\bbags\b', 3, 'bags'),
+    (r'\babs\b', 2, 'ABS'),
+]
+
+
+def km_num(km):
+    """Parse a km string like '32778' or '12,345' to int; None when absent."""
+    s = str(km or '')
+    m = re.search(r'([\d,]+)', s)
+    return int(m.group(1).replace(',', '')) if m else None
+
+
+def classify(title, desc):
+    """Return (tier, weight, label, veto_reason). Veto checks title only;
+    family match checks title + desc (mirrors the hunt ranker)."""
+    text = (title + ' ' + desc).lower()
+    ttext = title.lower()
+    for pat, why in VETO:
+        if re.search(pat, ttext):
+            return None, 0, None, why
+    best = None
+    for pat, weight, tier, label in FAMILIES:
+        if re.search(pat, text):
+            if best is None or weight > best[1]:
+                best = (pat, weight, tier, label)
+    if not best:
+        return None, 0, None, None
+    return best[2], best[1], best[3], None
+
+
+def score_item(title, desc, price_num, km, year=None):
+    """Return (score, family_label). Unmatched/vetoed rows score 0."""
+    if km == 0:
+        km = None  # 0 km is a placeholder, not delivery mileage
+    tier, base, label, _veto = classify(title, desc)
+    if tier is None:
+        return 0, None
+    s = base
+    text = (title + ' ' + desc).lower()
+    for pat, pts, _fl in EXTRAS:
+        if re.search(pat, text):
+            s += pts
+    if price_num:
+        # Kijiji v1 carries no cond/offer field, so every row is treated as
+        # used (matches the hunt ranker's behaviour on these state files).
+        if price_num <= 5000:
+            s += 5
+        elif price_num <= 6500:
+            s += 3
+        elif price_num <= 8000:
+            s += 1
+        elif price_num <= 10000:
+            s -= 1
+        else:
+            s -= 3
+    if km is not None:
+        if km < 5000:
+            s += 4
+        elif km < 15000:
+            s += 2
+        elif km > 90000:
+            s -= 5
+        elif km > 60000:
+            s -= 3
+    if year:
+        if year >= 2020:
+            s += 2
+        elif year >= 2015:
+            s += 1
+        elif year <= 2005:
+            s -= 1
+    return s, label
+
+
+def year_from_title(title):
+    """Extract a model year from the title, or None."""
+    m = re.search(r'\b(19\d\d|20[0-2]\d)\b', title or '')
+    return int(m.group(1)) if m else None

app/schema.sql

@@ -0,0 +1,54 @@
+-- Bike-Tinder v1 schema (SQLite). Locked per docs/DATA-MODEL.md.
+-- Applied by app.db.init_db(); WAL mode and busy_timeout are set there.
+
+CREATE TABLE IF NOT EXISTS listings (
+  id             INTEGER PRIMARY KEY AUTOINCREMENT,
+  region         TEXT NOT NULL CHECK (region IN ('ns','nb')),
+  src_id         TEXT NOT NULL,                    -- source ad id (Kijiji id in v1)
+  title          TEXT NOT NULL,
+  price_display  TEXT,                             -- "$5,000"; "Please Contact"
+  price_num      INTEGER,                          -- NULL when unpackable
+  km             INTEGER,                          -- NULL when not stated
+  location       TEXT,
+  url            TEXT NOT NULL,                    -- source ad page
+  img_url        TEXT,                             -- source CDN card image
+  desc           TEXT,                             -- card description snippet
+  family         TEXT,                             -- top rubric family label (taste signal)
+  score          INTEGER NOT NULL DEFAULT 0,       -- rubric score, recomputed each import
+  status         TEXT NOT NULL DEFAULT 'active' CHECK (status IN ('active','gone','dup')),
+  dup_of         INTEGER,                          -- canonical row id when status='dup'
+  first_seen     TEXT,
+  last_seen      TEXT,
+  created_at     TEXT DEFAULT (datetime('now')),
+  updated_at     TEXT,
+  UNIQUE (region, src_id)
+);
+
+CREATE INDEX IF NOT EXISTS idx_listings_deck ON listings (status, score DESC);
+CREATE INDEX IF NOT EXISTS idx_listings_dup  ON listings (dup_of) WHERE status = 'dup';
+
+CREATE TABLE IF NOT EXISTS swipes (
+  listing_id  INTEGER PRIMARY KEY REFERENCES listings(id) ON DELETE CASCADE,
+  action      TEXT NOT NULL CHECK (action IN ('like','nope','maybe')),
+  created_at  TEXT DEFAULT (datetime('now')),
+  updated_at  TEXT
+);
+
+CREATE TABLE IF NOT EXISTS import_runs (
+  id             INTEGER PRIMARY KEY AUTOINCREMENT,
+  started_at     TEXT NOT NULL,
+  ok             INTEGER NOT NULL,
+  partial        INTEGER NOT NULL DEFAULT 0,
+  src_fetched_ns INTEGER,
+  src_fetched_nb INTEGER,
+  total          INTEGER,
+  new            INTEGER,
+  gone           INTEGER,
+  dup            INTEGER,
+  error          TEXT
+);
+
+CREATE TABLE IF NOT EXISTS prefs (
+  key   TEXT PRIMARY KEY,
+  value TEXT
+);

app/selfcheck.py

@@ -0,0 +1,249 @@
+"""Dependency-free self-check for the Phase 1 data spine.
+
+  python3 -m app.selfcheck
+
+Covers: gate rejections, dedupe (incl. false-positive sanity), scoring
+spot-checks against the reference rubric, and gone-after-2-misses.
+"""
+import datetime
+import json
+import os
+import shutil
+import sys
+import tempfile
+
+from . import db, rubric
+from . import importer as imp
+
+
+def recent_iso():
+    return datetime.datetime.now(datetime.timezone.utc).strftime('%Y-%m-%dT%H:%M:%SZ')
+
+
+def make_item(title, price='$5,000', price_num=5000, km='10000', loc='Halifax',
+              url='https://www.kijiji.ca/v-x/1', desc='', img=''):
+    return {'title': title, 'price': price, 'price_num': price_num, 'loc': loc,
+            'km': km, 'url': url, 'desc': desc, 'img': img,
+            'first_seen': '2026-01-01T00:00:00Z', 'last_seen': '2026-01-01T00:00:00Z'}
+
+
+def make_state(items, ok=True, partial=False, fetched_at=None, absent=None):
+    return {'source': 'kijiji-ns', 'fetched_at': fetched_at or recent_iso(),
+            'ok': ok, 'fail_streak': 0, 'pages': {'1': len(items)},
+            'partial': partial, 'items': items, 'absent': absent or {}}
+
+
+def write_states(state_dir, ns, nb):
+    os.makedirs(state_dir, exist_ok=True)
+    for name, st in (('kijiji-ns', ns), ('kijiji-nb', nb)):
+        with open(os.path.join(state_dir, name + '.json'), 'w', encoding='utf-8') as fh:
+            json.dump(st, fh)
+
+
+def fresh_db():
+    d = tempfile.mkdtemp(prefix='bt-selfcheck-')
+    return d, os.path.join(d, 'test.sqlite3')
+
+
+# --- gate rejections -------------------------------------------------------
+
+def test_gate_rejections():
+    ok, err = imp.gate_region(make_state({'1': make_item('X')}, ok=False))
+    assert not ok and 'ok' in (err or ''), err
+    ok, err = imp.gate_region(make_state({'1': make_item('X')}, partial=True))
+    assert not ok and 'partial' in (err or ''), err
+    stale = (datetime.datetime.now(datetime.timezone.utc)
+             - datetime.timedelta(hours=30)).strftime('%Y-%m-%dT%H:%M:%SZ')
+    ok, err = imp.gate_region(make_state({'1': make_item('X')}, fetched_at=stale))
+    assert not ok and '26h' in (err or ''), err
+    ok, err = imp.gate_region(make_state({}))
+    assert not ok and 'empty' in (err or ''), err
+    ok, err = imp.gate_region(make_state({'1': make_item('X')}))
+    assert ok and err is None, err
+
+
+def test_gate_abort_leaves_db():
+    d, db_path = fresh_db()
+    try:
+        state_dir = os.path.join(d, 'state')
+        write_states(state_dir,
+                     make_state({'1': make_item('X')}, ok=False),   # ns fails
+                     make_state({'2': make_item('Y')}))             # nb ok
+        res = imp.run_import(state_dir, db_path)
+        assert res['ok'] is False
+        conn = db.connect(db_path)
+        db.init_db(conn)
+        n = conn.execute('SELECT COUNT(*) AS n FROM listings').fetchone()['n']
+        runs = conn.execute('SELECT ok, error FROM import_runs').fetchall()
+        conn.close()
+        assert n == 0, 'listings must be untouched on gate failure'
+        assert len(runs) == 1 and runs[0]['ok'] == 0 and 'ns' in runs[0]['error']
+    finally:
+        shutil.rmtree(d, ignore_errors=True)
+
+
+# --- dedupe ----------------------------------------------------------------
+
+def test_dedupe_twins():
+    d, db_path = fresh_db()
+    try:
+        state_dir = os.path.join(d, 'state')
+        twin = make_item('2021 Indian Scout Bobber', price_num=6500, km='8000')
+        write_states(state_dir,
+                     make_state({'ns1': twin}),
+                     make_state({'nb1': dict(twin)}))
+        res = imp.run_import(state_dir, db_path)
+        assert res['ok'] and res['total'] == 2 and res['dup'] == 1, res
+        conn = db.connect(db_path)
+        db.init_db(conn)
+        rows = conn.execute(
+            'SELECT id, region, status, dup_of FROM listings ORDER BY region').fetchall()
+        conn.close()
+        by_status = {r['status'] for r in rows}
+        assert by_status == {'active', 'dup'}, rows
+        dup = [r for r in rows if r['status'] == 'dup'][0]
+        active = [r for r in rows if r['status'] == 'active'][0]
+        assert dup['dup_of'] == active['id'], (dup, active)
+    finally:
+        shutil.rmtree(d, ignore_errors=True)
+
+
+def test_dedupe_false_positive_same_province():
+    # Same title+price but both in ONE province -> different bikes, not merged.
+    d, db_path = fresh_db()
+    try:
+        state_dir = os.path.join(d, 'state')
+        write_states(state_dir,
+                     make_state({'ns1': make_item('2025 BMW G 310 R', price_num=5444),
+                                 'ns2': make_item('2025 BMW G 310 R', price_num=5444)}),
+                     make_state({'nb_dummy': make_item('Some Other Bike', price_num=9999)}))
+        res = imp.run_import(state_dir, db_path)
+        assert res['ok'] and res['total'] == 3 and res['dup'] == 0, res
+    finally:
+        shutil.rmtree(d, ignore_errors=True)
+
+
+def test_dedupe_false_positive_price_diff():
+    # Same title, different price, cross-province -> not merged.
+    d, db_path = fresh_db()
+    try:
+        state_dir = os.path.join(d, 'state')
+        write_states(state_dir,
+                     make_state({'ns1': make_item('2021 Indian Scout', price_num=6500)}),
+                     make_state({'nb1': make_item('2021 Indian Scout', price_num=7000)}))
+        res = imp.run_import(state_dir, db_path)
+        assert res['ok'] and res['total'] == 2 and res['dup'] == 0, res
+    finally:
+        shutil.rmtree(d, ignore_errors=True)
+
+
+# --- scoring spot-checks ---------------------------------------------------
+
+def test_scoring():
+    s, fam = rubric.score_item('2021 Indian Scout Bobber', 'MVI done, low km',
+                               6500, 8000, 2021)
+    assert s == 105 and fam and 'Scout' in fam, (s, fam)
+    s, fam = rubric.score_item('Kawasaki Vulcan 900 Custom', 'whitewalls, windshield',
+                               5000, 20000, 2018)
+    assert s == 102 and 'Vulcan 900 Custom' in fam, (s, fam)
+    s, fam = rubric.score_item('Kawasaki Ninja 400', 'sport bike', 5000, 10000, 2020)
+    assert s == 0 and fam is None, (s, fam)  # vetoed -> unscored
+    s, fam = rubric.score_item('Some random bike', '', 3000, None, None)
+    assert s == 0 and fam is None, (s, fam)  # no family -> unscored
+    # km=0 placeholder -> treated as no km
+    s, _ = rubric.score_item('2021 Indian Scout', '', 5000, 0, 2021)
+    assert s == 92 + 5 + 2, s  # base + price(<=5000) + year(>=2020), no km bonus
+
+
+# --- gone after 2 misses ---------------------------------------------------
+
+def test_gone_after_two_misses():
+    d, db_path = fresh_db()
+    try:
+        conn = db.connect(db_path)
+        db.init_db(conn)
+        conn.execute(
+            "INSERT INTO listings (region, src_id, title, url, status, last_seen) "
+            "VALUES ('ns','gone1','X','http://x/1','active','2026-01-01T00:00:01Z')")
+        conn.commit()
+        # single miss: last_seen == prev_good -> not gone
+        gone = imp.mark_gone(conn, set(), '2026-01-01T00:00:01Z')
+        assert gone == 0, gone
+        st = conn.execute("SELECT status FROM listings WHERE src_id='gone1'").fetchone()['status']
+        assert st == 'active', st
+        # second miss: last_seen < prev_good -> gone
+        gone = imp.mark_gone(conn, set(), '2026-01-01T00:00:02Z')
+        assert gone == 1, gone
+        st = conn.execute("SELECT status FROM listings WHERE src_id='gone1'").fetchone()['status']
+        assert st == 'gone', st
+        conn.close()
+    finally:
+        shutil.rmtree(d, ignore_errors=True)
+
+
+def test_gone_full_flow():
+    # End-to-end: present -> absent (single miss) -> absent (gone).
+    d, db_path = fresh_db()
+    try:
+        state_dir = os.path.join(d, 'state')
+        orig = imp.now_iso
+        clock = {'now': '2026-01-01T00:00:01Z'}
+        imp.now_iso = lambda: clock['now']
+        try:
+            write_states(state_dir,
+                         make_state({'a1': make_item('2021 Indian Scout', price_num=6500)}),
+                         make_state({'nb_dummy': make_item('Some Other Bike', price_num=9999)}))
+            imp.run_import(state_dir, db_path)  # import 1: present
+
+            clock['now'] = '2026-01-01T00:00:02Z'
+            write_states(state_dir,
+                         make_state({'ns_other': make_item('Another Bike', price_num=8888)}),
+                         make_state({'nb_dummy': make_item('Some Other Bike', price_num=9999)}))
+            imp.run_import(state_dir, db_path)  # import 2: absent (single miss)
+            conn = db.connect(db_path)
+            db.init_db(conn)
+            st = conn.execute("SELECT status FROM listings WHERE src_id='a1'").fetchone()['status']
+            conn.close()
+            assert st == 'active', 'single miss must not mark gone'
+
+            clock['now'] = '2026-01-01T00:00:03Z'
+            imp.run_import(state_dir, db_path)  # import 3: absent (2nd miss)
+            conn = db.connect(db_path)
+            db.init_db(conn)
+            st = conn.execute("SELECT status FROM listings WHERE src_id='a1'").fetchone()['status']
+            conn.close()
+            assert st == 'gone', '2nd consecutive miss must mark gone'
+        finally:
+            imp.now_iso = orig
+    finally:
+        shutil.rmtree(d, ignore_errors=True)
+
+
+TESTS = [
+    ('gate rejections', test_gate_rejections),
+    ('gate abort leaves DB untouched', test_gate_abort_leaves_db),
+    ('dedupe cross-province twins', test_dedupe_twins),
+    ('dedupe false-positive (same province)', test_dedupe_false_positive_same_province),
+    ('dedupe false-positive (price differs)', test_dedupe_false_positive_price_diff),
+    ('scoring spot-checks', test_scoring),
+    ('gone after 2 misses (unit)', test_gone_after_two_misses),
+    ('gone after 2 misses (full flow)', test_gone_full_flow),
+]
+
+
+def main():
+    failures = 0
+    for name, fn in TESTS:
+        try:
+            fn()
+            print('PASS  %s' % name)
+        except Exception as e:
+            failures += 1
+            print('FAIL  %s: %s' % (name, e))
+    print('---')
+    print('%d/%d passed' % (len(TESTS) - failures, len(TESTS)))
+    return 1 if failures else 0
+
+
+if __name__ == '__main__':
+    sys.exit(main())