简介:面向网站内容采集和地图商家信息抓取需求,这套基于HTML/CSS/JavaScript的轻量级网页采集源码将前端页面与交互逻辑整合在一起,部署到服务器根目录即可运行,适合具备基础前端知识的数据采集开发人员、运营人员作为学习样例或二次开发起点。包内共14个文件,包含2个HTML页面、5个CSS样式文件、6个下载文件以及1个必看教程txt,整体仅195KB,结构紧凑、无冗余依赖。此外,源码内置了地图信息采集js版页面,展示如何通过脚本提取商家信息,并附带404错误页与使用说明,便于读者快速理解采集流程、字段组织方式和结果下载机制。目前已有178人学习下载,对于希望掌握网页数据提取、商家信息收集思路并搭建轻量采集工具的初中级开发者,这份源码能提供直观的参考和可运行的实践蓝本。
1. 网页采集软件的解压密码,藏在对采集链路的理解里
这类名为“地图商家采集网页源码”“网站内容采集价值5K源码”的压缩包,在资源站上常年排在下载榜前列。解压之后你会发现,里面通常是一套 PHP 管理后台加若干 Python 脚本,外加一堆采集规则 JSON,资源站把它标成“价值 5K 源码”,和各大站点收录的免费 Python 源码相比,区别只在链路完整度。真正值钱的不是代码本身,而是把“关键词配置 → 请求调度 → 页面解析 → 字段清洗 → 数据落盘”串成一条可跑通的链路。下面就把这条链路拆开讲:地图商家数据怎么从网页检索接口里拿到,网站内容怎么用规则引擎批量抽取,以及源码跑通之后怎么验证结果可靠。适合接采集单、做本地商家目录、以及想自己攒一套采集工具的人阅读。
2. 地图商家采集链路:POI 检索请求、结果解析与字段清洗
先明确一个概念:地图商家采集的本质是 POI(兴趣点)采集,拿的是商家名称、电话、地址、评分、坐标这类结构化字段。它和通用网页数据采集最大的不同,在于反爬重心不在页面结构,而在接口参数的构造和请求节奏的控制。
2.1 地图商家数据与网页内容采集,是同一链条的两个出口
市面上这些源码包的常规做法,是放弃官方开放接口,直接模拟网页版地图在前端发出的检索请求。官方开放接口有每日配额,还要申请 Key,对批量场景不友好;网页检索接口没有配额,但依赖浏览器环境参数,服务端会校验 Referer、签名和时间戳。两类入口的取舍可以先看一张表:
| 数据入口 | 配额 | 需要处理的干扰 | 典型用途 |
|---|---|---|---|
| 官方开放接口 | 有每日配额,并发受限 | Key 申请、配额耗尽 | 小批量、合规优先 |
| 网页检索接口 | 无硬配额,但有访问频控 | 参数签名、请求频率、风控跳转 | 中批量采集,源码包主流 |
网页检索接口的请求格式高度统一:关键字、区域编码、页码、每页条数,响应体是 JSON。所以地图商家采集的难点并不在“拿到网址”,而在“构造出一个能被服务端接受的模拟请求”。
2.2 从开发者工具抓包开始:复现一次地图商家搜索请求
任何源码包里给的地图接口地址,都可能因为站点改版而失效。可靠的做法是自己在浏览器里打开地图网页,按 F12 进入 Network 面板,搜索一个商家关键词,找到名为 search 或 poise 的 XHR 请求,把它的 URL 和参数复制出来替换到代码里。下面这段是一个可以直接改写的骨架。
import requests # 网页版某地图服务的 POI 检索接口占位地址 # 真实地址请在开发者工具 Network 面板里抓取后替换 URL = "https://map.example.com/poi/search" HEADERS = { "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36", "Referer": "https://map.example.com/", "Accept": "application/json, text/plain, */*", } def search_poi(keyword, adcode, page=1, size=20): params = { "keyword": keyword, # 商家业务关键词,如“口腔诊所”“火锅” "adcode": adcode, # 城市区域码,可在开放平台文档查表获得 "pno": page, # 页码从 1 开始 "size": size, # 每页条数,25 以内更稳 "scope": "2", # 2 返回评分、营业时间等扩展字段 } resp = requests.get(URL, params=params, headers=HEADERS, timeout=10) data = resp.json() pois = data.get("data", {}).get("pois", []) return pois, data.get("data", {}).get("count", 0) if __name__ == "__main__": items, total = search_poi("牙科", "110000") print("命中", total, "个商家,本页返回", len(items))这段代码的逻辑很直白:先用 params 字典组装查询参数,交给 requests 发送,再按 JSON 层级取商家列表和总数。参数里最容易出问题的是 adcode 和签名。adcode 是城市行政区划码,填错会直接返回空列表;签名则是多数地图接口约等于强制的一步,服务端要求 URL 带上由前端 JS 计算出的 sign 字段。遇到签名校验,不要试图在 Python 里完整重算签名算法,直接从抓包结果里完整复制一条带 sign 的真实请求试一试,多数情况下该请求在短时间内可以复用。
提示:接口地址、签名参数是源码包里最容易过期的东西。接手任何采集源码,第一件事不是跑,而是抓包核对这两项。
2.3 商家字段清洗:电话、坐标、经营范围为什么“脏”
拿到原始 POI 并不是终点。真实数据里,一个商家的电话字段会同时存在多条号码,用中文分号、逗号混着隔开;地址带省市区前缀,而同名商家会因分属多个分类被重复抓取。清洗是采集链路里最被低估的一环。我一般会在入库前跑一次归一化:
def clean_store(raw): # 电话清洗:把中文分号、逗号统一成英文分号,再拆分 raw_tel = raw.get("tel", "").replace(";", ";").replace(",", ";") phones = [p.strip() for p in raw_tel.split(";") if p.strip()] # 地址归一化:去掉省市前缀,保留门牌信息 addr = raw.get("address", "").strip() # 经营范围只取一级分类,避免“医疗;口腔”这种长串 return { "name": raw.get("name", "").strip(), "phone": "|".join(phones), "address": addr, "category": raw.get("type", "").split(";")[0], "lng": raw.get("lng"), "lat": raw.get("lat"), }清洗函数的关键是“先统一分隔符,再拆字段,最后取子串”。电话不一致、分类过长、坐标缺失这三类脏数据,往往要到 Excel 打开后才会被发现。与其在交付时补救,不如在入库前用同一个函数过滤一遍。还要注意坐标偏移:地图平台返回的多是加偏坐标系,直接和 GPS 原始坐标对比会有几百米偏差,交付时务必保留坐标来源字段,不要混用不同来源的坐标数据。
3. 从源码包拆出标准骨架:网页采集软件的四个必备模块
把“价值 5K 源码.zip”解压后,文件夹通常长这样:admin 目录放 PHP 后台,worker 目录放 Python 采集进程,rules 目录放采集规则,output 目录放导出数据。命名各有差异,但模块划分高度一致——任务管理、请求调度、抽取引擎、数据落地,缺一个都撑不起批量采集。
3.1 一份 5K 级采集源码的目录级解剖
先看整体。我按源码包最常见的实现,把模块和职责整理成一张表,方便对着手里的文件夹核对。
| 模块 | 常见实现 | 对应目录/文件 | 核心职责 |
|---|---|---|---|
| 任务管理 | PHP + MySQL 后台 | admin/, index.php | 配关键词、查任务、看日志 |
| 请求调度 | Python + SQLite | worker/, spider.py | 取任务、控频、重试 |
| 抽取引擎 | XPath/正则规则 | rules/*.json | 从 HTML/JSON 里提取字段 |
| 数据落地 | Excel/CSV 导出 | output/, export.py | 去重、写入、打包 |
这个组合之所以流行,是因为 PHP 后台几乎能在任何虚拟主机上直接部署,而 Python 侧靠 requests、lxml、pandas 三个库就覆盖了请求、解析、导出的全部场景,少数版本会把调度器用 Java 重写以支撑更高并发。源码本身很难造出多高深的东西,值钱的是把四个模块串起来的那层契约:任务表里每行任务带一个 rule_id,抽取引擎按 rule_id 加载规则 JSON,导出程序只认字段名。多数压缩包里还会附一份 README 格式的采集笔记,写明运行顺序和依赖安装,这份笔记往往比代码本身更能反映作者的工程习惯。把契约理清楚,这套源码才算真正归你所有。
3.2 网站内容采集的抽取规则引擎:把 XPath 放进 JSON
网站内容采集与地图商家采集的分水岭在这里。地图商家的响应是 JSON,解析简单;网页内容的响应是 HTML,标签结构千变万化,每个站点一套模板。把 XPath 硬编码进 Python 脚本是最差的方案,换一个列表页就要改代码。规则引擎的做法,是把字段名、XPath 表达式、回退表达式存成 JSON,运行时加载。
import json from lxml import html import requests def extract_by_rule(url, rule): """按 JSON 规则抽取一个页面的字段""" resp = requests.get(url, headers={"User-Agent": "Mozilla/5.0"}, timeout=10) # 部分站点声明的 charset 与实际不符,用 apparent_encoding 兜底 resp.encoding = resp.apparent_encoding doc = html.fromstring(resp.text) row = {"url": url} for field, exprs in rule["fields"].items(): # 规则允许配置一个 XPath 列表,依次回退,直到命中 if not isinstance(exprs, list): exprs = [exprs] for expr in exprs: nodes = doc.xpath(expr) if not nodes: continue node = nodes[0] # 元素节点取全部文本,属性节点直接取字符串 if isinstance(node, str): row[field] = node.strip() else: row[field] = "".join(node.itertext()).strip() break return row rule_example = { "name": "news_detail", "fields": { "title": ["//h1/text()", "//div[contains(@class,'title')]/text()"], "content": ["//div[contains(@class,'article')]//text()"], }, }这个函数有两个容易忽略的细节。第一是resp.encoding = resp.apparent_encoding,很多站点不写 charset 或声明错误,requests 默认按 ISO-8859-1 解码就会满屏乱码,apparent_encoding 靠 chardet 推断实际编码,能覆盖大多数页面。第二是节点文本的拼接方式,如果一个字段落在<div><span>a</span>b</div>这种嵌套里,xpath 的 text() 只能拿到 b,用 itertext() 才能拿全。回退列表的存在,就是为了应对同类站点页面结构不一致的问题。
3.3 调度与断点续采:崩了之后从哪里接上
采集任务跑一半被服务断开是常态,断点续采靠的是任务状态机。我对任务的定义是“一条 URL 加一条规则”,状态从 pending 到 running 再到 done,失败则回退 pending 并累加重试次数。SQLite 足够支撑几千条任务的并发拿取,不需要上消息队列。
import sqlite3 def claim_tasks(db_path, worker_id, batch=10): """从任务表里取出一批待处理任务,并标记为运行中""" con = sqlite3.connect(db_path) cur = con.cursor() cur.execute( """ UPDATE tasks SET state='running', worker=? WHERE id IN ( SELECT id FROM tasks WHERE state='pending' AND retry_times < 3 ORDER BY id LIMIT ? ) RETURNING id, url, rule_id """, (worker_id, batch), ) rows = cur.fetchall() con.commit() return rows用 UPDATE 钳制任务的好处在于,多个 worker 同时运行时不会拿到同一行,比“先 SELECT 再 UPDATE”的方式少一轮竞态。RETURNING 是 SQLite 3.35 之后才有的特性,老环境可以先查 id 再更新。任务调度里必须留一个 sleep 间隔参数,把两个请求之间的停顿交给配置控制,我一般默认 1 到 3 秒随机抖动——这既是给目标站点留余量,也是让采集进程更像人工访问。
注意:间隔参数不要设成 0,也不要设成固定值。固定间隔在服务端日志里比随机间隔更容易被识别。
4. 复现主流程:把网站内容采集与地图商家采集串成一条命令
前三章讲的是链路和组件,这一章把它们落成能直接运行的三段代码。建议先在自己电脑上跑通小批量,再按需加大参数。
4.1 网站内容采集的最小闭环:请求-解析-落盘一条龙
一个函数完成“抓列表页 → 逐条进详情页 → 抽取字段 → 写 CSV”。为了可复现,我把它写成不依赖外部规则文件的最简版。
import csv, time, random import requests from lxml import html def scrape_site(list_urls, detail_xpath, out_csv): headers = {"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64)"} rows = [] for list_url in list_urls: resp = requests.get(list_url, headers=headers, timeout=10) resp.encoding = resp.apparent_encoding doc = html.fromstring(resp.text) # 列表页里抓详情页 href,写一个宽松的 XPath 匹配 detail_links = doc.xpath("//a[contains(@href,'/detail/')]/@href") for link in detail_links[:10]: # 每页最多取前 10 条,避免误抓导航 if link.startswith("/"): link = "https://example-site.com" + link detail = requests.get(link, headers=headers, timeout=10) detail.encoding = detail.apparent_encoding ddoc = html.fromstring(detail.text) title = "".join(ddoc.xpath(detail_xpath["title"])).strip() content = "".join(ddoc.xpath(detail_xpath["content"])).strip() rows.append({"url": link, "title": title, "content": content}) time.sleep(random.uniform(0.5, 1.5)) # 随机间隔,不用固定值 with open(out_csv, "w", newline="", encoding="utf-8-sig") as f: writer = csv.DictWriter(f, fieldnames=["url", "title", "content"]) writer.writeheader() writer.writerows(rows) detail_xpath = { "title": "//h1//text()", "content": "//div[contains(@class,'content')]//text()", }注意两个参数。detail_links 的切片[:10]是最容易被删掉的一行,没有它,页面导航区里的“首页、上一页”这类链接会混进详情采集,产生大量无效请求。sleep 用 random.uniform 而非固定值,是为了让请求间隔分布更自然,避免大批任务在同一毫秒发出。CSV 写入用 utf-8-sig 编码,Excel 双击打开才不会出现中文乱码。
4.2 地图商家采集的排列组合:关键词 × 城市 × 分页
地图商家采集的规模来自“关键词乘城市”的笛卡尔积。比如做餐饮行业数据,关键词可能是“火锅、烧烤、川菜”,城市覆盖十个地级市,每个组合再翻页。下面这段代码处理这个三维循环,并控制请求节奏。
import time, random def crawl_map(keywords, adcodes, pages_per_keyword=5, delay=(1.0, 2.0)): results = [] seen = set() # 用 (名称, 地址) 去重,同名不同地址的店要保留 for kw in keywords: for ac in adcodes: for pno in range(1, pages_per_keyword + 1): pois, total = search_poi(kw, ac, page=pno, size=25) # 复用 2.2 的函数 if not pois: break # 返回空说明翻到末尾或被频控,停下来比硬跑好 for raw in pois: store = clean_store(raw) # 复用 2.3 的清洗函数 key = (store["name"], store["address"]) if key in seen: continue seen.add(key) results.append(store) time.sleep(random.uniform(*delay)) return results这个循环有三个值得说的点。第一,空结果用 break 而不是 continue,因为地图检索接口在某一页返回空,后续页码大概率也是空,继续翻页只浪费请求;但 break 会把“被频控”误判成“翻到底”,所以要在返回体里区分 count 为 0 和请求被拦截,后者靠状态码和响应体特征判断。第二,去重键用“名称+地址”,不能只用名称,同一品牌在多个商圈都有门店。第三,delay 参数化为换城市时的节奏调整留了入口,城市切换是新的访问维度,间隔应该比同城市翻页更保守。
4.3 数据落地的三种姿势:CSV、Excel、MySQL
采集结果最终要给非技术同事或客户看,导出格式决定交付体验。三种姿势各有各的坑,对比如下:
| 导出格式 | 最容易踩的坑 | 对策 |
|---|---|---|
| Excel | 电话列变科学计数法 | 转字符串后去掉尾部 .0 |
| CSV | Excel 打开中文乱码 | 用 utf-8-sig 编码 |
| MySQL | 字段顺序与表结构不一致 | 写入前按表字段排序 |
import pandas as pd def export(df, fmt="xlsx", path="output"): if fmt == "xlsx": # 电话列强制转字符串,Excel 才不会把 138xxxx 显示成科学计数法 if "phone" in df.columns: df["phone"] = df["phone"].astype(str).str.replace(r"\.0$", "", regex=True) df.to_excel(f"{path}/stores.xlsx", index=False, engine="openpyxl") elif fmt == "csv": df.to_csv(f"{path}/stores.csv", index=False, encoding="utf-8-sig") elif fmt == "mysql": # 批量写入要求字段顺序与表结构一致,先排序再 to_sql df = df[["name", "phone", "address", "category", "lng", "lat"]] df.to_sql("stores", con=engine, if_exists="append", index=False, chunksize=500)电话变成科学计数法是 Excel 导出最高频的问题。pandas 读入时若电话列混入空值和数字,整列会被推断为 float,导出就带上.0。上面的处理先把 phone 列转成字符串再删尾部.0,能兜住大部分情况。MySQL 写入用 chunksize 分块,避免一条超大 INSERT 语句卡死连接。坐标字段在入库前再检查一遍缺失情况,经纬度为空的记录到后续定位环节会直接变成废数据。
5. 网页采集源码跑通后,先做这三项可靠性验证再谈交付
源码能跑和跑得可靠是两码事。批量采集前先花半小时做三项验证,比事后拿一份满是空电话的清单返工划算得多。
5.1 用黄金样本校验抽取规则是否真的稳
从目标站点手工挑 5 到 10 个页面,人工记录标题、正文、电话、地址作为黄金样本。跑一遍抽取规则,把输出和人工结果逐字段对比,重点看两类差异:字段完全为空,以及字段里混入杂质。先修规则再扩容,比批量到一半才发现某类页面结构不同要省事很多。
5.2 把失败日志按阶段分层
每条失败记录至少要带任务 id、URL、状态码、失败阶段和重试次数。失败阶段分网络层、解析层、数据层:连接超时属于网络层,重试能解决大部分;XPath 空列表属于解析层,要改规则;导出写库失败属于数据层,查字段类型。日志第一字段打上阶段标签,批量跑完用一条 grep 就能定位主要矛盾。
5.3 增量采集的指纹字段与重复核验
增量采集的做法是给每条记录计算指纹,我一般取“名称 + 地址 + 电话”拼接后做 md5,入库时把指纹连同主键一起落表。下一轮采集先查指纹集合,命中就跳过。换城市、换行业之前,先用一句 SQL 核验历史数据有没有重复:
SELECT name, address, COUNT(*) FROM stores GROUP BY name, address HAVING COUNT(*) > 1;如果重复行数占比超过千分之一,先修去重键再开新一轮采集,否则增量指纹会被脏数据带偏。
本文还有配套的精品资源,点击获取