# 导入所需依赖库
import requests
from bs4 import BeautifulSoup
import re
import pandas as pd
import time
# 1. 配置全局请求头,模拟浏览器访问,规避基础反爬
HEADERS = {
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Geckocko) Chrome/120.0.0.0 Safari/537.36",
"Referer": "https://fxg.jinritemai.com/",
"Accept-Language": "zh-CN,zh;q=0.9"
}
# 2. 正则表达式匹配手机号、固定电话
PHONE_PATTERN = re.compile(r'1[3-9]\d{9}|0\d{2,3}-\d{7,8}')
# 3. 初始化存储列表
merchant_list = []
# 4. 定义核心采集函数
def get_douyin_merchant_data(page_url):
"""
采集抖音精选联盟商家数据
:param page_url: 精选联盟商家列表页面链接
:return: 单页商家数据列表
"""
try:
# 发送网络请求
response = requests.get(url=page_url, headers=HEADERS, timeout=10)
response.encoding = "utf-8"
html = response.text
# 解析网页DOM结构
soup = BeautifulSoup(html, "html.parser")
merchant_items = soup.find_all("div", class_="merchant-item-list")
# 遍历单页所有商家
for item in merchant_items:
merchant_info = {}
# 采集店铺基础信息
merchant_info["shop_name"] = item.find("div", class_="shop-name").get_text(strip=True) if item.find("div", class_="shop-name") else ""
merchant_info["shop_id"] = item.find("div", class_="shop-id").get_text(strip=True) if item.find("div", class_="shop-id") else ""
merchant_info["category"] = item.find("div", class_="shop-category").get_text(strip=True) if item.find("div", class_="shop-category") else ""
merchant_info["sales_volume"] = item.find("div", class_="sales-num").get_text(strip=True) if item.find("div", class_="sales-num") else ""
merchant_info["commission_rate"] = item.find("div", class_="commission").get_text(strip=True) if item.find("div", class_="commission") else ""
# 提取页面所有文本,匹配联系电话
page_text = soup.get_text(separator=" ", strip=True)
phone_res = PHONE_PATTERN.findall(page_text)
merchant_info["contact_phone"] = phone_res[0] if phone_res else "无公开电话"
merchant_list.append(merchant_info)
print(f"页面数据采集完成:{page_url}")
return merchant_list
except Exception as e:
print(f"页面采集异常:{str(e)}")
return None
# 5. 批量分页采集(可自定义页码范围)
def batch_collect_data(base_url, start_page, end_page):
"""批量遍历分页采集商家数据"""
for page in range(start_page, end_page + 1):
page_url = f"{base_url}&page={page}"
get_douyin_merchant_data(page_url)
time.sleep(2) # 延时请求,降低访问频率,规避反爬
# 6. 数据清洗与去重处理
def data_clean():
"""数据去重、缺失值统一处理"""
df = pd.DataFrame(merchant_list)
# 根据店铺ID、电话双维度去重
df = df.drop_duplicates(subset=["shop_id", "contact_phone"], keep="first")
# 统一填充空值
df["shop_name"] = df["shop_name"].fillna("未知店铺")
df["category"] = df["category"].fillna("未知类目")
return df
# 7. 数据导出函数
def export_data(df, save_path="抖音精选联盟商家数据.csv"):
"""导出结构化数据为CSV格式"""
df.to_csv(save_path, index=False, encoding="utf-8-sig")
print(f"数据导出完成,共采集有效商家数据:{len(df)}条")
# 主程序入口
if __name__ == "__main__":
# 此处替换为实际精选联盟商家列表接口/页面链接
BASE_URL = "https://fxg.jinritemai.com/backend/merchant/list"
# 采集1-10页数据,可自定义修改
batch_collect_data(BASE_URL, 1, 10)
# 数据清洗
clean_df = data_clean()
# 数据导出
export_data(clean_df)
来自 “ ITPUB博客 ” ,链接:https://blog.itpub.net/70050905/viewspace-3184467/ ,如需转载,请注明出处,否则将追究法律责任。