feat: Cast Iron Scout agent prototype

- Created autonomous cast iron deal scanner
- Scans eBay RSS feeds hourly for cast iron cookware
- Calculates FMV based on brand, type, size
- Sends Telegram alerts for deals ≥50% below FMV
- Identifies Griswold, Wagner, Wapak, Birmingham, Lodge, Victor
- Tracks seen items to avoid duplicate alerts
- Valuation engine with size multipliers
- Configurable preferences in config.json

Known issue: eBay RSS unreliable - next iteration will use proper scraping
This commit is contained in:
2026-04-09 17:45:36 -04:00
parent 674c2c3925
commit 06fb4a243e
10 changed files with 462 additions and 0 deletions

View File

@@ -0,0 +1,118 @@
#!/usr/bin/env python3
"""
Cast Iron Scout - Main Scanner Engine
Continuously scans for cast iron deals and alerts when good deals found
"""
import json
import subprocess
import sys
from datetime import datetime
from pathlib import Path
from sources.ebay_scanner import search_ebay_cast_iron
from valuation import is_good_deal, calculate_fmv
SCRIPT_DIR = Path(__file__).parent
STATE_FILE = SCRIPT_DIR / "state" / "seen_items.json"
CONFIG_FILE = SCRIPT_DIR / "config.json"
LOG_FILE = SCRIPT_DIR / "logs" / f"scanner-{datetime.now().strftime('%Y%m%d')}.log"
def load_config():
"""Load configuration"""
if CONFIG_FILE.exists():
return json.loads(CONFIG_FILE.read_text())
return {}
def load_state():
"""Load previously seen items to avoid duplicates"""
if STATE_FILE.exists():
return json.loads(STATE_FILE.read_text())
return {"seen_links": [], "last_scan": None}
def save_state(state):
"""Save state to file"""
state['last_scan'] = datetime.now().isoformat()
STATE_FILE.write_text(json.dumps(state, indent=2))
def log(message):
"""Log message"""
ts = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
log_line = f"[{ts}] {message}"
print(log_line)
with open(LOG_FILE, 'a') as f:
f.write(log_line + '\n')
def send_telegram_alert(item, fmv, discount):
"""Send Telegram alert for a good deal"""
config = load_config()
target = config.get('telegram_target', 'telegram:8269921691')
message = f"""🔥 *CAST IRON DEAL ALERT!*
*Item:* {item['title']}
*Price:* ${item['price']:.2f}
*FMV:* ${fmv:.2f}
*Discount:* {discount:.0f}% below FMV! 💰
*Source:* {item['source']}
*Found:* {item['found_at']}
🔗 {item['link']}
_Action: Buy now / Bid / Ignore_"""
try:
subprocess.run([
'openclaw', 'message', 'send',
'--channel', 'telegram',
'--target', target,
'--message', message
], capture_output=True, timeout=30)
log(f"✅ Alert sent for: {item['title'][:50]}")
except Exception as e:
log(f"❌ Failed to send alert: {e}")
def scan_all_sources():
"""Scan all sources for cast iron items"""
log("🔍 Starting cast iron scan...")
# Load config and state
config = load_config()
state = load_state()
seen_links = set(state.get('seen_links', []))
# Scan eBay
items = search_ebay_cast_iron()
log(f"Found {len(items)} items on eBay")
deals_found = 0
min_discount = config.get('min_discount_percent', 50)
for item in items:
# Skip if already seen
if item['link'] in seen_links:
continue
# Check if it's a good deal
is_deal, discount, fmv = is_good_deal(item['price'], item['title'], min_discount)
if is_deal:
log(f"🎯 DEAL FOUND: {item['title'][:50]} - ${item['price']} ({discount:.0f}% off)")
send_telegram_alert(item, fmv, discount)
deals_found += 1
# Mark as seen
seen_links.add(item['link'])
# Keep only last 1000 seen items to prevent state file from growing forever
if len(seen_links) > 1000:
seen_links = set(list(seen_links)[-1000:])
state['seen_links'] = list(seen_links)
save_state(state)
log(f"Scan complete. Deals found: {deals_found}, Total items processed: {len(items)}")
return deals_found
if __name__ == "__main__":
deals = scan_all_sources()
sys.exit(0)