191 lines
6.7 KiB
Python
191 lines
6.7 KiB
Python
"""
|
|
Quick auto-buyer test — runs a dry-run checkout on any product URL.
|
|
|
|
Usage (interactive):
|
|
python test_buy.py target https://www.target.com/p/...
|
|
python test_buy.py bestbuy https://www.bestbuy.com/site/...
|
|
python test_buy.py gamestop https://www.gamestop.com/products/...
|
|
|
|
Non-interactive (set in .env or environment):
|
|
TEST_USER_ID=1
|
|
TEST_PROFILE_PASSWORD=yourpassword
|
|
|
|
One-time login setup per user (saves browser session so future runs skip sign-in):
|
|
python test_buy.py --setup target --user-id 1
|
|
python test_buy.py --setup target --user-id 2
|
|
"""
|
|
|
|
import os
|
|
import sys
|
|
import logging
|
|
import getpass
|
|
from dotenv import load_dotenv
|
|
load_dotenv()
|
|
|
|
logging.basicConfig(
|
|
level=logging.INFO,
|
|
format="%(name)s — %(message)s",
|
|
handlers=[logging.StreamHandler()]
|
|
)
|
|
|
|
# ── Setup mode ────────────────────────────────────────────────────────────────
|
|
if len(sys.argv) >= 3 and sys.argv[1] == "--setup":
|
|
site = sys.argv[2].lower()
|
|
|
|
# Optional --user-id N
|
|
_uid = None
|
|
if "--user-id" in sys.argv:
|
|
_uid = int(sys.argv[sys.argv.index("--user-id") + 1])
|
|
|
|
if _uid is not None:
|
|
state_dir = os.path.join("data", "browser_states", str(_uid))
|
|
else:
|
|
state_dir = os.path.join("data", "browser_states")
|
|
state_file = os.path.join(state_dir, f"{site}.json")
|
|
os.makedirs(state_dir, exist_ok=True)
|
|
|
|
SITE_URLS = {
|
|
"target": "https://www.target.com/account",
|
|
"bestbuy": "https://www.bestbuy.com/identity/signin",
|
|
"gamestop": "https://www.gamestop.com/sign-in",
|
|
}
|
|
login_url = SITE_URLS.get(site, f"https://www.{site}.com")
|
|
|
|
uid_label = f"user {_uid}" if _uid is not None else "shared (no user)"
|
|
print(f"\n[setup] Opening {site} login page for {uid_label}.")
|
|
print(f"[setup] Log in manually in the browser window, then come back here and press Enter.")
|
|
print(f"[setup] Session will be saved to: {state_file}\n")
|
|
|
|
from playwright.sync_api import sync_playwright
|
|
with sync_playwright() as pw:
|
|
browser = pw.chromium.launch(channel="chrome", headless=False)
|
|
context = browser.new_context(viewport={"width": 1920, "height": 1080})
|
|
page = context.new_page()
|
|
page.goto(login_url)
|
|
input("[setup] Press Enter once you're logged in... ")
|
|
context.storage_state(path=state_file)
|
|
print(f"[setup] Session saved to {state_file}.")
|
|
browser.close()
|
|
sys.exit(0)
|
|
# ─────────────────────────────────────────────────────────────────────────────
|
|
|
|
if len(sys.argv) < 3:
|
|
print("Usage: python test_buy.py <site> <product_url>")
|
|
print(" python test_buy.py --setup <site>")
|
|
print(" e.g: python test_buy.py target https://www.target.com/p/pokemon-tcg-...")
|
|
sys.exit(1)
|
|
|
|
site = sys.argv[1].lower()
|
|
url = sys.argv[2]
|
|
|
|
# Force dry-run regardless of config
|
|
import config
|
|
config.AUTO_BUY_DRY_RUN = True
|
|
|
|
# ------------------------------------------------------------------
|
|
# Load credentials — prefer dashboard profile, fall back to .env
|
|
# ------------------------------------------------------------------
|
|
from src.database import get_database
|
|
from src.profile_store import decrypt_profile
|
|
|
|
db = get_database()
|
|
users = db.get_all_users()
|
|
|
|
shipping = None
|
|
payment = None
|
|
|
|
# Non-interactive mode: TEST_USER_ID + TEST_PROFILE_PASSWORD in env/.env
|
|
_env_user_id = os.environ.get("TEST_USER_ID", "").strip()
|
|
_env_password = os.environ.get("TEST_PROFILE_PASSWORD", "").strip()
|
|
|
|
if _env_user_id and _env_password:
|
|
user_id = int(_env_user_id)
|
|
if db.user_has_profile(user_id):
|
|
row = db.get_user_profile(user_id)
|
|
profile = decrypt_profile(_env_password, row['salt'], row['ciphertext'])
|
|
if profile is None:
|
|
print(f"[test_buy] TEST_PROFILE_PASSWORD is wrong for user {user_id}.")
|
|
sys.exit(1)
|
|
from src.profile_store import _unlocked
|
|
_unlocked[user_id] = profile
|
|
shipping = profile['shipping']
|
|
payment = profile['payment']
|
|
print(f"[test_buy] Non-interactive: loaded profile for user {user_id}.")
|
|
else:
|
|
print(f"[test_buy] No profile saved for TEST_USER_ID={user_id} — falling back to .env")
|
|
|
|
elif users:
|
|
# Interactive mode
|
|
print("\nAvailable users:")
|
|
for u in users:
|
|
has = db.user_has_profile(u['id'])
|
|
print(f" [{u['id']}] {u['name']}" + (" (profile saved)" if has else " (no profile)"))
|
|
|
|
choice = input("\nEnter user ID to use their profile (or press Enter to use .env): ").strip()
|
|
|
|
if choice:
|
|
user_id = int(choice)
|
|
if db.user_has_profile(user_id):
|
|
password = getpass.getpass(f"Password for user {user_id}: ")
|
|
row = db.get_user_profile(user_id)
|
|
profile = decrypt_profile(password, row['salt'], row['ciphertext'])
|
|
if profile is None:
|
|
print("Incorrect password.")
|
|
sys.exit(1)
|
|
from src.profile_store import _unlocked
|
|
_unlocked[user_id] = profile
|
|
shipping = profile['shipping']
|
|
payment = profile['payment']
|
|
print(f"Profile unlocked for user {user_id}.")
|
|
else:
|
|
print(f"No saved profile for user {user_id} — falling back to .env")
|
|
|
|
if not shipping:
|
|
from src.buyers.base_buyer import load_shipping, load_payment
|
|
shipping = load_shipping()
|
|
payment = load_payment()
|
|
print("Using .env credentials.")
|
|
|
|
# ------------------------------------------------------------------
|
|
# Run the buyer
|
|
# ------------------------------------------------------------------
|
|
from scrapers.base import Product
|
|
from src.buyers import get_buyer
|
|
|
|
buyer = get_buyer(site)
|
|
if not buyer:
|
|
print(f"No buyer for site: {site} (supported: target, bestbuy, gamestop)")
|
|
sys.exit(1)
|
|
|
|
# Inject credentials directly so the buyer uses them regardless of source
|
|
buyer.shipping = shipping
|
|
buyer.payment = payment
|
|
if _env_user_id:
|
|
buyer.user_id = int(_env_user_id)
|
|
elif 'user_id' in dir(): # set during interactive mode
|
|
buyer.user_id = user_id # type: ignore[possibly-undefined]
|
|
|
|
print(f"\n{'='*60}")
|
|
print(f" DRY RUN — {site.upper()}")
|
|
print(f" {url}")
|
|
print(f"{'='*60}\n")
|
|
|
|
product = Product(
|
|
name="Test Product",
|
|
url=url,
|
|
price="$19.99",
|
|
in_stock=True,
|
|
image_url=None,
|
|
site=site,
|
|
product_id="test",
|
|
)
|
|
|
|
result = buyer.buy_product(product)
|
|
|
|
print(f"\n{'='*60}")
|
|
print(f" Result : {'SUCCESS' if result.success else 'FAILED'}")
|
|
print(f" Message: {result.message}")
|
|
if result.order_number:
|
|
print(f" Order# : {result.order_number}")
|
|
print(f"{'='*60}\n")
|