配了代理 IP 却没变？Python requests 代理配置的 7 个致命坑

说真的，代理这东西你要是只用来"代理访问"那就太浪费了。这篇文章我带你用隧道代理搭两个真正能落地的系统——一个盯着全网价格给你省钱，一个盯着全网舆论给你挡刀。代码全是能直接跑的，别光收藏，动手试试。

一、先唠明白：为啥这俩事儿非得用代理？

别急着写代码，先想清楚一个问题：你不用代理行不行？

1.1 电商比价有多惨？

你想盯着某宝某东某多多上同一款耳机的价格，手动翻页？别闹了，几十上百个商品翻到眼花。写个爬虫直接跑？前50个请求还美滋滋，第51个直接给你甩个403——IP被封了，气不气？

电商平台的反爬三板斧：

IP频率限制：一个IP短时间狂刷商品页，不封你封谁？

行为分析：你请求间隔精确到毫秒，连个鼠标抖动都没有，不是机器人是啥？

地域杀：同一件商品北京卖599，上海卖549，你一个IP只能看到一个价，比了个寂寞。

没代理？要么被封，要么数据不全，二选一吧。

1.2 舆情监控有多难？

你想看看自家品牌在某博某乎某红书上口碑咋样，结果：

某博搜索页要登录，不登录？给你看个寂寞

某红书反爬狠到什么程度？数据中心IP过去基本是送人头，10个有9.9个被拦

同一个关键词你搜多了？账号直接限流，严重的给你封了

舆情监控的命门是又广又快——平台要多、发现负面要趁早。没个代理池撑着，根本跑不起来。

1.3 隧道代理为啥是这俩场景的天菜？

隧道代理的架构很简单，就三层：

你的代码 → 代理网关（地址固定不变）→ IP池（自动轮换，每次请求出口IP都不一样）→ 目标网站

好在哪呢？

地址固定，IP自动换：你代码里不用维护一大堆IP列表，填一个地址就行，剩下的网关帮你搞定

扛得住并发：网关层做负载均衡，并发量上去也不慌

匿名度拉满：不携带任何代理标识头，目标网站根本看不出你用了代理

想选哪的IP选哪的：指定出口城市/国家，电商地域价格问题直接解决

划重点：本文所有代码都是基于隧道代理模式写的，你把代理地址换成自己的就能跑，就这么简单。

二、开工前的准备：一个万能请求函数

不管是比价还是舆情，我们都需要一个靠谱的请求函数。代理配置、重试、超时、随机UA，全给它封装到一起，后面直接调用就行。

2.1 先装依赖

```bash

pip install requests beautifulsoup4 lxml pandas schedule snownlp jieba

```

2.2 封装请求函数

```python

import requests

from requests.adapters import HTTPAdapter

from urllib3.util.retry import Retry

import random

import time

# ========== 代理配置（把下面换成你自己的隧道代理地址）==========

PROXY_HOST = "你的代理网关地址"

PROXY_PORT = 端口号

PROXY_USER = "用户名"

PROXY_PASS = "密码"

PROXY_URL = f"http://{PROXY_USER}:{PROXY_PASS}@{PROXY_HOST}:{PROXY_PORT}"

PROXIES = {"http": PROXY_URL, "https": PROXY_URL}

# ========================================================

# UA池，每次请求随机挑一个，别老是用同一个

UA_POOL = [

"Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36",

"Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36",

"Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:121.0) Gecko/20100101 Firefox/121.0",

"Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/605.1.15 (KHTML, like Gecko) Version/17.2 Safari/605.1.15",

]

def create_session():

"""创建带自动重试的session，遇到502/503/504/429自动重试3次"""

session = requests.Session()

retries = Retry(

total=3,

backoff_factor=1,

status_forcelist=[502, 503, 504, 429],

)

session.mount("http://", HTTPAdapter(max_retries=retries))

session.mount("https://", HTTPAdapter(max_retries=retries))

return session

def fetch_url(url, session=None, timeout=15, referer=None):

"""

统一的带代理请求函数

返回: (response对象, 错误信息)，成功时error为None

"""

if session is None:

session = create_session()

headers = {

"User-Agent": random.choice(UA_POOL),

"Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,*/*;q=0.8",

"Accept-Language": "zh-CN,zh;q=0.9,en;q=0.8",

"Accept-Encoding": "gzip, deflate, br",

"Connection": "keep-alive",

}

if referer:

headers["Referer"] = referer

try:

resp = session.get(url, proxies=PROXIES, headers=headers, timeout=timeout)

resp.raise_for_status()

return resp, None

except requests.exceptions.ProxyError as e:

return None, f"代理连接失败: {e}"

except requests.exceptions.Timeout:

return None, "请求超时"

except requests.exceptions.HTTPError as e:

return None, f"HTTP错误: {e.response.status_code}"

except Exception as e:

return None, f"未知错误: {e}"

def human_delay(base=2.0, jitter=1.5):

"""

模拟人类操作的随机延迟，别老是精确等2秒

base是基础秒数，jitter是上下浮动范围

"""

delay = base + random.uniform(-jitter, jitter)

time.sleep(max(0.5, delay))

```

这个 fetch_url 函数是后面两个系统的基石，代理、重试、随机UA、超时、延迟全给你处理好了。说真的，你以后写任何爬虫都可以直接抄这个，省老事了。

三、实战一：搞一个多平台电商比价系统

3.1 我们要做啥？

很简单，就四件事：

1. 把你想监控的商品URL丢进去

2. 定时去各个平台抓价格，存到数据库

3. 价格降了或者达到心理价位了，自动提醒你

4. 生成比价报告，一眼看出哪个平台最便宜

就这么点事儿，但能帮你省下真金白银——毕竟谁的钱也不是大风刮来的对吧？

3.2 系统长啥样？

商品URL配置 → 定时采集引擎 → 价格历史数据库

↓ ↓

代理IP池 价格变动检测

↓

告警通知（邮件/Webhook）

3.3 数据库设计

用SQLite，轻量到爆炸，不用装任何额外服务。三张表就够了：

```python

import sqlite3

from datetime import datetime

DB_PATH = "price_monitor.db"

def init_db():

conn = sqlite3.connect(DB_PATH)

c = conn.cursor()

# 商品表：你要监控哪些东西

c.execute("""

CREATE TABLE IF NOT EXISTS products (

id INTEGER PRIMARY KEY AUTOINCREMENT,

sku TEXT UNIQUE, -- 商品唯一标识，自己起

name TEXT, -- 商品名称

platform TEXT, -- 平台标识

url TEXT, -- 商品页URL

target_price REAL, -- 心理价位，低于这个价就提醒

alert_enabled INTEGER DEFAULT 1, -- 开不开提醒

created_at TEXT

)

""")

# 价格历史表：每次抓到的价格都存在这里

c.execute("""

CREATE TABLE IF NOT EXISTS price_history (

id INTEGER PRIMARY KEY AUTOINCREMENT,

sku TEXT,

price REAL,

original_price REAL, -- 划线价/原价

currency TEXT DEFAULT 'CNY',

in_stock INTEGER, -- 有没有货

fetched_at TEXT,

FOREIGN KEY (sku) REFERENCES products(sku)

)

""")

# 告警记录表：每次触发提醒都记一笔

c.execute("""

CREATE TABLE IF NOT EXISTS alerts (

id INTEGER PRIMARY KEY AUTOINCREMENT,

sku TEXT,

alert_type TEXT, -- price_drop降价 / below_target达到目标价

message TEXT,

created_at TEXT

)

""")

conn.commit()

conn.close()

print("数据库初始化完成")

```

3.4 价格采集器：一个方法搞定多个平台

各大电商页面结构不一样，总不能写一堆if-else吧？用策略模式，每个平台一个解析函数，想加新平台就加一个方法，优雅得很。

```python

from bs4 import BeautifulSoup

import re

class PriceExtractor:

"""价格提取器：支持某东、某宝/某猫、某多多"""

@staticmethod

def extract_jd(html):

"""某东价格提取"""

soup = BeautifulSoup(html, "lxml")

# 先试试从页面内嵌的JSON里抠，这个最稳

script = soup.find("script", string=re.compile(r"pageConfig"))

if script:

match = re.search(r'"price":"([\d.]+)"', script.string)

if match:

return float(match.group(1))

# JSON里没有？那就从页面元素里找

price_elem = soup.select_one(".p-price .price, .price")

if price_elem:

price_text = price_elem.get_text(strip=True).replace("¥", "").replace(",", "")

return float(price_text)

return None

@staticmethod

def extract_taobao(html):

"""某宝/某猫价格提取"""

soup = BeautifulSoup(html, "lxml")

script = soup.find("script", string=re.compile(r"g_page_config"))

if script:

try:

match = re.search(r'"price":"([\d.]+)"', script.string)

if match:

return float(match.group(1))

except:

pass

price_elem = soup.select_one(".tm-price, .tb-rmb-num, [class*='price']")

if price_elem:

price_text = price_elem.get_text(strip=True).replace("¥", "").replace(",", "")

try:

return float(price_text)

except:

pass

return None

@staticmethod

def extract_pdd(html):

"""某多多价格提取"""

soup = BeautifulSoup(html, "lxml")

script = soup.find("script", string=re.compile(r"rawData"))

if script:

match = re.search(r'"min_on_sale_group_price":([\d]+)', script.string)

if match:

# 注意：某多多价格单位是分，要除以100

return int(match.group(1)) / 100

return None

@classmethod

def extract(cls, html, platform):

"""根据平台名调用对应的提取方法"""

methods = {

"jd": cls.extract_jd,

"taobao": cls.extract_taobao,

"pdd": cls.extract_pdd,

}

method = methods.get(platform)

if method:

return method(html)

return None

```

说句大实话：电商平台的页面结构说变就变，比女生变脸还快。上面的选择器是通用思路，实际用的时候建议先打开浏览器开发者工具，找到价格元素的真实选择器再填进去。遇到JS渲染的页面？那得上Playwright了，后面会说。

3.5 采集引擎+价格变动检测

```python

def add_product(sku, name, platform, url, target_price=None):

"""添加一个要监控的商品"""

conn = sqlite3.connect(DB_PATH)

c = conn.cursor()

try:

c.execute(

"INSERT INTO products (sku, name, platform, url, target_price, created_at) VALUES (?, ?, ?, ?, ?, ?)",

(sku, name, platform, url, target_price, datetime.now().isoformat())

)

conn.commit()

print(f"已添加监控: {name} ({platform})")

except sqlite3.IntegrityError:

print(f"这个商品已经在监控列表里了: {sku}")

finally:

conn.close()

def get_latest_price(sku):

"""查一下这个商品上次抓到的价格"""

conn = sqlite3.connect(DB_PATH)

c = conn.cursor()

c.execute("SELECT price FROM price_history WHERE sku=? ORDER BY fetched_at DESC LIMIT 1", (sku,))

row = c.fetchone()

conn.close()

return row[0] if row else None

def save_price(sku, price, original_price=None, in_stock=1):

"""把这次抓到的价格存起来"""

conn = sqlite3.connect(DB_PATH)

c = conn.cursor()

c.execute(

"INSERT INTO price_history (sku, price, original_price, in_stock, fetched_at) VALUES (?, ?, ?, ?, ?)",

(sku, price, original_price, in_stock, datetime.now().isoformat())

)

conn.commit()

conn.close()

def check_and_alert(sku, name, current_price, target_price):

"""看看这次价格有没有值得提醒的变化"""

alerts = []

last_price = get_latest_price(sku)

# 降价超过5%就提醒一下

if last_price and current_price < last_price:

drop_pct = (last_price - current_price) / last_price * 100

if drop_pct >= 5:

msg = f"【降价提醒】{name} 从 ¥{last_price} 降到 ¥{current_price}，便宜了 {drop_pct:.1f}%"

alerts.append(("price_drop", msg))

# 达到心理价位了，必须提醒！

if target_price and current_price <= target_price:

msg = f"【达到目标价】{name} 现在 ¥{current_price}，比你的心理价 ¥{target_price} 还低！冲不冲？"

alerts.append(("below_target", msg))

if alerts:

conn = sqlite3.connect(DB_PATH)

c = conn.cursor()

for alert_type, msg in alerts:

c.execute(

"INSERT INTO alerts (sku, alert_type, message, created_at) VALUES (?, ?, ?, ?)",

(sku, alert_type, msg, datetime.now().isoformat())

)

print(f"⚠️ {msg}")

conn.commit()

conn.close()

return alerts

def run_price_check():

"""跑一轮全量价格检查"""

conn = sqlite3.connect(DB_PATH)

c = conn.cursor()

c.execute("SELECT sku, name, platform, url, target_price FROM products WHERE alert_enabled=1")

products = c.fetchall()

conn.close()

session = create_session()

results = []

for sku, name, platform, url, target_price in products:

print(f"正在抓: {name} ({platform})...")

resp, error = fetch_url(url, session=session, timeout=20)

if error:

print(f" 抓失败了: {error}")

continue

price = PriceExtractor.extract(resp.text, platform)

if price is None:

print(f" 没提取到价格，可能页面结构变了或者需要登录")

continue

save_price(sku, price)

check_and_alert(sku, name, price, target_price)

results.append({"sku": sku, "name": name, "price": price, "platform": platform})

print(f" 当前价格: ¥{price}")

# 每个商品之间歇会儿，别太猛

human_delay(base=3, jitter=2)

return results

```

3.6 定时任务+比价报告

```python

import schedule

import pandas as pd

def generate_comparison_report():

"""生成比价报告：同一件商品哪个平台最便宜，一目了然"""

conn = sqlite3.connect(DB_PATH)

df = pd.read_sql_query("""

SELECT p.sku, p.name, p.platform, ph.price, ph.fetched_at

FROM products p

JOIN price_history ph ON p.sku = ph.sku

WHERE ph.fetched_at = (

SELECT MAX(fetched_at) FROM price_history WHERE sku = p.sku

)

ORDER BY p.sku, ph.price

""", conn)

conn.close()

if df.empty:

print("还没有价格数据呢，先跑一轮采集吧")

return

# 找出每个商品的最低价

idx = df.groupby("sku")["price"].idxmin()

cheapest = df.loc[idx, ["sku", "name", "platform", "price"]]

cheapest.columns = ["商品ID", "商品名称", "最低价平台", "最低价"]

print("\n" + "=" * 60)

print("📊 多平台比价报告")

print("=" * 60)

print(cheapest.to_string(index=False))

print("=" * 60)

cheapest.to_csv("comparison_report.csv", index=False, encoding="utf-8-sig")

print("报告已存到 comparison_report.csv")

def job_daily():

"""每日任务：抓价格 + 出报告"""

print(f"\n=== 开始今日价格检查 {datetime.now().strftime('%Y-%m-%d %H:%M')} ===")

run_price_check()

generate_comparison_report()

print("=== 检查完成 ===\n")

if __name__ == "__main__":

init_db()

# 举个例子：监控某款耳机在两个平台的价格

add_product("sku_001", "某品牌无线耳机", "jd", "https://item.example-jd.com/100012345678.html", target_price=599)

add_product("sku_002", "某品牌无线耳机", "taobao", "https://detail.example-tmall.com/item.htm?id=123456789", target_price=599)

# 先跑一次看看效果

job_daily()

# 然后定时每天10点和20点各跑一次

schedule.every().day.at("10:00").do(job_daily)

schedule.every().day.at("20:00").do(job_daily)

print("定时任务已启动，Ctrl+C退出")

while True:

schedule.run_pending()

time.sleep(60)

```

3.7 电商反爬应对小抄

电商的招 → 你的应对：

IP封得快 → 隧道代理自动换IP + 随机延迟，别一个IP猛冲

看请求头 → 随机UA池 + 完整的浏览器请求头，别裸奔

页面是JS渲染的 → 上Playwright/Puppeteer，模拟真实浏览器

要登录才能看 → 维护Cookie池，每个代理IP绑一个Cookie，别乱串

弹验证码 → OCR识别或打码服务（仅限合法场景哈）

不同地区价不同 → 代理指定出口地域，分别采集对比

四、实战二：搞一个全网舆情监控系统

4.1 我们要做啥？

也不复杂，五件事：

1. 设定你关心的关键词（比如你的品牌名、产品名）

2. 自动去多个平台搜相关的帖子和评论

3. 做情感分析，看看大家是在夸还是在骂

4. 负面声音变多或者热度突然飙升，自动预警

5. 每天出一份舆情日报，心里有数

做品牌的朋友应该懂，负面舆情这东西，早发现一小时和晚发现一小时，处理难度天差地别。

4.2 系统架构

关键词配置 → 多平台采集器 → 原始内容数据库

↓ ↓

代理IP池 数据清洗去重

↓

情感分析引擎

↓

热度评分+预警

↓

舆情日报

4.3 数据库设计

```python

DB_PATH_SENTIMENT = "sentiment_monitor.db"

def init_sentiment_db():

conn = sqlite3.connect(DB_PATH_SENTIMENT)

c = conn.cursor()

# 关键词表：你关心哪些词

c.execute("""

CREATE TABLE IF NOT EXISTS keywords (

id INTEGER PRIMARY KEY AUTOINCREMENT,

keyword TEXT UNIQUE,

category TEXT, -- 分类：品牌/产品/行业

priority INTEGER DEFAULT 1, -- 优先级，数字越大越紧急

enabled INTEGER DEFAULT 1

)

""")

# 舆情内容表：抓到的帖子都存在这

c.execute("""

CREATE TABLE IF NOT EXISTS posts (

id INTEGER PRIMARY KEY AUTOINCREMENT,

post_id TEXT UNIQUE, -- 内容唯一ID，用来去重

keyword TEXT, -- 匹配到的关键词

platform TEXT, -- 平台标识

author TEXT, -- 作者

title TEXT,

content TEXT,

url TEXT,

publish_time TEXT,

like_count INTEGER DEFAULT 0,

comment_count INTEGER DEFAULT 0,

repost_count INTEGER DEFAULT 0,

sentiment REAL, -- 情感分0-1，越接近1越正面

sentiment_label TEXT, -- positive正面/neutral中性/negative负面

hot_score REAL, -- 热度评分

fetched_at TEXT

)

""")

# 预警记录表

c.execute("""

CREATE TABLE IF NOT EXISTS sentiment_alerts (

id INTEGER PRIMARY KEY AUTOINCREMENT,

keyword TEXT,

alert_level TEXT, -- warning提醒/critical严重

message TEXT,

post_count INTEGER,

negative_ratio REAL,

created_at TEXT

)

""")

conn.commit()

conn.close()

print("舆情数据库初始化完成")

```

4.4 多平台采集器

和比价系统一样，策略模式安排上。这里以某博和某乎为例，想加某红书、小破站啥的，照葫芦画瓢就行。

```python

import hashlib

class PostCollector:

"""多平台内容采集器"""

@staticmethod

def collect_weibo(keyword, session, max_pages=3):

"""某博搜索采集"""

posts = []

for page in range(1, max_pages + 1):

url = f"https://s.example-weibo.com/search?q={keyword}&page={page}"

resp, error = fetch_url(url, session=session, timeout=15)

if error:

print(f" 某博第{page}页没抓到: {error}")

continue

soup = BeautifulSoup(resp.text, "lxml")

cards = soup.select(".card-wrap")

for card in cards:

try:

content_elem = card.select_one(".txt")

if not content_elem:

continue

content = content_elem.get_text(strip=True)

author_elem = card.select_one(".name")

author = author_elem.get_text(strip=True) if author_elem else "未知"

# 用内容哈希生成唯一ID，用来去重

post_id = hashlib.md5(f"weibo_{author}_{content[:50]}".encode()).hexdigest()

like = card.select_one(".like em")

comment = card.select_one(".comment em")

repost = card.select_one(".forward em")

posts.append({

"post_id": post_id,

"platform": "weibo",

"keyword": keyword,

"author": author,

"title": content[:30],

"content": content,

"url": url,

"publish_time": "",

"like_count": int(like.get_text(strip=True)) if like and like.get_text(strip=True).isdigit() else 0,

"comment_count": int(comment.get_text(strip=True)) if comment and comment.get_text(strip=True).isdigit() else 0,

"repost_count": int(repost.get_text(strip=True)) if repost and repost.get_text(strip=True).isdigit() else 0,

})

except Exception:

continue

human_delay(base=2, jitter=1)

return posts

@staticmethod

def collect_zhihu(keyword, session, max_pages=3):

"""某乎搜索采集"""

posts = []

for page in range(1, max_pages + 1):

url = f"https://www.example-zhihu.com/search?type=content&q={keyword}&page={page}"

resp, error = fetch_url(url, session=session, timeout=15)

if error:

print(f" 某乎第{page}页没抓到: {error}")

continue

soup = BeautifulSoup(resp.text, "lxml")

items = soup.select(".SearchResult-Card, .List-item")

for item in items:

try:

title_elem = item.select_one("h2 .ContentItem-title, .ContentItem-title")

content_elem = item.select_one(".RichContent-inner, .copyright RichText")

if not title_elem and not content_elem:

continue

title = title_elem.get_text(strip=True) if title_elem else ""

content = content_elem.get_text(strip=True) if content_elem else title

author_elem = item.select_one(".AuthorInfo-name")

author = author_elem.get_text(strip=True) if author_elem else "未知"

post_id = hashlib.md5(f"zhihu_{title[:30]}".encode()).hexdigest()

posts.append({

"post_id": post_id,

"platform": "zhihu",

"keyword": keyword,

"author": author,

"title": title,

"content": content,

"url": url,

"publish_time": "",

"like_count": 0,

"comment_count": 0,

"repost_count": 0,

})

except Exception:

continue

human_delay(base=2, jitter=1)

return posts

@classmethod

def collect(cls, keyword, platform, session, max_pages=3):

methods = {

"weibo": cls.collect_weibo,

"zhihu": cls.collect_zhihu,

}

method = methods.get(platform)

if method:

return method(keyword, session, max_pages)

return []

```

老规矩：某博某乎这些平台的页面结构也经常变，而且很多内容要登录才能看。上面的代码是框架，实际用的时候根据当前页面结构调选择器。需要登录的平台？维护Cookie池，每个代理IP绑一个Cookie会话，别让IP和Cookie串了，一串就露馅。

4.5 情感分析：让代码帮你判断是夸还是骂

用SnowNLP，中文情感分析的轻量神器，不用训练，开箱即用：

```python

from snownlp import SnowNLP

import jieba

import jieba.analyse

def analyze_sentiment(text):

"""

情感分析

返回: (情感分0-1, 标签, 关键词列表)

分数越接近1越正面，越接近0越负面

"""

if not text or len(text.strip()) < 2:

return 0.5, "neutral", []

try:

s = SnowNLP(text)

score = s.sentiments

if score >= 0.6:

label = "positive"

elif score <= 0.4:

label = "negative"

else:

label = "neutral"

# 顺便提取一下关键词

keywords = jieba.analyse.extract_tags(text, topK=5)

return score, label, keywords

except Exception:

return 0.5, "neutral", []

def calculate_hot_score(post):

"""

热度评分：点赞 + 评论×2 + 转发×3

负面内容热度×1.5，因为骂你的比夸你的更需要你关注

"""

base = post.get("like_count", 0) + post.get("comment_count", 0) * 2 + post.get("repost_count", 0) * 3

if post.get("sentiment_label") == "negative":

base *= 1.5

import math

return round(math.log1p(base), 4)

```

4.6 存数据+自动去重

```python

def save_posts(posts):

"""批量保存，重复的自动跳过"""

conn = sqlite3.connect(DB_PATH_SENTIMENT)

c = conn.cursor()

new_count = 0

for post in posts:

score, label, keywords = analyze_sentiment(post["content"])

post["sentiment"] = score

post["sentiment_label"] = label

post["hot_score"] = calculate_hot_score(post)

try:

c.execute("""

INSERT INTO posts (post_id, keyword, platform, author, title, content, url,

publish_time, like_count, comment_count, repost_count,

sentiment, sentiment_label, hot_score, fetched_at)

VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)

""", (

post["post_id"], post["keyword"], post["platform"], post["author"],

post["title"], post["content"], post["url"], post.get("publish_time", ""),

post["like_count"], post["comment_count"], post["repost_count"],

post["sentiment"], post["sentiment_label"], post["hot_score"],

datetime.now().isoformat()

))

new_count += 1

except sqlite3.IntegrityError:

# 重复了，跳过

continue

conn.commit()

conn.close()

return new_count

```

4.7 危机预警：负面声音一多就喊你

```python

def check_crisis_alert(keyword, window_hours=24):

"""

看看最近24小时这个关键词有没有危机

负面占比超30%提醒，超50%严重警告，负面热度突增也提醒

"""

conn = sqlite3.connect(DB_PATH_SENTIMENT)

c = conn.cursor()

c.execute("""

SELECT sentiment_label, hot_score, COUNT(*) as cnt

FROM posts

WHERE keyword=? AND fetched_at >= datetime('now', ?)

GROUP BY sentiment_label

""", (keyword, f"-{window_hours} hours"))

rows = c.fetchall()

conn.close()

total = sum(r[2] for r in rows)

if total == 0:

return None

negative_count = sum(r[2] for r in rows if r[0] == "negative")

negative_ratio = negative_count / total

negative_hot = sum(r[1] for r in rows if r[0] == "negative")

avg_negative_hot = negative_hot / negative_count if negative_count > 0 else 0

alert = None

# 严重警告：一半以上都在骂

if negative_ratio >= 0.5 and total >= 10:

alert_level = "critical"

msg = f"【严重预警】「{keyword}」近{window_hours}小时负面占比 {negative_ratio*100:.0f}%，共{total}条讨论，骂的有{negative_count}条，赶紧看看！"

# 一般提醒：三成以上负面

elif negative_ratio >= 0.3 and total >= 5:

alert_level = "warning"

msg = f"【舆情提醒】「{keyword}」近{window_hours}小时负面占比 {negative_ratio*100:.0f}%，共{total}条讨论，留意一下"

# 负面热度突然很高

elif avg_negative_hot > 2.0 and negative_count >= 3:

alert_level = "warning"

msg = f"【热度突增】「{keyword}」负面内容平均热度 {avg_negative_hot:.2f}，可能要发酵，建议关注"

if alert:

conn = sqlite3.connect(DB_PATH_SENTIMENT)

c = conn.cursor()

c.execute("""

INSERT INTO sentiment_alerts (keyword, alert_level, message, post_count, negative_ratio, created_at)

VALUES (?, ?, ?, ?, ?, ?)

""", (keyword, alert_level, msg, total, negative_ratio, datetime.now().isoformat()))

conn.commit()

conn.close()

print(f"🚨 {msg}")

return alert

```

4.8 舆情日报：每天一份，心里不慌

```python

def generate_daily_report():

"""生成每日舆情报告"""

conn = sqlite3.connect(DB_PATH_SENTIMENT)

df = pd.read_sql_query("""

SELECT keyword, platform, sentiment_label, COUNT(*) as cnt, AVG(hot_score) as avg_hot

FROM posts

WHERE fetched_at >= datetime('now', '-24 hours')

GROUP BY keyword, platform, sentiment_label

ORDER BY keyword, cnt DESC

""", conn)

df_negative = pd.read_sql_query("""

SELECT keyword, platform, author, title, content, hot_score, url

FROM posts

WHERE sentiment_label='negative' AND fetched_at >= datetime('now', '-24 hours')

ORDER BY hot_score DESC

LIMIT 10

""", conn)

conn.close()

report = []

report.append("=" * 60)

report.append(f"📰 舆情日报 {datetime.now().strftime('%Y-%m-%d')}")

report.append("=" * 60)

if df.empty:

report.append("近24小时没啥数据")

else:

report.append("\n【各关键词舆情分布】")

for keyword in df["keyword"].unique():

kw_data = df[df["keyword"] == keyword]

total = kw_data["cnt"].sum()

neg = kw_data[kw_data["sentiment_label"] == "negative"]["cnt"].sum()

pos = kw_data[kw_data["sentiment_label"] == "positive"]["cnt"].sum()

neu = kw_data[kw_data["sentiment_label"] == "neutral"]["cnt"].sum()

report.append(f"\n 关键词: {keyword} (共{total}条)")

report.append(f" 夸的: {pos} | 中立: {neu} | 骂的: {neg} (骂占比{neg/total*100:.0f}%)")

if not df_negative.empty:

report.append("\n【最需要关注的10条负面内容】")

for i, row in df_negative.iterrows():

report.append(f" {i+1}. [{row['platform']}] {row['author']}: {row['title'][:40]}")

report.append(f" 热度: {row['hot_score']:.2f} | {row['url']}")

report.append("\n" + "=" * 60)

report_text = "\n".join(report)

print(report_text)

with open(f"sentiment_report_{datetime.now().strftime('%Y%m%d')}.txt", "w", encoding="utf-8") as f:

f.write(report_text)

return report_text

```

4.9 跑起来！

```python

def run_sentiment_monitor():

"""跑一轮完整的舆情监测"""

print(f"\n=== 开始舆情监测 {datetime.now().strftime('%Y-%m-%d %H:%M')} ===")

conn = sqlite3.connect(DB_PATH_SENTIMENT)

c = conn.cursor()

c.execute("SELECT keyword FROM keywords WHERE enabled=1")

keywords = [r[0] for r in c.fetchall()]

conn.close()

if not keywords:

print("还没配置关键词呢，先加几个")

return

session = create_session()

platforms = ["weibo", "zhihu"] # 想加平台自己往这填

total_new = 0

for keyword in keywords:

print(f"\n正在盯: {keyword}")

for platform in platforms:

print(f" 扫平台: {platform}")

posts = PostCollector.collect(keyword, platform, session, max_pages=2)

if posts:

new_count = save_posts(posts)

total_new += new_count

print(f" 抓到{len(posts)}条，新的有{new_count}条")

human_delay(base=3, jitter=2)

# 每个关键词扫完都检查一下有没有危机

check_crisis_alert(keyword)

print(f"\n本轮搞定，新增{total_new}条内容")

generate_daily_report()

print("=== 监测完成 ===\n")

if __name__ == "__main__":

init_sentiment_db()

# 加几个你关心的关键词

conn = sqlite3.connect(DB_PATH_SENTIMENT)

c = conn.cursor()

for kw, cat in [("你的品牌名", "品牌"), ("你的产品名", "产品")]:

try:

c.execute("INSERT INTO keywords (keyword, category, priority) VALUES (?, ?, ?)", (kw, cat, 1))

except sqlite3.IntegrityError:

pass

conn.commit()

conn.close()

# 先跑一次

run_sentiment_monitor()

# 然后每2小时自动跑一次

schedule.every(2).hours.do(run_sentiment_monitor)

print("舆情监控已启动，Ctrl+C退出")

while True:

schedule.run_pending()

time.sleep(60)

```

4.10 两个场景的代理怎么选？对比一下

维度 → 电商比价 → 舆情监控：

IP类型 → 数据中心IP够用 → 建议住宅IP，社媒对数据中心IP封得狠

换IP频率 → 每次或每5次请求换 → 每次都换，搜索接口限流很严

地域要求 → 需要指定地域看当地价 → 国内IP就行

并发量 → 中等，商品数有限 → 高，多关键词×多平台×多页

要不要保持会话 → 不用 → 要，Cookie得跟IP绑一起

匿名度 → 高匿就行 → 高匿+住宅IP，双重保险

说真的，舆情监控对IP质量要求高不少。某博某红书这些平台对数据中心IP的识别率高到离谱，建议上住宅IP，而且每个IP绑一个独立Cookie会话，别串。

五、想更猛？这几个方向可以继续折腾

5.1 从单进程到分布式

监控的商品或关键词上百上千的时候，单进程就跑不动了。可以搞：

任务队列：Redis + Celery，把采集任务分给多个worker

代理池服务化：代理管理单独做成一个服务，大家共用

去重放Redis：用Redis的Set做去重，比数据库唯一约束快得多

5.2 从requests到Playwright

遇到JS渲染重、要登录交互的页面，requests+BeautifulSoup就抓瞎了。Playwright是真浏览器，TLS指纹天然没问题，还能模拟点击滚动。配代理也简单：

```python

from playwright.sync_api import sync_playwright

PROXY_SERVER = f"http://{PROXY_HOST}:{PROXY_PORT}"

with sync_playwright() as p:

browser = p.chromium.launch(

proxy={

"server": PROXY_SERVER,

"username": PROXY_USER,

"password": PROXY_PASS,

},

headless=True,

)

context = browser.new_context(

user_agent=random.choice(UA_POOL),

locale="zh-CN",

)

page = context.new_page()

page.goto("https://example.com", timeout=30000)

# 想干啥干啥

browser.close()

```

5.3 告警别光打印，推送到手机

本文的告警只做了打印和存数据库，实际用的时候可以接：

邮件（smtplib，HTML格式好看点）

企业微信/钉钉/飞书机器人（Webhook，团队都能看到）

Server酱/Telegram Bot（个人开发者最爱）

短信（严重危机才用，毕竟要钱）

六、最后说几句正经的

技术是中性的，但用技术的人得有底线：

1. 只抓公开能看到的数据，别想着绕登录偷人家非公开的东西

2. 遵守目标网站的robots.txt和用户协议，有些网站明说了不让爬

3. 控制频率，别把人家小网站搞崩了，做人留一线

4. 数据只用于合法的商业分析或个人研究，别搞不正当竞争

5. 舆情监控别碰个人隐私，别采集人家的身份证手机号啥的

6. 代理从正规渠道拿，确保IP来源合规

七、总结一下

这篇文章带你从零搭了两个能跑的系统：

电商比价系统：加商品 → 定时抓价 → 存历史 → 降价提醒 → 出比价报告。核心是多平台价格提取和历史对比，帮你省钱。

舆情监控系统：加关键词 → 多平台扫 → 情感分析 → 热度评分 → 危机预警 → 出日报。核心是多源数据整合和负面识别，帮你挡刀。

两个系统的地基都是隧道代理——解决IP被封、地域限制、并发分散这三个老大难问题。代理层不稳，上面业务逻辑写得再花也白搭。

想继续深入的话，这几个方向值得挖：

用Playwright重写采集器，对付JS渲染和登录态

上分布式架构，撑住万级监控规模

用大模型替代SnowNLP，情感分析更准还能做事件归类

整个可视化Dashboard，价格趋势和舆情热度实时看

说白了，代理这东西就像做菜的锅，锅好了才能炒出好菜。希望这篇文章能帮你把锅架起来，炒出自己的菜。有问题评论区聊，看到都会回。