app: add importer (gates, upsert, dedupe, gone-marking, scoring, import_runs)
This commit is contained in:
Torres
305
app/importer.py
Normal file
305
app/importer.py
Normal file
@@ -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())
|
||||
Reference in New Issue
Block a user