使用DeepSeek+本地知识库，尝试从0到1搭建高度定制化工作流（爬虫模块篇）
6.环境配置
python=3.9
langchain==0.1.13
langchain-community==0.0.29
torch==2.0.1
transformers>=4.38.2
timm>=0.9.16
accelerate
sentencepiece
attrdict
einops
# for gradio demo
gradio==3.48.0
gradio-client==0.6.1
mdtex2html==1.3.0
pypinyin==0.50.0
tiktoken==0.5.2
tqdm==4.64.0
colorama==0.4.5
Pygments==2.12.0
markdown==3.4.1
SentencePiece==0.1.96
requests
fake_useragent
hashlib
playwright
其中playwright安装后需要安装对应的浏览器驱动，使用
playwright install
进行安装。
7.模块实现
7.1. 代理池建立
-
目的：获取代理IP，用于爬取数据时使用，频繁使用本地的IP地址可能会导致IP被封禁。本文仅供展示，因此仅选取部分免费的代理IP爬取建立线程池，免费的代理IP大多数质量不高，因此需要定期更新代理池或使用付费代理。
-
实现思路： - 1.读取json文件中的代理IP获取网址，因为网址的每一页包含多条代理IP，并且网址的每一页的URL格式相同，因此可以采用多线程的方式快速获取所有代理IP。将获取到的代理IP存入txt文件中，每日定时更新。 - 2.模拟人为请求，在爬取过程中模拟访问浏览器，而非单纯使用request库发起请求，避免被网站封禁。
-
代码实现：
from playwright.sync_api import sync_playwright
import time
import random
from fake_useragent import UserAgent
import json
from concurrent.futures import ThreadPoolExecutor
def fetch_ip(url):
"""每个线程使用独立的浏览器实例"""
with sync_playwright() as p:
try:
# 初始化浏览器实例
browser = p.chromium.launch(
headless=True,
args=[
'--disable-blink-features=AutomationControlled',
'--no-sandbox',
'--disable-setuid-sandbox'
]
)
# 创建新的上下文
context = browser.new_context(
user_agent=get_stealth_headers()['User-Agent'],
viewport={'width': 1920, 'height': 1080},
extra_http_headers=get_stealth_headers()
)
# 反检测脚本
context.add_init_script("""
Object.defineProperty(navigator, 'webdriver', {
get: () => undefined
})
""")
page = context.new_page()
# 随机鼠标移动
page.mouse.move(
random.randint(0, 1920),
random.randint(0, 1080)
)
# 访问页面
page.goto(url, timeout=60000)
# 随机滚动
for _ in range(random.randint(1, 3)):
page.mouse.wheel(0, random.randint(300, 800))
time.sleep(0.5)
# 等待表格加载
page.wait_for_selector('table tbody tr', timeout=15000)
# 解析数据
rows = page.query_selector_all('table tbody tr')
ip_list = []
for row in rows:
try:
ip = row.query_selector('td:first-child').inner_text().strip()
port = row.query_selector('td:nth-child(2)').inner_text().strip()
ip_list.append(f'{ip}:{port}')
except:
continue
return ip_list
except Exception as e:
print(f'请求 {url} 失败: {str(e)}')
return []
finally:
# 确保资源释放
if 'page' in locals():
page.close()
if 'context' in locals():
context.close()
if 'browser' in locals():
browser.close()
def get_stealth_headers():
ua = UserAgent()
config = json.load(open('docs/ip_crawl_config.json', 'r', encoding='utf-8'))
return {
'User-Agent': ua.random,
'Accept-Language': 'en-US,en;q=0.9',
'Referer': config['referer']
}
def main():
config = json.load(open('docs/ip_crawl_config.json', 'r', encoding='utf-8'))
base_url = config['base_url']
urls = [f'{base_url}/{page_num}/' for page_num in range(1, 20)]
with ThreadPoolExecutor(max_workers=4) as executor:
results = executor.map(fetch_ip, urls)
with open('proxies.txt', 'w', encoding='utf-8') as f:
total = 0
for idx, ip_list in enumerate(results, 1):
if ip_list:
f.write('\n'.join(ip_list) + '\n')
print(f'第 {idx} 页获取到 {len(ip_list)} 条代理')
total += len(ip_list)
print(f'共获取 {total} 条代理')
if __name__ == '__main__':
main()
7.2. 数据爬取
-
目的：获取所要生成文案的相关数据，本项目所需要的是商品市场分析的有关数据，因此需要爬取对应平台以及第三方网站。
-
实现思路： - 1.分析网站数据爬取思路，若网站数据为静态加载，固定格式，可以直接通过python中的网页分析库进行爬取，例如beautifulsoup4等。 - 2.由于本项目所需爬取的网站结构相对复杂，因此使用了抓包的方式进行爬取。爬虫过程涉及js逆向分析，需要分析js代码，找到数据来源，通过python的requests库进行数据爬取。(感兴趣的朋友可以查看我先前发的文章，有详细讲解js逆向分析的过程) - 3.爬取数据后，对数据进行清洗，去除无用数据，将数据存储到本地json文件，方便后续使用。
-
代码实现：
'''
Author: yeffky
Date: 2025-02-09 17:18:05
LastEditTime: 2025-02-15 15:33:48
'''
import requests
import random
import time
import json
from fake_useragent import UserAgent
import hashlib
import time
from datetime import datetime
# 定义md5加密函数
def md5_hash(s):
return hashlib.md5(s.encode('utf-8')).hexdigest()
# Le函数，类似于JS中的Le函数
def getAuth(e):
# 第一次MD5，将输入转换为字符串并计算MD5
first_md5 = hashlib.md5(str(e).encode()).hexdigest()
# 拼接字符串并第二次计算MD5
second_md5 = hashlib.md5(combined.encode()).hexdigest()
return second_md5
def get_random_proxy():
# 尝试打开名为 'proxies.txt' 的文件
try:
with open('proxies.txt', 'r') as f:
# 读取文件内容并按行分割成列表
proxies = f.read().splitlines()
# 如果列表不为空，则随机选择一个代理并返回
if proxies:
return random.choice(proxies)
# 如果文件不存在，则捕获 FileNotFoundError 异常并忽略
except FileNotFoundError:
pass
# 如果文件不存在或列表为空，则返回 None
return None
def remove_invalid_proxy(proxy):
# 尝试执行以下代码块，如果发生异常则跳转到except块
try:
# 打开名为'proxies.txt'的文件，以读取模式('r')
with open('proxies.txt', 'r') as f:
proxies = f.read().splitlines()
if proxy in proxies:
# 检查传入的proxy是否在proxies列表中
proxies.remove(proxy)
with open('proxies.txt', 'w') as f:
f.write('\n'.join(proxies) + '\n')
except FileNotFoundError:
pass
def make_request(url, payload, auth, timestamp):
# 创建一个UserAgent对象，用于生成随机的User-Agent字符串
ua = UserAgent()
headers = {'User-Agent': ua.random, 'Auth': str(auth), 'Timestamp': str(timestamp), "Content-Type":"application/json",}
proxy = get_random_proxy() # 获取一个随机的代理
proxies = {'http': proxy, 'https': proxy} if proxy else None
try:
response = requests.post(
url,
headers=headers,
timeout=10,
proxies=proxies,
json=payload,
# verify=False
)
response.raise_for_status()
print(response)
return response.json()
except requests.exceptions.RequestException as e:
print(f"Request failed: {e}")
# If the request failed, remove the invalid proxy
if proxy:
remove_invalid_proxy(proxy)
# Try a new proxy
return None
def crawl_data():
# 加载爬虫配置文件
crawal_config = json.load(open('./docs/crawl_config.json', 'r'))
base_url = crawal_config['base_url'] # 获取基础URL
payload = crawal_config['payload'] # 获取请求负载
response_text = None
while not response_text: # Retry until the request is successful
# 时间戳
t = int((time.time()) * 1000) # 获取当前时间戳（毫秒）
auth = getAuth(t)
response_text = make_request(base_url, payload, auth, t)
if not response_text:
print("Retrying with a new proxy...")
time.sleep(random.randint(3, 5)) # Add a small delay before retrying
try:
data = response_text
items_list = data.get('data', {}).get('list', [])
# 获取今天的日期
today_date = datetime.now().strftime('%Y-%m-%d')
# 在文件名后加上今天的日期
filename = f'goods_{today_date}.json'
cnt = 0
with open(f'data/{filename}', 'w', encoding='utf-8') as f:
f.write('{"items": [' + '\n')
for item in items_list:
item['商品涨幅'] = item.pop('Priceincrease')
item['商品成交量'] = item.pop('SaleCount')
item['商品名称'] = item.pop('GoodsName')
item['商品最低价'] = item.pop('price')
item['商品图标'] = item.pop('iconUrl')
cnt += 1
item_str = json.dumps(item, ensure_ascii=False, indent=2)
if cnt == len(items_list):
f.write(item_str + '\n')
else:
f.write(item_str + ',\n\n')
f.write(']}' + '\n')
print(f"Successfully wrote {len(items_list)} items to goods.txt")
except json.JSONDecodeError:
print("Failed to decode JSON response")
if __name__ == "__main__":
crawl_data()