Files
endorphin-tabularium-docs/scripts/audit-ai-research.py

231 lines
17 KiB
Python
Raw Permalink Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
"""Summarize observed research coverage; does not claim an exhaustive AI census."""
import argparse
from collections import Counter
import json
from pathlib import Path
import sys
from urllib.parse import urlsplit
ROOT = Path(__file__).resolve().parents[1]
DATA = json.loads((ROOT / 'content/research/ai-companies.seed.json').read_text())
AUDIT = json.loads((ROOT / 'content/research/ai-companies.coverage.json').read_text())
SOURCES = json.loads((ROOT / 'content/research/ai-companies.sources.json').read_text())
TARGET = ROOT / '.artifacts/reports/02-coverage-and-gaps.md'
SOURCE_SLUG = 'ai-company-research/02-coverage-and-gaps'
BASE = '/docs/other-research/ai-company-research/overview'
def source_register():
sources = SOURCES['sources']
ids = {s['id'] for s in sources}
assert len(ids) == len(sources) and ids == {f'S{i:02}' for i in range(1, 17)}
assert SOURCES['scopeConfirmedAt'] <= DATA['asOf']
host_map = {}
for source in sources:
assert source['status'] in {'candidate-audit', 'targeted', 'pending', 'context', 'access-limited', 'per-company'}
assert source['boundary'] and source['nextCheck']
if source['url']:
assert urlsplit(source['url']).scheme in {'http', 'https'} and urlsplit(source['url']).netloc
for host in source['hosts']:
assert host not in host_map, f'Duplicate source host: {host}'
host_map[host] = source['id']
assigned = {key: [] for key in ids}
for evidence in DATA['evidence']:
owner = host_map.get(urlsplit(evidence['url']).netloc)
if owner is None:
owner = 'S12' if any(word in evidence['kind'] for word in ('官方', '公司自述', '公司披露')) else 'S16'
assigned[owner].append(evidence)
assert sum(len(v) for v in assigned.values()) == len(DATA['evidence'])
statuses = {'candidate-audit': '候选池逐条已读', 'targeted': '定向使用,未穷尽', 'pending': '候选台账待建',
'context': '背景研究', 'access-limited': '访问受限', 'per-company': '随公司逐条核对'}
lines = ['', '## 来源清单与验收边界', '', SOURCES['provenance'], '',
'每条证据在下表只计入一个渠道:先按所列域名归属,再将类型明确标为官方/公司自述/公司披露的剩余项计入S12,其余计入S16。公司可跨渠道出现,所以公司数不可跨行相加。零条证据表示尚未进入研究表,不表示该来源没有相关公司。背景报告不承担公司发现全集的角色。', '',
'| 来源 | 当前状态 | 证据 / 公司 / 观测数 | 已核对范围 | 剩余核查 |', '| --- | --- | --- | --- | --- |']
for source in sources:
evidence = assigned[source['id']]
evidence_ids = {e['id'] for e in evidence}
company_ids = {e['companyId'] for e in evidence if e['companyId']}
observations = sum(o['evidence'] in evidence_ids for o in DATA['observations'])
name = f"[{source['name']}]({source['url']})" if source['url'] else source['name']
lines.append(f"| {source['id']} {name} | {statuses[source['status']]} | {len(evidence)} / {len(company_ids)} / {observations} | {source['boundary']} | {source['nextCheck']} |")
lines += ['', 'Dealroom、Product Hunt、机器之心与36氪入口于2026-10-30重新访问。官网首页可读不证明已取得完整筛选结果;安全检测页面不作为文章内容。候选池之外的来源边界尚需建立可回查条目,完成前不宣布来源清单已全部覆盖。', '']
return lines
def candidate_register():
batches = DATA.get('candidateBatches', [])
companies = {c['id']: c for c in DATA['companies']}
categories = {c['id']: c['name'] for c in DATA['categories']}
known_sources = {s['id'] for s in SOURCES['sources']}
evidence = {e['id']: e for e in DATA['evidence']}
assert len({b['id'] for b in batches}) == len(batches)
all_ids = [r['id'] for b in batches for r in b['records']]
assert len(set(all_ids)) == len(all_ids)
lines = []
statuses = {'pending-year': '成立年待核', 'linked': '已关联公司',
'existing-outside-cohort': '已有关联,年份范围外',
'outside-cohort': '成立年范围外,不新增公司', 'promotion': '推广活动,非公司新增'}
for batch in batches:
assert batch['sourceId'] in known_sources
assert batch['accessedAt'] <= DATA['asOf']
assert any(e['url'] == batch['url'] for e in DATA['evidence'])
rows = batch['records']
counts = Counter(r['status'] for r in rows)
assert sum(counts.values()) == len(rows)
lines += ['', f"## {batch['sourceId']} 候选核对({batch['id']})", '',
f"[发现页面]({batch['url']}),读取于{batch['accessedAt']}。{batch['scope']}", '',
f"本批{len(rows)}条:{counts['linked']}条关联公司、{counts['existing-outside-cohort']}条关联范围外对照、{counts['outside-cohort']}条范围外不新增公司、{counts['pending-year']}条成立年待核、{counts['promotion']}条推广活动。完整来源覆盖标记:{'已完成' if batch['complete'] else '未完成'}。主营或产品任务的初分类不计入主公司表数量,只有已关联的公司记录进入对应公司明细。候选与缺项随研究版本保存在文档站同一SQLite的研究元数据中。", '',
'| 发现条目 | 主营或产品任务初分 | 核对状态 | 关联公司 | 核对证据 | 限制 |', '| --- | --- | --- | --- | --- | --- |']
for row in rows:
assert row['categoryId'] in categories
assert row['status'] in statuses
assert all(eid in evidence for eid in row.get('evidenceIds', []))
if row['status'] == 'outside-cohort':
assert row.get('evidenceIds') and not row.get('companyId')
name = f"[{row['name']}]({row['url']})" if row['url'] else row['name']
company = '—'
if row.get('companyId'):
assert row['companyId'] in companies
c = companies[row['companyId']]
company = f"[{c['name']}]({BASE}?company={c['id']}#research-company-detail)"
proof = '、'.join(f"[{eid}]({evidence[eid]['url']})" for eid in row.get('evidenceIds', [])) or '成立证据待补,产品页读取状态见候选记录'
category = f"[{categories[row['categoryId']]}]({BASE}?category={row['categoryId']})"
lines.append(f"| {name} | {category} | {statuses[row['status']]} | {company} | {proof} | {row['note']} |")
if batch['sourceId'] == 'S04':
lines += ['', '产品页可读只表示有可访问的资料;本批未将所有产品页完成公司、技术或财务深查。既有直接HTTP采集返回403,当前采用可读的公开页面响应,不声称取得底层全量列表;后续公开档案与未读页面继续保留缺项。', '']
else:
lines += ['', '发现条目和可读正文分开记录;登录外壳、读取失败或搜索摘录不充当已阅读全文。关联公司以列出的原始材料补证,仍保留技术、主体与收入未核事项。', '']
return lines
def directory_rechecks():
rows = DATA.get('directoryRechecks', [])
if not rows:
return []
original = {row['ycId']: row for row in AUDIT['records']}
companies = {c['id']: c for c in DATA['companies']}
evidence = {e['id']: e for e in DATA['evidence']}
assert len({row['ycId'] for row in rows}) == len(rows)
lines = ['', '## 原候选池的后续复核', '',
'原始1,428条读取快照与初筛计数保持不变;下表记录当时未入表、后续补足证据后进入公司表的项目。初筛未通过不是永久排除。后续入表不回写为当时已匹配,也不重复计入928条原始新增目录项。', '',
'| 原候选 | 原始状态 | 后续公司记录 | 复核证据 | 处理与限制 |', '| --- | --- | --- | --- | --- |']
for row in rows:
baseline = original[row['ycId']]
assert baseline['status'] == row['originalStatus']
assert row['reviewedAt'] <= DATA['asOf'] and row['reason']
assert row['originalStatus'] in {'ai-unclear', 'year-outside-or-missing', 'date-conflict'}
company = companies[row['companyId']]
assert row['evidenceIds'] and all(eid in evidence for eid in row['evidenceIds'])
proof = '、'.join(f"[{eid}]({evidence[eid]['url']})" for eid in row['evidenceIds'])
lines.append(f"| [{baseline['name']} · YC{row['ycId']}]({baseline['source']}) | {row['originalStatus']} | [{company['name']}]({BASE}?company={company['id']}#research-company-detail) | {proof} | {row['reviewedAt']}:{row['reason']} |")
lines += ['', f"已记录{len(rows)}条后续复核入表;其余未入表候选仍按原状态保留缺项,不能把这一批复核视为完成全部模糊项。", '']
return lines
def render():
cov = DATA['coverage']
companies = DATA['companies']
depth = sum(bool(c.get('profileOnly')) for c in companies)
origin = sum(bool(c.get('profileOnly')) or c.get('directoryOrigin') == 'yc-2025-2026' for c in companies)
promoted = origin - depth
reviewed = sum(bool(c.get('profileOnly')) and '人工复核' in c.get('classificationBasis', '') for c in companies)
years = Counter(c['year'] for c in companies)
income = Counter(c['revenueStatus'] for c in companies)
technical = Counter(c.get('technicalReview', {}).get('level', 'unreviewed') for c in companies)
assert cov['profilesRead'] + cov['failedProfiles'] == cov['candidateRecords']
assert cov['matchingProfiles'] == cov['existingProfiles'] + cov['newProfiles']
assert cov['matchingProfiles'] + cov['otherOrMissingYear'] + cov['unclearAI'] + cov['dateConflicts'] == cov['profilesRead']
assert origin == cov['newProfiles']
assert AUDIT['accessedAt'] == cov['accessedAt']
assert AUDIT['counts'] == cov
assert len(AUDIT['records']) == cov['profilesRead']
assert Counter(x['status'] for x in AUDIT['records']) == {
'imported': cov['newProfiles'], 'existing': cov['existingProfiles'],
'year-outside-or-missing': cov['otherOrMissingYear'],
'ai-unclear': cov['unclearAI'], 'date-conflict': cov['dateConflicts']}
assert len({x['ycId'] for x in AUDIT['records']}) == len(AUDIT['records'])
lines = ['# AI 公司资料覆盖与缺项', '',
f"截至 {DATA['asOf']},研究表含{len(companies):,}条公司或产品记录。YC候选池有逐条读取审计;其他来源为定向补充。清单尚不能作为全球AI公司普查,也不能将收入未知解释为零。", '',
'验收范围已于2026-10-30确认为:以确认的公开来源清单为边界,逐条核对并标明缺项。下方区分已逐条审计的候选池与定向补充来源;已使用某来源不代表该来源已被完整覆盖。', '',
f"[分类和数据工作台]({BASE})链接原始证据,[收入样本快照](/docs/other-research/ai-company-research/income-snapshot)展示当前可计算的MRR样本。", '',
'## 候选池与公司表的关系', '',
'| 环节 | 记录数 | 含义 |', '| --- | --- | --- |',
f"| YC批次候选 | {cov['candidateRecords']:,} | 从公开镜像发现,随后读取YC官方页面 |",
f"| 官方页面读取 | {cov['profilesRead']:,} | 此候选池读取失败{cov['failedProfiles']}条 |",
f"| 初筛匹配 | {cov['matchingProfiles']:,} | 成立年及AI相关描述通过初筛,尚非逐家商业验证 |",
f"| 与原表重合 | {cov['existingProfiles']} | URL匹配,未再次新增 |",
f"| 新增目录记录 | {cov['newProfiles']:,} | 原始入表来源,后续深化不改变发现计数 |",
f"| 年份不符或缺失 | {cov['otherOrMissingYear']} | 此轮不作为成立年匹配项 |",
f"| AI业务待确认 | {cov['unclearAI']} | 不强行把背景关键词当主营 |",
f"| 成立年与批次冲突 | {cov['dateConflicts']} | 另保留核查记录 |", '',
f"全表由{depth}条目录基础项和{len(companies)-depth}条个案或平台记录组成。候选池已匹配项之外另有{len(companies)-cov['matchingProfiles']}条记录,包含其他来源及发布/转型对照;其中已有{promoted}条原目录记录深化为个案,仍保留目录来源。两种统计边界不同。候选池数字只是该镜像快照的覆盖,Bud等后续定向补充不回填为原候选池已覆盖。", '',
'## 成立年与证据深度', '',
'| 维度 | 当前记录 | 解读 |', '| --- | --- | --- |',
f"| 记录为2025年成立 | {years[2025]} | 成立来源和精度见公司明细 |",
f"| 记录为2026年成立 | {years[2026]} | 不将产品发布日期代替成立日 |",
f"| 成立年未知或范围外 | {sum(v for k,v in years.items() if k not in (2025,2026))} | 发布/转型对照单列,不进入两年成立分组 |",
f"| 目录业务分类人工复核 | {reviewed} | 仅复核主营分类,不代表技术或财务审计 |",
f"| 其他目录基础项 | {depth-reviewed} | 规则初分或资料不足,继续待复核 |",
f"| 有团队数字 | {sum(c['team'] is not None for c in companies)} | 多为自报Team Size或Employees,时点与口径不统一 |",
f"| 专项技术证据记录 | {len(companies)-technical['unreviewed']} | 其中{technical['description']}条自述、{technical['public-artifact']}条公开材料、{technical['independent-test']}条第三方评测 |",
f"| 公司披露收入线索 | {income['self']} | 含历史、下界、币种不明及产品业务声明,不等于可比ARR |",
f"| 仅平台收入线索 | {income['platform']} | 同时有公司披露者计入上一行,不重复计数 |",
f"| 仅第三方估算 | {income['estimate']} | 不与直接披露混合 |",
f"| 暂无可用收入数字 | {income['unknown']} | 包含尚未逐家搜索的目录项,不代表没有收入 |", '',
'## 业务分类与数据入口', '',
'| 业务类别 | 记录数 |', '| --- | --- |',
]
counts = Counter(c['cat'] for c in companies)
for category in DATA['categories']:
lines.append(f"| [{category['name']}]({BASE}?category={category['id']}) | {counts[category['id']]} |")
lines += ['', '## 已使用来源的分布', '',
f"{len(DATA['evidence']):,}条证据记录对应{len({e['url'] for e in DATA['evidence']}):,}个不同URL。不同URL不自动构成独立验证;同一声明可能由公司、投资方和媒体重复转述。", '',
'| 来源 | 证据记录 | 不同URL | 用途及限制 |', '| --- | --- | --- | --- |',
]
families = [('YC官方目录', 'www.ycombinator.com', '公司自述的成立、团队与业务;英文及加速器样本偏多'),
('TrustMRR', 'trustmrr.com', '支付平台重建指标及方法;公开者存在选择偏差,非财务审计'),
('a16z Speedrun', 'speedrun.a16z.com', '投资机构中的公司介绍及收入声明;存在利益关系'),
('TechCrunch', 'techcrunch.com', '原始采访或报道;转述公司数字不等于查账'),
('Sacra', 'sacra.com', '第三方估算;与直接披露分开')]
included = set()
for name, host, note in families:
entries = [e for e in DATA['evidence'] if urlsplit(e['url']).netloc == host]
included.add(host)
lines.append(f"| {name} | {len(entries)} | {len({e['url'] for e in entries})} | {note} |")
others = [e for e in DATA['evidence'] if urlsplit(e['url']).netloc not in included]
lines.append(f"| 其他官网、评测及公开资料 | {len(others)} | {len({e['url'] for e in others})} | 按每条来源限制理解,不能统一当作独立证据 |")
lines += directory_rechecks()
lines += source_register()
lines += candidate_register()
lines += ['', '## 尚未满足的研究条件', '',
'行业ARR比较仍缺统一业务截止日、币种、经常性定义和业务边界。现有平台MRR可描述已采集产品,不足以代表全部新公司。技术评测只覆盖所列版本与任务,不宜跨领域汇总为统一强弱分数。', '',
'下一轮优先补收入线索的日期及合同定义、非YC和非英文来源、规则初分的人工复核,以及停止服务或被收购案例。这样能减少“只见赢家”和“只见融资公司”的偏差。公开来源池之外的未披露公司仍无法证明已被穷尽。', '',
]
return '\n'.join(lines)
def main():
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument('--check', action='store_true')
parser.add_argument('--document-export', type=Path)
args = parser.parse_args()
content = render()
if args.check:
if args.document_export is not None:
bundle = json.loads(args.document_export.read_text())
stored = next((d['content'] for d in bundle['documents'] if d['sourceSlug'] == SOURCE_SLUG), None)
if stored != content:
sys.exit('Research coverage report differs; review and regenerate')
print('Research coverage accounting verified; supplied document export compared when requested')
else:
TARGET.parent.mkdir(parents=True, exist_ok=True)
TARGET.write_text(content)
print(TARGET.relative_to(ROOT))
if __name__ == '__main__':
main()