npm下载量到潜在客户

npm下载量到潜在客户 是做什么的？

输入一个npm包列表。获取12周的下载数据。计算爆发速度。丰富维护者资料。输出一份针对每个爆发包的潜在客户简报，包含联系信号和推广信息。

---

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

---

常见错误

代理可能会...	为什么会错
为列表中的每个包获取GitHub资料	无token时速率限制为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:-未设置，未认证的速率限制适用（60次请求/小时——足够处理约10个包）}"

如果未设置GITHUB_TOKEN： 继续。通知用户：“GITHUB_TOKEN未设置。GitHub丰富受限于约10个包，之后将命中速率限制。请在github.com/settings/tokens添加一个token（无需任何权限）。”

无需密钥。npm API和npm注册表完全公开，无需认证。

---

第2步：收集输入

从对话中收集：

• 一个或多个npm包名（无作用域，如esbuild，或有作用域，如@hono/hono）
• 可选：简短的产品上下文字符串（用于个性化推广信息）

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

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

python3 << 'PYEOF'
import json, sys

packages_raw = "PACKAGES_HERE"  # 逗号或换行分隔
product_context = "CONTEXT_HERE"  # 可选，可以为空字符串

packages = [p.strip() for p in packages_raw.replace("\n", ",").split(",") if p.strip()]
if not packages:
    print("错误：未提供任何包。")
    sys.exit(1)

print(f"待分析的包数：{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步，因此您无需运行下面内联的代码块。

# 检查脚本是否存在
ls scripts/fetch.py 2>/dev/null && echo "脚本可用" || echo "未找到脚本"

如果脚本可用，直接运行它并跳至第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"))
# 脚本输出包含results数组——拆分为scored和enriched，用于第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"已加载 {len(scored)} 个包 | {out['breakout_count']} 个爆发 | {out['watching_count']} 个观察")
PYEOF

如果脚本不可用，运行下面的内联代码。

从npm下载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)  # 额外一周缓冲区，应对部分周
start_str = start_date.strftime("%Y-%m-%d")
end_str = end_date.strftime("%Y-%m-%d")

results = []
failed = []

for pkg in packages:
    # URL编码作用域包：@ -> %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())

        # 按ISO周聚合每日数据
        weekly = defaultdict(int)
        for entry in raw.get("downloads", []):
            day = datetime.strptime(entry["day"], "%Y-%m-%d")
            week_key = day.isocalendar()[:2]  # (年, 周数)
            weekly[week_key] += entry["downloads"]

        weeks = [v for k, v in sorted(weekly.items())]
        # 取最近12个完整的周桶
        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[-1]:,} 次下载")

    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}: 未找到 (404) —— 将被跳过")
        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} 错误")
    except Exception as e:
        failed.append(pkg)
        results.append({"package": pkg, "weeks": [], "total_weeks": 0, "current_weekly": 0, "status": f"error"})
        print(f"  {pkg}: 获取失败 ({e})")

    time.sleep(0.2)  # 温和的速率限制

json.dump(results, open("/tmp/npl-download-data.json", "w"), indent=2)
print(f"\n获取完成。成功：{len(results) - len(failed)} | 失败/未找到：{len(failed)}")
if failed:
    print(f"跳过：{', '.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": "数据不足",
                       "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)

    # 最佳区间乘数：500-500K周下载量
    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)

    # 分类
    if velocity_score > 80 and 500 < recent_4 < 500_000 and growth_ratio >= 1.5:
        tier = "爆发"
    elif velocity_score > 40 and recent_4 >= 500 and growth_ratio >= 1.2:
        tier = "观察"
    elif recent_4 < 500:
        tier = "为时过早"
    elif recent_4 >= 500_000:
        tier = "成熟"
    else:
        tier = "稳定"

    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
    })

# 按velocity_score降序排序
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"] == "爆发"]
watching = [p for p in scored if p["tier"] == "观察"]
too_early = [p for p in scored if p["tier"] == "为时过早"]

print(f"速度评分完成：")
print(f"  爆发：{len(breakout)}")
print(f"  观察：{len(watching)}")
print(f"  稳定/成熟：{len([p for p in scored if p['tier'] in ('稳定','成熟')])}")
print(f"  为时过早 (<500/周)：{len(too_early)}")
print()
for p in scored[:10]:
    print(f"  {p['tier']:12} {p['package']:30} 分数={p['velocity_score']:6.1f}  "
          f"{p['recent_4_avg']:>8,}/周  增长={p['growth_pct']:+.0f}%")

# 如果没有任何值得分析的内容则停止
if not breakout and not watching:
    all_too_early = all(p["tier"] in ("为时过早", "数据不足") for p in scored)
    if all_too_early:
        print("\n错误：所有包均低于每周500次下载的阈值，无法进行可靠的速度分析。")
        print("请尝试社区采用率更高的包。")
        import sys; sys.exit(1)
PYEOF

如果所有包均低于500/周： 停止并显示上述消息。

---

第5步：获取维护者资料

仅针对爆发和观察包。获取npm注册表元数据，然后获取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 ("爆发", "观察")]
print(f"正在为 {len(target_packages)} 个包（爆发 + 观察）获取资料...")

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注册表 ---
    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")]

        # 从仓库URL提取GitHub所有者
        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}: 注册表成功 | 维护者={profile['npm_maintainers'][:3]} | "
              f"github_所有者={profile['github_owner']}")
    except Exception as e:
        print(f"  {pkg}: 注册表获取失败 ({e})")

    time.sleep(0.1)

    # --- GitHub用户资料 ---
    candidates = []
    if profile["github_owner"]:
        candidates.append(profile["github_owner"])
    # 同时尝试npm维护者用户名（通常与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速率限制低（剩余{gh_rate_remaining}次请求）——跳过{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_data = json.loads(resp.read())
                gh_rate_remaining = int(resp.headers.get("X-RateLimit-Remaining", 0))

                user_info = {
                    "github_username": gh_data.get("login"),
                    "name": gh_data.get("name"),
                    "twitter_username": gh_data.get("twitter_username"),
                    "blog": gh_data.get("blog"),
                    "bio": gh_data.get("bio"),
                    "location": gh_data.get("location"),
                    "public_repos": gh_data.get("public_repos"),
                    "followers": gh_data.get("followers"),
                    "rate_limit_remaining": gh_rate_remaining
                }
                profile["github_users"].append(user_info)
                print(f"    GitHub用户{username}: 姓名={user_info['name']} | Twitter={user_info['twitter_username'] or '未找到'}")
        except urllib.error.HTTPError as e:
            if e.code == 404:
                print(f"    GitHub用户'{username}'未找到 (404)")
            else:
                print(f"    GitHub用户{username}: HTTP {e.code} 错误")
            gh_rate_remaining = int(e.headers.get("X-RateLimit-Remaining", 0)) if e.headers else 0
        except Exception as e:
            print(f"    GitHub用户{username}: 获取失败 ({e})")

        time.sleep(0.15)

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

json.dump(enriched, open("/tmp/npl-enriched.json", "w"), indent=2)
print(f"\n资料获取完成：{len(enriched)} 个包已丰富")
print(f"剩余GitHub速率限制：{gh_rate_remaining}")
if gh_rate_remaining < 10:
    print("注意：GitHub速率限制几乎耗尽。考虑添加GITHUB_TOKEN以获得更高的限制。")
PYEOF

---

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

python3 << 'PYEOF'
import json, sys

try:
    enriched = json.load(open("/tmp/npl-enriched.json"))
    scored = json.load(open("/tmp/npl-scored.json"))
except FileNotFoundError:
    print("错误：未找到丰富数据。确保已运行获取维护者资料步骤。")
    sys.exit(1)

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

breakout = [p for p in enriched if p["tier"] == "爆发"]
watching = [p for p in enriched if p["tier"] == "观察"]

if not breakout and not watching:
    print("\n警告：未找到爆发或观察包。没有潜在客户简报可生成。")
    print("输出空数组。")
    lead_briefs = []
else:
    lead_briefs = []

    print("\n" + "="*60)
    print(f"npm下载量到潜在客户 — 报告")
    print(f"产品上下文：{product_context or '(未提供)'}")
    print(f"#爆发：{len(breakout)} | #观察：{len(watching)}")
    print("="*60)

    for idx, item in enumerate(breakout + watching, 1):
        pkg = item["package"]
        tier = item["tier"]
        score = item["velocity_score"]
        growth = item["growth_pct"]
        recent = item["recent_4_avg"]
        prior = item["prior_4_avg"]
        profile = item.get("profile", {})

        # 收集联系信号
        contacts = []
        data_quality_flags = []

        for user in profile.get("github_users", []):
            contact = {"via": "GitHub", "username": user.get("github_username")}
            if user.get("twitter_username"):
                contact["twitter"] = user["twitter_username"]
                contacts.append(f"Twitter: @{user['twitter_username']}")
            else:
                data_quality_flags.append(f"{user.get('github_username','')}无Twitter — 未在GitHub上找到")
            if user.get("blog"):
                contacts.append(f"网站: {user['blog']}")
            if user.get("location"):
                contacts.append(f"地点: {user['location']}")
            contacts.append(f"GitHub: https://github.com/{user.get('github_username','')}")

        if not contacts:
            contacts = ["未找到联系信号——未在GitHub上找到用户资料"]
            data_quality_flags.append("无GitHub用户资料——未获取任何用户数据")

        # 生成推广文本
        if product_context:
            outreach = (f"您好，我看到您在维护{pkg}，该包在过去12周内增长迅猛"
                       f"（+{growth}%，从每周约{prior:,}次下载增至每周约{recent:,}次）。"
                       f"作为{product_context}的一部分，我认为这种采用的激增与我们的目标非常吻合。"
                       f"愿意聊聊吗？")
        else:
            outreach = (f"您好，我看到{pkg}正在快速增长"
                       f"（+{growth}%，从每周约{prior:,}次下载增至每周约{recent:,}次）。"
                       f"很想了解您在增长方面的经验。有空聊聊吗？")

        brief = {
            "rank": idx,
            "package": pkg,
            "tier": tier,
            "velocity_score": score,
            "growth_pct": growth,
            "recent_4_avg_weekly": recent,
            "prior_4_avg_weekly": prior,
            "description": profile.get("description", ""),
            "keywords": profile.get("keywords", []),
            "contact_signals": contacts,
            "data_quality_flags": data_quality_flags,
            "outreach_message": outreach
        }
        lead_briefs.append(brief)

        # 打印摘要
        print(f"\n# {idx}. {pkg} ({tier})")
        print(f"   速度分数: {score} | 增长: +{growth}%")
        print(f"   历史: {prior:,}/周 → {recent:,}/周")
        print(f"   描述: {profile.get('description','')[:80]}...")
        for c in contacts[:3]:
            print(f"   联系: {c}")
        if data_quality_flags:
            for f in data_quality_flags:
                print(f"   [标记] {f}")

# 始终输出JSON
output = {
    "lead_briefs": lead_briefs,
    "summary": {
        "total_packages_analyzed": len(scored),
        "breakout_count": len(breakout),
        "watching_count": len(watching),
        "too_early_count": len([p for p in scored if p["tier"] == "为时过早"]),
        "product_context": product_context,
        "empty_result": len(lead_briefs) == 0
    }
}

json.dump(output, open("/tmp/npl-lead-briefs.json", "w"), indent=2)
print(f"\n{'='*60}")
print(f"潜在客户简报已保存。爆发：{len(breakout)} | 观察：{len(watching)}")
print(f"输出：/tmp/npl-lead-briefs.json")
PYEOF

---

第7步：呈现结果

以视觉上可读的格式向用户展示结果。清晰地区分爆发包和观察包。包含所有数据质量标记。如果用户请求或清理未发现问题，不要重新开门。

---

第8步：提供跟进选项

在输出结束时询问：

下一步：
1. 为这些包重新运行（修改列表或上下文）
2. 为所选包深入分析（获取GitHub stars、提交等）
3. 导出为CSV

---

元数据

• 版本: 1.0.0
• 平台: Claude Code, Cursor, Codex
• 数据源: npm Downloads API（公开），npm Registry API（公开），GitHub REST API（需要GITHUB_TOKEN以保持高速率限制）
• 速率限制: npm：无认证，但实际上无限制；GitHub：无token时60次请求/小时，有token时5,000次请求/小时