金融爬虫实战：实时抓取雪球网股吧讨论热度的技术解析

1https://xueqiu.com/query/v1/symbol/search/status?symbol=SH000300&page=1&size=10

python1import requests
2import json
3
4headers = {
5    'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36',
6    'Cookie': 'xq_a_token=YOUR_TOKEN; xq_r_token=YOUR_RTOKEN',  # 需替换为实际值
7    'Referer': 'https://xueqiu.com/S/SH000300'
8}
9
10url = 'https://xueqiu.com/query/v1/symbol/search/status'
11params = {
12    'symbol': 'SH000300',
13    'page': 1,
14    'size': 10
15}
16
17response = requests.get(url, headers=headers, params=params)
18data = json.loads(response.text)

python1comments = []
2for item in data['list']:
3    comment = {
4        'text': item['text'].replace('<em>', '').replace('</em>', ''),  # 去除高亮标签
5        'time': item['timeBefore'],
6        'user_id': item['user']['id'],
7        'user_name': item['user']['screen_name']
8    }
9    comments.append(comment)

python1import time
2import random
3
4def fetch_comments(symbol, max_pages=5):
5    all_comments = []
6    for page in range(1, max_pages + 1):
7        params['page'] = page
8        try:
9            response = requests.get(url, headers=headers, params=params, timeout=10)
10            if response.status_code == 200:
11                data = json.loads(response.text)
12                if 'list' in data and data['list']:
13                    # 数据解析逻辑同上
14                    all_comments.extend(comments)
15            time.sleep(random.uniform(1, 3))  # 随机延迟
16        except Exception as e:
17            print(f"Page {page} error: {e}")
18            continue
19    return all_comments

python1import pymysql
2
3def save_to_mysql(comments):
4    conn = pymysql.connect(
5        host='localhost',
6        user='root',
7        password='your_password',
8        database='xueqiu',
9        charset='utf8mb4'
10    )
11    cursor = conn.cursor()
12    
13    sql = """
14    INSERT INTO comments 
15    (text, create_time, user_id, user_name) 
16    VALUES (%s, %s, %s, %s)
17    """
18    
19    for comment in comments:
20        cursor.execute(sql, (
21            comment['text'],
22            comment['time'],
23            comment['user_id'],
24            comment['user_name']
25        ))
26    conn.commit()
27    cursor.close()
28    conn.close()

python1# 使用代理池示例
2def get_proxy():
3    # 假设已搭建代理池API
4    response = requests.get('http://proxy-pool.com/random')
5    return response.text.strip()
6
7proxies = {
8    'http': f'http://{get_proxy()}',
9    'https': f'https://{get_proxy()}'
10}
11
12response = requests.get(url, headers=headers, params=params, proxies=proxies)

python1headers = {
2    'Accept': 'application/json, text/plain, */*',
3    'Accept-Language': 'zh-CN,zh;q=0.9',
4    'Connection': 'keep-alive',
5    'Sec-Fetch-Dest': 'empty',
6    'Sec-Fetch-Mode': 'cors',
7    'Sec-Fetch-Site': 'same-origin',
8    'X-Requested-With': 'XMLHttpRequest'
9}

python1from concurrent.futures import ThreadPoolExecutor
2
3def fetch_page(page):
4    params['page'] = page
5    # 请求逻辑同上
6    return comments
7
8with ThreadPoolExecutor(max_workers=5) as executor:
9    results = list(executor.map(fetch_page, range(1, 6)))
10all_comments = [item for sublist in results for item in sublist]

python1def remove_duplicates(comments):
2    seen = set()
3    unique_comments = []
4    for comment in comments:
5        # 使用评论ID或内容MD5作为唯一标识
6        identifier = comment['id'] if 'id' in comment else hash(comment['text'])
7        if identifier not in seen:
8            seen.add(identifier)
9            unique_comments.append(comment)
10    return unique_comments

python1def fetch_incremental(symbol, last_time):
2    all_comments = fetch_comments(symbol)
3    incremental = [c for c in all_comments if c['time'] > last_time]
4    return incremental

python1response.encoding = 'utf-8'  # 或 response.apparent_encoding