Varnan-Tech/opendirectory

npm 下載轉潛在客戶

取得一個 npm 套件清單。擷取 12 週的下載數據。計算爆發速度。豐富維護者資料。為每個爆發套件輸出排名後的潛在客戶簡報，包含聯繫訊號和推廣訊息。

---

關鍵規則： 輸出中的每個套件下載數據都必須來自 npm API 回應。每個維護者的 GitHub 使用者名稱或 Twitter 使用者名稱都必須來自 GitHub API 回應——而不是從 npm 使用者名稱猜測。如果 GitHub API 沒有回傳 twitter_username 欄位，請寫「在 GitHub 上找不到」——不要自行發明。

---

常見錯誤

代理可能會想要...	為什麼這是錯的
為清單中的每個套件擷取 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:-未設定，套用未驗證的速率限制 (每小時 60 次——大約足夠 ~10 個套件)}"

如果 GITHUB_TOKEN 未設定： 繼續。告知使用者：「GITHUB_TOKEN 未設定。GitHub 豐富處理僅限約 10 個套件，之後就會達到速率限制。請在 github.com/settings/tokens 新增一個令牌（不需要任何作用域）。」

沒有必需的密鑰。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("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，因此您無需執行下方的內嵌程式碼區塊

# 檢查腳本是否存在
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"))
# 腳本輸出包含 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"Loaded {len(scored)} packages | {out['breakout_count']} breakout | {out['watching_count']} watching")
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]  # (year, week_num)
            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, 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)  # 溫和的速率限制

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)

    # 最佳下載量範圍：每週 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 = "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
    })

# 按速度分數降序排序
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}%")

# 如果沒有值得分析的東西則停止
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 註冊表中繼資料，然後是 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 註冊表 ---
    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}: 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 使用者個人檔案 ---
    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 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", gh_rate_remaining))
                gh_data = json.loads(resp.read())

                gh_profile = {
                    "username": gh_data.get("login"),
                    "name": gh_data.get("name"),
                    "bio": (gh_data.get("bio") or "")[:200],
                    "company": gh_data.get("company"),
                    "location": gh_data.get("location"),
                    "email": gh_data.get("email"),
                    "twitter": gh_data.get("twitter_username"),
                    "blog": gh_data.get("blog"),
                    "public_repos": gh_data.get("public_repos"),
                    "followers": gh_data.get("followers"),
                    "created_at": gh_data.get("created_at"),
                    "data_source": "github-api"
                }
                if gh_profile["twitter"] is None:
                    gh_profile["twitter"] = "not_found_on_github"
                profile["github_users"].append(gh_profile)
                print(f"    GitHub: @{username} | name={gh_profile['name']} | twitter={gh_profile['twitter']}")
        except urllib.error.HTTPError as e:
            if e.code == 404:
                print(f"    GitHub: {username} not found (404)")
            elif e.code == 403:
                print(f"    GitHub: rate limited (403) -- remaining={gh_rate_remaining}")
                break
        except Exception as e:
            print(f"    GitHub: {username} fetch failed ({e})")

        time.sleep(0.1)

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

json.dump(enriched, open("/tmp/npl-enriched.json", "w"), indent=2)
print(f"\nProfile enrichment complete. Enriched {len(enriched)} packages.")
if not token:
    print("NOTE: No GITHUB_TOKEN set. Only a limited number of GitHub profiles could be fetched.")
PYEOF

---

步驟 6：呈現結果

在終端機中顯示格式化摘要，並寫入最終的 /tmp/npl-leads.json 檔案。

python3 << 'PYEOF'
import json, sys

scored = json.load(open("/tmp/npl-scored.json"))
try:
    enriched = json.load(open("/tmp/npl-enriched.json"))
except FileNotFoundError:
    enriched = []

with open("/tmp/npl-input.json") as f:
    inp = json.load(f)
product_context = inp.get("product_context", "")

data_quality_flags = []

# 檢查 GitHub 豐富處理是否不完整
if len(enriched) < len([p for p in scored if p["tier"] in ("breakout", "watching")]):
    data_quality_flags.append("GitHub enrichment incomplete -- rate limits may have been hit")

breakout_packages = [p for p in (enriched if enriched else scored) if p["tier"] == "breakout"]
watching_packages = [p for p in (enriched if enriched else scored) if p["tier"] == "watching"]

summary = {
    "package_count": len(scored),
    "breakout_count": len(breakout_packages),
    "watching_count": len(watching_packages),
    "data_quality_flags": data_quality_flags,
    "product_context": product_context
}

print("=" * 72)
print(f"  NPM DOWNLOADS TO LEADS REPORT")
if product_context:
    print(f"  Context: {product_context}")
print("=" * 72)
print(f"\n  Packages analyzed:   {summary['package_count']}")
print(f"  BREAKOUT:           {summary['breakout_count']}")
print(f"  WATCHING:           {summary['watching_count']}")
print(f"  Quality flags:      {len(data_quality_flags)}")

if data_quality_flags:
    print("\n  [!] Data quality notes:")
    for flag in data_quality_flags:
        print(f"      - {flag}")

print("\n" + "-" * 72)

# 為每個爆發套件寫入 JSON 結構
leads_output = []

for idx, item in enumerate(breakout_packages[:5]):
    profile = item.get("profile", {})
    gh_users = profile.get("github_users", [])
    gh_user = gh_users[0] if gh_users else {}
    maintainers = profile.get("npm_maintainers", [])

    lead = {
        "rank": idx + 1,
        "package": item["package"],
        "tier": item["tier"],
        "velocity_score": item["velocity_score"],
        "growth_pct": item["growth_pct"],
        "current_weekly": item["current_weekly"],
        "recent_4_avg": item["recent_4_avg"],
        "prior_4_avg": item["prior_4_avg"],
        "description": profile.get("description", ""),
        "keywords": profile.get("keywords", []),
        "npm_homepage": profile.get("npm_homepage", ""),
    }

    if gh_user:
        lead["maintainer"] = {
            "github_username": gh_user.get("username"),
            "name": gh_user.get("name"),
            "twitter": gh_user.get("twitter") if gh_user.get("twitter") != "not_found_on_github" else None,
            "twittermissing": gh_user.get("twitter") == "not_found_on_github",
            "company": gh_user.get("company"),
            "location": gh_user.get("location"),
            "email": gh_user.get("email"),
            "blog": gh_user.get("blog"),
            "followers": gh_user.get("followers"),
            "github_created_at": gh_user.get("created_at")
        }
    else:
        lead["maintainer"] = {
            "npm_usernames": maintainers[:3],
            "github_username": profile.get("github_owner"),
            "twitter": None,
            "twittermissing": True
        }

    # 建構推廣訊息
    context_note = f" about {product_context}" if product_context else ""
    outreach = (
        f"Hi {lead['maintainer'].get('name') or lead['maintainer'].get('github_username') or 'there'},\n\n"
        f"I've been following the growth of {item['package']} -- it's at {item['recent_4_avg']:,} weekly downloads, "
        f"growing {item['growth_pct']:.0f}% in the last four weeks (velocity score: {item['velocity_score']})."
        f"{context_note} Would you be open to a brief conversation?"
    )
    lead["outreach_message"] = outreach

    leads_output.append(lead)

# 排序並輸出
leads_output.sort(key=lambda x: x["velocity_score"], reverse=True)

json.dump({
    "summary": summary,
    "leads": leads_output,
    "watching": [
        {
            "package": p["package"],
            "velocity_score": p["velocity_score"],
            "growth_pct": p["growth_pct"],
            "recent_4_avg": p["recent_4_avg"]
        }
        for p in (enriched if enriched else scored) if p["tier"] == "watching"
    ],
    "too_early": [
        p["package"] for p in scored if p["tier"] in ("too_early", "insufficient_data")
    ]
}, open("/tmp/npl-leads.json", "w"), indent=2)

print(f"\n  OUTPUT: /tmp/npl-leads.json ({len(leads_output)} breakout leads)")
print("=" * 72)

# 終端機顯示
print(f"\n{'#':>3} {'Package':35} {'Score':>7} {'Growth':>8} {'Weekly':>10}  {'Maintainer':25} {'Twitter':20}")
print("-" * 72)
for lead in leads_output:
    m = lead["maintainer"]
    name = m.get("name") or m.get("github_username") or m.get("npm_usernames", [None])[0] or ""
    tw = m.get("twitter") or ("NOT FOUND" if m.get("twittermissing") else "")
    print(f"{lead['rank']:>3} {lead['package']:35} {lead['velocity_score']:>7.1f} {lead['growth_pct']:>+7.0f}% "
          f"{lead['recent_4_avg']:>8,}  {name[:25]:25} {tw[:20] if tw else 'N/A':20}")
PYEOF

---

步驟 7：維護者與 GitHub 映射（人工驗證）

提供一個表格，將 npm 使用者名稱映射到 GitHub 使用者名稱，並附上驗證方法。在回覆中使用此 Markdown 區塊。

### 維護者映射摘要

| npm 使用者名稱 | 推測的 GitHub 使用者名稱 | 驗證狀態 |
|---|---|---|
| `npm_username_a` | `github_username_a` | ✅ 已驗證（GitHub API 回傳相符的個人資料） |
| `npm_username_b` | `github_username_b` | ⚠️ 未驗證（#waited 但 GitHub 未回傳資料） |
| `npm_username_c` | `github_username_c` | ❌ 在 GitHub 上找不到 |

**注意事項：** npm 使用者名稱並不總是與 GitHub 使用者名稱相符。如果 GitHub API 未回傳個人資料，請不要假設它們相同。

---

步驟 8：回覆使用者

使用結果 /tmp/npl-leads.json 建構您的文字回覆。遵循此結構：

1. 摘要：分析了多少個套件，以及每個分類（爆發/值得關注/為時過早）的數量。
2. 速度排行榜：前 5 個套件的表格，包含速度分數、成長百分比和每週下載量。
3. 爆發潛在客戶：對於每個爆發套件，呈現：
   • 套件名稱和描述
   • 成長數據
   • 主要維護者資訊（真實姓名、GitHub 使用者名稱、Twitter 使用者名稱、公司、地點、電子郵件、追隨者數）
   • 推廣訊息
4. 值得關注的套件：值得監控但無需立即聯繫的簡短清單。
5. 為時過早的套件：低於每週 500 次下載的套件清單。
6. 數據品質標記：如果速率限制導致任何部分不完整。
7. 行動呼籲：詢問使用者是否要關注任何特定潛在客戶、調整排名、或分析另一組套件。

不要猜測。如果未找到 Twitter 使用者名稱，請如實陳述。不要發明 GitHub 使用者名稱。

---

輸出格式

最終的 /tmp/npl-leads.json 檔案包含：

• summary：總計、爆發、值得關注、為時過早的計數
• leads：爆發套件陣列，每個套件具有完整的豐富資料和推廣訊息
• watching：值得關注套件的精簡清單（僅套件名稱、分數）
• too_early：為時過早套件的名稱清單

如果此技能中斷或失敗： 如果可能，回覆部分結果。例如，如果 GitHub 速率限制被觸發，請呈現沒有豐富資料的速度排行榜。永遠不要發明任何東西——特別是不存在於 API 回應中的 Twitter 使用者名稱或 GitHub 使用者名稱。