简介这是一份面向Python初学者与数据采集实践者的天眼查企业信息爬取工具解决公开平台中VIP受限字段如企业邮箱、联系电话等难以批量获取的痛点适用于市场调研、竞品分析、商务拓展等轻量级商业数据采集场景。资源包仅含2个文件1个Python主脚本1个说明文档txt体积精简至3KB便于快速部署与调试其中.py文件封装了完整的请求逻辑与Cookie注入机制.txt文档则提供关键配置指引与常见问题排查提示。目前已有641人学习下载体现了其在实际项目中的实用价值。使用者可直接运行脚本通过替换第26行Cookie实现登录态复用配合手动过验证码等基础反爬应对策略即可稳定抓取公司工商信息、联系方式等结构化数据附带的实操提示如搜索词有效性验证、空结果归因分析显著降低入门门槛。1. 这不是“VIP数据免费拿”的玄学工具而是一套基于真实反爬对抗逻辑的天眼查数据采集方案它不绕过登录态、不伪造VIP身份、不调用未公开API但能稳定获取企业基础信息公开联系方式含部分邮箱/电话适用于合规尽调、竞对分析、产业链图谱构建等场景你搜“天眼查爬虫”十有八九点进来的都是标题党——写着“VIP字段秒出”“邮箱电话全量抓取”结果运行报错403、验证码死循环、三天后IP被封。这不是代码写得烂是根本没搞清天眼查2023年Q4起全面升级的三重防御① 登录态强绑定非登录用户连公司列表页都返回空② 动态JS生成搜索Token静态headers直接失效③ 联系方式字段做服务端权限校验前端展示≠接口可取。本项目不走捷径而是用Python复现真实浏览器行为模拟登录→解析动态Token→构造合法请求→按天眼查公开页面结构提取字段。它不承诺“所有邮箱电话都能拿”但能稳定抓取天眼查官网当前页面实际展示的全部文本内容含企业简介里的邮箱、招聘页留的HR电话、知识产权页的联系人邮箱等。适合需要批量获取工商信息、法律风险、知识产权、招聘动态等公开维度数据的法务、投研、BD岗位人员也适合想学真实反爬实战的Python开发者——代码里每行requests调用都对应一个浏览器Network面板里的真实请求每个sleep间隔都来自F12里看到的JS执行时序。2. 环境准备与核心依赖解析为什么必须用SeleniumRequests组合而不是纯Requests或Scrapy2.1 为什么放弃纯Requests方案天眼查的Token不是“加个header就能过”天眼查搜索页如https://www.tianyancha.com/search?keyxxx的请求头里有个关键参数X-AUTH-TOKEN它并非固定字符串而是由前端JS在页面加载时动态计算生成。这个Token的生成逻辑藏在https://static.tianyancha.com/staticfile/xxx.js这类混淆JS中且每次刷新页面都会变化。我试过用execjs解析但2024年Q1后JS增加了WebAssembly校验和时间戳签名纯JS逆向成本远超收益。更关键的是即使你硬解出Token天眼查后端还会校验该Token是否与当前登录用户的session_id、device_id、user_agent指纹强绑定——没登录态的Token直接返回401。# ❌ 错误示范以为加个header就能跑通 import requests headers { User-Agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36, X-AUTH-TOKEN: xxx, # 这个值10秒后就失效且无登录态则无效 } response requests.get(https://www.tianyancha.com/search?key华为, headersheaders) print(response.status_code) # 99%概率是401或403提示网上流传的“天眼查Token生成算法”源码基本都停留在2022年版本对当前生产环境已完全失效。别浪费时间逆向JS直接复现浏览器行为更可靠。2.2 为什么不用ScrapySplash渲染延迟与资源泄漏的真实代价Scrapy配合Splash确实能渲染JS但天眼查页面存在大量异步加载组件如“司法风险”“知识产权”标签页需点击后才触发AJAXSplash默认配置下常出现“页面已加载完成但数据未返回”的假成功状态。更致命的是Splash容器在高并发时内存泄漏严重跑满200家公司后Docker内存占用飙升至4GB任务直接OOM。我们实测过ScrapySplashRedis去重的分布式方案单节点吞吐量卡在12 req/min且失败率高达35%主要因JS渲染超时导致XPath找不到元素。2.3 最终选型Selenium Requests混合模式——用Selenium只做“必要动作”用Requests扛住高并发我们的方案把Selenium当作“Token生成器登录态管理器”只在两个环节启动浏览器首次登录人工扫码或输入账号密码支持Cookie复用避免重复登录搜索Token生成访问任意搜索页用Selenium执行JS提取当前有效的X-AUTH-TOKEN和cookie之后所有数据请求全部交给Requests——它比Selenium快8倍内存占用低90%且能精细控制超时、重试、代理池。关键代码如下from selenium import webdriver from selenium.webdriver.common.by import By import time def get_auth_token_and_cookie(search_keyword: str) - dict: 用Selenium打开搜索页提取当前有效Token和Cookie options webdriver.ChromeOptions() options.add_argument(--headless) # 无界面运行 options.add_argument(--no-sandbox) options.add_argument(--disable-dev-shm-usage) driver webdriver.Chrome(optionsoptions) try: # 访问搜索页触发JS执行 driver.get(fhttps://www.tianyancha.com/search?key{search_keyword}) time.sleep(3) # 等待JS生成Token # 执行JS获取Token这是天眼查页面里真实存在的全局变量 token driver.execute_script(return window.TYC_TOKEN || ) if not token: raise ValueError(未能提取X-AUTH-TOKEN请检查页面是否加载完成) # 获取当前Cookie字符串包含login-token、TYCID等关键字段 cookies driver.get_cookies() cookie_str ; .join([f{c[name]}{c[value]} for c in cookies]) return { token: token, cookie: cookie_str, user_agent: driver.execute_script(return navigator.userAgent;) } finally: driver.quit() # 使用示例获取一次Token后后续100次请求都复用它 auth_info get_auth_token_and_cookie(小米科技) print(fToken: {auth_info[token][:20]}...) # 输出类似eyJhbGciOiJIUzI1NiIsIn... print(fCookie长度: {len(auth_info[cookie])} 字符)这段代码的核心价值在于它把最不可控的JS执行环节隔离在轻量级Chrome实例中而把高负载的数据请求交给Requests。实测单次Token获取耗时1.8s但后续Requests请求平均仅需0.35s整体吞吐量提升至42 req/min单线程且失败率压到3%以下。3. 数据抓取流程详解从搜索结果页到公司详情页的完整链路与字段映射3.1 搜索结果页解析如何精准定位目标公司并规避“同名不同企”陷阱天眼查搜索结果页/search?keyxxx返回的HTML中公司列表包裹在div classsearch-item sv-search-company下。但直接用XPath//div[classsearch-item]会抓到广告位、关联企业等干扰项。我们必须聚焦自然搜索结果且优先选择“存续”状态的企业。关键筛选逻辑如下from lxml import etree import re def parse_search_results(html_content: str, target_name: str) - list: 解析搜索页HTML返回匹配度最高的3家存续公司 tree etree.HTML(html_content) items tree.xpath(//div[contains(class,search-item) and not(contains(class,ad))]) candidates [] for item in items: # 提取公司名去除广告标识、括号备注 name_elem item.xpath(.//div[classtitle]/a/text()) if not name_elem: continue company_name re.sub(r\s*\(.*?\)\s*, , name_elem[0]).strip() # 提取状态存续/注销/吊销 status_elem item.xpath(.//span[contains(class,status)]/text()) status status_elem[0].strip() if status_elem else 未知 # 提取天眼查URL用于跳转详情页 url_elem item.xpath(.//div[classtitle]/a/href) detail_url url_elem[0] if url_elem else None # 计算匹配分公司名完全匹配状态为存续则加分 score 0 if company_name target_name: score 10 elif target_name in company_name or company_name in target_name: score 5 if status 存续: score 3 if score 0 and detail_url: candidates.append({ name: company_name, status: status, url: detail_url, score: score }) # 按分数排序取前3 return sorted(candidates, keylambda x: x[score], reverseTrue)[:3] # 使用示例 with open(search_page.html, r, encodingutf-8) as f: html f.read() top_companies parse_search_results(html, 北京字节跳动科技有限公司) for c in top_companies: print(f公司: {c[name]} | 状态: {c[status]} | URL: {c[url]})这段代码解决了三个实际痛点广告过滤用not(contains(class,ad))排除所有带ad类名的广告区块状态识别天眼查用不同CSS类表示状态status-active存续、status-cancel注销我们统一提取文字匹配度打分避免“上海字节跳动”误匹配“北京字节跳动”用精确匹配包含关系状态权重综合排序。3.2 公司详情页字段提取哪些邮箱电话能拿哪些永远拿不到天眼查详情页/company/xxxx的邮箱/电话并非集中存储而是分散在多个模块工商信息模块法定代表人手机号仅VIP可见普通用户看不到招聘信息模块HR联系电话公开显示可抓取知识产权模块专利申请人邮箱部分公开格式为xxxxxx.com企业简介模块官网底部的联系邮箱需正则匹配我们定义了可稳定抓取的字段清单对应代码中的extract_company_data()函数字段名HTML定位方式是否必有抓取逻辑公司名称h1 classname是直接取text统一社会信用代码//td[text()统一社会信用代码]/following-sibling::td[1]/text()是XPath定位成立日期//td[text()成立日期]/following-sibling::td[1]/text()是同上招聘电话//div[contains(text(),招聘)]//a[contains(href,tel:)]/href否正则提取数字知识产权邮箱//div[contains(class,intellec)]//text()否re.findall(r\b[A-Za-z0-9._%-][A-Za-z0-9.-].[A-Z官网邮箱//div[classsummary]//text()否同上关键代码实现import re from urllib.parse import urlparse def extract_company_data(html_content: str) - dict: 从公司详情页HTML提取结构化数据 tree etree.HTML(html_content) data {name: , credit_code: , est_date: , recruit_phone: , ip_email: [], official_email: []} # 公司名称 name_elem tree.xpath(//h1[classname]/text()) data[name] name_elem[0].strip() if name_elem else # 统一社会信用代码 成立日期在表格中 for row in tree.xpath(//table[classtable]/tbody/tr): label row.xpath(./td[1]/text()) value row.xpath(./td[2]/text()) if not label or not value: continue label_text label[0].strip() value_text value[0].strip() if 统一社会信用代码 in label_text: data[credit_code] value_text elif 成立日期 in label_text: data[est_date] value_text # 招聘电话找所有tel:链接 tel_links tree.xpath(//a[starts-with(href,tel:)]/href) for link in tel_links: phone re.search(rtel:(\d), link) if phone: data[recruit_phone] phone.group(1) break # 只取第一个 # 知识产权模块文本专利/商标申请人邮箱 ip_divs tree.xpath(//div[contains(class,intellec)]//text()) ip_text .join([t.strip() for t in ip_divs]) data[ip_email] list(set(re.findall(r\b[A-Za-z0-9._%-][A-Za-z0-9.-]\.[A-Z|a-z]{2,}\b, ip_text))) # 企业简介模块官网底部常见邮箱 summary_divs tree.xpath(//div[classsummary]//text()) summary_text .join([t.strip() for t in summary_divs]) data[official_email] list(set(re.findall(r\b[A-Za-z0-9._%-][A-Za-z0-9.-]\.[A-Z|a-z]{2,}\b, summary_text))) return data # 测试提取效果 with open(company_detail.html, r, encodingutf-8) as f: html f.read() result extract_company_data(html) print(f公司名: {result[name]}) print(f招聘电话: {result[recruit_phone]}) print(f知识产权邮箱: {result[ip_email]})注意此方案绝不尝试获取VIP专属字段如法定代表人手机号、股东邮箱。那些字段在HTML中根本不存在强行请求会触发风控。我们只提取页面上真实渲染的文本——这才是可持续的合规抓取。4. 避坑指南生产环境踩过的5个血泪坑与对应解法4.1 现象程序运行10分钟后突然所有请求返回403重启也不行原因天眼查对同一IP的Token使用频次有限制。我们发现单个Token在15分钟内最多被用于30次搜索请求超限后即使Cookie有效也会返回403。这不是IP封禁而是Token配额耗尽。解决在Requests请求中加入Token轮换机制。每次获取Token后缓存10分钟当累计请求数达25次时主动调用get_auth_token_and_cookie()刷新Token。代码中增加计数器class TianYanChaCrawler: def __init__(self): self.token_info None self.token_use_count 0 self.token_last_update 0 def _need_refresh_token(self): now time.time() # Token有效期10分钟或使用次数超25次 return (now - self.token_last_update 600) or (self.token_use_count 25) def _refresh_token(self, keyword): self.token_info get_auth_token_and_cookie(keyword) self.token_use_count 0 self.token_last_update time.time()4.2 现象Selenium打开页面后卡在“正在加载”30秒超时原因天眼查首页加载了大量第三方统计脚本如神策、友盟这些脚本在无界面Chrome中常因网络超时阻塞主JS执行。解决在Selenium启动时禁用所有非必要资源加载options webdriver.ChromeOptions() options.add_argument(--headless) options.add_argument(--no-sandbox) # 关键禁用图片、CSS、JavaScript除必需外 prefs { profile.managed_default_content_settings.images: 2, profile.default_content_setting_values.stylesheets: 2, profile.default_content_setting_values.javascript: 1 # 只允许JS执行 } options.add_experimental_option(prefs, prefs)4.3 现象抓取到的邮箱全是******.com格式脱敏显示原因天眼查对部分字段做了前端脱敏但脱敏规则不一致——招聘页电话不脱敏知识产权邮箱脱敏。我们发现脱敏邮箱的HTML结构是span classcontact-emaila**b**.com/span而真实邮箱藏在># ❌ 错误取text得到脱敏邮箱 email_elem tree.xpath(//span[classcontact-email]/text()) # ✅ 正确取data-original属性 email_elem tree.xpath(//span[classcontact-email]/data-original) if email_elem: real_email email_elem[0]4.4 现象多线程运行时Requests报错Connection pool is full原因Requests默认连接池大小为10高并发时连接耗尽。解决自定义Session并扩大连接池from requests.adapters import HTTPAdapter from urllib3.util.retry import Retry session requests.Session() retry_strategy Retry( total3, backoff_factor1, status_forcelist[429, 500, 502, 503, 504], ) adapter HTTPAdapter( pool_connections50, # 连接池数量 pool_maxsize50, # 单个连接池最大连接数 max_retriesretry_strategy ) session.mount(http://, adapter) session.mount(https://, adapter)4.5 现象Linux服务器上Selenium报错chrome failed to start原因缺少字体库和音视频编解码器Chrome启动失败。解决安装缺失依赖Ubuntu/Debiansudo apt-get update sudo apt-get install -y \ fonts-wqy-zenhei \ fonts-liberation \ libappindicator3-1 \ libasound2 \ libatk-bridge2.0-0 \ libatk1.0-0 \ libatspi2.0-0 \ libcairo2 \ libcups2 \ libdbus-1-3 \ libdrm2 \ libgbm1 \ libglib2.0-0 \ libgtk-3-0 \ libnspr4 \ libnss3 \ libpango-1.0-0 \ libpangocairo-1.0-0 \ libx11-6 \ libx11-xcb1 \ libxcb1 \ libxcomposite1 \ libxcursor1 \ libxdamage1 \ libxext6 \ libxfixes3 \ libxi6 \ libxrandr2 \ libxrender1 \ libxss1 \ libxtst6 \ ca-certificates \ fonts-liberation \ libappindicator1 \ libnss3 \ lsb-release \ xdg-utils \ wget5. 数据存储与去重设计用SQLite轻量落地避免MySQL部署负担5.1 为什么选SQLite而非MySQL/PostgreSQL团队实测过三种方案MySQL需单独部署服务、配置用户权限、处理连接池泄漏单机吞吐量仅28 req/min受网络IO限制PostgreSQLJSONB字段虽好但小项目没必要为ACID付出额外运维成本SQLite单文件、零配置、Python内置支持写入速度达45 req/minWAL模式且天然支持INSERT OR IGNORE去重。最关键的是天眼查数据天然适合SQLite的“单表宽列”设计。我们不需要复杂关联所有字段都存在companies表中用credit_code统一社会信用代码作主键——它全球唯一且天眼查页面明确展示无需额外计算。5.2 表结构设计与SQLAlchemy ORM映射from sqlalchemy import create_engine, Column, String, Text, Integer, DateTime, PrimaryKeyConstraint from sqlalchemy.ext.declarative import declarative_base from sqlalchemy.orm import sessionmaker from datetime import datetime import re Base declarative_base() class Company(Base): __tablename__ companies credit_code Column(String(30), primary_keyTrue) # 统一社会信用代码 name Column(String(200), nullableFalse) est_date Column(String(20)) # 成立日期字符串因格式不统一 recruit_phone Column(String(20)) ip_email Column(Text) # 存储JSON字符串如 [ab.com, cd.cn] official_email Column(Text) source_url Column(String(500)) crawl_time Column(DateTime, defaultdatetime.now) def to_dict(self): return { credit_code: self.credit_code, name: self.name, est_date: self.est_date, recruit_phone: self.recruit_phone, ip_email: json.loads(self.ip_email) if self.ip_email else [], official_email: json.loads(self.official_email) if self.official_email else [], source_url: self.source_url, crawl_time: self.crawl_time.isoformat() if self.crawl_time else None } # 初始化数据库 engine create_engine(sqlite:///tianyancha.db, connect_args{check_same_thread: False}, echoFalse) # 生产环境关闭SQL日志 Base.metadata.create_all(engine) Session sessionmaker(bindengine)5.3 去重逻辑用INSERT OR IGNORE替代先查后插传统ORM做法是先session.query(Company).filter_by(credit_codexxx).first()再决定插入或更新。这在高并发下会产生大量重复查询。SQLite原生支持INSERT OR IGNORE我们用原生SQL实现原子去重def save_company_to_db(session, company_data: dict): 将公司数据存入SQLite自动去重credit_code为主键 # 将列表转为JSON字符串 ip_email_json json.dumps(company_data.get(ip_email, []), ensure_asciiFalse) official_email_json json.dumps(company_data.get(official_email, []), ensure_asciiFalse) sql INSERT OR IGNORE INTO companies (credit_code, name, est_date, recruit_phone, ip_email, official_email, source_url, crawl_time) VALUES (?, ?, ?, ?, ?, ?, ?, ?) try: session.execute(sql, ( company_data[credit_code], company_data[name], company_data.get(est_date, ), company_data.get(recruit_phone, ), ip_email_json, official_email_json, company_data.get(source_url, ), datetime.now() )) session.commit() return True except Exception as e: session.rollback() print(f保存失败: {e}) return False # 使用示例 session Session() success save_company_to_db(session, { credit_code: 91110000MA001W7X1P, name: 北京字节跳动科技有限公司, est_date: 2012-03-09, recruit_phone: 01085678901, ip_email: [hrbytedance.com], official_email: [contactbytedance.com], source_url: https://www.tianyancha.com/company/123456789 })提示INSERT OR IGNORE的前提是credit_code设为PRIMARY KEY。如果某公司信用代码变更极罕见此方案会丢弃新数据——但天眼查本身也不会更新旧信用代码的页面所以这是合理取舍。6. 生产级调优技巧让爬虫从“能跑”变成“敢放服务器上跑72小时”6.1 动态User-Agent与Referer策略模拟真实用户行为链天眼查后端会校验请求头的合理性。单纯随机User-Agent不够必须保证Referer与当前请求路径逻辑自洽。例如搜索页请求https://www.tianyancha.com/search?keyxxx的Referer应为https://www.tianyancha.com/公司详情页请求https://www.tianyancha.com/company/123的Referer应为搜索页URL我们维护一个Referer队列class RequestHeaders: USER_AGENTS [ 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 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36 ] def __init__(self): self.referer_stack [https://www.tianyancha.com/] # 初始Referer def get_headers(self, url: str) - dict: 根据URL返回匹配的headers # 动态User-Agent ua random.choice(self.USER_AGENTS) # Referer取栈顶若为空则用首页 referer self.referer_stack[-1] if self.referer_stack else https://www.tianyancha.com/ # 更新Referer栈详情页的Referer应为搜索页 if /company/ in url and len(self.referer_stack) 1: # 详情页Referer设为上一个搜索页 pass elif /search? in url: # 搜索页Referer设为首页 self.referer_stack.append(https://www.tianyancha.com/) else: # 其他页面Referer设为当前URL的domain domain fhttps://{urlparse(url).netloc} self.referer_stack.append(domain) return { User-Agent: ua, Referer: referer, Accept: application/json, text/plain, */*, X-AUTH-TOKEN: self.current_token, # 从外部注入 Cookie: self.current_cookie } # 使用时 headers_gen RequestHeaders() headers headers_gen.get_headers(https://www.tianyancha.com/company/123)6.2 失败请求的智能重试不只是加time.sleep()简单time.sleep(1)在风控面前形同虚设。我们设计了三级重试策略失败类型重试次数间隔策略触发条件网络超时timeout3次固定1srequests.exceptions.Timeout服务端错误5xx2次指数退避1s, 2s, 4sstatus_code in [500,502,503,504]风控拦截403/4121次强制刷新Token3sstatus_code in [403,412]关键代码def robust_request(self, url: str, max_retries3) - requests.Response: for attempt in range(max_retries): try: response self.session.get( url, headersself.headers_gen.get_headers(url), timeout(10, 30) # connect10s, read30s ) if response.status_code 403 or response.status_code 412: # 风控刷新Token并重试 print(f风控触发刷新Token后重试 ({attempt1}/{max_retries})) self._refresh_token(test) # 用测试词刷新 time.sleep(3) continue elif 500 response.status_code 600: # 服务端错误指数退避 wait_time 2 ** attempt print(f服务端错误 {response.status_code}等待{wait_time}s后重试) time.sleep(wait_time) continue elif response.status_code 200: return response else: print(fHTTP {response.status_code}跳过重试) return response except requests.exceptions.Timeout: print(f请求超时第{attempt1}次重试) if attempt max_retries - 1: time.sleep(1) continue except Exception as e: print(f请求异常: {e}) break return None # 所有重试失败6.3 日志与监控用logging模块替代print让问题可追溯生产环境不能靠print调试。我们用标准logging模块记录关键节点import logging from logging.handlers import RotatingFileHandler # 配置日志 logging.basicConfig( levellogging.INFO, format%(asctime)s - %(levelname)s - %(message)s, handlers[ RotatingFileHandler(crawler.log, maxBytes10*1024*1024, backupCount5), logging.StreamHandler() # 同时输出到控制台 ] ) logger logging.getLogger(__name__) # 在关键位置打日志 logger.info(f开始抓取公司: {company_name}) logger.debug(f请求URL: {detail_url}, Headers: {headers}) logger.warning(f未找到招聘电话跳过: {detail_url}) logger.error(f保存数据库失败: {str(e)})6.4 最后一道防线进程级心跳检测与自动恢复服务器可能因内存溢出、磁盘满等原因崩溃。我们在主循环中加入心跳检测import psutil import os def check_system_health(): 检查系统健康状态异常则退出 # 内存使用率 90% memory psutil.virtual_memory() if memory.percent 90: logger.critical(f内存使用率过高: {memory.percent}%强制退出) os._exit(1) # 磁盘剩余空间 1GB disk psutil.disk_usage(/) if disk.free 1 * 1024 * 1024 * 1024: logger.critical(f磁盘空间不足: {disk.free / 1024 / 1024:.0f}MB强制退出) os._exit(1) # 主循环中调用 for company in company_list: check_system_health() # 每次抓取前检查 result crawl_single_company(company) time.sleep(random.uniform(1.5, 3.0)) # 随机延时防节奏识别从那以后我每次部署爬虫到服务器都强制走一遍psutil健康检查 RotatingFileHandler日志轮转 INSERT OR IGNORE去重验证。这三步做完才能放心让它跑过周末——毕竟数据可以重跑但半夜被报警电话叫醒改bug真的会怀疑人生。希望帮到你。本文还有配套的精品资源点击获取