in 爬取

如何爬取Crunchbase上的公司数据

updated

如何爬取Crunchbase上的公司数据

在本教程中，我们将了解如何抓取Crunchbase——各种上市和私营公司和投资的最大的财务信息公共资源。 Crunchbase 包含数千家公司简介，其中包括投资数据、资金信息、领导职位、合并、新闻和行业趋势。 为了抓取 Crunchbase，我们将使用一种隐藏的网数据网络抓取方法，使用带有 HTTP 客户端库的 Python。 我们将主要关注捕获公司数据，尽管我们将学习的通用抓取算法可以很容易地应用于其他 Crunchbase 领域，例如人员或获取数据，而只需很少的努力。让我们开始吧！

为什么要抓取 Crunchbase.com？

Crunchbase 拥有庞大的业务数据集，可用于各种形式的市场分析和商业智能。例如，公司数据集包含公司的摘要详细信息（如描述、网站和地址）、公共财务信息（如收购、投资和）以及领导力和使用的技术数据。 此外，Crunchbase 数据包含许多用于潜在客户生成的数据点，例如公司的联系方式、领导层的社交资料和事件聚合。

项目设置

在本教程中，我们将使用 Python 和两个主要的社区包：

  • httpx – HTTP 客户端库，可以让我们与 crunchbase.com 的服务器进行通信
  • parsel – HTML 解析库，尽管我们在本教程中将做很少的 HTML 解析，主要是直接处理 JSON 数据。

我们还可以选择使用loguru – 一个漂亮的日志库，可以帮助我们通过漂亮的彩色日志跟踪正在发生的事情。 这些包可以通过pip命令轻松安装：

$ pip install httpx parsel loguru

或者，可以随意换成httpx任何其他 HTTP 客户端包，例如requests，因为我们只需要基本的 HTTP 功能，这些功能几乎可以在每个库中互换。至于，parsel另一个不错的选择是beautifulsoup包。

可用的 Crunchbase 目标

Crunchbase 包含多种数据类型：收购、人员、事件、中心、融资轮次和公司。在本教程中，我们将重点关注公司和人员数据，但我们将使用可应用于所有 Crunchbase 页面的通用解析技术。

crunchbase 发现页面
Crunchbase.com/discovery 页面显示了所有可用的数据集类型

您可以通过查看crunchbase.com/discover页面来探索可用的数据类型。

寻找 Crunchbase 公司和人员

要开始抓取 Crunchbase.com 的内容，我们需要找到一种方法来查找所有公司或人员的 URL。Crunchbase 确实提供了一个搜索系统，但是它只针对高级用户。那么，我们如何找到这些目标呢？ 由于 Crunchbase 希望被搜索引擎抓取和索引，因此它提供了一个包含所有目标 URL 的站点地图目录。让我们先看看crunchbase.com/robots.txt端点：

User-agent: *
Allow: /v4/md/applications/crunchbase
Disallow: /login
<...>
Sitemap: https://www.crunchbase.com/www-sitemaps/sitemap-index.xml

该/robots.txt页面指示各种网络爬虫（如 Google 等）的爬网建议。我们可以看到有一个站点地图索引，其中包含各种目标页面的索引：

<?xml version='1.0' encoding='UTF-8'?>
<sitemapindex xsi:schemaLocation="http://www.sitemaps.org/schemas/sitemap/0.9 http://www.sitemaps.org/schemas/sitemap/0.9/siteindex.xsd" xmlns="http://www.sitemaps.org/schemas/sitemap/0.9" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance">
    <sitemap>
      <loc>https://www.crunchbase.com/www-sitemaps/sitemap-acquisitions-2.xml.gz</loc>
      <lastmod>2022-07-06T06:05:33.000Z</lastmod>
    </sitemap>
    <...>
    <sitemap>
      <loc>https://www.crunchbase.com/www-sitemaps/sitemap-events-0.xml.gz</loc>
      <lastmod>2022-07-06T06:09:30.000Z</lastmod>
    </sitemap>
    <...>
    <sitemap>
      <loc>https://www.crunchbase.com/www-sitemaps/sitemap-funding_rounds-9.xml.gz</loc>
      <lastmod>2022-07-06T06:10:49.000Z</lastmod>
    </sitemap>
    <...>
    <sitemap>
      <loc>https://www.crunchbase.com/www-sitemaps/sitemap-hubs-1.xml.gz</loc>
      <lastmod>2022-07-06T06:05:10.000Z</lastmod>
    </sitemap>
    <...>
    <sitemap>
      <loc>https://www.crunchbase.com/www-sitemaps/sitemap-organizations-42.xml.gz</loc>
      <lastmod>2022-07-06T06:10:35.000Z</lastmod>
    </sitemap>
    <...>
    <sitemap>
      <loc>https://www.crunchbase.com/www-sitemaps/sitemap-people-29.xml.gz</loc>
      <lastmod>2022-07-06T06:09:25.000Z</lastmod>
    </sitemap>
</sitemapindex>

我们可以看到该页面包含收购、活动、融资轮次、中心以及公司（又名组织）和人员的站点地图索引页面。 每个站点地图索引最多可以包含 50 000 个 url，因此目前使用该索引我们可以找到超过 200 万家公司和近 150 万人！ 此外，我们可以看到节点还指示了最后更新日期<lastmod>，因此我们也有关于该索引最后一次更新时间的信息。 让我们来看看我们如何才能抓取所有这些。

抓取站点地图

为了抓取站点地图，我们将使用我们的客户端下载站点地图索引httpx并使用以下方法解析 URL parsel：

import gzip
from datetime import datetime
from typing import Iterator, List, Literal, Tuple

import httpx
from loguru import logger as log
from parsel import Selector


async def _scrape_sitemap_index(session: httpx.AsyncClient) -> List[str]:
    """scrape Crunchbase Sitemap index for all sitemap urls"""
    log.info("scraping sitemap index for sitemap urls")
    response = await session.get("https://www.crunchbase.com/www-sitemaps/sitemap-index.xml")
    sel = Selector(text=response.text)
    urls = sel.xpath("//sitemap/loc/text()").getall()
    log.info(f"found {len(urls)} sitemaps")
    return urls


def parse_sitemap(response) -> Iterator[Tuple[str, datetime]]:
    """parse sitemap for location urls and their last modification times"""
    sel = Selector(text=gzip.decompress(response.content).decode())
    urls = sel.xpath("//url")
    log.info(f"found {len(urls)} in sitemap {response.url}")
    for url_node in urls:
        url = url_node.xpath("loc/text()").get()
        last_modified = datetime.fromisoformat(url_node.xpath("lastmod/text()").get().strip("Z"))
        yield url, last_modified


async def discover_target(
    target: Literal["organizations", "people"], session: httpx.AsyncClient, min_last_modified=None
):
    """discover url from a specific sitemap type"""
    sitemap_urls = await _scrape_sitemap_index(session)
    urls = [url for url in sitemap_urls if target in url]
    log.info(f"found {len(urls)} matching sitemap urls (from total of {len(sitemap_urls)})")
    for url in urls:
        log.info(f"scraping sitemap: {url}")
        response = await session.get(url)
        for url, mod_time in parse_sitemap(response):
            if min_last_modified and mod_time < min_last_modified:
                continue  # skip
            yield url

上面，我们的代码检索中央站点地图索引并收集所有站点地图 URL。然后，我们抓取每个匹配人员或组织（又名公司）模式的站点地图 URL。让我们运行这段代码并查看它产生的值：

运行代码和示例输出
# append this to the previous code snippet to run it:
import asyncio
BASE_HEADERS = {
    "accept-language": "en-US,en;q=0.9",
    "user-agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/96.0.4664.110 Safari/537.36",
    "accept": "text/html,application/xhtml+xml,application/xml;q=0.9,image/webp,image/apng,*/*;q=0.8",
    "accept-language": "en-US;en;q=0.9",
    "accept-encoding": "gzip, deflate, br",
}


async def run():
    async with httpx.AsyncClient(
        limits=httpx.Limits(max_connections=5), timeout=httpx.Timeout(15.0), headers=BASE_HEADERS, http2=True
    ) as session:
        print('discovering companies:')
        async for url in discover_target("organization", session):
            print(url)
        print('discovering people:')
        async for url in discover_target("people", session):
            print(url)


if __name__ == "__main__":
    asyncio.run(run())
discovering companies:
INFO     | _scrape_sitemap_index - scraping sitemap index for sitemap urls
INFO     | _scrape_sitemap_index - found 89 sitemaps
INFO     | discover_target - found 43 matching sitemap urls (from total of 89)
INFO     | discover_target - scraping sitemap: https://www.crunchbase.com/www-sitemaps/sitemap-organizations-0.xml.gz
INFO     | parse_sitemap - found 50000 in sitemap https://www.crunchbase.com/www-sitemaps/sitemap-organizations-0.xml.gz
https://www.crunchbase.com/organization/tesla
<...>
discovering people:
INFO     | _scrape_sitemap_index - scraping sitemap index for sitemap urls
INFO     | _scrape_sitemap_index - found 89 sitemaps
INFO     | discover_target - found 30 matching sitemap urls (from total of 89)
INFO     | discover_target - scraping sitemap: https://www.crunchbase.com/www-sitemaps/sitemap-people-0.xml.gz
INFO     | parse_sitemap - found 50000 in sitemap https://www.crunchbase.com/www-sitemaps/sitemap-people-0.xml.gz
https://www.crunchbase.com/person/john-doe
<...>

我们可以看到，通过浏览 Crunchbase 站点地图，我们可以轻松快速地发现网站上列出的个人资料。现在我们可以找到公司和人员页面的 URL，让我们来看看如何抓取这些公共数据。

爬取 Crunchbase 公司

Crunchbase 公司页面包含大量分散在多个页面中的数据： 然而，我们可以深入了解页面源代码，而不是解析 HTML，我们可以看到相同的数据也可以在页面的应用程序状态变量中使用：

Crunchbase公司页面的页面源
页面源中存在的数据集（注意：为了便于阅读而未加引号，在原始源中一些字符被转义）

我们可以看到一个<script id="client-app-state">节点包含一个大型 JSON 文件，其中包含许多与我们在页面上看到的相同的详细信息。由于 Crunchbase 使用 Angular javascript 前端框架，它将 HTML 数据存储在页面状态缓存中，我们可以直接提取这些数据而不是解析 HTML 页面。让我们来看看我们如何做到这一点：

import json
from typing import Dict, List, TypedDict

from parsel import Selector

class CompanyData(TypedDict):
    """Type hint for data returned by Crunchbase company page parser"""

    organization: Dict
    employees: List[Dict]

def _parse_organization_data(data: Dict) -> Dict:
    """example that parses main company details from the whole company dataset"""
    properties = data['properties']
    cards = data['cards']
    parsed = {
        # theres meta data in the properties field:
        "name": properties['title'],
        "id": properties['identifier']['permalink'],
        "logo": "https://res.cloudinary.com/crunchbase-production/image/upload/" + properties['identifier']['image_id'],
        "description": properties['short_description'],
        # but most of the data is in the cards field:
        "semrush_global_rank": cards['semrush_summary']['semrush_global_rank'],
        "semrush_visits_latest_month": cards['semrush_summary']['semrush_visits_latest_month'],
        # etc... There's much more data!
    }
    return parsed

def _parse_employee_data(data: Dict) -> List[Dict]:
    """example that parses employee details from the whole employee dataset"""
    parsed = []
    for person in data['entities']:
        parsed.append({
            "name": person['properties']['name'],
            "linkedin": person['properties'].get('linkedin'),
            "job_levels": person['properties'].get('job_levels'),
            "job_departments": person['properties'].get('job_departments'),
            # etc...
        })
    return parsed

def _unescape_angular(text):
    """Helper function to unescape Angular quoted text"""
    ANGULAR_ESCAPE = {
        "&a;": "&",
        "&q;": '"',
        "&s;": "'",
        "&l;": "<",
        "&g;": ">",
    }
    for from_, to in ANGULAR_ESCAPE.items():
        text = text.replace(from_, to)
    return text


def parse_company(response) -> CompanyData:
    """parse company page for company and employee data"""

    sel = Selector(text=response.text)
    app_state_data = _unescape_angular(sel.css("script#client-app-state::text").get())
    app_state_data = json.loads(app_state_data)
    # there are multiple caches:
    cache_keys = list(app_state_data["HttpState"])
    # Organization data can be found in this cache:
    data_cache_key = next(key for key in cache_keys if "entities/organizations/" in key)
    # Some employee/contact data can be found in this key:
    people_cache_key = next(key for key in cache_keys if "/data/searches/contacts" in key)

    organization = app_state_data["HttpState"][data_cache_key]["data"]
    employees = app_state_data["HttpState"][people_cache_key]["data"]
    return {
        "organization": _parse_organization_data(organization),
        "employees": _parse_employee_data(employees),
    }

async def scrape_company(company_id: str, session: httpx.AsyncClient) -> CompanyData:
    """scrape crunchbase company page for organization and employee data"""
    # note: we use /people tab because it contains the most data:
    url = f"https://www.crunchbase.com/organization/{company_id}/people"
    response = await session.get(url)
    return parse_company(response)
运行代码和示例输出
# append this to the previous code snippet to run it:
import asyncio
import json

async def run():
    async with httpx.AsyncClient(
        limits=httpx.Limits(max_connections=5), timeout=httpx.Timeout(15.0), headers=BASE_HEADERS, http2=True
    ) as session:
        data = await scrape_company("tesla-motors", session=session)
        print(json.dumps(data, indent=2, ensure_ascii=False))


if __name__ == "__main__":
    asyncio.run(run())
{
  "organization": {
    "name": "Tesla",
    "id": "tesla-motors",
    "logo": "https://res.cloudinary.com/crunchbase-production/image/upload/v1459804290/mkxozts4fsvkj73azuls.png",
    "description": "Tesla Motors specializes in developing a full range of electric vehicles.",
    "semrush_global_rank": 3462,
    "semrush_visits_latest_month": 34638116
  },
  "employees": [
    {
      "name": "Kenneth Rogers",
      "linkedin": "kenneth-rogers-07a7b149",
      "job_levels": [
        "l_500_exec"
      ],
      "job_departments": [
        "management"
      ]
    },
    ...
  ]
}

上面我们定义了我们公司的 scraper，你可以看到它主要是解析代码。让我们在这里快速解压我们的流程：

  1. 我们检索组织的“人员”标签页，例如/organization/tesla-motors。我们使用此页面是因为所有组织子页面（又名选项卡）都包含相同的缓存，除了人员页面外还包含一些员工数据。
  2. 我们在其中找到缓存数据<script id="app-state-data">并取消引用它，因为它使用特殊的 Angular 引用。
  3. 将我们作为 JSON 加载到 Python 的字典中，并从数据集中选择一些重要的字段。请注意，缓存中有大量数据 – 大部分在页面上可见的数据等等 – 但对于本演示，我们坚持使用几个基本字段。

如您所见，由于我们直接抓取 Angular 缓存而不是解析 HTML，我们只需几行代码就可以轻松获取整个数据集！我们可以将其应用于抓取托管在 Crunchbase 上的其他数据类型吗？

抓取其他 Crunchbase 数据类型

Crunchbase 不仅包含公司的详细信息，还包含行业新闻、投资者（人）、融资轮次和收购的详细信息。因为我们选择通过 Angular 缓存而不是 HTML 本身来进行解析，所以我们可以轻松地调整我们的解析器以从这些其他端点提取数据集：

import json
from typing import Dict, List, TypedDict

from parsel import Selector


class PersonData(TypedDict):
    id: str
    name: str


def parse_person(response) -> PersonData:
    """parse person/investor profile from Crunchbase person's page"""
    sel = Selector(text=response.text)
    app_state_data = _unescape_angular(sel.css("script#client-app-state::text").get())
    app_state_data = json.loads(app_state_data)
    cache_keys = list(app_state_data["HttpState"])
    dataset_key = next(key for key in cache_keys if "data/entities" in key)
    dataset = app_state_data["HttpState"][dataset_key]["data"]
    parsed = {
        # we can get metadata from properties field:
        "title": dataset['properties']['title'],
        "description": dataset['properties']['short_description'],
        "type": dataset['properties']['layout_id'],
        # the rest of the data can be found in the cards field:
        "investing_overview": dataset['cards']['investor_overview_headline'],
        "socials": {k: v['value'] for k, v in dataset['cards']['overview_fields2'].items()},
        "positions": [{
            "started": job.get('started_on', {}).get('value'),
            "title": job['title'],
            "org": job['organization_identifier']['value'],
            # etc.
        } for job in dataset['cards']['current_jobs_image_list']],
        # etc... there are many more fields to parse
    }
    return parsed


async def scrape_person(person_id: str, session: httpx.AsyncClient) -> PersonData:
    """scrape Crunchbase.com investor's profile"""
    url = f"https://www.crunchbase.com/person/{person_id}"
    response = await session.get(url)
    return parse_person(response)

上面的例子使用了我们用来抓取公司数据来抓取投资者数据的相同技术。通过从 Angular 应用程序状态中提取数据，我们只需几行代码就可以抓取任何 Crunchbase 端点的数据集！

常问问题

为了总结本指南，让我们看一下有关网络抓取Crunchbase.com 的一些常见问题：

是的。Crunchbase 数据是公开的，我们不会提取任何私人信息。以缓慢、尊重的速度抓取 Crunchbase.com 属于道德抓取定义。话虽如此，在抓取个人数据（例如个人（投资者）数据）时，应注意欧盟的 GDRP 合规性。

你能抓取 Crunchbase.com 吗？

是的，爬crunchbase的方法有很多种。然而，由于 Crunchbase 拥有丰富的站点地图基础设施，因此爬取是不必要的。

概括

在本教程中，我们构建了一个Crunchbase爬虫。我们已经了解了如何通过 Crunchbase 的站点地图功能发现公司和人员页面。然后，我们为 Crunchbase 本身等 Angular 支持的网站编写了一个通用数据集解析器，并将其用于抓取公司和人员数据。为此，我们将 Python 与一些工具包（如httpx ）一起使用。  

Python爬虫工具包

Written by 河小马

河小马是一位杰出的数字营销行业领袖，广告中国论坛的重要成员，其专业技能涵盖了PPC广告、域名停放、网站开发、联盟营销以及跨境电商咨询等多个领域。作为一位资深程序开发者，他不仅具备强大的技术能力，而且在出海网络营销方面拥有超过13年的经验。