npm-downloads-to-leads

npm 下载量转化为潜在客户

输入一份 npm 包列表。获取 12 周的下载数据。计算突破速度。补充维护者资料。为每个突破包输出一份按排名排列的潜在客户简报，包含联系信号和外联消息。

---

**关键规则：**输出中的每个包下载量数据都必须来自 npm API 响应。每个维护者的 GitHub 用户名或 Twitter 用户名都必须来自 GitHub API 响应——不能根据 npm 用户名猜测。如果 GitHub API 未返回 twitter_username 字段，请写“GitHub 上未找到”，不要编造。

---

常见错误

Agent 会想要……	为什么不对
为列表中的每个包获取 GitHub 资料	没有令牌时，速率限制为每小时 60 次请求。为表现稳定或正在下滑的包补充资料，会在找到突破包之前浪费请求额度。只为突破包和观察中的包获取资料。
按每周原始下载量给包排名	原始下载量会偏向 React 和 lodash，但它们并不是潜在客户。一个包的下载量从每周 1K 增至 8K，比 React 每周 50M 的下载量更值得采取行动。速度评分才是信号。
跳过对作用域包进行 URL 编码	不编码的 @org/pkg 会导致 npm API 返回 404。对每个作用域包名称，将 @ 编码为 %40，将 / 编码为 %2F。
GitHub 触及速率限制时就停止技能	应优雅降级。展示基于 npm 数据的速度排行榜，跳过剩余的 GitHub 补充步骤，并向 data_quality_flags 添加标记。不要中止。
编写外联消息时不点明具体包名	泛泛的“我看到了你的项目”消息通常得不到回复。每条外联消息都必须写出包名、增长数据，以及与用户提供的背景信息之间的具体关联。
将每周下载量低于 500 的包列为潜在客户	每周低于 500 次下载属于噪声。维护者尚未拥有有意义的受众。标记为“时机尚早”，但不要将其列为潜在客户。

---

第 1 步：设置检查

echo "GITHUB_TOKEN: ${GITHUB_TOKEN:-not set, unauthenticated rate limit applies (60 req/hr -- enough for ~10 packages)}"

**如果未设置 GITHUB_TOKEN：**继续。告知用户：“未设置 GITHUB_TOKEN。GitHub 补充资料最多只能处理约 10 个包，之后会触及速率限制。请在 github.com/settings/tokens 添加令牌（无需任何权限范围）。”

不需要任何密钥。npm API 和 npm registry 完全公开，无需身份验证。

---

第 2 步：收集输入

从对话中收集：

• 一个或多个 npm 包名称（无作用域，例如 esbuild；或有作用域，例如 @hono/hono）
• 可选：简短的产品背景信息（用于个性化外联消息）

如果用户提供 npmjs.com URL，只提取包名。保留完整的作用域名称，包括 @ 和组织前缀——编码将在第 3 步处理。

**如果未提供包名：**询问：“你想分析哪些 npm 包？可以提供自己的包、竞争对手的包，或两者混合。例如：esbuild、@hono/hono、zod、valibot”

python3 << 'PYEOF'
import json, sys

packages_raw = "PACKAGES_HERE"  # comma or newline separated
product_context = "CONTEXT_HERE"  # optional, can be empty string

packages = [p.strip() for p in packages_raw.replace("\n", ",").split(",") if p.strip()]
if not packages:
    print("ERROR: No packages provided.")
    sys.exit(1)

print(f"Packages to analyze: {len(packages)}")
for p in packages:
    print(f"  {p}")

with open("/tmp/npl-input.json", "w") as f:
    json.dump({"packages": packages, "product_context": product_context}, f)
PYEOF

---

第 3 步：获取 12 周下载数据

如果有独立脚本，请使用它——它会通过一次调用处理第 3、4 和 5 步，因此无需运行下面的内嵌代码块。

# Check if the script exists
ls scripts/fetch.py 2>/dev/null && echo "script available" || echo "script not found"

**如果有该脚本，**直接运行并跳到第 6 步：

python3 scripts/fetch.py PACKAGES_HERE --context "CONTEXT_HERE" --output /tmp/npl-script-out.json

然后将输出载入第 6 步所需的增强格式：

python3 << 'PYEOF'
import json

out = json.load(open("/tmp/npl-script-out.json"))
# Script output has results array -- split into scored and enriched for Steps 6-8
enriched = [r for r in out["results"] if "profile" in r]
scored = out["results"]
json.dump(scored, open("/tmp/npl-scored.json", "w"), indent=2)
json.dump(enriched, open("/tmp/npl-enriched.json", "w"), indent=2)
json.dump({"packages": [r["package"] for r in scored], "product_context": out.get("product_context", "")},
          open("/tmp/npl-input.json", "w"), indent=2)
print(f"Loaded {len(scored)} packages | {out['breakout_count']} breakout | {out['watching_count']} watching")
PYEOF

**如果没有该脚本，**运行下面的内嵌代码。

从 npm Downloads API 获取每个包的每日下载数据。汇总为每周数据。

python3 << 'PYEOF'
import json, urllib.request, sys, time
from datetime import datetime, timedelta, timezone
from collections import defaultdict
import urllib.parse

data = json.load(open("/tmp/npl-input.json"))
packages = data["packages"]

end_date = datetime.now(tz=timezone.utc)
start_date = end_date - timedelta(weeks=13)  # extra week buffer for partial weeks
start_str = start_date.strftime("%Y-%m-%d")
end_str = end_date.strftime("%Y-%m-%d")

results = []
failed = []

for pkg in packages:
    # URL-encode scoped packages: @ -> %40, / -> %2F
    encoded = pkg.replace("@", "%40").replace("/", "%2F")
    url = f"https://api.npmjs.org/downloads/range/{start_str}:{end_str}/{encoded}"

    try:
        req = urllib.request.Request(url, headers={"User-Agent": "npm-downloads-to-leads/1.0"})
        with urllib.request.urlopen(req, timeout=20) as resp:
            raw = json.loads(resp.read())

        # Aggregate daily to weekly by ISO week
        weekly = defaultdict(int)
        for entry in raw.get("downloads", []):
            day = datetime.strptime(entry["day"], "%Y-%m-%d")
            week_key = day.isocalendar()[:2]  # (year, week_num)
            weekly[week_key] += entry["downloads"]

        weeks = [v for k, v in sorted(weekly.items())]
        # Take last 12 complete weekly buckets
        weeks = weeks[-12:]

        results.append({
            "package": pkg,
            "weeks": weeks,
            "total_weeks": len(weeks),
            "current_weekly": weeks[-1] if weeks else 0,
            "status": "ok"
        })
        print(f"  {pkg}: {len(weeks)} weeks, latest week {weeks[-1]:,} downloads")

    except urllib.error.HTTPError as e:
        if e.code == 404:
            failed.append(pkg)
            results.append({"package": pkg, "weeks": [], "total_weeks": 0, "current_weekly": 0, "status": "not_found"})
            print(f"  {pkg}: NOT FOUND (404) -- will be skipped")
        else:
            failed.append(pkg)
            results.append({"package": pkg, "weeks": [], "total_weeks": 0, "current_weekly": 0, "status": f"error_{e.code}"})
            print(f"  {pkg}: HTTP {e.code} error")
    except Exception as e:
        failed.append(pkg)
        results.append({"package": pkg, "weeks": [], "total_weeks": 0, "current_weekly": 0, "status": f"error"})
        print(f"  {pkg}: fetch failed ({e})")

    time.sleep(0.2)  # gentle rate limiting

json.dump(results, open("/tmp/npl-download-data.json", "w"), indent=2)
print(f"\nFetch complete. OK: {len(results) - len(failed)} | Failed/Not found: {len(failed)}")
if failed:
    print(f"Skipped: {', '.join(failed)}")
PYEOF

**如果所有包都返回 404 或错误：**停止。告知用户：“无法获取下载数据。请检查包名是否正确，以及包是否存在于 npmjs.com。作用域包必须包含完整名称：@org/package。”

---

第 4 步：计算速度评分

无需 API 调用。纯 Python。计算速度评分、增长比率，并为每个包分类。

python3 << 'PYEOF'
import json

raw_results = json.load(open("/tmp/npl-download-data.json"))
scored = []

for item in raw_results:
    pkg = item["package"]
    weeks = item["weeks"]
    status = item["status"]

    if status != "ok" or len(weeks) < 4:
        scored.append({**item, "velocity_score": 0, "growth_pct": 0, "tier": "insufficient_data",
                       "recent_4_avg": 0, "prior_4_avg": 0})
        continue

    recent_4 = sum(weeks[-4:]) / 4
    prior_4 = sum(weeks[-8:-4]) / max(len(weeks) - 4, 1) if len(weeks) >= 8 else sum(weeks[:4]) / max(len(weeks[:4]), 1)
    recent_2 = sum(weeks[-2:]) / 2
    mid_2 = sum(weeks[-4:-2]) / 2 if len(weeks) >= 4 else recent_2

    growth_ratio = recent_4 / max(prior_4, 1)
    acceleration = recent_2 / max(mid_2, 1)
    growth_pct = round((growth_ratio - 1) * 100, 1)

    # Sweet spot multiplier: 500-500K weekly downloads
    if recent_4 < 500:
        noise_factor = max(recent_4 / 500, 0.1)
    elif recent_4 > 500_000:
        noise_factor = max(500_000 / recent_4, 0.1)
    else:
        noise_factor = 1.0

    velocity_score = round(growth_ratio * acceleration * noise_factor * 100, 1)

    # Classify
    if velocity_score > 80 and 500 < recent_4 < 500_000 and growth_ratio >= 1.5:
        tier = "breakout"
    elif velocity_score > 40 and recent_4 >= 500 and growth_ratio >= 1.2:
        tier = "watching"
    elif recent_4 < 500:
        tier = "too_early"
    elif recent_4 >= 500_000:
        tier = "established"
    else:
        tier = "steady"

    scored.append({
        **item,
        "velocity_score": velocity_score,
        "growth_pct": growth_pct,
        "recent_4_avg": round(recent_4),
        "prior_4_avg": round(prior_4),
        "tier": tier
    })

# Sort by velocity_score descending
scored.sort(key=lambda x: x["velocity_score"], reverse=True)

json.dump(scored, open("/tmp/npl-scored.json", "w"), indent=2)

breakout = [p for p in scored if p["tier"] == "breakout"]
watching = [p for p in scored if p["tier"] == "watching"]
too_early = [p for p in scored if p["tier"] == "too_early"]

print(f"Velocity scoring complete:")
print(f"  BREAKOUT: {len(breakout)}")
print(f"  WATCHING: {len(watching)}")
print(f"  STEADY/ESTABLISHED: {len([p for p in scored if p['tier'] in ('steady','established')])}")
print(f"  TOO EARLY (<500/week): {len(too_early)}")
print()
for p in scored[:10]:
    print(f"  {p['tier'].upper():12} {p['package']:30} score={p['velocity_score']:6.1f}  "
          f"{p['recent_4_avg']:>8,}/wk  growth={p['growth_pct']:+.0f}%")

# Stop if nothing worth analyzing
if not breakout and not watching:
    all_too_early = all(p["tier"] in ("too_early", "insufficient_data") for p in scored)
    if all_too_early:
        print("\nERROR: All packages are below the 500 weekly downloads threshold for reliable velocity analysis.")
        print("Try packages with more community adoption.")
        import sys; sys.exit(1)
PYEOF

**如果所有包的每周下载量都低于 500：**显示上面的消息后停止。

---

第 5 步：获取维护者资料

只针对突破包和观察中的包。先获取 npm registry 元数据，再获取 GitHub 用户资料。

python3 << 'PYEOF'
import json, urllib.request, re, os, time

scored = json.load(open("/tmp/npl-scored.json"))
token = os.environ.get("GITHUB_TOKEN", "")

gh_headers = {"Accept": "application/vnd.github+json", "User-Agent": "npm-downloads-to-leads/1.0"}
if token:
    gh_headers["Authorization"] = f"Bearer {token}"

target_packages = [p for p in scored if p["tier"] in ("breakout", "watching")]
print(f"Fetching profiles for {len(target_packages)} packages (breakout + watching)...")

gh_rate_remaining = 999
enriched = []

for item in target_packages:
    pkg = item["package"]
    profile = {"package": pkg, "npm_maintainers": [], "description": "", "keywords": [],
               "github_owner": None, "github_repo": None, "github_users": [], "npm_homepage": ""}

    # --- npm registry ---
    encoded = pkg.replace("@", "%40").replace("/", "%2F")
    reg_url = f"https://registry.npmjs.org/{encoded}"
    try:
        req = urllib.request.Request(reg_url, headers={"User-Agent": "npm-downloads-to-leads/1.0"})
        with urllib.request.urlopen(req, timeout=20) as resp:
            reg = json.loads(resp.read())

        profile["description"] = reg.get("description", "")
        profile["keywords"] = (reg.get("keywords") or [])[:6]
        profile["npm_homepage"] = reg.get("homepage", "")
        profile["npm_maintainers"] = [m.get("name", "") for m in reg.get("maintainers", []) if m.get("name")]

        # Extract GitHub owner from repository URL
        repo_field = reg.get("repository") or {}
        if isinstance(repo_field, dict):
            repo_url = repo_field.get("url", "")
        else:
            repo_url = str(repo_field)
        gh_match = re.search(r"github\.com[/:]([^/]+)/([^/.]+)", repo_url)
        if gh_match:
            profile["github_owner"] = gh_match.group(1)
            profile["github_repo"] = gh_match.group(2).rstrip(".git")

        print(f"  {pkg}: registry OK | maintainers={profile['npm_maintainers'][:3]} | "
              f"github_owner={profile['github_owner']}")
    except Exception as e:
        print(f"  {pkg}: registry fetch failed ({e})")

    time.sleep(0.1)

    # --- GitHub user profiles ---
    candidates = []
    if profile["github_owner"]:
        candidates.append(profile["github_owner"])
    # Also try npm maintainer usernames (often match GitHub)
    for m in profile["npm_maintainers"][:2]:
        if m and m not in candidates:
            candidates.append(m)

    for username in candidates[:3]:
        if gh_rate_remaining <= 5:
            print(f"  GitHub rate limit low ({gh_rate_remaining} remaining) -- skipping {username}")
            break

        gh_url = f"https://api.github.com/users/{username}"
        req = urllib.request.Request(gh_url, headers=gh_headers)
        try:
            with urllib.request.urlopen(req, timeout=15) as resp:
                gh_rate_remaining = int(resp.headers.get("X-RateLimit-Remaining", 999))
                gh_data = json.loads(resp.read())

            profile["github_users"].append({
                "username": username,
                "name": gh_data.get("name") or username,
                "twitter_username": gh_data.get("twitter_username") or "not found on GitHub",
                "bio": gh_data.get("bio") or "",
                "blog": gh_data.get("blog") or "",
                "company": gh_data.get("company") or "",
                "followers": gh_data.get("followers", 0),
                "public_repos": gh_data.get("public_repos", 0),
                "github_url": gh_data.get("html_url", f"https://github.com/{username}")
            })
            print(f"    GitHub @{username}: {gh_data.get('followers', 0)} followers | "
                  f"twitter={gh_data.get('twitter_username') or 'none'} | rate_remaining={gh_rate_remaining}")
        except urllib.error.HTTPError as e:
            if e.code == 404:
                print(f"    GitHub @{username}: not found")
            else:
                print(f"    GitHub @{username}: HTTP {e.code}")
        except Exception as e:
            print(f"    GitHub @{username}: failed ({e})")

        time.sleep(0.2)

    enriched.append({**item, "profile": profile})

json.dump(enriched, open("/tmp/npl-enriched.json", "w"), indent=2)
json.dump(scored, open("/tmp/npl-scored.json", "w"), indent=2)
print(f"\nEnrichment complete. Profiles fetched: {len(enriched)}")
print(f"GitHub rate limit remaining: {gh_rate_remaining}")
PYEOF

步骤 6：生成潜在客户简报

打印扩充后的突增和关注软件包，然后生成潜在客户简报和外联消息。

python3 << 'PYEOF'
import json

enriched = json.load(open("/tmp/npl-enriched.json"))
input_data = json.load(open("/tmp/npl-input.json"))
product_context = input_data.get("product_context", "")

breakout = [p for p in enriched if p["tier"] == "breakout"]
watching = [p for p in enriched if p["tier"] == "watching"]

print("=== DATA FOR LEAD BRIEF GENERATION ===")
print(f"Product context: {product_context or '(none provided)'}")
print()

for item in breakout + watching:
    pkg = item["package"]
    prof = item.get("profile", {})
    gh_users = prof.get("github_users", [])
    primary_gh = gh_users[0] if gh_users else {}

    print(f"PACKAGE: {pkg} ({item['tier'].upper()})")
    print(f"  Velocity score: {item['velocity_score']} | Growth: {item['growth_pct']:+.0f}%")
    print(f"  Recent 4-week avg: {item['recent_4_avg']:,}/week | Prior 4-week avg: {item['prior_4_avg']:,}/week")
    print(f"  Weekly trend (last 8): {item['weeks'][-8:]}")
    print(f"  Description: {prof.get('description', 'none')}")
    print(f"  Keywords: {', '.join(prof.get('keywords', []))}")
    print(f"  npm maintainers: {', '.join(prof.get('npm_maintainers', []))}")
    if primary_gh:
        print(f"  GitHub: @{primary_gh.get('username')} | {primary_gh.get('followers')} followers | "
              f"{primary_gh.get('public_repos')} repos")
        print(f"  Twitter: {primary_gh.get('twitter_username')}")
        print(f"  Bio: {primary_gh.get('bio')}")
        print(f"  Company: {primary_gh.get('company')}")
    else:
        print(f"  GitHub: no profile found")
    print()
PYEOF

使用上方打印的软件包数据，为每个 BREAKOUT 和 WATCHING 软件包生成一份潜在客户简报。

规则：

• 简报中的所有增长数字都必须来自打印的数据，不要四舍五入或修改
• 所有 GitHub 账号和 Twitter 用户名都必须来自打印的数据；如果字段中这样写，就写“未在 GitHub 上找到”
• “现在联系的理由”必须引用数据中具体的增长拐点（周数、数字）
• “建议的首条消息”必须提及软件包及其增长情况；如果提供了 product_context，则必须将其与该上下文具体关联
• 不得使用长破折号。不得使用禁用词：强大、稳健、无缝、创新、颠覆性、简化流程、利用、转型

请按以下精确结构将潜在客户简报写入 /tmp/npl-briefs.json：

{
  "lead_briefs": [
    {
      "package": "pkg-name",
      "tier": "breakout",
      "growth_summary": "1-sentence summary of the growth numbers",
      "maintainer_handle": "@github_handle or npm username if no GitHub found",
      "twitter": "@handle or not found on GitHub",
      "github_followers": 0,
      "why_now": "2-3 sentences specific to this package's inflection point",
      "suggested_message": "2-4 sentences. Names the package, the growth, and connects to product_context if provided."
    }
  ]
}

写入文件后，请通过以下内容确认：

python3 -c "
import json
d = json.load(open('/tmp/npl-briefs.json'))
print(f'Lead briefs generated: {len(d.get(\"lead_briefs\", []))}')
for b in d['lead_briefs']:
    print(f'  {b[\"package\"]} ({b[\"tier\"]}): maintainer={b[\"maintainer_handle\"]}')
"

---

步骤 7：自我检查

python3 << 'PYEOF'
import json

scored = json.load(open("/tmp/npl-scored.json"))
enriched = json.load(open("/tmp/npl-enriched.json"))
briefs = json.load(open("/tmp/npl-briefs.json"))

failures = []

# Verify: every brief has a real package name from the scored list
real_packages = {p["package"] for p in scored}
for brief in briefs.get("lead_briefs", []):
    if brief.get("package") not in real_packages:
        failures.append(f"Brief for unknown package '{brief.get('package')}' -- removed")

briefs["lead_briefs"] = [b for b in briefs.get("lead_briefs", []) if b.get("package") in real_packages]

# Verify: velocity leaderboard is sorted correctly (checked on scored, not briefs)
sorted_scores = sorted([(p["package"], p["velocity_score"]) for p in scored], key=lambda x: -x[1])
if scored[0]["velocity_score"] < scored[-1]["velocity_score"]:
    failures.append("Scored list not sorted by velocity_score -- re-sorted")
    scored.sort(key=lambda x: x["velocity_score"], reverse=True)

# Verify: no GitHub/Twitter handles in briefs that weren't in GitHub API responses
enriched_gh = {}
for item in enriched:
    for gh_user in item.get("profile", {}).get("github_users", []):
        enriched_gh[gh_user["username"]] = gh_user.get("twitter_username", "not found on GitHub")

for brief in briefs.get("lead_briefs", []):
    twitter = brief.get("twitter", "")
    if twitter and twitter not in ("not found on GitHub", "") and not twitter.startswith("not found"):
        # Verify it came from the API
        found = any(twitter.lstrip("@") == v.lstrip("@") for v in enriched_gh.values() if v != "not found on GitHub")
        if not found:
            failures.append(f"Warning: Twitter handle '{twitter}' for {brief['package']} not verified in GitHub API data")

# Check required fields
for brief in briefs.get("lead_briefs", []):
    for field in ["package", "tier", "growth_summary", "maintainer_handle", "twitter", "why_now", "suggested_message"]:
        if not brief.get(field):
            failures.append(f"Missing field '{field}' in brief for {brief.get('package', '?')}")

# Check for em dashes
briefs_str = json.dumps(briefs)
if "\u2014" in briefs_str:
    briefs_str = briefs_str.replace("\u2014", " - ")
    briefs = json.loads(briefs_str)
    failures.append("Fixed: em dash characters removed from briefs")

# Check for forbidden words
forbidden = ["powerful", "robust", "seamless", "innovative", "game-changing", "streamline", "leverage", "transform"]
full_text = json.dumps(briefs).lower()
for word in forbidden:
    if word in full_text:
        failures.append(f"Warning: forbidden word '{word}' found in briefs -- review before presenting")

output = {
    "scored": scored,
    "enriched": enriched,
    "briefs": briefs,
    "data_quality_flags": failures
}

json.dump(output, open("/tmp/npl-output.json", "w"), indent=2)
print(f"QA complete. Issues: {len(failures)}")
for f in failures:
    print(f"  - {f}")
if not failures:
    print("All QA checks passed.")
PYEOF

---

步骤 8：保存并呈现输出

python3 << 'PYEOF'
import json, os
from datetime import datetime, timezone

output = json.load(open("/tmp/npl-output.json"))
scored = output["scored"]
enriched_map = {e["package"]: e for e in output["enriched"]}
briefs_map = {b["package"]: b for b in output["briefs"].get("lead_briefs", [])}
flags = output["data_quality_flags"]
date_str = datetime.now(tz=timezone.utc).strftime("%Y-%m-%d")

breakout = [p for p in scored if p["tier"] == "breakout"]
watching = [p for p in scored if p["tier"] == "watching"]
too_early = [p for p in scored if p["tier"] == "too_early"]
established = [p for p in scored if p["tier"] == "established"]

lines = [
    f"## npm Breakout Report",
    f"Packages analyzed: {len(scored)} | Breakout: {len(breakout)} | Watching: {len(watching)} | Date: {date_str}",
    "",
    "---",
    "",
    "### Velocity Leaderboard",
    "",
    "| Rank | Package | Weekly Downloads | 8-Week Growth | Velocity Score | Status |",
    "|---|---|---|---|---|---|",
]

for i, pkg in enumerate(scored[:15], 1):
    status_label = {"breakout": "BREAKOUT", "watching": "WATCHING", "steady": "steady",
                    "established": "established", "too_early": "too early", "insufficient_data": "no data"}.get(pkg["tier"], pkg["tier"])
    growth_str = f"{pkg['growth_pct']:+.0f}%" if pkg.get("growth_pct") else "n/a"
    lines.append(
        f"| {i} | {pkg['package']} | {pkg['recent_4_avg']:,} | {growth_str} | "
        f"{pkg['velocity_score']} | {status_label} |"
    )

lines += ["", "---", ""]

if breakout or watching:
    lines += ["### Lead Briefs", ""]

    for item in breakout + watching:
        pkg = item["package"]
        brief = briefs_map.get(pkg, {})
        profile = enriched_map.get(pkg, {}).get("profile", {})
        gh_users = profile.get("github_users", [])
        primary_gh = gh_users[0] if gh_users else {}

        lines.append(f"#### {pkg} ({item['tier'].upper()})")
        lines.append(f"Weekly downloads: {item['recent_4_avg']:,}/week (was {item['prior_4_avg']:,} -- {item['growth_pct']:+.0f}% growth over 8 weeks)")
        if profile.get("description"):
            lines.append(f"What it does: {profile['description']}")
        if profile.get("keywords"):
            lines.append(f"Keywords: {', '.join(profile['keywords'])}")
        lines.append("")

        if primary_gh:
            lines.append(f"**Maintainer: @{primary_gh.get('username')}**")
            lines.append(f"- GitHub: {primary_gh.get('followers', 0):,} followers | {primary_gh.get('public_repos', 0)} public repos")
            lines.append(f"- Twitter: {primary_gh.get('twitter_username', 'not found on GitHub')}")
            if primary_gh.get("bio"):
                lines.append(f"- Bio: \"{primary_gh['bio']}\"")
            if primary_gh.get("company"):
                lines.append(f"- Company: {primary_gh['company']}")
            if primary_gh.get("blog"):
                lines.append(f"- Website: {primary_gh['blog']}")
        elif profile.get("npm_maintainers"):
            lines.append(f"**Maintainer (npm only):** {', '.join(profile['npm_maintainers'][:3])}")
            lines.append("- GitHub profile: not found")

        lines.append("")
        if brief.get("why_now"):
            lines.append(f"**Why reach out now:** {brief['why_now']}")
        if brief.get("suggested_message"):
            lines.append(f"\n**Suggested first message:**")
            lines.append(f"> {brief['suggested_message']}")
        lines.append("")
        lines.append("---")
        lines.append("")

if too_early:
    lines += [f"### Too Early ({len(too_early)} packages below 500 weekly downloads)", ""]
    for p in too_early:
        lines.append(f"- {p['package']}: ~{p['recent_4_avg']:,}/week -- revisit when above 500/week")
    lines.append("")

if established:
    lines += [f"### Established Packages (above 500K/week, velocity less meaningful)", ""]
    for p in established:
        lines.append(f"- {p['package']}: ~{p['recent_4_avg']:,}/week")
    lines.append("")

lines += ["---", ""]
lines.append(f"Data quality notes: {'; '.join(flags) if flags else 'None'}")

output_path = f"docs/npm-leads/{date_str}.md"
os.makedirs("docs/npm-leads", exist_ok=True)
open(output_path, "w").write("\n".join(lines))

print("\n".join(lines))
print(f"\nSaved to: {output_path}")
PYEOF

清理临时文件：

rm -f /tmp/npl-input.json /tmp/npl-download-data.json /tmp/npl-scored.json \
      /tmp/npl-enriched.json /tmp/npl-briefs.json /tmp/npl-output.json