亮数据与 AI 深度集成：构建电商策略自动化系统新范式 - qidao123.com技术社区-IT企服评测·应用市场

from selenium.webdriver import Remote, ChromeOptions
from selenium.webdriver.common.by import By
from selenium.webdriver.chromium.remote_connection import ChromiumRemoteConnection
import pandas as pd
# 替换为你自己的信息即可
AUTH = 'brd-customer-您的客户 ID-zone-您的区域:您的密码'
SBR_WEBDRIVER = f'https://{AUTH}@brd.superproxy.io:9515'
def main():
print('连接到 Scraping Browser...')
sbr_connection = ChromiumRemoteConnection(SBR_WEBDRIVER, 'goog', 'chrome')
with Remote(sbr_connection, options=ChromeOptions()) as driver:
print('连接成功！正在导航到亚马逊电脑产品列表...')
driver.get('https://www.amazon.com/s?k=laptop')
print('页面加载完成！正在提取商品链接...')
driver.implicitly_wait(10)
# 获取所有商品的容器
products = driver.find_elements(By.XPATH, '//div[@data-component-type="s-search-result"]')
product_links = []
for link in product_links:
print(f'正在访问商品页面: {link}')
driver.get(link)
try:
# 找到评价页面链接
review_link = WebDriverWait(driver, 10).until(
EC.presence_of_element_located((By.XPATH, '//a[@data-hook="see-all-reviews-link-foot"]'))
).get_attribute('href')
driver.get(review_link)
# 获取该款式的所有评价数
try:
total_reviews_count = WebDriverWait(driver, 10).until(
EC.presence_of_element_located((By.XPATH, '//div[@data-hook="total-review-count"]'))
).text
except:
total_reviews_count = None
# 获取评价容器
reviews = WebDriverWait(driver, 10).until(
EC.presence_of_all_elements_located((By.XPATH, '//div[@data-hook="review"]'))
)
for review in reviews:
try:
# 提取评价标题
title = review.find_element(By.XPATH, './/a[@data-hook="review-title"]').text
except:
title = None
try:
# 提取评价内容
content = review.find_element(By.XPATH, './/span[@data-hook="review-body"]').text
except:
content = None
try:
# 提取评价星级
rating = review.find_element(By.XPATH, './/i[@data-hook="review-star-rating"]').text
except:
rating = None
try:
# 提取评价者
reviewer = review.find_element(By.XPATH, './/span[@class="a-profile-name"]').text
except:
reviewer = None
try:
# 提取评价时间
review_date = review.find_element(By.XPATH, './/span[@data-hook="review-date"]').text
except:
review_date = None
try:
# 提取评价的评价数
review_helpful_count = review.find_element(By.XPATH, './/span[@data-hook="helpful-vote-statement"]').text
except:
review_helpful_count = None
all_reviews.append({
'评价网址': review_link,
'评价标题': title,
'评价内容': content,
'评价星级': rating,
'评价者': reviewer,
'评价时间': review_date,
'该款式的所有评价数': total_reviews_count,
'评价的评价数': review_helpful_count
})
except:
print('未找到该商品的评价页面')
# 将数据保存到 DataFrame
df = pd.DataFrame(all_reviews)
# 将数据保存到 CSV 文件
df.to_csv('amazon_laptop_reviews.csv', index=False, encoding='utf-8-sig')
print('数据提取完成，已保存到 amazon_laptop_reviews.csv')
if __name__ == '__main__':
main()

复制代码

# 实时数据采集与AI分析管道
from brightdata_scraper import ScrapingBrowser
from deepseek_ai import StrategyModel
import pandas as pd
import json
class AutoStrategySystem:
def __init__(self, auth_token):
self.sb_client = ScrapingBrowser(auth_token)
self.ai_model = StrategyModel(api_key="DEEPSEEK_API_KEY")
self.data_pipeline = DataProcessor()
def execute_pipeline(self, target_url):
# 自动化数据采集
raw_data = self.sb_client.scrape(
url=target_url,
params={
"output_format": "json",
"anti_block": {"proxy_rotation": True},
"captcha_solver": "auto"
}
)
# 实时数据清洗
processed_data = self.data_pipeline.transform(
raw_data,
rules={
"field_mapping": {
"user_rating": "评价星级",
"purchase_time": "购买时间戳"
},
"anonymize_fields": ["user_id"]
}
)
# AI策略生成
strategy_report = self.ai_model.generate_strategy(
input_data=processed_data,
business_goals={
"revenue_growth": 25,
"retention_rate": 40
}
)
return strategy_report
# 实例化并执行（亚马逊笔记本电脑品类）
system = AutoStrategySystem(auth_token="BRD_AUTH_TOKEN")
strategy = system.execute_pipeline(
"https://www.amazon.com/s?k=laptop"
)
print(json.dumps(strategy, indent=2))

复制代码

# 智能IP轮换算法（动态调整代理策略）
def adaptive_proxy_rotation(attempts):
base_delay = 2**attempts + random.uniform(0, 1) # 指数退避策略
geo_distribution = get_user_geo_distribution() # 获取用户地域分布
return {
"proxy_pool": select_optimal_proxies(geo_distribution), # 按地域优选代理IP
"request_interval": base_delay * 0.8, # 动态调整请求间隔
"retry_strategy": "exponential_backoff" # 指数级重试策略
}

复制代码

# 实时行为特征编码（提取用户核心行为指标）
def extract_behavior_features(logs):
return pd.DataFrame({
'session_duration': logs['page_leave_time'] - logs['page_enter_time'], # 会话时长
'price_sensitivity': np.log(logs['view_count'] / logs['purchase_count']), # 价格敏感度（浏览-购买比）
'category_affinity': calculate_entropy(logs['browse_categories']) # 品类偏好熵值
})

复制代码

# 自动生成营销策略代码（基于用户价值分层）
def compile_strategy(params):
template = f"""
if user_value == 'high':
apply_discount({params['high_discount']}) # 高价值用户专属折扣
recommend_{params['high_product']}() # 推荐高端产品
elif user_value == 'mid':
show_countdown('{params['mid_deadline']}') # 倒计时促销
apply_coupon({params['mid_coupon']}) # 发放中等面额优惠券
"""
return dynamic_code_generator(template) # 动态生成可执行代码

复制代码