携程景点评论爬虫实战：POIID提取与批量数据采集

简介：这是一款基于Python开发的携程景点评论爬虫工具，面向旅游数据分析人员、景点运营者及爬虫学习者，解决手动收集评论效率低、数据零散的问题。工具可从携程景点详情页自动识别真实POIID，并批量下载该景点全部用户评论，最终输出结构化数据包，便于后续统计与文本分析。压缩包共6个文件，约45KB，包含py爬虫主程序、md与txt说明文档、docx附赠资源及license授权文件，覆盖源码、运行指引与扩展参考。已有75人学习下载。使用者可获取完整可运行的爬虫脚本，理解POIID提取与评论批量抓取的核心逻辑，并借助说明文档快速上手、按需修改代码以适配不同抓取任务；附赠文档还提供额外教程与常见问题解答，帮助排查运行问题、规范数据处理流程。需注意，使用时应遵守携程服务条款与相关法律法规，在合法合规前提下使用数据。

1. 携程景点评论爬虫：从 POIID 到全量评论的落地路径

做旅游数据分析的同行大概率都遇到过这个场景：老板丢过来一个 Excel，里面列着几十个热门景区的名字，要求两天内交出这些景点的评论情感分析报告。手动复制？一个景点动辄几千条评论，翻页翻到手抽筋。这时候一套能自动提取 POIID、批量拉取评论的 Python 脚本就成了刚需。这份携程景点评论爬虫工具的核心价值就在于此——它把「景点详情页 → 真实 POIID → 全量评论数据」这条链路串了起来，省去了人工找 ID 和逐页翻评论的重复劳动。适合做旅游舆情监控、竞品分析、学术研究的数据从业者，也适合刚接触 requests 爬虫、想找一个真实项目练手的 Python 学习者。需要说明的是，工具本身只负责数据采集环节，后续的清洗、存储、可视化需要自己按需接上。

2. 拆解爬虫链路：POIID 提取与请求构造的底层逻辑

2.1 为什么 POIID 是整个流程的钥匙

携程的景点评论接口不是靠景点名称来定位的，而是靠一个叫 POIID 的数字标识。你可以把它理解成景点在携程数据库里的身份证号——同一个景点在不同城市可能有重名，但 POIID 是唯一的。很多新手一上来就想着用景点名称去拼 URL，结果要么请求不到数据，要么拿到的是别的景点的评论，这就是没搞清楚 POIID 的作用。

POIID 通常藏在景点详情页的 HTML 源码里，常见的位置有几个：页面头部的window.__INITIAL_STATE__变量、某个<script>标签里的 JSON 配置、或者页面 URL 的 query 参数中。不同页面模板的存放位置可能不一样，所以提取逻辑不能写死一个正则就完事，得做多模式匹配。

我一般会按这个优先级来找：先看 URL 里有没有明显的数字 ID，没有再去 HTML 里搜poiId、poi_id、sightId这几个关键词，最后兜底用正则从整页源码里捞长数字串。下面是一个提取 POIID 的参考实现：

import re import requests HEADERS = { "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) " "AppleWebKit/537.36 (KHTML, like Gecko) " "Chrome/120.0.0.0 Safari/537.36", "Referer": "https://you.ctrip.com/", } def extract_poi_id(detail_url: str) -> str | None: """从景点详情页 HTML 中提取 POIID，支持多种页面模板""" resp = requests.get(detail_url, headers=HEADERS, timeout=15) resp.raise_for_status() html = resp.text # 模式一：URL 路径中直接带数字 ID，如 /sight/xxx/12345.html url_match = re.search(r"/sight/[^/]+/(\d+)\.html", detail_url) if url_match: return url_match.group(1) # 模式二：页面内嵌 JSON 中的 poiId 字段 json_match = re.search(r'"poiId"\s*:\s*"?(\d+)"?', html) if json_match: return json_match.group(1) # 模式三：兜底，从 script 标签里捞 6 位以上的数字串 fallback = re.search(r'poi[_-]?id["\']?\s*[:=]\s*["\']?(\d{5,})', html, re.I) if fallback: return fallback.group(1) return None

这段代码的逻辑说明：extract_poi_id接收景点详情页的完整 URL，先发一次 GET 请求拿到 HTML。然后按三种模式依次尝试匹配——URL 路径匹配最快最准，内嵌 JSON 匹配覆盖面广，兜底正则用来应对页面改版后的意外情况。参数方面，timeout=15是防止某个页面卡住导致整个批量任务挂起，HEADERS里的Referer建议带上，部分接口会校验来源。

提示：如果三种模式都返回 None，大概率是页面结构变了或者触发了验证码，这时候不要硬跑，先把 HTML 存下来人工看一眼。

2.2 评论接口的请求参数怎么拼

拿到 POIID 之后，下一步是构造评论列表的请求。携程的评论数据一般通过一个独立的接口返回，返回格式是 JSON，分页由pageIndex或page参数控制。常见的请求参数包括：

实际请求时，接口地址和参数名可能随携程前端版本变化。我一般会先用浏览器开发者工具抓一次真实请求，把 URL 和参数原样复制到代码里，跑通之后再改成循环翻页。下面是一个翻页拉取评论的骨架：

import time import json def fetch_comments(poi_id: str, max_pages: int = 50) -> list[dict]: """按页拉取指定景点的评论，返回评论列表""" api_url = "https://m.ctrip.com/restapi/soa2/13444/json/getCommentCollapseList" all_comments = [] for page in range(1, max_pages + 1): payload = { "poiId": poi_id, "pageIndex": page, "pageSize": 20, "sortType": 1, "channelType": 2, } try: resp = requests.post(api_url, json=payload, headers=HEADERS, timeout=15) data = resp.json() except Exception as e: print(f"[page {page}] 请求异常: {e}") break items = data.get("result", {}).get("items", []) if not items: print(f"[page {page}] 无更多数据，停止翻页") break all_comments.extend(items) print(f"[page {page}] 获取 {len(items)} 条，累计 {len(all_comments)} 条") time.sleep(1.5) # 控制频率，避免触发风控 return all_comments

逻辑说明：fetch_comments接收 POIID 和最大页数，循环构造 POST 请求。每页拿到数据后检查items是否为空，为空说明已经到底了，直接 break。time.sleep(1.5)是必须的——不加延迟连续请求，轻则被限流返回空数据，重则 IP 被临时封禁。参数方面，max_pages建议设一个上限，防止死循环；pageSize如果接口支持 20 就用 20，能减少请求次数。

2.3 数据落盘：JSON 还是 CSV

评论拿到手之后，落盘格式的选择取决于后续用途。如果要做情感分析、关键词提取，JSON 保留字段最完整；如果要丢进 Excel 给非技术同事看，CSV 更友好。我一般两个都存——原始 JSON 做备份，清洗后的 CSV 做交付。

import csv def save_comments(comments: list[dict], poi_id: str): """将评论数据分别保存为 JSON 和 CSV""" # 原始 JSON 备份 with open(f"comments_{poi_id}_raw.json", "w", encoding="utf-8") as f: json.dump(comments, f, ensure_ascii=False, indent=2) # 清洗后的 CSV fields = ["commentId", "content", "score", "publishTime", "userNick"] with open(f"comments_{poi_id}.csv", "w", encoding="utf-8-sig", newline="") as f: writer = csv.DictWriter(f, fieldnames=fields, extrasaction="ignore") writer.writeheader() for c in comments: writer.writerow(c) print(f"已保存 {len(comments)} 条评论到 comments_{poi_id}.csv")

这里有个细节：CSV 的编码用utf-8-sig而不是utf-8，原因是 Excel 打开纯 UTF-8 的 CSV 时中文会乱码，加 BOM 头能解决这个问题。extrasaction="ignore"的作用是当评论字典里有 fields 之外的键时自动忽略，不会报错。

3. 批量任务编排：多景点循环与断点续跑

3.1 用配置文件管理待爬景点列表

单景点跑通之后，下一步就是批量。最忌讳的做法是把景点 URL 硬编码在脚本里，改一个就得动代码。常见做法是单独维护一个配置文件，把景点名称和详情页 URL 列进去，脚本读取后逐个处理。

import yaml def load_sights(config_path: str = "sights.yaml") -> list[dict]: """从 YAML 配置文件读取待爬景点列表""" with open(config_path, "r", encoding="utf-8") as f: data = yaml.safe_load(f) return data.get("sights", []) # sights.yaml 示例结构： # sights: # - name: 西湖 # url: https://you.ctrip.com/sight/hangzhou14/12345.html # - name: 故宫 # url: https://you.ctrip.com/sight/beijing1/67890.html

用 YAML 而不是 JSON 做配置，主要是因为 YAML 支持注释，团队协作时可以在每个景点旁边标注优先级或备注。load_sights返回一个列表，每个元素包含name和url两个字段，主流程遍历这个列表即可。

3.2 断点续跑：别让一个失败拖垮整批任务

批量任务最怕的就是跑到第 30 个景点时网络抖动，整个脚本崩了，前面 29 个的数据虽然存了，但重新跑又得从头来。解决办法是维护一个进度文件，记录哪些景点已经完成，下次启动时跳过。

import os PROGRESS_FILE = "progress.txt" def load_progress() -> set: """读取已完成的景点标识""" if not os.path.exists(PROGRESS_FILE): return set() with open(PROGRESS_FILE, "r", encoding="utf-8") as f: return set(line.strip() for line in f if line.strip()) def mark_done(poi_id: str): """标记某个景点已完成""" with open(PROGRESS_FILE, "a", encoding="utf-8") as f: f.write(poi_id + "\n") def run_batch(sights: list[dict]): """批量执行爬取任务，支持断点续跑""" done = load_progress() for sight in sights: poi_id = extract_poi_id(sight["url"]) if not poi_id: print(f"[跳过] {sight['name']} 未提取到 POIID") continue if poi_id in done: print(f"[跳过] {sight['name']} ({poi_id}) 已完成") continue print(f"[开始] {sight['name']} ({poi_id})") comments = fetch_comments(poi_id) save_comments(comments, poi_id) mark_done(poi_id) print(f"[完成] {sight['name']} 共 {len(comments)} 条")

这套机制的核心是progress.txt文件，每完成一个景点就追加一行 POIID。下次运行时load_progress把已完成的 ID 读进集合，遍历时直接跳过。注意mark_done用的是追加模式"a"，不是覆盖模式，否则每次都会把之前的记录冲掉。

3.3 异常重试与日志记录

网络请求失败是常态，关键是要有重试机制和清晰的日志。我一般用requests配合urllib3的重试适配器，或者自己写一个简单的重试装饰器。

import logging from functools import wraps logging.basicConfig( level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s", handlers=[ logging.FileHandler("spider.log", encoding="utf-8"), logging.StreamHandler(), ], ) logger = logging.getLogger(__name__) def retry(times: int = 3, delay: float = 2.0): """简单的重试装饰器""" def decorator(func): @wraps(func) def wrapper(*args, **kwargs): for attempt in range(1, times + 1): try: return func(*args, **kwargs) except Exception as e: logger.warning(f"{func.__name__} 第 {attempt} 次失败: {e}") if attempt == times: raise time.sleep(delay * attempt) return wrapper return decorator

retry装饰器的参数times控制最大重试次数，delay是基础等待时间，实际等待时间按delay * attempt递增，避免短时间内反复冲击目标接口。日志同时输出到文件和控制台，排查问题时直接翻spider.log就行。

4. 避坑与排查：那些跑起来才知道的坑

4.1 返回数据为空但状态码是 200

现象：请求评论接口，HTTP 状态码 200，但items字段是空数组，翻到第二页还是空。

原因：大概率是触发了携程的风控机制。携程对评论接口有频率限制，短时间内大量请求会被静默限流——不返回 403，但也不给你数据。

解决：降低请求频率，time.sleep从 1.5 秒加到 3 秒以上；如果还是空，换一个时间段再跑。另外检查HEADERS里的User-Agent和Referer是否完整，缺Referer有时也会被判定为异常请求。

4.2 POIID 提取到了但评论接口返回 404

现象：POIID 明明提取成功，但请求评论接口时返回 404 或参数错误。

原因：POIID 提取对了，但接口地址或参数格式变了。携程的评论接口有多个版本，PC 端和移动端的接口地址不同，参数名也可能有差异。

解决：用浏览器开发者工具重新抓一次真实请求，对比接口 URL、请求方法（GET 还是 POST）、参数名是否一致。特别注意channelType这个参数，不同渠道的值不一样，填错了接口会拒绝。

4.3 翻页到某一页突然中断

现象：前 10 页正常返回，第 11 页开始报连接超时或返回非 JSON 格式的内容。

原因：可能是目标接口对翻页深度有限制，超过一定页数后不再返回数据；也可能是网络抖动导致单次请求失败。

解决：在翻页循环里加 try-except，单页失败不中断整个任务，记录失败页码后续补跑。同时设一个合理的max_pages上限，一般景点评论超过 50 页的情况很少，如果确实需要全量，考虑按时间分段拉取。

4.4 CSV 中文乱码

现象：用 Excel 打开生成的 CSV 文件，中文全部变成乱码。

原因：Excel 默认按 GBK 编码打开 CSV，而 Python 写文件时用的是 UTF-8。

解决：写 CSV 时编码参数用utf-8-sig，加 BOM 头后 Excel 能正确识别。如果已经生成了乱码文件，用记事本打开另存为 ANSI 编码也能救回来，但不如一开始就写对。

4.5 脚本跑着跑着内存暴涨

现象：批量爬取几十个景点后，Python 进程占用内存从几十 MB 涨到几个 GB。

原因：所有评论数据都堆在all_comments列表里没释放，景点越多内存占用越大。

解决：每个景点处理完后及时落盘并清空列表，不要等所有景点跑完再统一保存。如果确实需要汇总，用生成器逐条 yield，或者分批写入数据库。

5. 进阶技巧：评论数据去重与增量更新

5.1 基于 commentId 的去重策略

同一批评论数据重复拉取时，最直接的去重依据是commentId。每条评论在携程系统里都有唯一 ID，用集合做去重是最快的。

def dedup_comments(comments: list[dict], seen_ids: set) -> list[dict]: """基于 commentId 去重，返回新增评论""" new_comments = [] for c in comments: cid = str(c.get("commentId", "")) if cid and cid not in seen_ids: seen_ids.add(cid) new_comments.append(c) return new_comments

seen_ids集合可以持久化到本地文件或数据库，每次增量爬取时先加载历史 ID，只保留新增部分。这样重复跑同一个景点时不会产生冗余数据。

5.2 增量更新：只拉最新评论

如果只是想做舆情监控，不需要每次全量拉取，可以只拉第一页或前几页的最新评论，和本地已有数据做对比。具体做法是记录上次爬取时间，翻页时遇到发布时间早于上次爬取时间的评论就停止。

from datetime import datetime def fetch_incremental(poi_id: str, last_crawl_time: str) -> list[dict]: """增量拉取：遇到早于上次爬取时间的评论就停止""" new_comments = [] last_dt = datetime.fromisoformat(last_crawl_time) for page in range(1, 10): payload = {"poiId": poi_id, "pageIndex": page, "pageSize": 20, "sortType": 1} resp = requests.post(API_URL, json=payload, headers=HEADERS, timeout=15) items = resp.json().get("result", {}).get("items", []) if not items: break stop = False for c in items: pub_time = c.get("publishTime", "") try: pub_dt = datetime.fromisoformat(pub_time.replace("Z", "+00:00")) except ValueError: new_comments.append(c) continue if pub_dt <= last_dt: stop = True break new_comments.append(c) if stop: break time.sleep(2) return new_comments

这段代码的关键在于sortType=1按最新排序，然后逐条比对发布时间。一旦遇到早于last_crawl_time的评论就停止翻页，避免做无用功。last_crawl_time建议存在配置文件或数据库里，每次增量跑完后更新。

5.3 验证数据完整性的小技巧

爬完之后怎么确认数据没漏？我一般会做两个检查：一是对比接口返回的总条数和实际保存的条数是否一致，二是随机抽几条评论去页面上人工核对。接口返回的 JSON 里通常有一个totalCount字段，拿它和len(all_comments)对比，差距超过 5% 就说明翻页逻辑有问题，得回去查。

从那以后我每次跑批量任务前，都会先用一个景点做全流程验证，确认 POIID 提取、评论拉取、落盘、去重四个环节都正常，再放开跑全量。这个习惯帮我省了很多事后补数据的麻烦。希望帮到你。

本文还有配套的精品资源，点击获取