monster/discgod
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
1945c19fb3
discgod/server/discgod_server.py
type-two 1945c19fb3 discgod: API-token Discogs seller-inventory tracker 
Lean MV3 extension + standalone single-file ingest server that pulls watched
sellers' full inventories from the Discogs API into discogs_full (mp_* schema),
tracking price changes and sales over time. Successor to pliceclogs/marketwatcher;
runs alongside the legacy plice server on a separate port (5057).

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-06-13 19:54:33 +10:00

786 lines
31 KiB
Python
Raw Blame History

#!/usr/bin/env python3
"""
discgod ingest server — STANDALONE, single file.
Everything (schema, Discogs-API→DB mapping, upsert/price-diff/sold-detection,
watchlist, stats, HTTP server) lives here so it's drop-in: copy it to the box,
run it, done. It coexists with the legacy pliceclogs server on :5002 — discgod
listens on :5057, writes the same `discogs_full` DB, and touches none of the
legacy endpoints. If you ever want one server, the handlers below paste straight
into pliceclogs_server.py.
The extension's background worker pulls each watched seller's full inventory
from api.discogs.com (client-side, with your token) and POSTs the raw listing
pages here; this server maps + upserts them into the proven `mp_*` tables and
detects price changes / sales.
Stdlib + psycopg2 only. DB auth: plain psycopg2.connect(dbname=...) — local peer
auth on the Ultra (run as johnking). Off-box: set PGHOST/PGUSER/PGPASSWORD.
Run:
python3 discgod_server.py
# or, like the legacy server:
nohup /opt/homebrew/bin/python3 discgod_server.py > discgod_server.log 2>&1 &
Endpoints
GET /discgod/health
GET /discgod/stats
GET /discgod/tracked
POST /discgod/track {username, note?, filter?, enabled?, seller_id?}
POST /discgod/untrack {username}
POST /discgod/ingest {username, seller_id?, seller?, page, total?, crawl_id, filter?, listings:[...]}
POST /discgod/crawl_complete {username, crawl_id, filter?, treat_complete?}
"""
import json
import os
import sys
from datetime import datetime
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
import psycopg2
import psycopg2.extras
try:
sys.stdout.reconfigure(line_buffering=True)
except Exception:
pass
DB_NAME = os.environ.get("DISCGOD_DB", os.environ.get("PLICE_DB", "discogs_full"))
PORT = int(os.environ.get("DISCGOD_PORT", "5057"))
# Params that describe the search *filter*, not a position within it.
NON_FILTER_PARAMS = {"page", "pages", "per_page", "limit", "sort", "sort_order", "status", "token"}
def get_conn():
return psycopg2.connect(dbname=DB_NAME)
# ===========================================================================
# Schema (idempotent). mp_* mirror the marketwatcher schema exactly so existing
# queries and the join to the XML `release` table keep working.
# ===========================================================================
SCHEMA = """
CREATE TABLE IF NOT EXISTS mp_seller (
seller_id BIGINT PRIMARY KEY,
username TEXT NOT NULL,
rating NUMERIC(3,1),
rating_percent NUMERIC(5,2),
ratings_count INTEGER,
ships_from TEXT,
first_seen TIMESTAMPTZ DEFAULT now(),
last_seen TIMESTAMPTZ DEFAULT now()
);
CREATE INDEX IF NOT EXISTS idx_mp_seller_username ON mp_seller (username);
CREATE TABLE IF NOT EXISTS mp_listing (
item_id BIGINT PRIMARY KEY,
release_id BIGINT NOT NULL,
seller_id BIGINT,
seller_username TEXT,
title TEXT,
label TEXT,
cat_no TEXT,
media_condition TEXT,
sleeve_condition TEXT,
notes TEXT,
price_value NUMERIC(12,2),
price_currency TEXT,
shipping_value NUMERIC(12,2),
converted_value NUMERIC(12,2),
have_count INTEGER,
want_count INTEGER,
ships_from TEXT,
status TEXT NOT NULL DEFAULT 'active',
gone_at TIMESTAMPTZ,
first_seen TIMESTAMPTZ DEFAULT now(),
last_seen TIMESTAMPTZ DEFAULT now(),
times_seen INTEGER DEFAULT 1,
last_crawl_id TEXT
);
CREATE INDEX IF NOT EXISTS idx_mp_listing_release ON mp_listing (release_id);
CREATE INDEX IF NOT EXISTS idx_mp_listing_seller ON mp_listing (seller_username, status);
CREATE INDEX IF NOT EXISTS idx_mp_listing_seen ON mp_listing (last_seen DESC);
CREATE TABLE IF NOT EXISTS mp_listing_event (
id BIGSERIAL PRIMARY KEY,
item_id BIGINT NOT NULL,
release_id BIGINT,
seller_username TEXT,
event TEXT NOT NULL,
old_price NUMERIC(12,2),
new_price NUMERIC(12,2),
currency TEXT,
details JSONB,
recorded_at TIMESTAMPTZ DEFAULT now()
);
CREATE INDEX IF NOT EXISTS idx_mp_event_item ON mp_listing_event (item_id, recorded_at DESC);
CREATE INDEX IF NOT EXISTS idx_mp_event_type ON mp_listing_event (event, recorded_at DESC);
CREATE INDEX IF NOT EXISTS idx_mp_event_release ON mp_listing_event (release_id);
CREATE TABLE IF NOT EXISTS mp_seller_snapshot (
id BIGSERIAL PRIMARY KEY,
seller_id BIGINT,
username TEXT NOT NULL,
total_listings BIGINT,
source TEXT NOT NULL,
filter_key TEXT,
filter_params JSONB,
captured_at TIMESTAMPTZ DEFAULT now()
);
CREATE INDEX IF NOT EXISTS idx_mp_snapshot_series
ON mp_seller_snapshot (username, source, filter_key, captured_at DESC);
CREATE TABLE IF NOT EXISTS mp_page_log (
id BIGSERIAL PRIMARY KEY,
url TEXT,
page_type TEXT,
seller_page TEXT,
params JSONB,
page INTEGER,
total_results BIGINT,
item_count INTEGER,
item_ids BIGINT[],
new_items INTEGER,
price_changes INTEGER,
crawl_id TEXT,
scraped_at TIMESTAMPTZ DEFAULT now()
);
CREATE INDEX IF NOT EXISTS idx_mp_page_log_time ON mp_page_log (scraped_at DESC);
-- discgod watchlist: who to auto-crawl, shared by the extension + cron crawler.
CREATE TABLE IF NOT EXISTS discgod_seller_watch (
username TEXT PRIMARY KEY,
seller_id BIGINT,
enabled BOOLEAN DEFAULT TRUE,
note TEXT,
filter_params JSONB,
added_at TIMESTAMPTZ DEFAULT now(),
last_crawled_at TIMESTAMPTZ,
last_listing_count INTEGER,
last_status TEXT
);
CREATE INDEX IF NOT EXISTS idx_discgod_watch_enabled ON discgod_seller_watch (enabled);
"""
def ensure_schema(conn):
with conn.cursor() as cur:
cur.execute(SCHEMA)
cur.execute("ALTER TABLE mp_listing ADD COLUMN IF NOT EXISTS last_crawl_id TEXT")
cur.execute("ALTER TABLE mp_page_log ADD COLUMN IF NOT EXISTS crawl_id TEXT")
conn.commit()
# ===========================================================================
# Coercion helpers
# ===========================================================================
def _num(val):
if val is None or val == "":
return None
try:
return float(val)
except (TypeError, ValueError):
return None
def _int(val):
if val is None or val == "":
return None
try:
return int(val)
except (TypeError, ValueError):
return None
def _txt(val):
if val is None:
return None
s = str(val).strip()
return s if s else None
def _cents(val):
return None if val is None else int(round(val * 100))
def filter_key_from_params(params):
if not params:
return ""
items = sorted((k, str(v)) for k, v in params.items() if k not in NON_FILTER_PARAMS)
return "&".join(f"{k}={v}" for k, v in items)
# ===========================================================================
# Map a raw Discogs API inventory listing -> our flat item dict
# ===========================================================================
def map_listing(raw, seller_username=None, seller_id=None):
"""
Convert one Discogs API listing object (from /users/{u}/inventory) into the
flat shape our upsert expects. Defensive about missing/renamed fields;
returns None if it lacks the two required keys.
"""
item_id = _int(raw.get("id"))
release = raw.get("release") or {}
release_id = _int(release.get("id"))
if not item_id or not release_id:
return None
price = raw.get("price") or {}
# original_price = price converted into the authed account's currency
# (Discogs' confusingly-named field). Good as "converted_value".
converted = raw.get("original_price") or {}
shipping = raw.get("shipping_price") or {}
stats = release.get("stats") or {}
community = stats.get("community") or release.get("community") or {}
seller = raw.get("seller") or {}
sstats = seller.get("stats") or {}
return {
"item_id": item_id,
"release_id": release_id,
"seller_id": _int(seller.get("id")) or _int(seller_id),
"seller_username": _txt(seller.get("username")) or _txt(seller_username),
# Discogs: stats.stars = 0-5 star score (-> rating NUMERIC(3,1));
# stats.rating = percent positive, e.g. 100.0 (-> rating_percent).
"seller_rating": _num(sstats.get("stars")),
"seller_percent": _num(sstats.get("rating")),
"seller_ratings_count": _int(sstats.get("total")),
"title": _txt(release.get("description")),
"label": _txt(release.get("label")),
"cat_no": _txt(release.get("catalog_number") or release.get("catno")),
"media_condition": _txt(raw.get("condition")),
"sleeve_condition": _txt(raw.get("sleeve_condition")),
"notes": _txt(raw.get("comments")),
"price_value": _num(price.get("value")),
"price_currency": _txt(price.get("currency")),
"shipping_value": _num(shipping.get("value")),
"converted_value": _num(converted.get("value")),
"have_count": _int(community.get("in_collection")),
"want_count": _int(community.get("in_wantlist")),
"ships_from": _txt(raw.get("ships_from")),
}
# ===========================================================================
# Upserts (adapted from the marketwatcher engine, mp_* compatible)
# ===========================================================================
def upsert_seller_profile(cur, username, seller_id, profile):
"""Record seller identity + location from the /users/{username} blob.
Reputation (stars/percent/count) comes from each listing's seller.stats,
not from this profile resource (whose `rating` field is ambiguous and can
overflow rating NUMERIC(3,1)); COALESCE preserves it."""
username = _txt(username)
seller_id = _int(seller_id) or _int((profile or {}).get("id"))
if not seller_id or not username:
return 0
profile = profile or {}
cur.execute("""
INSERT INTO mp_seller (seller_id, username, ships_from)
VALUES (%s, %s, %s)
ON CONFLICT (seller_id) DO UPDATE SET
username = EXCLUDED.username,
ships_from = COALESCE(EXCLUDED.ships_from, mp_seller.ships_from),
last_seen = now()
""", (seller_id, username, _txt(profile.get("location"))))
return 1
def upsert_sellers_from_items(cur, items):
"""Upsert sellers gleaned from listing items (when listings carry seller)."""
sellers = {}
for it in items:
sid = _int(it.get("seller_id"))
if not sid or not _txt(it.get("seller_username")):
continue
sellers[sid] = (
sid,
_txt(it.get("seller_username")),
_num(it.get("seller_rating")),
_num(it.get("seller_percent")),
_int(it.get("seller_ratings_count")),
_txt(it.get("ships_from")),
)
if not sellers:
return 0
psycopg2.extras.execute_values(cur, """
INSERT INTO mp_seller (seller_id, username, rating, rating_percent, ratings_count, ships_from)
VALUES %s
ON CONFLICT (seller_id) DO UPDATE SET
username = EXCLUDED.username,
rating = COALESCE(EXCLUDED.rating, mp_seller.rating),
rating_percent = COALESCE(EXCLUDED.rating_percent, mp_seller.rating_percent),
ratings_count = COALESCE(EXCLUDED.ratings_count, mp_seller.ratings_count),
ships_from = COALESCE(EXCLUDED.ships_from, mp_seller.ships_from),
last_seen = now()
""", list(sellers.values()))
return len(sellers)
def upsert_listings(cur, items, crawl_id=None):
"""Upsert listings, emitting events for new / price-changed / condition-
changed / relisted items. Returns (new, price_changes, relisted)."""
by_id = {}
for it in items:
if _int(it.get("item_id")) and _int(it.get("release_id")):
by_id[_int(it["item_id"])] = it
valid = list(by_id.values())
if not valid:
return 0, 0, 0
ids = list(by_id.keys())
cur.execute("""
SELECT item_id, price_value, media_condition, sleeve_condition, status
FROM mp_listing WHERE item_id = ANY(%s)
FOR UPDATE
""", (ids,))
existing = {row[0]: row for row in cur.fetchall()}
rows, events = [], []
new_count = price_changes = relists = 0
for it in valid:
item_id = _int(it["item_id"])
release_id = _int(it["release_id"])
price = _num(it.get("price_value"))
currency = _txt(it.get("price_currency"))
media = _txt(it.get("media_condition"))
sleeve = _txt(it.get("sleeve_condition"))
seller = _txt(it.get("seller_username"))
prev = existing.get(item_id)
if prev is None:
new_count += 1
events.append((item_id, release_id, seller, "listed", None, price, currency,
json.dumps({"media": media, "sleeve": sleeve})))
else:
_, prev_price, prev_media, prev_sleeve, prev_status = prev
prev_price = float(prev_price) if prev_price is not None else None
if prev_status == "gone":
relists += 1
events.append((item_id, release_id, seller, "relisted", prev_price, price, currency, None))
if price is not None and prev_price is not None and _cents(price) != _cents(prev_price):
price_changes += 1
events.append((item_id, release_id, seller, "price_change", prev_price, price, currency, None))
if (media != prev_media and media and prev_media) or \
(sleeve != prev_sleeve and sleeve and prev_sleeve):
events.append((item_id, release_id, seller, "condition_change", None, None, None,
json.dumps({"old_media": prev_media, "new_media": media,
"old_sleeve": prev_sleeve, "new_sleeve": sleeve})))
rows.append((
item_id, release_id, _int(it.get("seller_id")), seller,
_txt(it.get("title")), _txt(it.get("label")), _txt(it.get("cat_no")),
media, sleeve, _txt(it.get("notes")),
price, currency, _num(it.get("shipping_value")), _num(it.get("converted_value")),
_int(it.get("have_count")), _int(it.get("want_count")), _txt(it.get("ships_from")),
crawl_id,
))
psycopg2.extras.execute_values(cur, """
INSERT INTO mp_listing
(item_id, release_id, seller_id, seller_username, title, label, cat_no,
media_condition, sleeve_condition, notes,
price_value, price_currency, shipping_value, converted_value,
have_count, want_count, ships_from, last_crawl_id)
VALUES %s
ON CONFLICT (item_id) DO UPDATE SET
release_id = EXCLUDED.release_id,
seller_id = COALESCE(EXCLUDED.seller_id, mp_listing.seller_id),
seller_username = COALESCE(EXCLUDED.seller_username, mp_listing.seller_username),
title = COALESCE(EXCLUDED.title, mp_listing.title),
label = COALESCE(EXCLUDED.label, mp_listing.label),
cat_no = COALESCE(EXCLUDED.cat_no, mp_listing.cat_no),
media_condition = COALESCE(EXCLUDED.media_condition, mp_listing.media_condition),
sleeve_condition = COALESCE(EXCLUDED.sleeve_condition, mp_listing.sleeve_condition),
notes = COALESCE(EXCLUDED.notes, mp_listing.notes),
price_value = COALESCE(EXCLUDED.price_value, mp_listing.price_value),
price_currency = COALESCE(EXCLUDED.price_currency, mp_listing.price_currency),
shipping_value = COALESCE(EXCLUDED.shipping_value, mp_listing.shipping_value),
converted_value = COALESCE(EXCLUDED.converted_value, mp_listing.converted_value),
have_count = COALESCE(EXCLUDED.have_count, mp_listing.have_count),
want_count = COALESCE(EXCLUDED.want_count, mp_listing.want_count),
ships_from = COALESCE(EXCLUDED.ships_from, mp_listing.ships_from),
last_crawl_id = COALESCE(EXCLUDED.last_crawl_id, mp_listing.last_crawl_id),
status = 'active',
gone_at = NULL,
last_seen = now(),
times_seen = mp_listing.times_seen + 1
""", rows)
if events:
psycopg2.extras.execute_values(cur, """
INSERT INTO mp_listing_event
(item_id, release_id, seller_username, event, old_price, new_price, currency, details)
VALUES %s
""", events)
return new_count, price_changes, relists
SNAPSHOT_DEDUP_HOURS = 6
def record_seller_snapshot(cur, username, seller_id, total, source, params):
"""Insert an inventory-size snapshot unless an identical recent one exists."""
if not username or total is None:
return False
fkey = filter_key_from_params(params)
cur.execute("""
SELECT total_listings FROM mp_seller_snapshot
WHERE username = %s AND source = %s AND filter_key = %s
AND captured_at > now() - make_interval(hours => %s)
ORDER BY captured_at DESC LIMIT 1
""", (username, source, fkey, SNAPSHOT_DEDUP_HOURS))
row = cur.fetchone()
if row and row[0] == total:
return False
cur.execute("""
INSERT INTO mp_seller_snapshot (seller_id, username, total_listings, source, filter_key, filter_params)
VALUES (%s, %s, %s, %s, %s, %s::jsonb)
""", (seller_id, username, total, source, fkey, json.dumps(params or {})))
return True
def mark_gone(cur, username, crawl_id, filtered, treat_complete):
"""After a full inventory crawl, mark this seller's active listings the crawl
did NOT touch (last_crawl_id != crawl_id) as gone — sold or delisted. Only
safe when the crawl was the complete inventory. Returns count, or None when
skipped."""
username = _txt(username)
crawl_id = _txt(crawl_id)
if not username or not crawl_id:
return None
if filtered and not treat_complete:
return None
cur.execute("""
UPDATE mp_listing
SET status = 'gone', gone_at = now()
WHERE seller_username = %s AND status = 'active'
AND last_crawl_id IS DISTINCT FROM %s
RETURNING item_id, release_id, price_value, price_currency
""", (username, crawl_id))
gone = cur.fetchall()
if gone:
psycopg2.extras.execute_values(cur, """
INSERT INTO mp_listing_event
(item_id, release_id, seller_username, event, old_price, new_price, currency, details)
VALUES %s
""", [
(item_id, release_id, username, "gone", price, None, currency,
json.dumps({"presumed": "sold_or_delisted", "crawl_filtered": filtered}))
for item_id, release_id, price, currency in gone
])
return len(gone)
# ===========================================================================
# High-level operations (own their own transaction unless a conn is passed)
# ===========================================================================
def ingest_inventory_page(payload, conn=None):
"""Process one page of a seller's inventory.
payload = {username, seller_id?, seller?, page, total?, crawl_id, filter?,
listings:[raw API objects]}"""
owns = conn is None
conn = conn or get_conn()
try:
username = _txt(payload.get("username"))
seller_id = _int(payload.get("seller_id"))
profile = payload.get("seller")
page = _int(payload.get("page")) or 1
total = _int(payload.get("total"))
crawl_id = _txt(payload.get("crawl_id"))
params = payload.get("filter") or {}
raw_listings = payload.get("listings") or []
items = [m for m in (map_listing(r, username, seller_id) for r in raw_listings) if m]
with conn.cursor() as cur:
sellers = 0
if profile or seller_id:
sellers += upsert_seller_profile(cur, username, seller_id, profile)
sellers += upsert_sellers_from_items(cur, items)
new_count, price_changes, relists = upsert_listings(cur, items, crawl_id)
snapshots = 0
if page == 1 and total is not None:
if record_seller_snapshot(cur, username, seller_id, total, "api_inventory", params):
snapshots += 1
item_ids = [it["item_id"] for it in items]
cur.execute("""
INSERT INTO mp_page_log
(url, page_type, seller_page, params, page, total_results,
item_count, item_ids, new_items, price_changes, crawl_id)
VALUES (%s, %s, %s, %s::jsonb, %s, %s, %s, %s, %s, %s, %s)
""", (
f"api:/users/{username}/inventory?page={page}",
"api_inventory", username, json.dumps(params), page, total,
len(items), item_ids, new_count, price_changes, crawl_id,
))
conn.commit()
return {"ok": True, "username": username, "page": page,
"items": len(items), "new": new_count,
"price_changes": price_changes, "relisted": relists,
"sellers": sellers, "snapshots": snapshots}
finally:
if owns:
conn.close()
def finish_crawl(payload, conn=None):
"""Close out a full inventory crawl: mark untouched listings gone.
payload = {username, crawl_id, filter?, treat_complete?}"""
owns = conn is None
conn = conn or get_conn()
try:
username = _txt(payload.get("username"))
crawl_id = _txt(payload.get("crawl_id"))
params = payload.get("filter") or {}
filtered = bool(filter_key_from_params(params))
treat_complete = bool(payload.get("treat_complete"))
if not username or not crawl_id:
return {"ok": False, "error": "username and crawl_id required"}
with conn.cursor() as cur:
marked = mark_gone(cur, username, crawl_id, filtered, treat_complete)
conn.commit()
if marked is None:
return {"ok": True, "marked_gone": 0,
"skipped": "filtered crawl not treated as complete inventory"}
return {"ok": True, "marked_gone": marked}
finally:
if owns:
conn.close()
# ---- watchlist ------------------------------------------------------------
def list_tracked(conn=None, enabled_only=False):
owns = conn is None
conn = conn or get_conn()
try:
with conn.cursor(cursor_factory=psycopg2.extras.RealDictCursor) as cur:
cur.execute(f"""
SELECT w.username, w.seller_id, w.enabled, w.note, w.filter_params,
w.added_at, w.last_crawled_at, w.last_listing_count, w.last_status,
(SELECT count(*) FROM mp_listing l
WHERE l.seller_username = w.username AND l.status = 'active') AS active_listings
FROM discgod_seller_watch w
{"WHERE w.enabled" if enabled_only else ""}
ORDER BY w.last_crawled_at DESC NULLS LAST, w.username
""")
rows = cur.fetchall()
for r in rows:
for k in ("added_at", "last_crawled_at"):
if r.get(k):
r[k] = r[k].isoformat()
return rows
finally:
if owns:
conn.close()
def track_seller(username, note=None, filter_params=None, enabled=True, seller_id=None, conn=None):
owns = conn is None
conn = conn or get_conn()
try:
username = _txt(username)
if not username:
return {"ok": False, "error": "username required"}
with conn.cursor() as cur:
cur.execute("""
INSERT INTO discgod_seller_watch (username, seller_id, enabled, note, filter_params)
VALUES (%s, %s, %s, %s, %s::jsonb)
ON CONFLICT (username) DO UPDATE SET
enabled = EXCLUDED.enabled,
note = COALESCE(EXCLUDED.note, discgod_seller_watch.note),
filter_params = COALESCE(EXCLUDED.filter_params, discgod_seller_watch.filter_params),
seller_id = COALESCE(EXCLUDED.seller_id, discgod_seller_watch.seller_id)
""", (username, _int(seller_id), bool(enabled), _txt(note),
json.dumps(filter_params) if filter_params else None))
conn.commit()
return {"ok": True, "username": username}
finally:
if owns:
conn.close()
def untrack_seller(username, conn=None):
owns = conn is None
conn = conn or get_conn()
try:
with conn.cursor() as cur:
cur.execute("DELETE FROM discgod_seller_watch WHERE username = %s", (_txt(username),))
n = cur.rowcount
conn.commit()
return {"ok": True, "removed": n}
finally:
if owns:
conn.close()
def mark_watch_crawled(username, listing_count, status="ok", conn=None):
owns = conn is None
conn = conn or get_conn()
try:
with conn.cursor() as cur:
cur.execute("""
UPDATE discgod_seller_watch
SET last_crawled_at = now(), last_listing_count = %s, last_status = %s
WHERE username = %s
""", (_int(listing_count), _txt(status), _txt(username)))
conn.commit()
finally:
if owns:
conn.close()
# ---- dashboard stats ------------------------------------------------------
def stats(conn=None):
owns = conn is None
conn = conn or get_conn()
try:
with conn.cursor() as cur:
cur.execute("""
SELECT
(SELECT count(*) FROM mp_listing WHERE status = 'active'),
(SELECT count(*) FROM mp_listing WHERE status = 'gone'),
(SELECT count(*) FROM discgod_seller_watch WHERE enabled),
(SELECT count(*) FROM mp_seller),
(SELECT count(*) FROM mp_page_log
WHERE page_type = 'api_inventory' AND scraped_at > now() - interval '24 hours'),
(SELECT max(scraped_at) FROM mp_page_log),
(SELECT count(*) FROM mp_listing_event
WHERE event = 'gone' AND recorded_at > now() - interval '7 days'),
(SELECT count(*) FROM mp_listing_event
WHERE event = 'price_change' AND recorded_at > now() - interval '7 days')
""")
(active, gone, watched, sellers, pages_24h, last_capture,
gone_7d, price_changes_7d) = cur.fetchone()
return {"ok": True, "active_listings": active, "gone_listings": gone,
"sellers_watched": watched, "sellers_known": sellers,
"pages_24h": pages_24h,
"last_capture": last_capture.isoformat() if last_capture else None,
"gone_7d": gone_7d, "price_changes_7d": price_changes_7d}
finally:
if owns:
conn.close()
# ===========================================================================
# HTTP server
# ===========================================================================
def _log(msg):
print(f"[{datetime.now().strftime('%H:%M:%S')}] {msg}")
class Handler(BaseHTTPRequestHandler):
def log_message(self, *_):
pass
def _cors(self):
self.send_header("Access-Control-Allow-Origin", "*")
self.send_header("Access-Control-Allow-Headers", "Content-Type")
self.send_header("Access-Control-Allow-Methods", "POST, GET, OPTIONS")
self.send_header("Access-Control-Allow-Private-Network", "true")
def _send(self, obj, status=200):
body = json.dumps(obj).encode()
self.send_response(status)
self._cors()
self.send_header("Content-Type", "application/json")
self.send_header("Content-Length", str(len(body)))
self.end_headers()
self.wfile.write(body)
def _read_json(self):
length = int(self.headers.get("Content-Length", 0))
body = self.rfile.read(length) if length else b""
return json.loads(body) if body else {}
def _safe(self, fn, label):
try:
self._send(fn())
except Exception as e:
import traceback
traceback.print_exc()
_log(f"ERROR ({label}): {e}")
self._send({"ok": False, "error": str(e)}, status=500)
def do_OPTIONS(self):
self.send_response(200)
self._cors()
self.end_headers()
def do_GET(self):
if self.path in ("/discgod/health", "/discgod", "/"):
self._send({"ok": True, "server": "discgod", "db": DB_NAME})
elif self.path == "/discgod/stats":
self._safe(stats, "stats")
elif self.path == "/discgod/tracked":
self._safe(lambda: {"ok": True, "sellers": list_tracked()}, "tracked")
else:
self.send_response(404)
self.end_headers()
def do_POST(self):
if self.path == "/discgod/ingest":
def run():
res = ingest_inventory_page(self._read_json())
_log(f"[ingest] {res.get('username')} p{res.get('page')} "
f"{res.get('items')} items ({res.get('new')} new, "
f"{res.get('price_changes')} price changes, {res.get('snapshots')} snap)")
return res
self._safe(run, "ingest")
elif self.path == "/discgod/crawl_complete":
def run():
payload = self._read_json()
res = finish_crawl(payload)
if res.get("ok") and payload.get("username"):
mark_watch_crawled(payload["username"], payload.get("listing_count"), "ok")
_log(f"[crawl_complete] {payload.get('username')} -> "
f"{res.get('marked_gone', 0)} gone {res.get('skipped','')}")
return res
self._safe(run, "crawl_complete")
elif self.path == "/discgod/track":
def run():
p = self._read_json()
return track_seller(p.get("username"), p.get("note"), p.get("filter"),
p.get("enabled", True), p.get("seller_id"))
self._safe(run, "track")
elif self.path == "/discgod/untrack":
def run():
return untrack_seller(self._read_json().get("username"))
self._safe(run, "untrack")
else:
self.send_response(404)
self.end_headers()
def main():
conn = get_conn()
ensure_schema(conn)
conn.close()
_log(f"discgod server listening on 0.0.0.0:{PORT} (db={DB_NAME})")
ThreadingHTTPServer(("0.0.0.0", PORT), Handler).serve_forever()
if __name__ == "__main__":
main()
Reference in New Issue View Git Blame Copy Permalink