窥见隔壁塔吉特邮编大法半价购买培根果汁等等的神秘交易，以半价购得无数水果生鲜。然而，总需等待他人的邮编，遂写下一个隐秘脚本，遍历所有商店，查询最终价格。为免重复劳动，我将所有塔吉特商店的信息存入store_data.csv文件，此处共享。运行神秘程序target.py，输入欲购买商品的TCIN编号（即塔吉特商品网址中A-后的数字串），程序即会遍历所有商店价格，输出打折的store_id。所有商店价格亦可保存至对应csv文件中，手动根据store_id在store_data.csv中查询店铺位置与邮编即可。
target.py 1.0
import requests
import pandas as pd
import time
import random
from tqdm import tqdm
import socks
import socket
import threading
import csv
USING_PROXY = True
TRYING_TIMES = 5
# Function to fetch and parse the latest SOCKS5 proxy list
def fetch_proxies():
url = "https://raw.githubusercontent.com/proxifly/free-proxy-list/main/proxies/protocols/socks5/data.txt"
response = requests.get(url)
if response.status_code == 200:
proxies = response.text.splitlines()
# Validate and clean proxy list
valid_proxies = [proxy.strip() for proxy in proxies if proxy.startswith("socks5://")]
return valid_proxies
else:
print(f"Failed to download proxy list with status code: {response.status_code}")
return []
# Function to periodically update the proxy list
def update_proxies():
global proxies_list
while True:
print("Updating proxy list...")
proxies_list = fetch_proxies()
print(f"Updated proxy list with {len(proxies_list)} proxies.")
time.sleep(300) # Update every 5 minutes
if USING_PROXY:
# Initialize proxy list
proxies_list = fetch_proxies()
# Start a background thread to update the proxy list periodically
proxy_thread = threading.Thread(target=update_proxies, daemon=True)
proxy_thread.start()
# Function to generate random headers
def generate_random_headers():
user_agent_templates = [
"Mozilla/5.0 (Windows NT {win_ver}; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/{chrome_ver} Safari/537.36",
"Mozilla/5.0 (Windows NT {win_ver}; Win64; x64; rv:{firefox_ver}) Gecko/20100101 Firefox/{firefox_ver}",
"Mozilla/5.0 (Macintosh; Intel Mac OS X {mac_ver}) AppleWebKit/605.1.15 (KHTML, like Gecko) Version/{safari_ver} Safari/605.1.15",
"Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/{chrome_ver} Safari/537.36",
"Mozilla/5.0 (Windows NT {win_ver}; WOW64; rv:{firefox_ver}) Gecko/20100101 Firefox/{firefox_ver}"
]
accept_language_options = [
"en-US,en;q=0.9",
"en-GB,en;q=0.9",
"en-CA,en;q=0.9",
"en-AU,en;q=0.9",
"fr-FR,fr;q=0.9,en;q=0.8"
]
accept_encoding_options = [
"gzip, deflate, br",
"gzip, deflate",
"br"
]
template = random.choice(user_agent_templates)
win_ver = f"{random.randint(6, 10)}.{random.randint(0, 1)}"
chrome_ver = f"{random.randint(70, 90)}.0.{random.randint(3000, 4000)}.{random.randint(100, 150)}"
firefox_ver = f"{random.randint(60, 89)}.0"
mac_ver = f"10_{random.randint(10, 15)}_{random.randint(1, 6)}"
safari_ver = f"{random.randint(10, 14)}.{random.randint(0, 1)}.{random.randint(1, 3)}"
user_agent = template.format(win_ver=win_ver, chrome_ver=chrome_ver, firefox_ver=firefox_ver, mac_ver=mac_ver, safari_ver=safari_ver)
accept_language = random.choice(accept_language_options)
accept_encoding = random.choice(accept_encoding_options)
headers = {
"User-Agent": user_agent,
"Accept": "application/json, text/plain, */*",
"Accept-Encoding": accept_encoding,
"Accept-Language": accept_language,
"Connection": "keep-alive"
}
return headers
def set_proxy(proxy):
proxy_parts = proxy.replace("socks5://", "").split(":")
socks.set_default_proxy(socks.SOCKS5, proxy_parts[0], int(proxy_parts[1]))
socket.socket = socks.socksocket
def reset_proxy():
socks.set_default_proxy(None)
socket.socket = socket.socket
def remove_invalid_proxy(proxy):
global proxies_list
proxies_list = [p for p in proxies_list if p != proxy]
def fetch_data():
# 提示用户输入 tcin
print("提示: 14911563: T Bone, 54273119: Ribeye, 54640786: New York Strip")
tcin = input("请输入 tcin: ")
# 目标URL的基础部分
url_base = "https://redsky.target.com/redsky_aggregations/v1/web/pdp_client_v1?key=9f36aeafbe60771e321a7cc95a78140772ab3e96&tcin={tcin}&is_bot=false&store_id={store_id}&pricing_store_id={pricing_store_id}&has_pricing_store_id=true"
# Read existing store IDs
try:
store_data = pd.read_csv("store_data.csv")
store_ids = store_data['store_id'].tolist()
postal_codes = store_data['postal_code'].tolist()
location_names = store_data['location_name'].tolist()
except FileNotFoundError:
print("Store data CSV file not found.")
return
# 创建一个CSV文件来存储结果
filename = f"price_data_tcin_{tcin}.csv"
with open(filename, mode='w', newline='', encoding='utf-8') as file:
writer = csv.writer(file)
# 写入表头
writer.writerow(["store_id", "location_name", "postal_code", "current_retail", "formatted_current_price", "reg_retail"])
if USING_PROXY:
proxy = random.choice(proxies_list)
set_proxy(proxy)
# 遍历 store_id
for store_id, postal_code, location_name in tqdm(zip(store_ids, postal_codes, location_names), desc=f"Fetching price data for tcin {tcin}", unit="request", total=len(store_ids)):
headers = generate_random_headers()
for _ in range(TRYING_TIMES):
# 格式化URL
url = url_base.format(tcin=tcin, store_id=store_id, pricing_store_id=store_id)
# 发送GET请求
try:
response = requests.get(url, headers=headers, timeout=10)
# 确认请求成功
if response.status_code == 200:
# 解析JSON响应
data = response.json()
# 获取"price"字段
price_info = data.get("data", {}).get("product", {}).get("price", {})
# 提取需要的信息
current_retail = price_info.get("current_retail")
formatted_current_price = price_info.get("formatted_current_price")
reg_retail = price_info.get("reg_retail")
# 写入CSV文件
writer.writerow([store_id, location_name, postal_code, current_retail, formatted_current_price, reg_retail])
if "formatted_current_price_type" in price_info and price_info["formatted_current_price_type"] != "reg":
# 更新进度条后缀信息
tqdm.write(f"Store ID: {store_id}, Location Name: {location_name}, Postal Code: {postal_code}, Current Price: {formatted_current_price}")
break
elif response.status_code == 404:
# 换一个User-Agent
if USING_PROXY:
remove_invalid_proxy(proxy)
proxy = random.choice(proxies_list)
set_proxy(proxy)
headers = generate_random_headers()
print(f"Received 404 for store_id {store_id}, changing Proxy and User-Agent.")
else:
print(f"Request failed for store_id {store_id} with status code: {response.status_code}")
break
except requests.exceptions.RequestException as e:
print(f"Request exception for store_id {store_id}: {e}")
if USING_PROXY:
remove_invalid_proxy(proxy)
proxy = random.choice(proxies_list)
set_proxy(proxy)
headers = generate_random_headers()
else:
break
# 避免发送请求过快，加入随机延迟
# time.sleep(random.uniform(0.5, 1.5))
reset_proxy()
print(f"数据已成功保存到 {filename} 文件中")
# Run the main function
fetch_data()
Update：提示：用多了，可能会被Target封IP，可能过两天就好了。或者修改 USING_PROXY = True （不确保代理是否有效）
Update：版本0.2： 为了避免封ip， 默认启用代理 USING_PROXY = True （可以自行关闭False，提升速度，风险自理），设置 TRYING_TIMES = 5 （每个store_id最大尝试5次）。
Update：
Update：版本1.0：
以及 target 店铺位置名称
