feat(snipe): auction support + easter eggs (Konami, The Steal, de-emphasis)

Auction metadata:
- Listing model gains buying_format + ends_at fields
- Migration 002 adds columns to existing databases
- scraper.py: parse s-item__time-left → absolute ends_at ISO timestamp
- normaliser.py: extract buyingOptions + itemEndDate from Browse API
- store.py: save/get updated for new fields

Easter eggs (app/ui/components/easter_eggs.py):
- Konami code detector (JS → URL param → Streamlit rerun)
- Web Audio API snipe call synthesis, gated behind sidebar checkbox
  (disabled by default for safety/accessibility)
- "The Steal" gold shimmer: trust ≥ 90, price 15–30% below market,
  no suspicious_price flag
- Auction de-emphasis: soft caption when > 1h remaining

UI updates:
- listing_row: steal banner + auction notice per row
- Search: inject CSS, check snipe mode, "Ending soon" sort option,
  pass market_price from comp cache to row renderer
- app.py: Konami detector + audio enable/disable sidebar toggle

Tests: 22 new tests (72 total, all green)
This commit is contained in:
pyr0ball 2026-03-25 14:27:02 -07:00
parent 68a9879191
commit 6ec0f957b9
11 changed files with 521 additions and 16 deletions

View file

@ -15,5 +15,16 @@ if not wizard.is_configured():
wizard.run()
st.stop()
from app.ui.components.easter_eggs import inject_konami_detector
inject_konami_detector()
with st.sidebar:
st.divider()
audio_enabled = st.checkbox(
"🔊 Enable audio easter egg",
value=False,
help="Plays a synthesised sound on Konami code activation. Off by default.",
)
from app.ui.Search import render
render()
render(audio_enabled=audio_enabled)

View file

@ -0,0 +1,3 @@
-- Add auction metadata to listings (v0.1.1)
ALTER TABLE listings ADD COLUMN buying_format TEXT NOT NULL DEFAULT 'fixed_price';
ALTER TABLE listings ADD COLUMN ends_at TEXT;

View file

@ -29,6 +29,8 @@ class Listing:
url: str
photo_urls: list[str] = field(default_factory=list)
listing_age_days: int = 0
buying_format: str = "fixed_price" # "fixed_price", "auction", "best_offer"
ends_at: Optional[str] = None # ISO8601 auction end time; None for fixed-price
id: Optional[int] = None
fetched_at: Optional[str] = None
trust_score_id: Optional[int] = None

View file

@ -48,19 +48,21 @@ class Store:
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 (?,?,?,?,?,?,?,?,?,?)",
"seller_platform_id, url, photo_urls, listing_age_days, buying_format, ends_at) "
"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),
json.dumps(listing.photo_urls), listing.listing_age_days,
listing.buying_format, listing.ends_at),
)
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 "
"seller_platform_id, url, photo_urls, listing_age_days, id, fetched_at, "
"buying_format, ends_at "
"FROM listings WHERE platform=? AND platform_listing_id=?",
(platform, platform_listing_id),
).fetchone()
@ -72,6 +74,8 @@ class Store:
listing_age_days=row[9],
id=row[10],
fetched_at=row[11],
buying_format=row[12] or "fixed_price",
ends_at=row[13],
)
# --- MarketComp ---

View file

@ -25,6 +25,22 @@ def normalise_listing(raw: dict) -> Listing:
except ValueError:
pass
options = raw.get("buyingOptions", [])
if "AUCTION" in options:
buying_format = "auction"
elif "BEST_OFFER" in options:
buying_format = "best_offer"
else:
buying_format = "fixed_price"
ends_at = None
end_raw = raw.get("itemEndDate", "")
if end_raw:
try:
ends_at = datetime.fromisoformat(end_raw.replace("Z", "+00:00")).isoformat()
except ValueError:
pass
seller = raw.get("seller", {})
return Listing(
platform="ebay",
@ -37,6 +53,8 @@ def normalise_listing(raw: dict) -> Listing:
url=raw.get("itemWebUrl", ""),
photo_urls=photos,
listing_age_days=listing_age_days,
buying_format=buying_format,
ends_at=ends_at,
)

View file

@ -41,6 +41,7 @@ _HEADERS = {
_SELLER_RE = re.compile(r"^(.+?)\s+\(([0-9,]+)\)\s+([\d.]+)%")
_PRICE_RE = re.compile(r"[\d,]+\.?\d*")
_ITEM_ID_RE = re.compile(r"/itm/(\d+)")
_TIME_LEFT_RE = re.compile(r"(?:(\d+)d\s*)?(?:(\d+)h\s*)?(?:(\d+)m\s*)?(?:(\d+)s\s*)?left", re.I)
# ---------------------------------------------------------------------------
@ -71,6 +72,26 @@ def _parse_seller(text: str) -> tuple[str, int, float]:
return m.group(1).strip(), int(m.group(2).replace(",", "")), float(m.group(3)) / 100.0
def _parse_time_left(text: str) -> Optional[timedelta]:
"""Parse eBay time-left text into a timedelta.
Handles '3d 14h left', '14h 23m left', '23m 45s left'.
Returns None if text doesn't match (i.e. fixed-price listing).
"""
if not text:
return None
m = _TIME_LEFT_RE.search(text)
if not m or not any(m.groups()):
return None
days = int(m.group(1) or 0)
hours = int(m.group(2) or 0)
minutes = int(m.group(3) or 0)
seconds = int(m.group(4) or 0)
if days == hours == minutes == seconds == 0:
return None
return timedelta(days=days, hours=hours, minutes=minutes, seconds=seconds)
def scrape_listings(html: str) -> list[Listing]:
"""Parse eBay search results HTML into Listing objects."""
soup = BeautifulSoup(html, "lxml")
@ -104,6 +125,14 @@ def scrape_listings(html: str) -> list[Listing]:
if img_el:
photo_url = img_el.get("data-src") or img_el.get("src") or ""
# Auction detection: presence of s-item__time-left means auction format
time_el = item.select_one("span.s-item__time-left")
time_remaining = _parse_time_left(time_el.text) if time_el else None
buying_format = "auction" if time_remaining is not None else "fixed_price"
ends_at = None
if time_remaining is not None:
ends_at = (datetime.now(timezone.utc) + time_remaining).isoformat()
results.append(Listing(
platform="ebay",
platform_listing_id=platform_listing_id,
@ -115,6 +144,8 @@ def scrape_listings(html: str) -> list[Listing]:
url=url,
photo_urls=[photo_url] if photo_url else [],
listing_age_days=0, # not reliably in search HTML
buying_format=buying_format,
ends_at=ends_at,
))
return results

View file

@ -10,6 +10,10 @@ from app.platforms import PlatformAdapter, SearchFilters
from app.trust import TrustScorer
from app.ui.components.filters import build_filter_options, render_filter_sidebar, FilterState
from app.ui.components.listing_row import render_listing_row
from app.ui.components.easter_eggs import (
inject_steal_css, check_snipe_mode, render_snipe_mode_banner,
auction_hours_remaining,
)
log = logging.getLogger(__name__)
@ -72,7 +76,12 @@ def _passes_filter(listing, trust, seller, state: FilterState) -> bool:
return True
def render() -> None:
def render(audio_enabled: bool = False) -> None:
inject_steal_css()
if check_snipe_mode():
render_snipe_mode_banner(audio_enabled)
st.title("🔍 Snipe — eBay Listing Search")
col_q, col_price, col_btn = st.columns([4, 2, 1])
@ -115,14 +124,21 @@ def render() -> None:
opts = build_filter_options(pairs)
filter_state = render_filter_sidebar(pairs, opts)
sort_col = st.selectbox("Sort by", ["Trust score", "Price ↑", "Price ↓", "Newest"],
label_visibility="collapsed")
sort_col = st.selectbox(
"Sort by",
["Trust score", "Price ↑", "Price ↓", "Newest", "Ending soon"],
label_visibility="collapsed",
)
def sort_key(pair):
l, t = pair
if sort_col == "Trust score": return -(t.composite_score if t else 0)
if sort_col == "Price ↑": return l.price
if sort_col == "Price ↓": return -l.price
if sort_col == "Trust score": return -(t.composite_score if t else 0)
if sort_col == "Price ↑": return l.price
if sort_col == "Price ↓": return -l.price
if sort_col == "Ending soon":
h = auction_hours_remaining(l)
# Non-auctions sort to end; auctions sort ascending by time left
return (h if h is not None else float("inf"))
return l.listing_age_days
sorted_pairs = sorted(pairs, key=sort_key)
@ -132,9 +148,14 @@ def render() -> None:
st.caption(f"{len(visible)} results · {hidden_count} hidden by filters")
import hashlib
query_hash = hashlib.md5(query.encode()).hexdigest()
comp = store.get_market_comp("ebay", query_hash)
market_price = comp.median_price if comp else None
for listing, trust in visible:
seller = store.get_seller("ebay", listing.seller_platform_id)
render_listing_row(listing, trust, seller)
render_listing_row(listing, trust, seller, market_price=market_price)
if hidden_count:
if st.button(f"Show {hidden_count} hidden results"):
@ -142,4 +163,4 @@ def render() -> None:
for listing, trust in sorted_pairs:
if (listing.platform, listing.platform_listing_id) not in visible_ids:
seller = store.get_seller("ebay", listing.seller_platform_id)
render_listing_row(listing, trust, seller)
render_listing_row(listing, trust, seller, market_price=market_price)

View file

@ -0,0 +1,219 @@
"""Easter egg features for Snipe.
Three features:
1. Konami code Snipe Mode JS detector sets ?snipe_mode=1 URL param,
Streamlit detects it on rerun. Audio is synthesised client-side via Web
Audio API (no bundled file; local-first friendly). Disabled by default
for accessibility / autoplay-policy reasons; requires explicit sidebar opt-in.
2. The Steal shimmer a listing with trust 90, price 1530 % below market,
and no suspicious_price flag gets a gold shimmer banner.
3. Auction de-emphasis auctions with > 1 h remaining show a soft notice
because live prices are misleading until the final minutes.
"""
from __future__ import annotations
import json
from datetime import datetime, timezone
from typing import Optional
import streamlit as st
from app.db.models import Listing, TrustScore
# ---------------------------------------------------------------------------
# 1. Konami → Snipe Mode
# ---------------------------------------------------------------------------
_KONAMI_JS = """
<script>
(function () {
const SEQ = [38,38,40,40,37,39,37,39,66,65];
let idx = 0;
document.addEventListener('keydown', function (e) {
if (e.keyCode === SEQ[idx]) {
idx++;
if (idx === SEQ.length) {
idx = 0;
const url = new URL(window.location.href);
url.searchParams.set('snipe_mode', '1');
window.location.href = url.toString();
}
} else {
idx = (e.keyCode === SEQ[0]) ? 1 : 0;
}
});
})();
</script>
"""
_SNIPE_AUDIO_JS = """
<script>
(function () {
if (window.__snipeAudioPlayed) return;
window.__snipeAudioPlayed = true;
try {
const ctx = new (window.AudioContext || window.webkitAudioContext)();
// Short "sniper scope click" high sine blip followed by a lower resonant hit
function blip(freq, start, dur, gain) {
const osc = ctx.createOscillator();
const env = ctx.createGain();
osc.connect(env); env.connect(ctx.destination);
osc.type = 'sine'; osc.frequency.setValueAtTime(freq, ctx.currentTime + start);
env.gain.setValueAtTime(0, ctx.currentTime + start);
env.gain.linearRampToValueAtTime(gain, ctx.currentTime + start + 0.01);
env.gain.exponentialRampToValueAtTime(0.0001, ctx.currentTime + start + dur);
osc.start(ctx.currentTime + start);
osc.stop(ctx.currentTime + start + dur + 0.05);
}
blip(880, 0.00, 0.08, 0.3);
blip(440, 0.10, 0.15, 0.2);
blip(220, 0.20, 0.25, 0.15);
} catch (e) { /* AudioContext blocked silent fail */ }
})();
</script>
"""
_SNIPE_BANNER_CSS = """
<style>
@keyframes snipe-scan {
0% { background-position: -200% center; }
100% { background-position: 200% center; }
}
.snipe-mode-banner {
background: linear-gradient(
90deg,
#0d1117 0%, #0d1117 40%,
#39ff14 50%,
#0d1117 60%, #0d1117 100%
);
background-size: 200% auto;
animation: snipe-scan 2s linear infinite;
color: #39ff14;
font-family: monospace;
font-size: 13px;
letter-spacing: 0.15em;
padding: 6px 16px;
border-radius: 4px;
margin-bottom: 10px;
text-align: center;
text-shadow: 0 0 8px #39ff14;
}
</style>
"""
def inject_konami_detector() -> None:
"""Inject the JS Konami sequence detector into the page (once per load)."""
st.components.v1.html(_KONAMI_JS, height=0)
def check_snipe_mode() -> bool:
"""Return True if ?snipe_mode=1 is present in the URL query params."""
return st.query_params.get("snipe_mode", "") == "1"
def render_snipe_mode_banner(audio_enabled: bool) -> None:
"""Render the Snipe Mode activation banner and optionally play the audio cue."""
st.markdown(_SNIPE_BANNER_CSS, unsafe_allow_html=True)
st.markdown(
'<div class="snipe-mode-banner">🎯 SNIPE MODE ACTIVATED — TARGET ACQUIRED</div>',
unsafe_allow_html=True,
)
if audio_enabled:
st.components.v1.html(_SNIPE_AUDIO_JS, height=0)
# ---------------------------------------------------------------------------
# 2. The Steal shimmer
# ---------------------------------------------------------------------------
_STEAL_CSS = """
<style>
@keyframes steal-glow {
0% { box-shadow: 0 0 6px 1px rgba(255,215,0,0.5); }
50% { box-shadow: 0 0 18px 4px rgba(255,215,0,0.9); }
100% { box-shadow: 0 0 6px 1px rgba(255,215,0,0.5); }
}
.steal-banner {
background: linear-gradient(
90deg,
transparent 0%,
rgba(255,215,0,0.12) 30%,
rgba(255,215,0,0.35) 50%,
rgba(255,215,0,0.12) 70%,
transparent 100%
);
border: 1px solid rgba(255,215,0,0.6);
animation: steal-glow 2.2s ease-in-out infinite;
border-radius: 6px;
padding: 4px 12px;
font-size: 12px;
color: #ffd700;
font-weight: 600;
margin-bottom: 6px;
letter-spacing: 0.05em;
}
</style>
"""
def inject_steal_css() -> None:
"""Inject the steal-shimmer CSS (idempotent — Streamlit deduplicates)."""
st.markdown(_STEAL_CSS, unsafe_allow_html=True)
def is_steal(listing: Listing, trust: Optional[TrustScore], market_price: Optional[float]) -> bool:
"""Return True when this listing qualifies as 'The Steal'.
Criteria (all must hold):
- trust composite 90
- no suspicious_price flag
- price is 1530 % below the market median
(deeper discounts are suspicious, not steals)
"""
if trust is None or market_price is None or market_price <= 0:
return False
if trust.composite_score < 90:
return False
flags = json.loads(trust.red_flags_json or "[]")
if "suspicious_price" in flags:
return False
discount = (market_price - listing.price) / market_price
return 0.15 <= discount <= 0.30
def render_steal_banner() -> None:
"""Render the gold shimmer steal banner above a listing row."""
st.markdown(
'<div class="steal-banner">✦ THE STEAL — significantly below market, high trust</div>',
unsafe_allow_html=True,
)
# ---------------------------------------------------------------------------
# 3. Auction de-emphasis
# ---------------------------------------------------------------------------
def auction_hours_remaining(listing: Listing) -> Optional[float]:
"""Return hours remaining for an auction listing, or None for fixed-price / no data."""
if listing.buying_format != "auction" or not listing.ends_at:
return None
try:
ends = datetime.fromisoformat(listing.ends_at)
delta = ends - datetime.now(timezone.utc)
return max(delta.total_seconds() / 3600, 0.0)
except (ValueError, TypeError):
return None
def render_auction_notice(hours: float) -> None:
"""Render a soft de-emphasis notice for auctions with significant time remaining."""
if hours >= 1.0:
h = int(hours)
label = f"{h}h left" if h < 24 else f"{h // 24}d {h % 24}h left"
st.caption(
f"⏰ Auction · {label} — price not final until last few minutes"
)

View file

@ -1,10 +1,15 @@
"""Render a single listing row with trust score, badges, and error states."""
from __future__ import annotations
import json
import streamlit as st
from app.db.models import Listing, TrustScore, Seller
from typing import Optional
import streamlit as st
from app.db.models import Listing, TrustScore, Seller
from app.ui.components.easter_eggs import (
is_steal, render_steal_banner, render_auction_notice, auction_hours_remaining,
)
def _score_colour(score: int) -> str:
if score >= 80: return "🟢"
@ -29,7 +34,17 @@ def render_listing_row(
listing: Listing,
trust: Optional[TrustScore],
seller: Optional[Seller] = None,
market_price: Optional[float] = None,
) -> None:
# Easter egg: The Steal shimmer
if is_steal(listing, trust, market_price):
render_steal_banner()
# Auction de-emphasis (if > 1h remaining, price is not meaningful yet)
hours = auction_hours_remaining(listing)
if hours is not None:
render_auction_notice(hours)
col_img, col_info, col_score = st.columns([1, 5, 2])
with col_img:

View file

@ -4,7 +4,10 @@ Uses a minimal HTML fixture that mirrors eBay's search results structure.
No HTTP requests are made all tests operate on the pure parsing functions.
"""
import pytest
from app.platforms.ebay.scraper import scrape_listings, scrape_sellers, _parse_price, _parse_seller
from datetime import timedelta
from app.platforms.ebay.scraper import (
scrape_listings, scrape_sellers, _parse_price, _parse_seller, _parse_time_left,
)
# ---------------------------------------------------------------------------
# Minimal eBay search results HTML fixture
@ -149,3 +152,59 @@ class TestScrapeSellers:
"""category_history_json is always '{}' from scraper — signals partial score."""
sellers = scrape_sellers(_EBAY_HTML)
assert all(s.category_history_json == "{}" for s in sellers.values())
# ---------------------------------------------------------------------------
# _parse_time_left
# ---------------------------------------------------------------------------
class TestParseTimeLeft:
def test_days_hours(self):
td = _parse_time_left("3d 14h left")
assert td == timedelta(days=3, hours=14)
def test_hours_minutes(self):
td = _parse_time_left("14h 23m left")
assert td == timedelta(hours=14, minutes=23)
def test_minutes_seconds(self):
td = _parse_time_left("23m 45s left")
assert td == timedelta(minutes=23, seconds=45)
def test_days_only(self):
td = _parse_time_left("2d left")
assert td == timedelta(days=2)
def test_no_match_returns_none(self):
assert _parse_time_left("Buy It Now") is None
def test_empty_string_returns_none(self):
assert _parse_time_left("") is None
def test_all_zeros_returns_none(self):
# Regex can match "0d 0h 0m 0s left" — should treat as no time left = None
assert _parse_time_left("0d 0h 0m 0s left") is None
def test_auction_listing_sets_ends_at(self):
"""scrape_listings should set ends_at for an auction item."""
auction_html = """
<html><body><ul class="srp-results">
<li class="s-item">
<h3 class="s-item__title"><span>Test Item</span></h3>
<a class="s-item__link" href="https://www.ebay.com/itm/999"></a>
<span class="s-item__price">$100.00</span>
<span class="s-item__time-left">2h 30m left</span>
</li>
</ul></body></html>
"""
listings = scrape_listings(auction_html)
assert len(listings) == 1
assert listings[0].buying_format == "auction"
assert listings[0].ends_at is not None
def test_fixed_price_listing_no_ends_at(self):
"""scrape_listings should leave ends_at=None for fixed-price items."""
listings = scrape_listings(_EBAY_HTML)
fixed = [l for l in listings if l.buying_format == "fixed_price"]
assert len(fixed) > 0
assert all(l.ends_at is None for l in fixed)

View file

@ -0,0 +1,122 @@
"""Tests for easter egg helpers (pure logic — no Streamlit calls)."""
from __future__ import annotations
import json
from datetime import datetime, timedelta, timezone
import pytest
from app.db.models import Listing, TrustScore
from app.ui.components.easter_eggs import is_steal, auction_hours_remaining
def _listing(**kwargs) -> Listing:
defaults = dict(
platform="ebay",
platform_listing_id="1",
title="Test",
price=800.0,
currency="USD",
condition="used",
seller_platform_id="seller1",
url="https://ebay.com/itm/1",
buying_format="fixed_price",
ends_at=None,
)
defaults.update(kwargs)
return Listing(**defaults)
def _trust(score: int, flags: list[str] | None = None) -> TrustScore:
return TrustScore(
listing_id=1,
composite_score=score,
account_age_score=20,
feedback_count_score=20,
feedback_ratio_score=20,
price_vs_market_score=20,
category_history_score=score - 80 if score >= 80 else 0,
red_flags_json=json.dumps(flags or []),
)
# ---------------------------------------------------------------------------
# is_steal
# ---------------------------------------------------------------------------
class TestIsSteal:
def test_qualifies_when_high_trust_and_20_pct_below(self):
listing = _listing(price=840.0) # 16% below 1000
trust = _trust(92)
assert is_steal(listing, trust, market_price=1000.0) is True
def test_fails_when_trust_below_90(self):
listing = _listing(price=840.0)
trust = _trust(89)
assert is_steal(listing, trust, market_price=1000.0) is False
def test_fails_when_discount_too_deep(self):
# 35% below market — suspicious, not a steal
listing = _listing(price=650.0)
trust = _trust(95)
assert is_steal(listing, trust, market_price=1000.0) is False
def test_fails_when_discount_too_shallow(self):
# 10% below market — not enough of a deal
listing = _listing(price=900.0)
trust = _trust(95)
assert is_steal(listing, trust, market_price=1000.0) is False
def test_fails_when_suspicious_price_flag(self):
listing = _listing(price=840.0)
trust = _trust(92, flags=["suspicious_price"])
assert is_steal(listing, trust, market_price=1000.0) is False
def test_fails_when_no_market_price(self):
listing = _listing(price=840.0)
trust = _trust(92)
assert is_steal(listing, trust, market_price=None) is False
def test_fails_when_no_trust(self):
listing = _listing(price=840.0)
assert is_steal(listing, None, market_price=1000.0) is False
def test_boundary_15_pct(self):
listing = _listing(price=850.0) # exactly 15% below 1000
trust = _trust(92)
assert is_steal(listing, trust, market_price=1000.0) is True
def test_boundary_30_pct(self):
listing = _listing(price=700.0) # exactly 30% below 1000
trust = _trust(92)
assert is_steal(listing, trust, market_price=1000.0) is True
# ---------------------------------------------------------------------------
# auction_hours_remaining
# ---------------------------------------------------------------------------
class TestAuctionHoursRemaining:
def _auction_listing(self, hours_ahead: float) -> Listing:
ends = (datetime.now(timezone.utc) + timedelta(hours=hours_ahead)).isoformat()
return _listing(buying_format="auction", ends_at=ends)
def test_returns_hours_for_active_auction(self):
listing = self._auction_listing(3.0)
h = auction_hours_remaining(listing)
assert h is not None
assert 2.9 < h < 3.1
def test_returns_none_for_fixed_price(self):
listing = _listing(buying_format="fixed_price")
assert auction_hours_remaining(listing) is None
def test_returns_none_when_no_ends_at(self):
listing = _listing(buying_format="auction", ends_at=None)
assert auction_hours_remaining(listing) is None
def test_returns_zero_for_ended_auction(self):
ends = (datetime.now(timezone.utc) - timedelta(hours=1)).isoformat()
listing = _listing(buying_format="auction", ends_at=ends)
h = auction_hours_remaining(listing)
assert h == 0.0