Downloads npm para Leads
Receba uma lista de pacotes npm. Busque 12 semanas de dados de download. Calcule a velocidade de crescimento (breakout velocity). Enriqueça os perfis dos mantenedores. Gere um resumo de lead classificado por pacote em crescimento com sinais de contato e uma mensagem de abordagem.
Regra crítica: Cada número de download de pacote na saída deve vir da resposta da API npm. Cada handle do GitHub ou nome de usuário do Twitter do mantenedor deve vir da resposta da API do GitHub – não adivinhado a partir do nome de usuário npm. Se a API do GitHub não retornar o campo twitter_username, escreva "não encontrado no GitHub" – não invente um.
Erros Comuns
| O agente vai querer... | Por que isso está errado |
|---|---|
| Buscar perfis do GitHub para cada pacote na lista | O limite de taxa é de 60 requisições/hora sem token. Enriquecer pacotes estáveis ou em declínio gasta o orçamento antes de chegar aos que estão em crescimento. Busque apenas perfis para pacotes em crescimento (breakout) e em observação (watching). |
| Classificar pacotes por downloads semanais brutos | Downloads brutos favorecem React e lodash, que não são leads. Um pacote que vai de 1K para 8K/semana é mais acionável do que React com 50M/semana. A pontuação de velocidade é o sinal. |
| Pular a codificação de URL para pacotes com escopo | @org/pkg sem codificação causa um 404 da API npm. Codifique @ como %40 e / como %2F para cada nome de pacote com escopo. |
| Parar a skill quando o limite de taxa do GitHub for atingido | Degrade de forma graciosa. Apresente o ranking de velocidade a partir dos dados npm, pule as complementações restantes do GitHub e adicione uma flag em data_quality_flags. Não interrompa. |
| Escrever mensagens de abordagem sem nomear o pacote específico | Mensagens genéricas do tipo "Vi seu projeto" ficam sem resposta. Cada mensagem de abordagem deve nomear o pacote, seus números de crescimento e uma conexão específica com o contexto fornecido pelo usuário. |
| Incluir pacotes abaixo de 500 downloads semanais como leads | Abaixo de 500/semana é ruído. O mantenedor ainda não tem um público significativo. Sinalize como "muito cedo", mas não apresente como lead. |
Passo 1: Verificação de Configuração
echo "GITHUB_TOKEN: ${GITHUB_TOKEN:-não definido, limite de taxa não autenticado se aplica (60 req/hora – suficiente para ~10 pacotes)}"
Se GITHUB_TOKEN não estiver definido: Continue. Informe o usuário: "GITHUB_TOKEN não está definido. O enriquecimento do GitHub está limitado a ~10 pacotes antes de atingir o limite de taxa. Adicione um token em github.com/settings/tokens (nenhum escopo necessário)."
Nenhuma chave obrigatória. As APIs npm e o registro npm são totalmente públicos, sem autenticação.
Passo 2: Coletar Entrada
Colete da conversa:
- Um ou mais nomes de pacotes npm (sem escopo como
esbuild, ou com escopo como@hono/hono) - Opcional: uma pequena string de contexto do produto (usada para personalizar mensagens de abordagem)
Se o usuário fornecer uma URL npmjs.com, extraia apenas o nome do pacote. Preserve o nome completo com escopo, incluindo @ e o prefixo da organização – a codificação é tratada no Passo 3.
Se nenhum pacote for fornecido: Pergunte: "Quais pacotes npm você gostaria de analisar? Forneça os seus, de concorrentes ou uma mistura. Exemplo: esbuild, @hono/hono, zod, valibot"
python3 << 'PYEOF'
import json, sys
packages_raw = "PACKAGES_AQUI" # separado por vírgula ou nova linha
product_context = "CONTEXTO_AQUI" # opcional, pode ser string vazia
packages = [p.strip() for p in packages_raw.replace("\n", ",").split(",") if p.strip()]
if not packages:
print("ERRO: Nenhum pacote fornecido.")
sys.exit(1)
print(f"Pacotes a analisar: {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
Passo 3: Buscar Dados de Download de 12 Semanas
Use o script standalone se disponível – ele lida com os Passos 3, 4 e 5 em uma única chamada, então você não precisa executar os blocos de código inline abaixo.
# Verifique se o script existe
ls scripts/fetch.py 2>/dev/null && echo "script disponível" || echo "script não encontrado"
Se o script estiver disponível, execute-o diretamente e pule para o Passo 6:
python3 scripts/fetch.py PACOTES_AQUI --context "CONTEXTO_AQUI" --output /tmp/npl-script-out.json
Depois, carregue a saída no formato enriquecido que o Passo 6 espera:
python3 << 'PYEOF'
import json
out = json.load(open("/tmp/npl-script-out.json"))
# A saída do script tem um array results – divida em scored e enriched para os Passos 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"Carregados {len(scored)} pacotes | {out['breakout_count']} em crescimento | {out['watching_count']} em observação")
PYEOF
Se o script não estiver disponível, execute o código inline abaixo.
Busque dados diários de download para cada pacote da API npm Downloads. Agregue em buckets semanais.
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) # semana extra de buffer para semanas parciais
start_str = start_date.strftime("%Y-%m-%d")
end_str = end_date.strftime("%Y-%m-%d")
results = []
failed = []
for pkg in packages:
# Codifica URL para pacotes com escopo: @ -> %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())
# Agrega diário para semanal por semana ISO
weekly = defaultdict(int)
for entry in raw.get("downloads", []):
day = datetime.strptime(entry["day"], "%Y-%m-%d")
week_key = day.isocalendar()[:2] # (ano, numero_semana)
weekly[week_key] += entry["downloads"]
weeks = [v for k, v in sorted(weekly.items())]
# Pega as últimas 12 semanas completas
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)} semanas, última semana {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}: NÃO ENCONTRADO (404) – será ignorado")
else:
failed.append(pkg)
results.append({"package": pkg, "weeks": [], "total_weeks": 0, "current_weekly": 0, "status": f"error_{e.code}"})
print(f" {pkg}: Erro 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}: falha na busca ({e})")
time.sleep(0.2) # limitação de taxa suave
json.dump(results, open("/tmp/npl-download-data.json", "w"), indent=2)
print(f"\nBusca concluída. OK: {len(results) - len(failed)} | Falha/Não encontrado: {len(failed)}")
if failed:
print(f"Ignorados: {', '.join(failed)}")
PYEOF
Se todos os pacotes retornarem 404 ou erros: Pare. Diga ao usuário: "Não foi possível buscar dados de download. Verifique se os nomes dos pacotes estão corretos e existem em npmjs.com. Pacotes com escopo devem incluir o nome completo: @org/package."
Passo 4: Calcular Pontuações de Velocidade
Sem chamada de API. Python puro. Calcule a pontuação de velocidade, a taxa de crescimento e classifique cada pacote.
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)
# Multiplicador de ponto ideal: 500-500K downloads semanais
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)
# Classifica
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
})
# Ordena por velocity_score descendente
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"Classificação de velocidade concluída:")
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/semana): {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,}/sem crescimento={p['growth_pct']:+.0f}%")
# Para se não houver nada que vale a pena analisar
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("\nERRO: Todos os pacotes estão abaixo do limite de 500 downloads semanais para uma análise de velocidade confiável.")
print("Tente pacotes com maior adoção pela comunidade.")
import sys; sys.exit(1)
PYEOF
Se todos os pacotes estiverem abaixo de 500/semana: Pare com a mensagem acima.
Passo 5: Buscar Perfis dos Mantenedores
Apenas para pacotes em crescimento (breakout) e em observação (watching). Busque metadados do registro npm e, em seguida, perfis de usuários do 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"Buscando perfis para {len(target_packages)} pacotes (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": ""}
# --- registro 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")]
# Extrai owner do GitHub a partir da URL do repositório
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}: registro OK | mantenedores={profile['npm_maintainers'][:3]} | "
f"github_owner={profile['github_owner']}")
except Exception as e:
print(f" {pkg}: falha na busca do registro ({e})")
time.sleep(0.1)
# --- Perfis de usuários do GitHub ---
candidates = []
if profile["github_owner"]:
candidates.append(profile["github_owner"])
# Tenta também nomes de usuário dos mantenedores npm (frequentemente coincidem com 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" Limite de taxa do GitHub baixo ({gh_rate_remaining} restantes) – ignorando {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", gh_rate_remaining))
user_info = {
"username": gh_data.get("login", username),
"name": gh_data.get("name"),
"twitter": gh_data.get("twitter_username"),
"bio": gh_data.get("bio"),
"location": gh_data.get("location"),
"email": gh_data.get("email"),
"public_repos": gh_data.get("public_repos"),
"followers": gh_data.get("followers"),
"html_url": gh_data.get("html_url"),
}
profile["github_users"].append(user_info)
print(f" {username}: GitHub OK | seguidores={gh_data.get('followers', '?')} twitter={gh_data.get('twitter_username', 'não encontrado')}")
except urllib.error.HTTPError as e:
if e.code == 403:
gh_rate_remaining = 0
print(f" {username}: Limite de taxa do GitHub atingido (403)")
break
elif e.code == 404:
print(f" {username}: Não encontrado no GitHub")
else:
print(f" {username}: Erro HTTP {e.code}")
except Exception as e:
print(f" {username}: falha na busca ({e})")
time.sleep(0.2)
enriched.append({
**item,
"profile": profile
})
# Se o limite de taxa estiver esgotado, pare com este pacote e marque
if gh_rate_remaining <= 5:
print("")
print("⚠ Limite de taxa do GitHub esgotado. Os pacotes restantes não serão enriquecidos.")
print(" Os dados de download npm ainda serão apresentados.")
break
json.dump(enriched, open("/tmp/npl-enriched.json", "w"), indent=2)
print(f"\nEnriquecimento concluído: {len(enriched)} pacotes enriquecidos")
print(f" Chamadas de API do GitHub restantes: {gh_rate_remaining}")
PYEOF
Passo 6: Resumo de Saída
Mostre ao usuário:
- Ranking de velocidade para todos os pacotes
- Breves de leads detalhados para pacotes breakout
- Pacotes watching (classificados, sem breves completos)
- Pacotes ignorados com motivo
- Bandeiras de qualidade dos dados
python3 << 'PYEOF'
import json
scored = json.load(open("/tmp/npl-scored.json"))
enriched = json.load(open("/tmp/npl-enriched.json"))
try:
inp = json.load(open("/tmp/npl-input.json"))
context = inp.get("product_context", "")
except:
context = ""
# Mapeia pacotes enriquecidos para acesso fácil
enriched_map = {e["package"]: e for e in enriched}
# === 1. Ranking de Velocidade ===
print("=" * 72)
print("RANKING DE VELOCIDADE DE DOWNLOADS npm")
print("=" * 72)
print(f"{"PACOTE":30} {"VELOCIDADE":10} {"SEMANAIS":>10} {"CRESC %":>8} {"NÍVEL":14}")
print("-" * 72)
for item in scored:
pkg = item["package"]
vs = item["velocity_score"]
avg = item["recent_4_avg"]
gp = item["growth_pct"]
tier = item["tier"]
print(f"{pkg:30} {vs:>8.1f} {avg:>8,}/sem {gp:>+7.0f}% {tier:14}")
print()
# === 2. Breves de Leads ===
breakout_items = [item for item in scored if item["tier"] == "breakout"]
if breakout_items:
print("=" * 72)
print("📈 BREVES DE LEADS – PACOTES EM CRESCIMENTO (BREAKOUT)")
print("=" * 72)
for item in breakout_items:
pkg = item["package"]
prof = enriched_map.get(pkg, {}).get("profile", {})
print(f"\n{'─' * 72}")
print(f" 📦 {pkg} | Pontuação: {item['velocity_score']} | {item['recent_4_avg']:,}/sem | Crescimento: {item['growth_pct']:+.0f}%")
print(f"{'─' * 72}")
if prof.get("description"):
print(f" Descrição: {prof['description'][:200]}")
# Sinais de contato
github_users = prof.get("github_users", [])
npm_maint = prof.get("npm_maintainers", [])
print()
print(" 📋 SINAIS DE CONTATO:")
if github_users:
for u in github_users:
twitter = u.get("twitter")
twitter_str = f"@{twitter}" if twitter else "não encontrado no GitHub"
print(f" • GitHub: {u['html_url']}")
print(f" • Nome: {u.get('name', 'não encontrado no GitHub')}")
print(f" • Twitter: {twitter_str}")
print(f" • Localização: {u.get('location', 'não encontrado no GitHub')}")
print(f" • Bio: {(u.get('bio') or '')[ :100]}")
else:
print(" • Mantenedores npm (possíveis handles do GitHub):")
for m in npm_maint[:3]:
print(f" - {m}")
print(" • GitHub: Perfil não enriquecido (limite de taxa ou sem repositório GitHub)")
# Mensagem de abordagem
print()
print(" 💬 MENSAGEM DE ABORDAGEM SUGERIDA:")
maint_name = ""
if github_users:
maint_name = f"{github_users[0].get('name', '') or github_users[0]['username']}"
if len(github_users) > 1:
maint_name += f" e {github_users[1].get('name', '') or github_users[1]['username']}"
msg = f"Oi {maint_name or 'equipe do pacote ' + pkg},"
msg += f"\n\nPercebi que {pkg} está ganhando força – {item['growth_pct']:+.0f}% de crescimento com {item['recent_4_avg']:,} downloads semanais."
if context:
msg += f"\n\nEstamos trabalhando em [{context}] e acredito que seu pacote se alinha bem com o que estamos construindo."
msg += f" Gostaria de conversar sobre colaboração?"
else:
msg += f"\n\nGostaria de conversar sobre como poderíamos colaborar ou trocar ideias?"
msg += f"\n\nAbraço," if "equipe" not in maint_name else ""
print(f" {msg}")
print()
# === 3. Pacotes em Observação ===
watching_items = [item for item in scored if item["tier"] == "watching"]
if watching_items:
print()
print("=" * 72)
print("🔭 PACOTES EM OBSERVAÇÃO (WATCHING) – Classificados, sem breve completo")
print("=" * 72)
for item in watching_items:
print(f" • {item['package']:30} Velocidade: {item['velocity_score']:>6.1f} | {item['recent_4_avg']:>8,}/sem | Crescimento: {item['growth_pct']:+.0f}%")
prof = enriched_map.get(item["package"], {}).get("profile", {})
if prof.get("description"):
print(f" {prof['description'][:150]}")
# === 4. Pacotes Ignorados ===
ignored_items = [item for item in scored if item["tier"] in ("too_early", "insufficient_data", "established")]
if ignored_items:
print()
print("=" * 72)
print("⏭ PACOTES IGNORADOS (com motivo)")
print("=" * 72)
for item in ignored_items:
reason = {
"too_early": "Muito cedo (<500/semana)",
"insufficient_data": "Dados insuficientes",
"established": "Pacote estabelecido (>500K/semana)",
"steady": "Crescimento estável ou baixo"
}.get(item["tier"], item["tier"])
print(f" • {item['package']:30} Motivo: {reason} | {item['recent_4_avg']:>8,}/sem")
# === 5. Bandeiras de Qualidade dos Dados ===
print()
print("=" * 72)
print("🏁 BANDEIRAS DE QUALIDADE DOS DADOS")
print("=" * 72)
flags = []
token = __import__('os').environ.get("GITHUB_TOKEN", "")
if not token:
flags.append("GitHub: Sem token – limite de 60 req/hora; pode ter enriquecimento incompleto")
# Verifica pacotes não encontrados
not_found = [item["package"] for item in scored if item.get("status") == "not_found"]
if not_found:
flags.append(f"npm: {len(not_found)} pacote(s) não encontrado(s): {', '.join(not_found)}")
# Verifica pacotes com dados insuficientes
insufficient = [item["package"] for item in scored if item["tier"] == "insufficient_data"]
if insufficient:
flags.append(f"Dados: {len(insufficient)} pacote(s) com dados insuficientes (<4 semanas de dados)")
# Verifica enriquecimento parcial (se o limite de taxa foi atingido)
if enriched and len(enriched) < len([p for p in scored if p["tier"] in ("breakout", "watching")]):
flags.append("GitHub: Enriquecimento incompleto – limite de taxa atingido. Perfis ausentes para alguns pacotes em crescimento/observação.")
if not flags:
flags.append("Nenhum problema de qualidade dos dados identificado.")
for flag in flags:
print(f" ⚑ {flag}")
print()
print("=" * 72)
print("✅ Análise concluída. Use os breves de leads acima para inspirar divulgação.")
print("=" * 72)
PYEOF
Passo 7: Entrega
Apresente todas as saídas do Passo 6 ao usuário como a resposta final. Nenhuma ação extra é necessária.
Se o usuário perguntar sobre quaisquer recomendações específicas ou próximos passos, sugira:
- "Quer que eu aprofunde a análise de qualquer um desses leads?"
- "Posso personalizar as mensagens de abordagem para um pacote específico."
- "Que tal adicionar mais pacotes para análise?"
Modo de Depuração
Defina DEBUG=1 como variável de ambiente para habilitar logs detalhados durante a execução:
export DEBUG=1 # Habilita logs de depuração para todos os passos
Com DEBUG=1, scripts adicionais imprimirão:
- Dados brutos de resposta da API
- Detalhes do cálculo da pontuação de velocidade
- Estrutura completa dos perfis enriquecidos
Sem depuração, o comportamento padrão permanece inalterado.