データ駆動型採用：ウェブスクレイピングを通じてスケーラブルなタレントインテリジェンスプラットフォームを構築する

basket.yaml                       (アナリスト定義: 企業 × ソース × 地域)
       │
       ▼
┌──────────────────┐
│   オーケストレーター   │  （企業、ソース、地域）ごとに1つのタスク; 制約されたファンアウト
└──────┬───────────┘
       │
       ▼
┌──────────────────┐
│   スクラペレス     │  connect_over_cdp → ホームページをウォームアップ → 検索URLをロード
│  (クラウドブラウザ) │  米国住宅イーグレス、JSレンダリング、検出防止
└──────┬───────────┘
       │  レンダリングされたHTML
       ▼
┌──────────────────┐
│   ノーマライザー     │  ソースごとの抽出器 → 標準掲載スキーマ
└──────┬───────────┘
       │
       ▼
postings.ndjson           （1行ごとに公開掲載観察）
       │
       ▼
データウェアハウスのロード + 速度の導出 / バックフィル / 機能ミックス + アラート

import os
from urllib.parse import urlencode
from playwright.sync_api import sync_playwright

def scraping_browser_url(proxy_country="US", session_ttl=240):
    params = urlencode({
        "token": os.environ["SCRAPELESS_API_KEY"],
        "sessionTTL": session_ttl,
        "proxyCountry": proxy_country,
    })
    return f"wss://browser.scrapeless.com/api/v2/browser?{params}"

from playwright.sync_api import sync_playwright

def render_search_page(homepage_url: str, search_url: str,
                       proxy_country: str = "US") -> str:
    """ホームページをウォームアップした後、クラウドで公開の求人検索ページをレンダリングします。"""

with sync_playwright() as p:
        browser = p.chromium.connect_over_cdp(
            scraping_browser_url(proxy_country=proxy_country)
        )
        context = browser.contexts[0] if browser.contexts else browser.new_context()
        page = context.pages[0] if context.pages else context.new_page()

        # 1) まずはホームページでセッションを温めます。
        page.goto(homepage_url, wait_until="domcontentloaded", timeout=60_000)
        page.wait_for_timeout(1_500)

        # 2) 次に、ターゲットの公開検索ページを読み込みます。グリッドはハイドレーションの後に描画されます。
        page.goto(search_url, wait_until="networkidle", timeout=60_000)
        page.wait_for_selector("[data-posting], article, li", timeout=20_000)

        html = page.content()
        browser.close()
        return html

# basket.yaml
regions:
  - US

companies:
  - company: target_company_a
    sources:
      - source: company_careers
        homepage: "https://careers.example-company-a.com/"
        search:
          US: "https://careers.example-company-a.com/jobs?country=US"
      - source: public_aggregator
        homepage: "https://jobs.example-aggregator.com/"
        search:
          US: "https://jobs.example-aggregator.com/search?q=engineering&loc=US"
  - company: target_company_b
    sources:
      - source: company_careers
        homepage: "https://careers.example-company-b.com/"
        search:
          US: "https://careers.example-company-b.com/openings"

from dataclasses import dataclass, asdict
from datetime import datetime, timezone
from typing import Optional
from lxml import html as lxml_html

@dataclass
class PostingRecord:
    company: str
    source: str
    region: str
    posting_id: str                 # ソースごとの安定したIDまたはタイトル+場所のハッシュ
    role_title: str
    function: Optional[str]          # "engineering" | "sales" | "ops" | ...
    seniority: Optional[str]         # "junior" | "mid" | "senior" | "staff" | None
    location: Optional[str]          # 市 / メトロ / "リモート"
    posted_date: Optional[str]       # ページがレンダリングするISO日付文字列、またはNone
    salary_band: Optional[str]       # ページが公開している範囲
    captured_at: str                 # ISO-8601 UTC、読み取り時間に記載

def extract_company_careers(html: str, company: str, source: str,
                            region: str) -> list[PostingRecord]:
    doc = lxml_html.fromstring(html)
    records: list[PostingRecord] = []

    for card in doc.cssselect("[data-posting], article.job-card"):
        title_el = card.cssselect(".job-title, h3")
        loc_el = card.cssselect(".job-location, [data-location]")
        date_el = card.cssselect("time, [data-posted]")
        pay_el = card.cssselect(".salary, [data-comp]")

        title = title_el[0].text_content().strip() if title_el else ""
        if not title:
            continue  # 非投稿カードをスキップ；タイトルが存在しない場合は実際のリスティングではありません

        location = loc_el[0].text_content().strip() if loc_el else None
        posted = (date_el[0].get("datetime") or date_el[0].text_content().strip()) if date_el else None

        records.append(PostingRecord(
            company=company,
            source=source,
            region=region,
            posting_id=_posting_id(card, title, location),
            role_title=title,
            function=_classify_function(title),
            seniority=_classify_seniority(title),

分類ヘルパーは、DOMではなく抽出器で導出を保持します。彼らは役割のタイトルを粗い機能とシニオリティバンドにマッピングします — 個別のデータは含まれていません：

```python
import hashlib

_FUNCTION_KEYWORDS = {
    "エンジニアリング": ("エンジニア", "デベロッパー", "SRE", "プラットフォーム", "ML ", "データ "),
    "営業": ("営業", "アカウントエグゼクティブ", "アカウントマネージャー", "SDR"),
    "マーケティング": ("マーケティング", "成長", "ブランド", "コンテンツ"),
    "オペレーション": ("オペレーション", "サプライ", "ロジスティクス", "サポート"),
}
_SENIORITY_KEYWORDS = {
    "スタッフ": ("スタッフ", "プリンシパル", "ディスティングイッシュド"),
    "シニア": ("シニア", "SR.", "リード"),
    "ジュニア": ("ジュニア", "JR.", "インターン", "エントリー"),
}

def _classify(title: str, table: dict) -> Optional[str]:
    low = title.lower()
    for label, kws in table.items():
        if any(kw in low for kw in kws):
            return label
    return None

def _classify_function(title: str) -> Optional[str]:
    return _classify(title, _FUNCTION_KEYWORDS)

def _classify_seniority(title: str) -> Optional[str]:
    return _classify(title, _SENIORITY_KEYWORDS) or "ミッド"

def _posting_id(card, title: str, location: str | None) -> str:
    native = card.get("data-posting-id") or card.get("id")
    if native:
        return native.strip()
    # タイトル + ロケーションの安定したハッシュは、実行を通じて再掲された役割を識別します。
    basis = f"{title}|{location or ''}".lower().encode("utf-8")
    return hashlib.sha1(basis).hexdigest()[:16]

import yaml

def load_basket(path: str = "basket.yaml") -> dict:
    with open(path, encoding="utf-8") as f:
        return yaml.safe_load(f)

def walk_basket(basket: dict):
    """すべての（会社、ソース、地域）エントリのPostingRecordリストを生成します。"""
    for item in basket["companies"]:
        for src in item["sources"]:
            for region, search_url in src["search"].items():
                html = render_search_page(
                    homepage_url=src["homepage"],
                    search_url=search_url,
                    proxy_country=region,
                )
                yield extract_company_careers(
                    html, item["company"], src["source"], region
                )

import json
from pathlib import Path

def append_records(records: list[PostingRecord], out_path: str = "postings.ndjson"):
    Path(out_path).parent.mkdir(parents=True, exist_ok=True)
    with open(out_path, "a", encoding="utf-8") as f:
        for r in records:
            f.write(json.dumps(asdict(r)) + "\n")

-- 採用速度: 会社ごとの新規投稿数、先週7日間 vs 前の7日間
WITH obs AS (
  SELECT company, function, posting_id,

CAST(posted_date AS DATE) AS posted_date,
         CAST(captured_at AS DATE) AS captured_date
  FROM talent_postings
  WHERE posted_date IS NOT NULL
),
windowed AS (
  SELECT company, function,
    COUNT(DISTINCT CASE WHEN posted_date >= CURRENT_DATE - 7  THEN posting_id END) AS reqs_last_7,
    COUNT(DISTINCT CASE WHEN posted_date >= CURRENT_DATE - 14
                         AND posted_date <  CURRENT_DATE - 7  THEN posting_id END) AS reqs_prior_7
  FROM obs
  GROUP BY company, function
)
SELECT company, function, reqs_last_7, reqs_prior_7,
       reqs_last_7 - reqs_prior_7 AS velocity_delta
FROM windowed
ORDER BY velocity_delta DESC;

-- バックフィル信号：同じ会社のために消えた後に再び現れる posting_id
SELECT company, posting_id, role_title,
       COUNT(*)                       AS times_observed,
       MIN(CAST(captured_at AS DATE)) AS first_seen,
       MAX(CAST(captured_at AS DATE)) AS last_seen
FROM talent_postings
GROUP BY company, posting_id, role_title
HAVING MAX(CAST(captured_at AS DATE)) - MIN(CAST(captured_at AS DATE)) > 21
   AND COUNT(DISTINCT CAST(captured_at AS DATE)) >= 2
ORDER BY company, times_observed DESC;

{
  "company": "target_company_a",
  "source": "company_careers",
  "region": "US",
  "posting_id": "a1b2c3d4e5f60718",
  "role_title": "シニアプラットフォームエンジニア",
  "function": "engineering",
  "seniority": "senior",
  "location": "Austin, TX",
  "posted_date": "<ソースが描画するISO日付、またはnull>",
  "salary_band": "$180k–$220k",
  "captured_at": "<読み取り時に書き込まれたISO-8601 UTCタイムスタンプ>"
}

- **合法な根拠と地域法。** 公共ページから取得した集約された企業の採用データは通常、最も敏感な個人データのカテゴリに該当しませんが、GDPR、CCPAおよび同等の法令は、個人を特定できるものに対して依然として適用されます。私たちはデータセットを企業の属性に設定しているため、合法な根拠が明確です。スコープを拡大する場合は、まず法律相談を行って合法な根拠を確認してください。
- **サイトの利用規約とロボット指令を尊重する。** 訪問者のためにサイトが公開しているHTML検索ページをレンダリングしてください。サイトがロボットファイルに予約している内部クエリエンドポイントをターゲットにせず、クローラの遅延ガイダンスを遵守してください。
- **公開された雇用主レビューのデータは集約のままに。** レビューサイトが感情的な文脈を提供する場合、配布レベルの評価を収集します — 個々のレビューアのIDや個人に結びついたレビューのテキストは決して収集しません。

エージェント駆動の同様のコレクションプリミティブのフレームは、[AIエージェントのユースケースガイド](https://www.scrapeless.com/ja/blog/ai-agent-use-cases-scrapeless-2026?utm_source=website&utm_medium=blog&utm_campaign=scrapingbrowser&utm_term=talent-market-intelligence-scrapeless) で、同じScraping Browserツールに基づいた求人エージェントを示しています。

---

## 結論: タレントマーケットインテリジェンスパイプラインを拡大する

パイプラインは次の6つのステップに簡素化されます: 企業とソースのバスケットを定義 → クラウドブラウザに接続し、セッションをウォームアップ → 各公共検索ページを米国のエグレスに固定してレンダリング → 企業の属性アポジションスキーマに抽出 → NDJSONにストリーミング → 速度、機能ミックスを導出し、ウェアハウスでバックフィルします。各ステップは読むために十分小さいです; 構成は、1日の単一のスケジュールで複数の採用面で十数の企業を扱います。

米国のエグレスを固定し、ターゲット検索ページの前にホームページをウォームアップし、レンダリング → 抽出のパターンに従い、欠落しているフィールドをnullableとして扱い、すべてのフィールドを企業属性として保持します — 会社と役職、個人ではなく。

---

## AI搭載のデータパイプラインを構築する準備はできましたか？

私たちのコミュニティに参加して無料プランを請求し、タレントおよび競争インテリジェンスパイプラインを構築する開発者とつながりましょう: [Discord](https://discord.gg/VU2vtbq7Q2) · [Telegram](https://t.me/scrapeless)。

[app.scrapeless.com](https://app.scrapeless.com/passport/login/?utm_source=website&utm_medium=blog&utm_campaign=scrapingbrowser&utm_term=talent-market-intelligence-scrapeless) にサインアップして無料のScraping Browserランタイムを取得し、上記のパターンをパイプラインが必要とする企業、採用面、および地域に適応させてください。価格詳細は [scrapeless.com/en/pricing](https://www.scrapeless.com/ja/pricing?utm_source=website&utm_medium=blog&utm_campaign=scrapingbrowser&utm_term=talent-market-intelligence-scrapeless); Scraping Browser製品ページは [scrapeless.com/en/product/scraping-browser](https://www.scrapeless.com/ja/product/scraping-browser?utm_source=website&utm_medium=blog&utm_campaign=scrapingbrowser&utm_term=talent-market-intelligence-scrapeless); 完全な接続およびプロキシの参照は [docs.scrapeless.com](https://docs.scrapeless.com?utm_source=website&utm_medium=blog&utm_campaign=scrapingbrowser&utm_term=talent-market-intelligence-scrapeless) です。

---

## FAQ

**Q: タレントマーケットインテリジェンスを収集することは合法ですか？ そして、個人データについては？**

合法性は収集する内容に完全に依存します。このパイプラインは、公共ページから企業属性および役職レベルのデータを収集します — 職務名、機能、位置、投稿数、公共の給与範囲 — そして意図的に個人データを収集しません: 名前、連絡先詳細、個人の雇用履歴はありません。公に見えるデータは一般的にアクセス可能ですが、GDPR、CCPAおよび同等の法律は、個人を特定できるものに依然として適用され、サイトの利用規約はすべてに適用されます。スキーマを企業の属性として維持することが合法な根拠を明確に保つことになります; スコープを拡大する前に法的助言を受けてください。

**Q: プロキシは必要ですか？ どの国を固定すべきですか？**

はい。アグリゲーターやキャリアページは地域およびIPの評判によって結果をローカライズするため、接続URLの`proxyCountry`で計測する地域にエグレス国を固定してください。米国のエグレスリクエストが地域制限されたページに送信されると、フォールバックまたはジオブロックが返される場合があります。Scrapelessの住宅用プロキシは195か国以上で典型的なバスケットをカバーしており、別のプロキシプロバイダーをスタックに持ち込むことなく済みます。

**Q: 検索ページが空またはアクセスチャレンジを表示します — クリーンなレンダリングを取得するにはどうすればよいですか？**

米国の住宅用エグレスを固定し、ターゲットロードの前にセッションをウォームアップします: 最初に同じクラウドブラウザセッション内でサイトのホームページに移動し、それを安定させてから、公共の検索URLに移動し、`networkidle`と投稿コンテナセレクタで待機します。ウォーミングは、検索ページが期待するクライアントサイドの状態を確立しますので、グリッドが完全に描画される代わりに半分の水分補給されたシェルを返すことはありません。

**Q: ソースがDOMを回転させた場合はどうなりますか？**
ステップ4で変更されるのは、ソースごとのエクストラクターのみです。標準的なポスティングスキーマ、倉庫テーブル、派生クエリ、およびアラートルールはすべて影響を受けません。ソースがリリースを出荷する際には、セレクターを再確認して厳密にしできるだけ`data-*`属性を使用することをお勧めします。エクストラクターは揮発性の層として扱い、スキーマは安定した層として扱ってください。

**Q: 個人を追跡せずに採用の速度とバックフィルをどのように導き出しますか？**

両方の指標は、ポスティング日付と安定した`posting_id`から派生し、人々からではありません。速度は、時間のウィンドウ内で会社ごとに機能ごとに開かれた異なるポスティングの数です。バックフィルプレッシャーは、同じ`posting_id`が会社全体で再現されることです。数学的な処理は、ファーモグラフィック観察に基づいて倉庫（ステップ7）で行われます — 個人は決して特定されず、追跡されません。

**Q: 複数の会社とソースを同時に実行できますか？**

はい。各(会社、ソース、地域)のエントリは、自身のセッションを温めて描画するため、オーケストレーターはスレッドプール全体にタスクを分配します。ファンアウトが礼儀正しい状態に保たれるよう、ワーカー数はホストごとに3以下にしてください。並行性はワーカーが追加されることでスケールしますが、セッションを共有することでスケールするのではありません。

**Q: パイプラインはどのくらいの頻度で実行すべきですか？**

日次は、採用速度の追跡における標準的なリズムです。なぜならポスティングは数日ごとに回転するからです。機能ミックスと拡張信号が遅い場合は週次で問題ありません。ステップ7の派生ウィンドウは日次キャプチャを想定しており、スケジュールは意思決定層が実際に消費する最速の信号に合わせて調整してください。