Python脚本自动获取最佳c1s offer

Update 2025/11/13: 改进保存的offer 优化判断只保存最新最佳优惠 (更新c1s_web_discount_finder.py)

众所周知c1s有各种隐藏offer, 需要通过特殊方式获取

c1s 官网搜索关键词 Capital One Shopping 新的刷Offer办法 （附如何 70% 返现 TargetOptical Meta眼镜Gen2）
比价offer 【疑似DEAD】三折以下买chewy的办法
【引用自 bigfar】:
Capital One Shopping Offer 经验贴以及奇技淫巧
3. 比价Offer
这类offer通常可以通过在Amazon上浏览物品或者在https://capitaloneshopping.com/search?q=xxxxx 上搜索来获得。这种方式通常会在页面底部显示多个商家的offer，一般都比较接近当前后台的最优解。比如泥潭非常著名的Chewy 25% cashback，就可以通过这种方式稳定地刷出来
以及其他各种offer
【引用自 未知】:
Capital One Shopping Offer 经验贴以及奇技淫巧 败家
经常逛泥潭的朋友们应该都知道C1S这个大羊毛了。在当前Amex和Chase严格限制撸卡的2025年，C1S无疑为我们提供了一种稳定的返现渠道。本文将分享我从去年黑五前开始研究C1S offer以来的一些经验和技巧，希望能帮助大家更有效地利用这一工具。
C1S简介
简单来说，Capital One Shopping(C1S)是一个类似Rakuten的返现网站，属于Capital One旗下，但并不…
比较讨厌的是这些offer是随机发放的 需要靠脸来一直刷 所以自然就写了一个脚本来帮我刷offer了

比价offer已经自己用测试了超过一个月 非常稳定 每次也都能track上

注: 根据经验 (YMMV) 20次没刷出来你就刷不出来这个offer了

python脚本以及配置文件

api_client.py
"""
API Client for C1S Simulator
Handles all HTTP requests to Capital One Shopping API.
"""

import requests
import logging
from typing import Dict, Optional, Any
from config_manager import ConfigManager
from utils import setup_cookies, make_request

class C1SApiClient:
"""API client for Capital One Shopping requests"""

def __init__(self, cookies_path: str, config_manager: ConfigManager):
self.cookies_path = cookies_path
self.config_manager = config_manager
self.session = requests.Session()
self.logger = logging.getLogger(__name__)
self._setup_session()

def _setup_session(self):
"""Setup the requests session with cookies and headers"""
self.session = setup_cookies(self.session, self.cookies_path)

# Set default headers
headers = self.config_manager.get_headers()
self.session.headers.update(headers)

# Add session-specific headers
wb_session = self._extract_wb_session()
if wb_session:
self.session.headers['x-wb-session'] = wb_session

def _extract_wb_session(self) -> Optional[str]:
"""Extract wb_session value from cookies for x-wb-session header"""
wb_session_cookie = self.session.cookies.get('wb_session')
if wb_session_cookie:
# URL decode the session value
import urllib.parse
return urllib.parse.unquote(wb_session_cookie)
return None

def _make_request(self, url: str, method: str = 'GET', **kwargs) -> Optional[Dict[str, Any]]:
"""Make HTTP request with error handling"""
return make_request(self.session, url, method, **kwargs)

def get_amazon_programs(self) -> Optional[Dict[str, Any]]:
"""Request 1: Get Amazon programs"""
endpoints = self.config_manager.get_api_endpoints()
url = endpoints.get("amazon_programs")

if not url:
self.logger.error("Amazon programs endpoint not configured")
return None

# Add if-modified-since header as in the original request
headers = {"if-modified-since": "Sat, 13 Sep 2025 17:21:49 GMT"}

return self._make_request(url, headers=headers)

def get_program_coupons(self, program_id: str) -> Optional[Dict[str, Any]]:
"""Request 2: Get program coupons"""
endpoints = self.config_manager.get_api_endpoints()
url_template = endpoints.get("program_coupons")

if not url_template:
self.logger.error("Program coupons endpoint not configured")
return None

url = url_template.format(program_id=program_id)
headers = {"if-modified-since": "Sat, 13 Sep 2025 10:00:17 GMT"}

return self._make_request(url, headers=headers)

def get_program_rewards(self, program_id: str) -> Optional[Dict[str, Any]]:
"""Request 3: Get program rewards"""
endpoints = self.config_manager.get_api_endpoints()
url_template = endpoints.get("program_rewards")

if not url_template:
self.logger.error("Program rewards endpoint not configured")
return None

url = url_template.format(program_id=program_id)
headers = {"if-modified-since": "Fri, 12 Sep 2025 22:59:07 GMT"}

return self._make_request(url, headers=headers)

def get_program_info(self, program_id: str) -> Optional[Dict[str, Any]]:
"""Request 4: Get program info"""
endpoints = self.config_manager.get_api_endpoints()
url_template = endpoints.get("program_info")

if not url_template:
self.logger.error("Program info endpoint not configured")
return None

url = url_template.format(program_id=program_id)
headers = {"if-modified-since": "Sat, 13 Sep 2025 16:00:46 GMT"}

return self._make_request(url, headers=headers)

def get_product_data(self, asin: str) -> Optional[Dict[str, Any]]:
"""Request 5: Get product pricing data"""
endpoints = self.config_manager.get_api_endpoints()
url_template = endpoints.get("product_check")

if not url_template:
self.logger.error("Product check endpoint not configured")
return None

url = url_template.format(asin=asin)

# Add additional headers for product request
headers = {
"content-type": "application/json",
"x-wb-extension": "0.1.1303"
}

return self._make_request(url, headers=headers)

def get_session_cookies(self) -> requests.cookies.RequestsCookieJar:
"""Get the session cookies for use in other requests"""
return self.session.cookies

c1s_simulator.py
#!/usr/bin/env python3
"""
Capital One Shopping (C1S) Plugin Simulator
Main entry point for the C1S offer fetching system.
"""

import json
import time
import uuid
import logging
from typing import Dict, List, Optional
from dataclasses import dataclass

from api_client import C1SApiClient
from config_manager import ConfigManager
from offer_checker import OfferChecker

@dataclass
class OfferResult:
"""Result of an offer check"""
asin: str
discount_percentage: float
savings: float
origin_total: float
match_total: float
match_vendor: str
quote_id: str
meets_criteria: bool
timestamp: str

class C1SSimulator:
"""Main simulator class for Capital One Shopping plugin behavior"""

def __init__(self, config_path: str = "config.json", cookies_path: str = "cookies.txt"):
self.config_manager = ConfigManager(config_path)
self.api_client = C1SApiClient(cookies_path, self.config_manager)
self.offer_checker = OfferChecker(self.config_manager)

# Setup logging
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(levelname)s - %(message)s'
)
self.logger = logging.getLogger(__name__)
self.count = 0

def run_simulation(self) -> OfferResult:
"""Run the complete simulation for all configured products"""
self.logger.info("Starting C1S simulation...")

results = []
config = self.config_manager.get_config()

# For amazon, program_id is always 40ec8518-6d29-441a-8d5f-48177e64aac9
program_id = "40ec8518-6d29-441a-8d5f-48177e64aac9"

# Execute preliminary requests (requests 1-4)
self._execute_preliminary_requests(program_id)

product = config["products"][self.count % len(config["products"])]
self.logger.info(f"Checking offers for product ASIN {product['asin']}")

result = self._check_product_offers(product, config["target_website"])

if not result.meets_criteria:
self.logger.info(f"Product {product['asin']}, quote {result.quote_id} doesn't meet criteria. Will retry after delay {config['retry_delay_seconds']} seconds.")
retry_delay = config.get("retry_delay_seconds", 30)
time.sleep(retry_delay)
self.count += 1
result = None
else:
pass

self.logger.info("Simulation completed.")
return result

def _execute_preliminary_requests(self, program_id: str):
"""Execute the first 4 requests to mimic plugin behavior"""
self.logger.info("Executing preliminary requests...")

# Request 1: Get Amazon programs
self.api_client.get_amazon_programs()

# Request 2: Get program coupons
self.api_client.get_program_coupons(program_id)

# Request 3: Get program rewards
self.api_client.get_program_rewards(program_id)

# Request 4: Get program info
self.api_client.get_program_info(program_id)

self.logger.info("Preliminary requests completed.")

def _check_product_offers(self, product: Dict, target_website: str) -> OfferResult:
"""Check offers for a specific product"""
# Request 5: Get product pricing data
product_data = self.api_client.get_product_data(product["asin"])

if not product_data:
return OfferResult(
asin=product["asin"],
discount_percentage=0.0,
savings=0.0,
origin_total=0.0,
match_total=0.0,
match_vendor="",
quote_id="",
meets_criteria=False,
timestamp=time.strftime("%Y-%m-%d %H:%M:%S")
)

# Calculate discount percentage
discount_percentage = self.offer_checker.calculate_discount_percentage(product_data, target_website)
meets_criteria = self.offer_checker.meets_discount_criteria(discount_percentage)

return OfferResult(
asin=product["asin"],
discount_percentage=discount_percentage,
savings=product_data.get("lastQuote", {}).get("savings", 0),
origin_total=product_data.get("lastQuote", {}).get("origin_total", 0),
match_total=product_data.get("lastQuote", {}).get("match_total", 0),
match_vendor=product_data.get("lastQuote", {}).get("match_vendor", ""),
quote_id=product_data.get("lastQuote", {}).get("quote_id", ""),
meets_criteria=meets_criteria,
timestamp=time.strftime("%Y-%m-%d %H:%M:%S")
)

def main():
"""Main entry point"""
simulator = C1SSimulator()
result = None
while not result:
result = simulator.run_simulation()

# Print results
print("\n" + "="*80)
print("C1S SIMULATION RESULTS")
print("="*80)

print(f"\nASIN: {result.asin}")
print(f"Discount: {result.discount_percentage:.2f}%")
print(f"Savings: ${result.savings/100:.2f}")
print(f"Origin Total: ${result.origin_total/100:.2f}")
print(f"Match Vendor: {result.match_vendor}")
print(f"Quote ID: {result.quote_id}")
print(f"Meets Criteria: {'✓' if result.meets_criteria else '✗'}")
print(f"Timestamp: {result.timestamp}")
print("-" * 40)

if __name__ == "__main__":
main()

c1s_web_discount_finder.py
#!/usr/bin/env python3
"""
Capital One Shopping (C1S) Web Simulator
Main entry point for the C1S offer fetching system.
"""

import json
import time
import uuid
import logging
from typing import Dict, List, Optional
from dataclasses import dataclass
from offer_checker import OfferChecker
from web_client import *

class C1SWebSimulator:
"""Main simulator class for Capital One Shopping web behavior"""

def __init__(self, config_path: str = "config.json", cookies_path: str = "web_cookies.txt"):
self.config_manager = ConfigManager(config_path)
self.web_client = WebClient(cookies_path, self.config_manager)
self.offer_checker = OfferChecker(self.config_manager)

# Setup logging
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(levelname)s - %(message)s'
)
self.logger = logging.getLogger(__name__)
self.count = 0

def run_simulation(self):
"""Run the complete simulation for all configured products"""
search_keyword = self.config_manager.get_search_keyword()
self.logger.info("Starting C1S Web simulation...")
self.logger.info(f"Searching for keyword: {search_keyword}")
search_results = self.web_client.get_search_results(search_keyword)
self.logger.info(f"Found {len(search_results.get('SearchPage', {}).get('results', []))} search results.")
self.logger.info("Starting offer checks...")
for item in search_results.get('SearchPage', {}).get('results', []):
title = item.get('product', {}).get('title', 'N/A')
asin = item.get('asin', 'N/A')
cpid = item.get('cpid', 'N/A')
id = item.get('id', 'N/A')
url = item.get('url', '')
self.logger.info(f"Title: {title}, ASIN: {asin}, CPID: {cpid}, URL: {url}")
discount_result = self.web_client.get_discount_by_wbid(id)
if not discount_result:
self.logger.info(f"No discount data found for product {title}.")
self.logger.info("sleep for 5 seconds")
time.sleep(5)
continue
id = discount_result.get('id', 'N/A')
results = discount_result.get('items', [{}])[0]
for result in results.get('results', []):
reward = result.get('pricing', {}).get('reward', {})
if reward == {}:
continue
redirect_url = result.get('redirectUrl', '')
vendor = result.get('vendor', 'N/A')
reward_source_url = f"https://capitaloneshopping.com{url}?run={id}"
exclusions = reward.get('exclusions', "")
if reward.get('type') == "percentage":
discount_percentage = reward.get('amount', 0) / 100
self.logger.info(f"Found discount for {vendor}: {discount_percentage}% off at {redirect_url} from {reward_source_url}")
save_data = {
"title": title,
"vendor": vendor,
"discount_percentage": discount_percentage,
"exclusions": exclusions,
"redirect_url": redirect_url,
"reward_source_url": reward_source_url
}
self.save_result_to_file("reward_results.json", save_data)
elif reward.get('type') == "fixed":
amount = reward.get('amount', 0)
if amount == 0:
continue
amount_off = reward.get('amount', 0) / 100
threshold = reward.get('threshold', 0) / 100
self.logger.info(f"Found discount for {vendor}: ${amount_off} off at {redirect_url} from {reward_source_url}")
save_data = {
"title": title,
"vendor": vendor,
"amount_off": amount_off,
"threshold": threshold,
"exclusions": exclusions,
"redirect_url": redirect_url,
"reward_source_url": reward_source_url
}
self.save_result_to_file("reward_results.json", save_data)
else:
self.logger.info(f"Found reward for {vendor}: a{reward} at {redirect_url} from {reward_source_url}")
save_data = {
"title": title,
"vendor": vendor,
"reward_raw": reward,
"redirect_url": redirect_url,
"reward_source_url": reward_source_url
}
self.save_result_to_file("reward_results.json", save_data)
self.logger.info("sleep for 5 seconds before checking next item")
time.sleep(5)
return

def save_result_to_file(self, filename: str, data: Dict):
"""Save result data to a JSON file"""
existing_data = []
try:
with open(filename, 'r', encoding='utf-8') as f:
existing_data = json.load(f)
except:
pass # File does not exist or is invalid, will create new
# Compare offer, only save the best offer per vendor
if data.get("discount_percentage"): # 论坛似乎有满300-300的offer 不知道是什么样的response
vendor = data.get("vendor")
for existing_offer in existing_data:
if existing_offer.get("vendor") == vendor:
if "discount_percentage" not in existing_offer:
continue # Existing offer has no discount percentage, skip
if existing_offer.get("discount_percentage", 0) > data.get("discount_percentage", 0):
return # Existing offer is better, do not save
else:
existing_data.remove(existing_offer) # Remove worse offer, keep the latest same offer
break
if data.get("amount_off"):
vendor = data.get("vendor")
for existing_offer in existing_data:
if existing_offer.get("vendor") == vendor:
if "amount_off" not in existing_offer:
continue # Existing offer has no amount_off, skip
# Same amount, keep the minimum threshold, same threshold, keep the maximum amount
if existing_offer.get("threshold", 0) is None:
del existing_offer["threshold"]
if data.get("threshold", 0) is None:
del data["threshold"]
if (existing_offer.get("amount_off", 0) > data.get("amount_off", 0)) or \
(existing_offer.get("amount_off", 0) == data.get("amount_off", 0) and existing_offer.get("threshold", 0) < data.get("threshold", 0)):
return # Existing offer is better
else:
existing_data.remove(existing_offer) # Remove worse offer, keep the latest same offer
break
existing_data.append(data)
with open(filename, 'w', encoding='utf-8') as f:
json.dump(existing_data, f, indent=4)

def clean_up_result_file(self, filename: str):
"""Clean up result file by removing duplicates and keeping best offers"""
try:
with open(filename, 'r', encoding='utf-8') as f:
existing_data = json.load(f)
except:
return # File does not exist or is invalid, nothing to clean
for data in existing_data:
self.save_result_to_file(filename, data)
return

def main():
"""Main entry point"""
simulator = C1SWebSimulator()
simulator.clean_up_result_file("reward_results.json")
simulator.run_simulation()

if __name__ == "__main__":
main()

config.json
{
"target_website": "Chewy",
"minimum_discount_percentage": 28,
"retry_delay_seconds": 10,
"products": [
{
"asin": "B006FSUUTC"
},
{
"asin": "B07SRXLMH4"
},
{
"asin": "B01NA9TN3D"
},
{
"asin": "B000634CO6"
}
],
"headers": {
"accept": "*/*",
"accept-language": "en-US,en;q=0.9",
"priority": "u=1, i",
"sec-fetch-dest": "empty",
"sec-fetch-mode": "cors",
"sec-fetch-site": "none",
"sec-fetch-storage-access": "active",
"sec-gpc": "1",
"user-agent": "填写你自己的user-agent"
},
"api_endpoints": {
"amazon_programs": "https://capitaloneshopping.com/api/v1/programs/amazon.com",
"program_coupons": "https://capitaloneshopping.com/api/v1/program/{program_id}/coupons?extension=chrome&ranker=trash_day_ctrl",
"program_rewards": "https://capitaloneshopping.com/api/v1/program/{program_id}/rewards?re=t",
"program_info": "https://capitaloneshopping.com/api/v1/program/{program_id}",
"product_check": "https://capitaloneshopping.com/api/v1/productV2?asin={asin}&priceCheckId=undefined&retry=undefined"
},
"prime": true,
"zip_code": "填写你的邮编",
"search_keyword": "搜索关键词"
}

config_manager.py
"""
Configuration Manager for C1S Simulator
Handles loading and managing configuration from JSON file.
"""

import json
import logging
from typing import Dict, Any

class ConfigManager:
"""Manages configuration loading and access"""

def __init__(self, config_path: str):
self.config_path = config_path
self.config = None
self.logger = logging.getLogger(__name__)
self._load_config()

def _load_config(self):
"""Load configuration from JSON file"""
try:
with open(self.config_path, 'r', encoding='utf-8') as f:
self.config = json.load(f)
self.logger.info(f"Configuration loaded from {self.config_path}")
except FileNotFoundError:
self.logger.error(f"Configuration file not found: {self.config_path}")
raise
except json.JSONDecodeError as e:
self.logger.error(f"Invalid JSON in configuration file: {e}")
raise

def get_config(self) -> Dict[str, Any]:
"""Get the complete configuration"""
if self.config is None:
self._load_config()
return self.config

def get_target_website(self) -> str:
"""Get the target website for offers"""
return self.config.get("target_website", "")

def get_minimum_discount_percentage(self) -> float:
"""Get the minimum discount percentage threshold"""
return self.config.get("minimum_discount_percentage", 0.0)

def get_retry_delay_seconds(self) -> int:
"""Get the retry delay in seconds"""
return self.config.get("retry_delay_seconds", 30)

def get_products(self) -> list:
"""Get the list of products to check"""
return self.config.get("products", [])

def get_headers(self) -> Dict[str, str]:
"""Get the HTTP headers for requests"""
return self.config.get("headers", {})

def get_api_endpoints(self) -> Dict[str, str]:
"""Get the API endpoints configuration"""
return self.config.get("api_endpoints", {})

def get_prime(self) -> bool:
"""Get whether user has prime membership"""
return self.config.get("prime", False)

def get_zip_code(self) -> str:
"""Get the user's zip code"""
return self.config.get("zip_code", "")

def get_search_keyword(self) -> str:
"""Get the search keyword for product search"""
return self.config.get("search_keyword", "")

def reload_config(self):
"""Reload configuration from file"""
self._load_config()

offer_checker.py
"""
Offer Checker for C1S Simulator
Handles discount calculation and criteria checking.
"""

import logging
import requests
import re
from typing import Dict, Any, Optional, Tuple
from bs4 import BeautifulSoup
from config_manager import ConfigManager

class OfferChecker:
"""Handles offer validation and discount calculations"""

def __init__(self, config_manager: ConfigManager):
self.config_manager = config_manager
self.logger = logging.getLogger(__name__)

def calculate_discount_percentage(self, product_data: Dict[str, Any], match_vendor: str) -> float:
"""Calculate discount percentage from product data"""
try:
last_quote = product_data.get("lastQuote", {})

if not last_quote:
self.logger.warning("No lastQuote data found in product response")
return 0.0

partial_url = product_data.get("url", "")
if not partial_url:
self.logger.warning("No URL found in product response")
return 0.0

quote_id = last_quote.get("quote_id", "")
if not quote_id:
self.logger.warning("No quote_id found in lastQuote")
return 0.0

if match_vendor == 'chewy.com':
match_vendor = 'Chewy'

full_url = f"https://capitaloneshopping.com{partial_url}?extReq=true&run={quote_id}"

self.logger.info(f"Scraping discount data from: {full_url}")

# Scrape the website for discount information
discount_data = self._scrape_discount_data(full_url, match_vendor)

if not discount_data:
self.logger.warning("Could not extract discount data from website")
return 0.0

# Get the discount percentage directly from the scraped data
discount_percentage = discount_data.get("discount_percentage", 0.0)
vendor_name = discount_data.get("vendor_name", match_vendor)
offer_text = discount_data.get("offer_text", "")

self.logger.info(f"Found discount: {discount_percentage:.2f}% for {vendor_name} "
f"(Offer: '{offer_text}')")

return discount_percentage

except Exception as e:
self.logger.error(f"Error calculating discount percentage: {e}")
return 0.0

def meets_discount_criteria(self, discount_percentage: float) -> bool:
"""Check if discount percentage meets the minimum criteria"""
minimum_discount = self.config_manager.get_minimum_discount_percentage()
meets_criteria = discount_percentage >= minimum_discount

self.logger.info(f"Discount check: {discount_percentage:.2f}% "
f"{'≥' if meets_criteria else '<'} {minimum_discount}% "
f"({'PASS' if meets_criteria else 'FAIL'})")

return meets_criteria

def _scrape_discount_data(self, url: str, target_vendor: str) -> Optional[Dict[str, float]]:
"""Scrape discount data from the C1S website, focusing on the target vendor's offer percentage"""
try:
# Make request to the C1S page
headers = {
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/140.0.0.0 Safari/537.36'
}

response = requests.get(url, headers=headers, timeout=30)
response.raise_for_status()

soup = BeautifulSoup(response.content, 'html.parser')

# Find the main comparison table
table_wrapper = soup.find('div', class_='table-wrapper')
if not table_wrapper:
self.logger.warning("Could not find comparison table on page")
return None

# Look for the target vendor row specifically
rows = table_wrapper.find_all('div', class_='affiliate-row-component')

for row in rows:
vendor_element = row.find('div', class_='vendor')
if not vendor_element:
continue

vendor_name_element = vendor_element.find('span', class_='charcoal-link')
if not vendor_name_element:
continue

vendor_name = vendor_name_element.get_text(strip=True)

# Only process the target vendor row
if vendor_name.lower() == target_vendor.lower():
self.logger.info(f"Found target vendor row: {vendor_name}")

# Look for special offers section with percentage back
special_offers_element = row.find('div', class_='special-offers')
if special_offers_element:
# Look for the "get X% back" text
offer_text_element = special_offers_element.find('h4', class_='offer-text')
if offer_text_element:
offer_text = offer_text_element.get_text(strip=True)
self.logger.info(f"Found offer text: {offer_text}")

# Extract percentage from text like "get 25% back*"
discount_percentage = self._extract_percentage_back(offer_text)

if discount_percentage > 0:
self.logger.info(f"Extracted discount percentage: {discount_percentage}%")
return {
'discount_percentage': discount_percentage,
'vendor_name': vendor_name,
'offer_text': offer_text
}

# If no special offers found, log and continue
self.logger.warning(f"No special offers found for {vendor_name}")
break

self.logger.warning(f"Target vendor '{target_vendor}' not found in comparison table")
return None

except Exception as e:
self.logger.error(f"Error scraping discount data: {e}")
return None

def _extract_percentage_back(self, offer_text: str) -> float:
"""Extract percentage from text like 'get 25% back*'"""
try:
# Look for pattern like "get X% back" or "X% back"
percentage_match = re.search(r'get\s+(\d+(?:\.\d+)?)%\s+back', offer_text.lower())
if not percentage_match:
# Try alternative pattern without "get"
percentage_match = re.search(r'(\d+(?:\.\d+)?)%\s+back', offer_text.lower())

if percentage_match:
return float(percentage_match.group(1))

self.logger.warning(f"Could not extract percentage from offer text: {offer_text}")
return 0.0
except (ValueError, AttributeError) as e:
self.logger.error(f"Error extracting percentage from '{offer_text}': {e}")
return 0.0

def get_target_website(self) -> str:
"""Get the target website for comparison"""
return self.config_manager.get_target_website()

def validate_product_data(self, product_data: Dict[str, Any]) -> bool:
"""Validate that product data contains required fields"""
if not product_data:
return False

required_fields = ["asin", "title", "lastQuote"]
for field in required_fields:
if field not in product_data:
self.logger.warning(f"Missing required field: {field}")
return False

last_quote = product_data.get("lastQuote", {})
required_quote_fields = ["origin_total", "match_total", "match_vendor", "savings"]
for field in required_quote_fields:
if field not in last_quote:
self.logger.warning(f"Missing required quote field: {field}")
return False

return True

requirements.txt
requests>=2.31.0
beautifulsoup4>=4.12.0

utils.py
import logging
from typing import Optional, Dict, Any

import requests
from requests import Session

logger = logging.getLogger(__name__)

def setup_cookies(session: Session, cookies_path: str) -> Session:
"""Setup the requests session with cookies and headers"""
# Load cookies
try:
with open(cookies_path, 'r', encoding='utf-8') as f:
cookie_string = f.read().strip()

# Parse cookies and add to session
for cookie in cookie_string.split('; '):
if '=' in cookie:
name, value = cookie.split('=', 1)
session.cookies.set(name, value)

logger.info(f"Loaded cookies from {cookies_path}")
except FileNotFoundError:
logger.error(f"Cookies file not found: {cookies_path}")
raise
return session

def make_request(session: Session, url: str, method: str = 'GET', **kwargs) -> Optional[Dict[str, Any]]:
"""Make HTTP request with error handling"""
try:
logger.debug(f"Making {method} request to: {url}")

response = session.request(method, url, **kwargs)
response.raise_for_status()

# Try to parse JSON response
try:
return response.json()
except ValueError:
# Return text if not JSON
return {"text": response.text}

except requests.exceptions.RequestException as e:
logger.error(f"Request failed: {e}")
return None

web_client.py
import base64
import json
import logging
import time
import uuid
from typing import Optional, Dict, Any

import requests

from config_manager import ConfigManager
from utils import setup_cookies, make_request

class WebClient:
def __init__(self, cookies_path: str, config_manager: ConfigManager):
self.cookies_path = cookies_path
self.config_manager = config_manager
self.session = requests.Session()
self.logger = logging.getLogger(__name__)
self._setup_session()

def _setup_session(self):
"""Setup the requests session with cookies and headers"""
self.session = setup_cookies(self.session, self.cookies_path)

# Set default headers
headers = self.config_manager.get_headers()
self.session.headers.update(headers)

def _make_request(self, url: str, method: str = 'GET', **kwargs) -> Optional[Dict[str, Any]]:
"""Make HTTP request with error handling"""
return make_request(self.session, url, method, **kwargs)

def _get_search_url(self, keyword: str) -> str:
"""Construct search URL for the given keyword"""
base_url = "https://capitaloneshopping.com/search"
keyword_dict = {"match":keyword}
keyword_json_b64 = base64.b64encode(json.dumps(keyword_dict, separators=(',', ':')).encode('utf-8')).decode('utf-8')
uuid_str = uuid.uuid4()
search_url = f"{base_url}?q={keyword_json_b64}&searchId={uuid_str}&_data=routes%2F__app%2Fsearch"
return search_url

def get_search_results(self, keyword: str) -> Optional[Dict[str, Any]]:
"""Request search results for the given keyword"""
url = self._get_search_url(keyword)
return self._make_request(url)

def get_suggest_results(self, keyword: str) -> Optional[Dict[str, Any]]:
"""Request suggest results for the given keyword"""
url = "https://capitaloneshopping.com/api/v1/merchant/suggest"
payload = {"country": "US", "term": keyword, "tag": "store_page"}
return self._make_request(url, method='POST', json=payload)

def get_discount_by_wbid(self, wbid: str) -> Optional[Dict[str, Any]]:
"""Request discount information by WBID"""
url = "https://capitaloneshopping.com/api/v1/run/runV3"
payload = {
"wbid": wbid,
"prime": self.config_manager.get_prime(),
"zipcode": self.config_manager.get_zip_code()
}
resp = self._make_request(url, method='POST', json=payload)
estimate_run_id = resp.get("estimateRunId") if resp else None
update_pricing_run_id = resp.get("updatedPricingRunId") if resp else None
if not estimate_run_id or not update_pricing_run_id:
self.logger.warning("Could not get run IDs from discount response")
return None
TIMEOUT = 30 # 30 seconds
self.logger.info("Waiting for pricing run to complete...")
for _ in range(TIMEOUT):
if self._check_for_run_complete(update_pricing_run_id):
break
time.sleep(1)
else:
self.logger.warning("Timeout waiting for pricing run to complete")
return None
return self._get_run_results(update_pricing_run_id, wbid)

def _check_for_run_complete(self, update_pricing_run_id: str) -> bool:
"""Check if the pricing run is complete"""
url = f"https://capitaloneshopping.com/api/v1/run/{update_pricing_run_id}/isComplete"
resp = self._make_request(url)
return resp.get("isComplete", False) if resp else False

def _get_run_results(self, update_pricing_run_id: str, product_id: str) -> Optional[Dict[str, Any]]:
"""Get the results of a completed run"""
url = "https://capitaloneshopping.com/api/v1/cart"
payload = {"cartClient": "web", "pla": False, "items": [{"productId": product_id, "run_id": update_pricing_run_id}]}
return self._make_request(url, method='POST', json=payload)

使用方法:

不传github/gitlab是为了防止DMCA被删库

将上面的几个python文件保存在电脑里 用title的命名存储这些py文件

或者使用mega下载: 41.2 KB folder on MEGA

入口文件是c1s_simulator.py以及c1s_web_discount_finder.py

c1s_simulator.py用于模拟插件访问亚马逊商品 (只支持亚马逊搜索)
c1s_web_discount_finder.py用于模拟c1s官网关键词搜索

需要修改config.json填写邮编, UA (user-agent), 以及搜索关键词

config.json自带chewy 30% offer的检索 (设置搜索28%以上的offer 不建议设置刚好的数值 防止浮点数碍事)

除此之外, 需要获取插件cookies存在文件 cookies.txt 以及浏览器cookies存在文件 web_cookies.txt

由于上面代码块太长 编辑已经很卡了 抓包以及插件配置教程在楼下

插件抓包

打开浏览器设置

brave://settings/ chrome://settings/

进入插件tab

image250×932 19.9 KB

打开开发者模式

image281×128 765 Bytes

管理插件

image990×586 19.9 KB

或者直接使用类似链接brave://extensions/ chrome://extensions/

找到c1s 点击details

image1485×874 75.2 KB

点击Service workers

image1062×874 28.5 KB

弹出插件console

image913×871 27.2 KB

随意访问一个网页 比如亚马逊 就能看到一堆请求

image680×590 29 KB

找到coupons?xxx 里面点开cookies就是你插件的cookies了

image902×441 27.2 KB

浏览器抓包

F12打开console, 刷新c1s网页, 可以从里面找到cookies

image1787×756 158 KB

获取UA

直接访问 What is my user agent? - WhatIsMyBrowser.com 就能复制到自己的UA了

部分配置文件教程

target_website: 爬取的网站

例如 Free Automatic Savings on Purina ONE Plus Tender Cuts in Gravy Healthy Weight Lamb and Brown Rice Entree in Wet Dog Food Gravy - (12) 13 Oz. Cans

image1618×205 10.4 KB

填写Chewy或者Amazon 大小写也必须一致

minimum_discount_percentage: 最低折扣爬取

上面的例子里面我要爬chewy 31%, 我填写28, 可以避免浮点数导致的匹配错误

c1s_simulator.py会反复爬取直到检索到比你期望更高的折扣

products: 亚马逊商品asin

amazon/dp/xxx xxx的部分就是asin, 用于模拟访问这个亚马逊商品的时候的插件行为

search_keyword: c1s网页上搜索的关键词 比如想搜dyson的优惠 填写dyson, 比如想搜meta眼睛的优惠 填写meta ray-ban

彩蛋

api_client.py类已经完全模拟插件的行为了 也就是说用这个类可以实现邮件offer (自己写个入口函数就行, 或者让chatgpt帮你 我没这个需求 懒得写 )

c1s_web_discount_finder.py执行完会生成一个reward_results.json, 百分比 off的会存储最高的offer, 其他offer全部保存

鉴于我没刷到上面说的Dyson 300-300 offer, 我也不知道返回的rewards structure会是怎么样的 所以没法择优保存

PS.

模拟插件是很久之前写的 刚让AI确认了一下插件和网页的cookies, 理论上几乎一致, 可以只抓网页的 删掉一部分就是插件的

跑了一下，对比“只看 key（忽略 value）”的结果如下：

共同的 keys：ajs_anonymous_id, ajs_group_id, ajs_user_id, wb_profile_id, wb_session_status
仅网页里有、插件里没有的 keys：__cf_bm, _dd_s, ktlvDW7IG5ClOcxYTbmY, w82S5kL1, wb_extension_version, wb_session
仅插件里的 keys：无

结论：两组 cookies 的 key 不一致（）。如果你需要一致，插件这边缺少上面那 6 个 key。

PPS.

请勿转载/丢AI转语言

前排占座

前排关注！

感觉这个得去白金区啊 不然大家一起刷感觉很容易overload 然后就一起亡了

Nbnb！

不至于 这个只是省下了手动的部分 这些操作在上面的引用的帖子里都有教程

感觉就是省去了手动部分才比较容易被abuse 他们那个系统爬不同网站拉offer那个过程感觉还挺费劲的 不过anyway只是我个人看法

反正C1会自己封撸多的人，没啥区别

但是高offer的购买“投入”相对少

苦了我们这些不会技术了，看不懂，放弃

不难的 所有教程都只需要做一次

【引用自 uhhhh】:
DMCA
爬不到论坛吗？

论坛会理DMCA?

谢谢，我是技术白痴

插眼，zszs

这个跟技术无关

DMCA是他们发投诉 要求你删掉这个repository

要怎么DMCA论坛呢? 1. 他哪里有论坛的官方正式联系方式? 2. 论坛收到他的要求论坛会理他吗

他最多能做的就是找谷歌发DMCA把这个帖子的链接拉黑 (拉黑是按链接的 不能整个论坛拉黑)

有venture x会增加刷到大offer的概率吗 以前确实看到过一些很不错的offer但是后来变少了

喜闻乐见的是 拉黑的链接只是隐藏 你照样能看到

Tricks:

谷歌会显示隐藏了一些链接 然后点详情会让你看到具体DMCA的请求 里面会列出具体的链接 从里面点进去就行 搜盗版的东西这个trick非常好用 而且里面的链接比谷歌的检索结果更佳 是直接链接到你希望的准确结果

依旧是后排disclaimer: 被c1s杀了别找我

【引用自 uhhhh】:
点详情会让你看到具体DMCA的请求
请问是页面最下方小字那里吗

yes zszs

Lz离收到C1的大包不远了

杀全家大礼包?

技术上来说 抓不到的 发的请求跟插件完全一致

按频率之类的特征抓的话 等着一堆误封吧

加点什么时间延迟

你可以自己修改入口函数 反正核心代码都在里面了 自己微调一下就完全可以非常拟人了 比如随机1~3个线程 随机延迟 模拟你同时开几个tab等offer

自动化威力不小的，python AA查票被封应该有自动化的原因

不, offer大包

我肯定不会上这么低级的当 上个这样干的是G胖

撸爆c1给桑 @令狐冲 报仇

多谢楼主！ 运行完有Ouput simulation results, 然后怎么用呢。。。。 而且并没有生成reward_results.json。

看不懂，先mark一下

simulator不会输出文件 会直接显示你要的结果 这个是模拟插件刷offer

看上去很牛逼 先mark一把～

嘎嘎好用

能刷att 开户offer吗，hh

可以的 但是那玩意一天只刷新一次 所以跟手动没区别

我账号里track 到几个自己没买的order, 是网络错误吗

emmmm 你爬的cookies跟下单的账号不同吗 还是爬完分享给其他人了?

我没用帖子里的工具，还没开始研究。

之前买vineyard我自己4个order 最后track出6个。

没听过这种BUG… 不过类似的意思 你分享你的链接给别人用了吗?

哦 或者是你点链接进去之后 又激活插件的返现 我经常这样做 可以刷一大片

我是来学it技术的

C1除了vx都是屎

可达鸭兄是脚本仙人。。。

所以刷出offer后，会保存一个url？然后我通过url下单？

yes 也有保存原始链接 有需要可以先原始链接截图好跟客服撕逼

会被ban么？

特征跟你手动一致 这个ban你手动也等着ban吧

火钳刘明！

前排占楼

六名 zs

target optical单独把meta列出来

是不是 70% 不适用了已经

是通过搜索出来的吗 如果是而且还被exlude了那就是dead了 我至今没刷出来

还没出来70%，在加码

但是meta只有百分之个位数

【引用自 uhhhh】:
user-agent
请问user agent是啥？是邮箱吗？

浏览器标识

感谢楼主，稍微改了一下接口cls_web_discount_finder.py，支持 keywords 用 cli传参，目前多个client 并行爬，很香。

对于 0 编程基础的我真的挺难的 看不太懂。。。

这个太牛了，改天试一下

请问cookie里面有无时间，过期日之类是参数会经常会变化？你用不变的cookie发起请求会不会导致露馅？

你浏览器要重新登录了就重新抓个包呗

插件那边的cookie从来不会过期，我有半年没登录过也不会过期，

我担心的是即使它不提示过期，它的插件cookie里有些字段也是动态变化的，而你模拟的是不变的，会不会被检测到？

插件cookies其实就是网页的加了一些字段而已

cookies是会过期的 只是他们能用过期cookies match到你 但是为了防止意外我下单前都会检查网页登陆有没有掉 掉了我都会重新登录

【引用自 uhhhh】:
点击Service workers
image1062×874 28.5 KB
我抓不了包怎么办

image830×1006 26.2 KB

不知道为什么你这个不显示service worker诶

我这个是chrome了哦，windows和mac都没有哦

image281×128 765 Bytes

右上角开发者模式试试?

我截图是用的brave, chromium浏览器 我平时用chrome 2个都有这个

Brave我没开开发者模式 但是只能死马当活马医 试试看吧

哦 测试了一下这个开发者模式得打开才能看到

不好意思，有幾個問題

插件cookies和浏览器cookies內容除了最後面稍有不同好像差不多是正常的嗎？我用的是火狐
如想刷出特定廠商的返現優惠像是AT&T光纖，Xfinity網路或是Mint Mobile的話要怎麼設定呢？

【引用自 元素轉換】:
插件cookies和浏览器cookies內容除了最後面稍有不同好像差不多是正常的嗎？我用的是火狐
是的
【引用自 元素轉換】:
如想刷出特定廠商的返現優惠像是AT&T光纖，Xfinity網路或是Mint Mobile的話要怎麼設定呢？
好问题 搜索关键词我也不知道要用什么 你得问问那些刷出来的人是怎么刷出来的

花了不少時間才搞懂json裡變數名target_website其實是Python碼裡的target vendor。後悔沒看清楚部分配置文件教程

發現目前程式本身並不支援固定額度返現優惠，不過我拿來嘗試的商品最優解並不是沒任何優惠的Amazon也不是有優惠的Walmart而是沒任何優惠的第三個網站
INFO - Found target vendor row: Walmart
INFO - Found offer text: Spend $20.00, get $5.00 in Rewards*
WARNING - Could not extract percentage from offer text: Spend $20.00, get $5.00 in Rewards*
WARNING - No special offers found for Walmart
WARNING - Target vendor 'Walmart' not found in comparison table
WARNING - Could not extract discount data from website
INFO - Discount check: 0.00% < 28% (FAIL)

插件cookies即使在同一個session裡都一直在變化。我嘗試多次用F5重新載入同一個網頁結果每次抓到的插件cookies都是開頭部份一樣但後面不一樣甚至長度都差很多

program_coupons在火狐上抓到的結尾是ranker=trash_day而不是ranker=trash_day_ctrl。不知道是瀏覽器的差別還是插件最新版改了結尾

实现邮件offer方面能展開來說說嗎？是一跑就讓C1S直接寄offer信嗎？即使目前有尚未過期的offer信也能讓C1S寄嗎？

【引用自 元素轉換】:
發現目前程式本身並不支援固定額度返現優惠
我一开始是为了chewy, 所以写死百分比了 :\
【引用自 元素轉換】:
不過我拿來嘗試的商品最優解並不是沒任何優惠的Amazon也不是有優惠的Walmart而是沒任何優惠的第三個網站
同上 为了chewy 所以只写了amazon的逻辑 但是改一改event id和网站就应该可以适配其他网站了
【引用自 元素轉換】:
实现邮件offer方面能展開來說說嗎？是一跑就讓C1S直接寄offer信嗎？即使目前有尚未過期的offer信也能讓C1S寄嗎？
网站query和event id query就是触发邮件offer的关键了 你说的寄信的细节我就不清楚了

【引用自 元素轉換】:
邮件offer
所以關於邮件offer你是在說程式自己拼出邮件offer的連結而不是觸發C1S寄邮件offer？

不是 脚本有你需要的代码 你可以调用来模拟访问网页 触发c1s决定给你发邮件offer

有没有傻瓜版，适合只会用浏览器看视频的玩家

学到了学到了

有没有多个账号的

自己让AI改一下

2樓不是說別丟AI？

【引用自 uhhhh】:
请勿转载/丢AI转语言
核心是转语言 转语言指的是我现在这个是用python写的 不要丢AI让他换个语言重写

感谢楼主！非常有用

请教下楼主这个脚本能刷email offer吗？那种email offer都会说

Exclusive Email Bonus offer (limit 1 Email Bonus per user)

这个怎么理解，是说只能用这个商家的一个offer （别的商家的offer还可以再用）？还是说一个user只能用一个email offer？

谢谢 我试一下

可以刷email offer 但是你要自己写一个main (或者让AI写) 来模拟访问网页触发c1s给你发邮件
per user 这个新用户的只会给你一次 不管商家的

求问cookie是每次都要换吗

等你浏览器登录掉了才需要更新