日本企業のためのGTMエンジニアリング実践｜EDINET・gBizINFO・顧問ネットワークで営業優位を作る

import os
import requests

EDINET_BASE_URL = "https://api.edinet-fsa.go.jp/api/v2"
EDINET_API_KEY = os.environ["EDINET_API_KEY"]

def get_annual_reports(date_str: str) -> list[dict]:
    """指定日の有価証券報告書候補を取得する。"""
    resp = requests.get(
        f"{EDINET_BASE_URL}/documents.json",
        params={
            "date": date_str,
            "type": 2,
            "Subscription-Key": EDINET_API_KEY,
        },
        timeout=60,
    )
    resp.raise_for_status()
    payload = resp.json()

    if payload.get("metadata", {}).get("status") != "200":
        raise RuntimeError(f"EDINET API error: {payload}")

    return [
        doc
        for doc in payload.get("results", [])
        if doc.get("docTypeCode") == "120"
        and doc.get("xbrlFlag") == "1"
    ]

import io
import re
import zipfile
from lxml import etree

def download_xbrl(doc_id: str) -> bytes:
    """EDINET文書IDからXBRL ZIPを取得する。"""
    resp = requests.get(
        f"{EDINET_BASE_URL}/documents/{doc_id}",
        params={
            "type": 1,
            "Subscription-Key": EDINET_API_KEY,
        },
        timeout=120,
    )
    resp.raise_for_status()
    return resp.content

def extract_business_risks(xbrl_zip: bytes) -> str | None:
    """XBRL ZIPから事業等のリスク本文を抽出する。"""
    with zipfile.ZipFile(io.BytesIO(xbrl_zip)) as archive:
        xbrl_files = [
            name
            for name in archive.namelist()
            if name.startswith("XBRL/PublicDoc/")
            and name.endswith(".xbrl")
        ]
        if not xbrl_files:
            return None

        with archive.open(xbrl_files[0]) as file:
            tree = etree.parse(file)
            root = tree.getroot()

    jpcrp_ns = None
    for prefix, uri in root.nsmap.items():
        if "jpcrp_cor" in (prefix or "") or "jpcrp_cor" in uri:
            jpcrp_ns = uri
            break

    if jpcrp_ns is None:
        return None

    elements = root.findall(f".//{{{jpcrp_ns}}}BusinessRisksTextBlock")
    if not elements:
        return None

    raw_text = etree.tostring(elements[0], encoding="unicode", method="text")
    return re.sub(r"\s+", " ", raw_text).strip()

RISK_KEYWORDS = {
    "security_saas": [
        "サイバーセキュリティ",
        "情報漏えい",
        "不正アクセス",
        "ランサムウェア",
        "個人情報",
        "サイバー攻撃",
    ],
    "hr_saas": [
        "人材確保",
        "人的資本",
        "離職",
        "人材育成",
        "働き方",
        "人手不足",
    ],
    "manufacturing_dx": [
        "サプライチェーン",
        "デジタルトランスフォーメーション",
        "生産性",
        "設備老朽化",
        "カーボンニュートラル",
    ],
}

def find_new_keywords(
    current_text: str,
    previous_text: str | None,
    category: str,
) -> list[str]:
    """前年に無く、今年のリスク本文に出たキーワードを返す。"""
    keywords = RISK_KEYWORDS.get(category, [])
    current_hits = [kw for kw in keywords if kw in current_text]

    if not previous_text:
        return current_hits

    previous_hits = [kw for kw in keywords if kw in previous_text]
    return [kw for kw in current_hits if kw not in previous_hits]

def score_risk_change(
    current_text: str,
    previous_text: str | None,
    category: str,
) -> dict:
    """営業仮説用の簡易スコアを作る。"""
    new_keywords = find_new_keywords(current_text, previous_text, category)
    volume_change = 0.0

    if previous_text:
        volume_change = (
            len(current_text) - len(previous_text)
        ) / max(len(previous_text), 1)

    return {
        "new_keywords": new_keywords,
        "keyword_count": len(new_keywords),
        "volume_change_rate": round(volume_change, 3),
        "needs_human_review": len(new_keywords) > 0 or volume_change > 0.2,
    }

import os
import requests

GBIZINFO_BASE_URL = "https://info.gbiz.go.jp/hojin/v1"
GBIZINFO_API_TOKEN = os.environ["GBIZINFO_API_TOKEN"]

def get_company_profile(corporate_number: str) -> dict | None:
    """企業番号からgBizINFOの基本情報を取得する。"""
    resp = requests.get(
        f"{GBIZINFO_BASE_URL}/hojin/{corporate_number}",
        headers={"X-hojinInfo-api-token": GBIZINFO_API_TOKEN},
        timeout=30,
    )
    if resp.status_code != 200:
        return None

    payload = resp.json()
    records = payload.get("hojin-infos", [])
    return records[0] if records else None

def get_gbiz_records(corporate_number: str, resource: str) -> list[dict]:
    """gBizINFOのサブリソースを取得する共通関数。"""
    resp = requests.get(
        f"{GBIZINFO_BASE_URL}/hojin/{corporate_number}/{resource}",
        headers={"X-hojinInfo-api-token": GBIZINFO_API_TOKEN},
        timeout=30,
    )
    if resp.status_code != 200:
        return []

    payload = resp.json()
    key = f"{resource}-infos"
    return payload.get(key, [])

def build_activity_signals(corporate_number: str) -> dict:
    """営業仮説に使う企業活動シグナルを作る。"""
    subsidies = get_gbiz_records(corporate_number, "subsidy")
    certifications = get_gbiz_records(corporate_number, "certification")
    procurements = get_gbiz_records(corporate_number, "procurement")

    return {
        "has_subsidy_records": len(subsidies) > 0,
        "has_certification_records": len(certifications) > 0,
        "has_procurement_records": len(procurements) > 0,
        "review_questions": [
            "直近の投資テーマと自社商材の接点はあるか",
            "認定や調達の履歴から担当部門を推測できるか",
            "初回接点で確認すべき公開事実は何か",
        ],
    }

def calculate_gtm_signal_score(
    risk_result: dict,
    activity_signals: dict,
    has_customer_similarity: bool,
    has_relationship_path: bool,
) -> dict:
    """営業確認の優先順位を作るための簡易スコア。"""
    score = 0
    reasons = []

    if risk_result.get("new_keywords"):
        score += 3
        reasons.append("事業リスクに新しい関連語がある")

    if risk_result.get("volume_change_rate", 0) > 0.2:
        score += 1
        reasons.append("事業リスクの記述量が増えている")

    if activity_signals.get("has_subsidy_records"):
        score += 1
        reasons.append("公的支援の履歴がある")

    if activity_signals.get("has_certification_records"):
        score += 1
        reasons.append("認定の履歴がある")

    if has_customer_similarity:
        score += 2
        reasons.append("既存顧客と似た属性がある")

    if has_relationship_path:
        score += 2
        reasons.append("紹介・顧問・パートナー接点がある")

    return {
        "score": score,
        "priority": "A" if score >= 6 else "B" if score >= 3 else "C",
        "reasons": reasons,
    }

対象企業: 株式会社サンプル
見つけた公開シグナル: 有価証券報告書の事業リスクでサイバー対策の記述が増加
確認したい仮説: 監査・委託先管理を含むセキュリティ体制の見直しが進んでいるか
紹介してほしい相手: 情報システム、監査、リスク管理の責任者
お願いしたいこと: 初回15分の課題確認