Circuit-Forge/snipe
2
0
Fork 0
Code Issues 13 Pull requests Projects Releases Packages Wiki Activity Actions

feat: add data models, migrations, and store

Browse source
This commit is contained in:
pyr0ball 2026-03-25 12:39:39 -07:00
parent ac114da5e7
commit 675146ff1a
6 changed files with 335 additions and 0 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

0
app/db/__init__.py Normal file
View file

76
app/db/migrations/001_init.sql Normal file
View file

@ -0,0 +1,76 @@
CREATE TABLE IF NOT EXISTS sellers (
id INTEGER PRIMARY KEY AUTOINCREMENT,
platform TEXT NOT NULL,
platform_seller_id TEXT NOT NULL,
username TEXT NOT NULL,
account_age_days INTEGER NOT NULL,
feedback_count INTEGER NOT NULL,
feedback_ratio REAL NOT NULL,
category_history_json TEXT NOT NULL DEFAULT '{}',
fetched_at TEXT DEFAULT CURRENT_TIMESTAMP,
UNIQUE(platform, platform_seller_id)
);
CREATE TABLE IF NOT EXISTS listings (
id INTEGER PRIMARY KEY AUTOINCREMENT,
platform TEXT NOT NULL,
platform_listing_id TEXT NOT NULL,
title TEXT NOT NULL,
price REAL NOT NULL,
currency TEXT NOT NULL DEFAULT 'USD',
condition TEXT,
seller_platform_id TEXT,
url TEXT,
photo_urls TEXT NOT NULL DEFAULT '[]',
listing_age_days INTEGER DEFAULT 0,
fetched_at TEXT DEFAULT CURRENT_TIMESTAMP,
trust_score_id INTEGER REFERENCES trust_scores(id),
UNIQUE(platform, platform_listing_id)
);
CREATE TABLE IF NOT EXISTS trust_scores (
id INTEGER PRIMARY KEY AUTOINCREMENT,
listing_id INTEGER NOT NULL REFERENCES listings(id),
composite_score INTEGER NOT NULL,
account_age_score INTEGER NOT NULL DEFAULT 0,
feedback_count_score INTEGER NOT NULL DEFAULT 0,
feedback_ratio_score INTEGER NOT NULL DEFAULT 0,
price_vs_market_score INTEGER NOT NULL DEFAULT 0,
category_history_score INTEGER NOT NULL DEFAULT 0,
photo_hash_duplicate INTEGER NOT NULL DEFAULT 0,
photo_analysis_json TEXT,
red_flags_json TEXT NOT NULL DEFAULT '[]',
score_is_partial INTEGER NOT NULL DEFAULT 0,
scored_at TEXT DEFAULT CURRENT_TIMESTAMP
);
CREATE TABLE IF NOT EXISTS market_comps (
id INTEGER PRIMARY KEY AUTOINCREMENT,
platform TEXT NOT NULL,
query_hash TEXT NOT NULL,
median_price REAL NOT NULL,
sample_count INTEGER NOT NULL,
fetched_at TEXT DEFAULT CURRENT_TIMESTAMP,
expires_at TEXT NOT NULL,
UNIQUE(platform, query_hash)
);
CREATE TABLE IF NOT EXISTS saved_searches (
id INTEGER PRIMARY KEY AUTOINCREMENT,
name TEXT NOT NULL,
query TEXT NOT NULL,
platform TEXT NOT NULL DEFAULT 'ebay',
filters_json TEXT NOT NULL DEFAULT '{}',
created_at TEXT DEFAULT CURRENT_TIMESTAMP,
last_run_at TEXT
);
-- PhotoHash: perceptual hash store for cross-search dedup (v0.2+). Schema present in v0.1.
CREATE TABLE IF NOT EXISTS photo_hashes (
id INTEGER PRIMARY KEY AUTOINCREMENT,
listing_id INTEGER NOT NULL REFERENCES listings(id),
photo_url TEXT NOT NULL,
phash TEXT NOT NULL,
first_seen_at TEXT DEFAULT CURRENT_TIMESTAMP,
UNIQUE(listing_id, photo_url)
);

84
app/db/models.py Normal file
View file

@ -0,0 +1,84 @@
"""Dataclasses for all Snipe domain objects."""
from __future__ import annotations
from dataclasses import dataclass, field
from typing import Optional
@dataclass
class Seller:
platform: str
platform_seller_id: str
username: str
account_age_days: int
feedback_count: int
feedback_ratio: float # 0.0–1.0
category_history_json: str # JSON blob of past category sales
id: Optional[int] = None
fetched_at: Optional[str] = None
@dataclass
class Listing:
platform: str
platform_listing_id: str
title: str
price: float
currency: str
condition: str
seller_platform_id: str
url: str
photo_urls: list[str] = field(default_factory=list)
listing_age_days: int = 0
id: Optional[int] = None
fetched_at: Optional[str] = None
trust_score_id: Optional[int] = None
@dataclass
class TrustScore:
listing_id: int
composite_score: int # 0–100
account_age_score: int # 0–20
feedback_count_score: int # 0–20
feedback_ratio_score: int # 0–20
price_vs_market_score: int # 0–20
category_history_score: int # 0–20
photo_hash_duplicate: bool = False
photo_analysis_json: Optional[str] = None
red_flags_json: str = "[]"
score_is_partial: bool = False
id: Optional[int] = None
scored_at: Optional[str] = None
@dataclass
class MarketComp:
platform: str
query_hash: str
median_price: float
sample_count: int
expires_at: str # ISO8601 — checked against current time
id: Optional[int] = None
fetched_at: Optional[str] = None
@dataclass
class SavedSearch:
"""Schema scaffolded in v0.1; background monitoring wired in v0.2."""
name: str
query: str
platform: str
filters_json: str = "{}"
id: Optional[int] = None
created_at: Optional[str] = None
last_run_at: Optional[str] = None
@dataclass
class PhotoHash:
"""Perceptual hash store for cross-search dedup (v0.2+). Schema scaffolded in v0.1."""
listing_id: int
photo_url: str
phash: str # hex string from imagehash
id: Optional[int] = None
first_seen_at: Optional[str] = None

97
app/db/store.py Normal file
View file

@ -0,0 +1,97 @@
"""Thin SQLite read/write layer for all Snipe models."""
from __future__ import annotations
import json
from datetime import datetime, timezone
from pathlib import Path
from typing import Optional
from circuitforge_core.db import get_connection, run_migrations
from .models import Listing, Seller, TrustScore, MarketComp
MIGRATIONS_DIR = Path(__file__).parent / "migrations"
class Store:
def __init__(self, db_path: Path):
self._conn = get_connection(db_path)
run_migrations(self._conn, MIGRATIONS_DIR)
# --- Seller ---
def save_seller(self, seller: Seller) -> None:
self._conn.execute(
"INSERT OR REPLACE INTO sellers "
"(platform, platform_seller_id, username, account_age_days, "
"feedback_count, feedback_ratio, category_history_json) "
"VALUES (?,?,?,?,?,?,?)",
(seller.platform, seller.platform_seller_id, seller.username,
seller.account_age_days, seller.feedback_count, seller.feedback_ratio,
seller.category_history_json),
)
self._conn.commit()
def get_seller(self, platform: str, platform_seller_id: str) -> Optional[Seller]:
row = self._conn.execute(
"SELECT platform, platform_seller_id, username, account_age_days, "
"feedback_count, feedback_ratio, category_history_json, id, fetched_at "
"FROM sellers WHERE platform=? AND platform_seller_id=?",
(platform, platform_seller_id),
).fetchone()
if not row:
return None
return Seller(*row[:7], id=row[7], fetched_at=row[8])
# --- Listing ---
def save_listing(self, listing: Listing) -> None:
self._conn.execute(
"INSERT OR REPLACE INTO listings "
"(platform, platform_listing_id, title, price, currency, condition, "
"seller_platform_id, url, photo_urls, listing_age_days) "
"VALUES (?,?,?,?,?,?,?,?,?,?)",
(listing.platform, listing.platform_listing_id, listing.title,
listing.price, listing.currency, listing.condition,
listing.seller_platform_id, listing.url,
json.dumps(listing.photo_urls), listing.listing_age_days),
)
self._conn.commit()
def get_listing(self, platform: str, platform_listing_id: str) -> Optional[Listing]:
row = self._conn.execute(
"SELECT platform, platform_listing_id, title, price, currency, condition, "
"seller_platform_id, url, photo_urls, listing_age_days, id, fetched_at "
"FROM listings WHERE platform=? AND platform_listing_id=?",
(platform, platform_listing_id),
).fetchone()
if not row:
return None
return Listing(
*row[:8],
photo_urls=json.loads(row[8]),
listing_age_days=row[9],
id=row[10],
fetched_at=row[11],
)
# --- MarketComp ---
def save_market_comp(self, comp: MarketComp) -> None:
self._conn.execute(
"INSERT OR REPLACE INTO market_comps "
"(platform, query_hash, median_price, sample_count, expires_at) "
"VALUES (?,?,?,?,?)",
(comp.platform, comp.query_hash, comp.median_price,
comp.sample_count, comp.expires_at),
)
self._conn.commit()
def get_market_comp(self, platform: str, query_hash: str) -> Optional[MarketComp]:
row = self._conn.execute(
"SELECT platform, query_hash, median_price, sample_count, expires_at, id, fetched_at "
"FROM market_comps WHERE platform=? AND query_hash=? AND expires_at > ?",
(platform, query_hash, datetime.now(timezone.utc).isoformat()),
).fetchone()
if not row:
return None
return MarketComp(*row[:5], id=row[5], fetched_at=row[6])

0
tests/db/__init__.py Normal file
View file

78
tests/db/test_store.py Normal file
View file

@ -0,0 +1,78 @@
import pytest
from pathlib import Path
from app.db.store import Store
from app.db.models import Listing, Seller, TrustScore, MarketComp
@pytest.fixture
def store(tmp_path):
return Store(tmp_path / "test.db")
def test_store_creates_tables(store):
# If no exception on init, tables exist
pass
def test_save_and_get_seller(store):
seller = Seller(
platform="ebay",
platform_seller_id="user123",
username="techseller",
account_age_days=730,
feedback_count=450,
feedback_ratio=0.991,
category_history_json="{}",
)
store.save_seller(seller)
result = store.get_seller("ebay", "user123")
assert result is not None
assert result.username == "techseller"
assert result.feedback_count == 450
def test_save_and_get_listing(store):
listing = Listing(
platform="ebay",
platform_listing_id="ebay-123",
title="RTX 4090 FE",
price=950.00,
currency="USD",
condition="used",
seller_platform_id="user123",
url="https://ebay.com/itm/123",
photo_urls=["https://i.ebayimg.com/1.jpg"],
listing_age_days=3,
)
store.save_listing(listing)
result = store.get_listing("ebay", "ebay-123")
assert result is not None
assert result.title == "RTX 4090 FE"
assert result.price == 950.00
def test_save_and_get_market_comp(store):
comp = MarketComp(
platform="ebay",
query_hash="abc123",
median_price=1050.0,
sample_count=12,
expires_at="2026-03-26T00:00:00",
)
store.save_market_comp(comp)
result = store.get_market_comp("ebay", "abc123")
assert result is not None
assert result.median_price == 1050.0
def test_get_market_comp_returns_none_for_expired(store):
comp = MarketComp(
platform="ebay",
query_hash="expired",
median_price=900.0,
sample_count=5,
expires_at="2020-01-01T00:00:00", # past
)
store.save_market_comp(comp)
result = store.get_market_comp("ebay", "expired")
assert result is None