简介面向需要批量获取马蜂窝酒店、美食、景点评论及游记数据的开发者或数据分析人员这份Python爬虫资源输入目的地或关键词即可运行自动抓取公开页面上的评论与游记信息。评论数据包含用户昵称、ID、用户等级、评论内容、有用性、评论时间等字段游记数据包含地址、浏览量、评论量、游玩时间、游玩天数、同行人等字段字段覆盖较完整。压缩包共11个文件主体为9个Python脚本配合1个cfg配置文件用于Scrapy工程设定1个Markdown说明文档用于快速上手整包仅13KB轻量易读。当前已有4605人学习下载代码结构清晰启动、爬取和配置分工明确适合有一定爬虫基础并希望直接改造复用的读者快速搭建马蜂窝数据采集任务。1. 马蜂窝爬虫的切入点三类数据与两个入口评论区是一个站十年积累下来的账本游记则是用户主动写出来的内容资产。爬马蜂窝旅游数据最常见的诉求是把酒店、美食、景点的评论和游记对起来看一家店是不是被刷好评一条线路在游记里被反复提及这些信号比单纯的评分可靠得多。马蜂窝的接口设计整体比大众点评温和评论数据通过XHR接口分页返回游记正文则是服务端渲染的静态HTML两种结构决定了完全不同的采集策略。整个采集链路可以压缩成三步定位数据真实来源、构造带分页参数的请求、按字段解析落盘。全程用 requests 加标准库就能完成不需要为了这个体量接 Scrapy。但如果你要扩展到多个城市、多类POI并持续增量更新就不得不考虑并发设计和断点续爬。本篇文章的代码基线锁定在 2021.6.28 前后可用的接口规则这个时间点的意义在于接口路径和参数名在之后有过微调固定一个基线方便后续做差异对比。2. 马蜂窝请求接口分析与景点评论爬取实现2.1 从页面到接口定位数据真实来源的常规做法打开任意马蜂窝景点详情页评论区域在页面中下部。很多人第一反应是直接解析HTML但评论是通过异步请求拉取的首屏HTML里只包含前几条兜底数据。常见的做法是打开浏览器的开发者工具切到Network面板过滤XHR类型然后在页面上点击评论翻页这时能看到类似poi/comment/list的请求路径。返回的JSON里包含了评论列表、用户信息和分页数据。请求格式是典型的GETquery参数至少包含poi_id、page、pageSize三个核心字段。poi_id是关键定位值在景点详情页URL的poi段后面能看到一串数字page从1开始递增pageSize建议设置在20到40之间调太大会返回空列表。请求头里必须带上Referer指向对应POI的详情页否则部分接口在服务端校验不通过。表景点评论接口主要参数参数类型说明poi_idint景点唯一ID详情页URLpoi段后的数字pageint页码从1开始pageSizeint每页条数建议20到40sortint排序方式0按时间倒序1按热度imgsint是否返回图片地址0为不返回2.2 用 requests 构造最小可运行的请求代码下面这段代码是验证接口连通性的最小闭环目标只有一件事把第一页评论拉下来并解析成字典列表。不要在这个阶段贪多接口通没通、字段对不对跑一次就知道了。import requests HEADERS { 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, Referer: https://www.mafengwo.cn/poi/3473.html, X-Requested-With: XMLHttpRequest, } def fetch_poi_comments(poi_id: int, page: int 1, page_size: int 20) - list: url fhttps://www.mafengwo.cn/poi/comment/list_{poi_id}.html params { page: page, pageSize: page_size, sort: 0, imgs: 0, } resp requests.get(url, paramsparams, headersHEADERS, timeout10) resp.raise_for_status() data resp.json() return data.get(comment_list, [])这段代码里有三个关键点。第一params单独传而不是拼进URLrequests会自动做URL编码避免特殊字符导致请求失败。第二imgs置为0能缩小响应体纯采集评论文本时不需要图片地址。第三.get(comment_list, [])做了防御性取值即使接口返回结构变化、缺失该字段也不会直接抛异常中断流程。2.3 分页停止条件与评论解析的3个关键字段评论列表的字段结构不复杂但有三处容易踩坑。第一个是comment_id这是去重的唯一依据落库时必须加唯一约束第二个是user_info里的昵称字段匿名评论下可能是空对象{}直接.get(name)会拿到None而不是报错但如果代码写成item[user_info][name]就会在空对象上抛类型错误第三个是time字段马蜂窝返回的是Unix时间戳而不是可读字符串。def parse_comment(item: dict) - dict: user_info item.get(user_info) or {} return { comment_id: item.get(comment_id), poi_id: item.get(poi_id), user_name: user_info.get(name, ), content: item.get(content, ), rating: item.get(rating, 0), time_ts: item.get(time, 0), reply_count: item.get(reply_count, 0), }(item.get(user_info) or {})这行写法是在防御user_info为None的情况马蜂窝的旧接口里确实会有user_info: null的返回。解析完字段后分页停止条件优先看当前返回条数是否小于pageSize次选看接口是否返回total_page。有条数的判断优先用条数因为total_page不是所有接口都稳定返回。3. 酒店与美食评论的差异化字段处理3.1 酒店评论与景点评论的字段差异与映射表设计酒店POI的评论接口路径和景点几乎复用同一套规则但返回字段会有明显增加酒店评论会携带room_type房型、stay_date入住时间、cost消费金额三个业务字段。如果你只用一个通用解析函数这些额外信息会被静默丢弃。另一个问题是不同POI类型的评分结构不统一景点的评分是单一分值酒店和美食则会拆成多个维度。我通常的做法是维护一张POI类型到额外字段的映射表按poi_type分发到对应的解析逻辑POI_TYPE_EXTRA_FIELDS { hotel: [room_type, stay_date, cost], food: [taste_score, environment_score, service_score], attraction: [visit_date], } def parse_comment_with_extra(item: dict, poi_type: str) - dict: base parse_comment(item) for field in POI_TYPE_EXTRA_FIELDS.get(poi_type, []): base[field] item.get(field, ) return base这段代码的好处是改动局部化。新增一个POI类型时只需要往映射表里加字段列表不用动parse_comment的基础逻辑。字段缺失时用空字符串兜底保证所有评论记录都有统一的列方便后续合并成DataFrame或写数据库。3.2 美食评论的三维度评分结构处理美食POI的评分体系最特殊拆成味道、环境、服务三个维度。实际接口返回中这三种维度的字段结构并不统一一种情况是嵌套对象{score: {taste: 4.5, environment: 4.0, service: 4.5}}另一种是平铺字段直接挂在评论对象上。采集时两套结构都会遇到。def flatten_food_score(item: dict) - dict: score item.get(score) if score and isinstance(score, dict): return { taste_score: score.get(taste), environment_score: score.get(environment), service_score: score.get(service), } return { taste_score: item.get(taste_score), environment_score: item.get(environment_score), service_score: item.get(service_score), }这个兜底逻辑不复杂但避免了两种结构混跑时数据丢失。另外美食评论的图片占比相当高很多用户上传的图片地址是images.mafengwo.cn下的CDN路径带上Referer: https://www.mafengwo.cn/才能正常下载否则大概率返回403。图片不要直接落盘先在CSV里存URL后续单独起一个下载任务处理。3.3 增量判断按最大时间戳提前终止分页增量爬取最常见的错误是每次都全量重抓浪费带宽和时间。评论是持续产生的数据合理的做法是记录当前库中的最大time_ts只抓取时间戳大于这个值的新评论。import time def incremental_fetch_comments(poi_id: int, max_ts: int, max_pages: int 30): results [] for page in range(1, max_pages 1): comments fetch_poi_comments(poi_id, pagepage, page_size20) if not comments: break page_max_ts max(c[time] for c in comments) if page_max_ts max_ts: break results.extend([c for c in comments if c[time] max_ts]) time.sleep(0.5) return results这里的提前终止成立有一个前提请求必须按时间倒序返回也就是sort0。如果按热度排序那最大时间戳不会集中在第一页增量逻辑就会失效。这也是在上一个模块里把sort0当作固定参数的原因。每页之间 sleep 0.5秒避免每页连发导致IP被暂时限制。4. 游记数据的分页抓取与正文提取4.1 游记列表页的分页与总页数获取游记是另一个数据源页面结构和评论完全不同。评论区走JSON接口游记列表页是服务端渲染的HTML。列表页URL里通常包含目的地ID和页码实际抓取时先拉第一页从分页控件里解析总页数再按页遍历。import re from bs4 import BeautifulSoup def get_total_pages(first_html: str) - int: soup BeautifulSoup(first_html, html.parser) page_links soup.select(a.page-link) page_numbers [] for a in page_links: text a.get_text(stripTrue) if text.isdigit(): page_numbers.append(int(text)) return max(page_numbers) if page_numbers else 1page-link是马蜂窝分页控件使用的类名解析不到数字时直接返回1表示只有一页。这个函数有个隐含假设总页数会以数字形式出现在分页控件里且最后一页的数字是最大值。实际使用中这个假设都成立但如果你发现某次结果异常应该把HTML里的分页HTML片段打印出来看一下而不是直接修代码。4.2 从游记详情页提取正文的通用解析方法游记详情页正文包裹在带特定class的容器里常见的是.at_entry。解析正文不要用正则硬抠标签嵌套复杂时正则很容易误伤图片和换行。用 BeautifulSoup 的选择器先把容器选出来再用get_text带分隔符合并文本。from bs4 import BeautifulSoup def extract_travel_note(html: str) - dict: soup BeautifulSoup(html, html.parser) entry soup.select_one(.at_entry) or soup.select_one(#content) if entry is None: return {} title_tag soup.select_one(h1.travel-title) images [] for img in entry.select(img): url img.get(data-original) or img.get(src) if url: images.append(url) return { title: title_tag.get_text(stripTrue) if title_tag else , content: entry.get_text(\n, stripTrue)[:8000], images: images, }>import os import requests def download_travel_images(image_urls: list, save_dir: str): os.makedirs(save_dir, exist_okTrue) headers {Referer: https://www.mafengwo.cn/} for idx, url in enumerate(image_urls[:50]): try: resp requests.get(url, headersheaders, timeout10) ext url.split(?)[0].split(.)[-1] ext ext if len(ext) 4 else jpg with open(os.path.join(save_dir, f{idx:03d}.{ext}), wb) as f: f.write(resp.content) except Exception: continue单篇游记限制下载50张避免有人在游记里塞大量图片把带宽打满。写失败直接跳过不阻塞后续图片。扩展名从URL里解析但CDN地址经常带签名参数所以过滤掉问号后的部分再取扩展名。提示游记详情页如果频繁访问会触发安全验证验证页的特征是HTML里出现特定的JS跳转逻辑。发现extract_travel_note返回空字典时先检查返回的HTML内容判断是否命中验证再决定要不要调低频率。5. 分布式爬虫的并发设计线程池、去重与反爬应对5.1 requests ThreadPoolExecutor 设定并发上限单线程循环爬马蜂窝不是不行但效率太低。一个景点评论接口一页只有20条单线程每秒处理一个请求一小时也就3600条。用ThreadPoolExecutor加并发能明显提升吞吐量但线程数要克制。我一般把max_workers设置在5到10之间超过这个值触发限流的概率会显著上升。from concurrent.futures import ThreadPoolExecutor, as_completed def crawl_multiple_pois(poi_ids: list, max_workers: int 8): results [] with ThreadPoolExecutor(max_workersmax_workers) as executor: future_to_poi { executor.submit(fetch_poi_comments, pid, 1, 20): pid for pid in poi_ids } for future in as_completed(future_to_poi): pid future_to_poi[future] try: comments future.result() except Exception as exc: print(fpoi {pid} 抓取失败: {exc}) continue results.append((pid, comments)) return resultsas_completed在任何一个Future完成时就返回结果不用等全部结束。这样可以边爬边处理不用把所有结果堆积在内存里。8个线程配合0.3秒的单请求间隔实际吞吐比20个线程不设间隔更稳定——后者经常会触发300多个请求后被临时封禁。5.2 代理池设计与请求头轮换马蜂窝对单IP的请求频次是有限制的高频访问一段时间的IP会被限制对评论接口的访问。绕开限制最直接的手段是IP轮换。生产里的做法通常是接入代理服务从一个动态代理API拉取IP列表维护在本地池子里供爬虫取用。class ProxyPool: def __init__(self, proxy_api_url: str): self.proxy_api_url proxy_api_url self._proxies [] def refresh(self): resp requests.get(self.proxy_api_url, timeout5) data resp.json() self._proxies [fhttp://{item[ip]}:{item[port]} for item in data.get(data, [])] def get(self) - dict: if not self._proxies: self.refresh() if not self._proxies: raise RuntimeError(代理池为空) proxy self._proxies.pop() return {http: proxy, https: proxy}代理池最基础的操作就是这三个refresh拉新IP、get取一个可用代理、失败时把它丢掉。实际使用中代理请求失败要标记该代理并再次调get()换下一个。代理池更大的价值是在分布式场景下让多个爬虫进程从同一个池子取IP避免互相抢同一个代理导致大量冲突。5.3 断点续爬按任务粒度记录完成状态断点续爬的关键设计原则是记录“已完成任务”而不是记录“抓到了第几页”。任务的粒度可以是poi_id、游记ID或列表页页码粒度越细断点越准确。以POI评论抓取为例每个poi_id就是一个独立任务成功写入CSV后把poi_id追加到完成列表文件中。import os def load_done(filepath: str) - set: if not os.path.exists(filepath): return set() with open(filepath, r, encodingutf-8) as f: return {line.strip() for line in f if line.strip()} def mark_done(filepath: str, item_id: int): with open(filepath, a, encodingutf-8) as f: f.write(f{item_id}\n)写入用追加模式a避免重复读写整个文件。下次启动时读入done_set在任务提交前判断是否跳过。这个方案的容错边界很清晰进程崩溃最多丢失当前正在执行的那个任务重启后自动从断点续跑。比用数据库记录状态更轻量CSV场景完全够用。6. CSV落地与时效性校验脚本采集的数据最终落到CSV然后用一个小脚本做质量校验。这个校验在每轮抓取结束后跑一遍主要干三件事去重、过滤无效内容、检查时间戳合法性。import pandas as pd def validate_csv(filepath: str): df pd.read_csv(filepath, dtype{comment_id: str}) original_count len(df) df.drop_duplicates(subsetcomment_id, keepfirst, inplaceTrue) df df[df[content].notna() (df[content].str.len() 5)] df df[(df[time_ts] 0) (df[time_ts] 2**31)] df.to_csv(filepath.replace(.csv, _clean.csv), indexFalse, encodingutf-8-sig) print(f原始 {original_count} 条清洗后 {len(df)} 条丢弃 {original_count - len(df)} 条)utf-8-sig编码是为了适配Excel直接打开CSV不乱码如果你后续用程序读CSV普通utf-8即可。内容长度过滤掉“不错的”“还行”这类5字以下的无效短评time_ts的上下界判断则是把明显异常的时间戳剔除。关于时效性校验有一个直白的技巧用清洗后的数据统计最近7天的评论数量并把它作为接口存活的指标。如果最近7天评论数长期为零而景点本身是热门目的地说明当前接口已经被调整。这时把第2章里面的 URL、参数名和当前接口做一次diff——2021.6.28 版本的接口快照如果你存过一份 JSON 注释文本对照着看会省掉大量重新分析的工夫。本文还有配套的精品资源点击获取