gh-issue-to-demand-signal

GitHub Issue 需求信号

选取竞争对手公开的 GitHub 仓库，获取其未关闭的 issue。在本地过滤噪声，将其归为 6 个需求类别，并根据真实互动情况评分。输出排名靠前的需求缺口报告和 GTM 消息简报。

---

关键规则： 输出中的每个 issue 标题都必须与 GitHub API 响应中的内容完全一致。每个聚类主题名称都必须源自该聚类中的实际 issue 标题。若过滤噪声后剩余的 issue 少于 10 个，请停止并告知用户：该仓库规模太小，无法进行可靠的聚类。任何地方都不得编造 issue 内容。

---

常见错误

Agent 容易犯的错误	错在哪里
不做过滤，就把全部 200 个原始 issue 发给 AI	机器人 issue、PR 和零互动噪声会夸大聚类数量并浪费上下文。应先在本地过滤。
将评论数用作主要需求信号	评论中包含维护者回复、无关讨论和垃圾内容。reactions["+1"] 是最干净的买家信号。
总结聚类时改写 issue 标题	改写会丢失买家的原话，而这正是整个分析的重点。请使用与原文完全一致的 issue 标题。
若过滤后剩余 issue 少于 10 个，仍继续执行第 4 步	少于 10 个 issue，说明仓库规模太小或提供的 URL 有误。基于稀疏数据进行聚类会产生毫无意义的类别。
将拉取请求纳入分析	GitHub Issues 端点也会返回 PR。请检查 issue 对象中是否不存在 pull_request 键。
未检查全部 3 项标准，就将某个 issue 标记为被忽视的需求	三项必须全部满足：reactions >= 10、时间 >= 180 天，且没有 planned/in-progress/roadmap 标签。缺少任何一项都不符合条件。

---

第 1 步：检查设置

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

如果未设置 GITHUB_TOKEN： 继续执行。告知用户：“尚未设置 GITHUB_TOKEN。未认证的速率限制为每小时 60 个请求，在触及限制前足以获取 2 次。若需重复使用，请在 github.com/settings/tokens 添加令牌（访问公开仓库无需任何作用域）。”

---

第 2 步：收集输入

需要提供：

• GitHub 仓库 URL（例如 https://github.com/owner/repo）或 owner/repo 标识（例如 facebook/react）

从输入中解析 owner 和 repo：

python3 << 'PYEOF'
import re, sys, os

raw = "REPO_INPUT_HERE"

# Normalize to owner/repo
if raw.startswith("http"):
    m = re.search(r"github\.com/([^/]+)/([^/?\s]+)", raw)
    if not m:
        print("ERROR: Could not parse GitHub URL. Expected format: https://github.com/owner/repo")
        sys.exit(1)
    owner, repo = m.group(1), m.group(2).rstrip("/")
elif "/" in raw:
    parts = raw.strip().split("/")
    owner, repo = parts[0], parts[1]
else:
    print("ERROR: Input must be a GitHub URL or owner/repo slug (e.g. vercel/next.js)")
    sys.exit(1)

print(f"Owner: {owner}")
print(f"Repo: {repo}")

with open("/tmp/ghd-target.txt", "w") as f:
    f.write(f"{owner}/{repo}")
PYEOF

如果解析失败： 停止。询问：“请提供 GitHub 仓库 URL（https://github.com/owner/repo）或 owner/repo 标识（例如 vercel/next.js）。”

---

第 3 步：从 GitHub REST API 获取 Issues

最多获取 200 个 issue（每页 100 个，共 2 页）。首次获取后检查速率限制。

python3 << 'PYEOF'
import json, urllib.request, os, sys
from datetime import datetime, timezone

target = open("/tmp/ghd-target.txt").read().strip()
owner_repo = target
token = os.environ.get("GITHUB_TOKEN", "")

headers = {"Accept": "application/vnd.github+json", "User-Agent": "gh-issue-demand-signal/1.0"}
if token:
    headers["Authorization"] = f"Bearer {token}"

all_issues = []
rate_limit_hit = False

for page in [1, 2]:
    url = f"https://api.github.com/repos/{owner_repo}/issues?state=open&per_page=100&page={page}"
    req = urllib.request.Request(url, headers=headers)

    try:
        with urllib.request.urlopen(req, timeout=30) as resp:
            # Check rate limit after first page
            if page == 1:
                remaining = int(resp.headers.get("X-RateLimit-Remaining", 999))
                reset_ts = resp.headers.get("X-RateLimit-Reset", "")
                if remaining == 0:
                    reset_str = datetime.fromtimestamp(int(reset_ts), tz=timezone.utc).strftime("%H:%M UTC") if reset_ts else "unknown"
                    print(f"ERROR: GitHub rate limit exhausted. Resets at {reset_str}.")
                    print("Add GITHUB_TOKEN to your .env file to get 5000 req/hr. See github.com/settings/tokens (no scopes needed).")
                    sys.exit(1)
                print(f"Rate limit remaining: {remaining}")

                # Check for 404/403
                status = resp.status
                if status == 404:
                    print(f"ERROR: Repo '{owner_repo}' not found. Check the URL or slug.")
                    sys.exit(1)

            page_data = json.loads(resp.read())
            if not page_data:
                print(f"Page {page}: empty, stopping.")
                break
            all_issues.extend(page_data)
            print(f"Page {page}: {len(page_data)} issues fetched")
    except urllib.error.HTTPError as e:
        if e.code == 404:
            print(f"ERROR: Repo '{owner_repo}' not found (404). Check the URL or slug.")
        elif e.code == 403:
            print(f"ERROR: Access denied (403). Repo may be private or rate limit hit.")
        else:
            print(f"ERROR: GitHub API returned HTTP {e.code}")
        sys.exit(1)
    except Exception as e:
        print(f"ERROR: Failed to fetch page {page}: {e}")
        sys.exit(1)

print(f"Total raw issues fetched: {len(all_issues)}")
json.dump(all_issues, open("/tmp/ghd-raw-issues.json", "w"), indent=2)
PYEOF

如果 GitHub 返回 404： 停止。告知用户：“未找到仓库。请检查 URL 或标识，然后重试。未经身份验证且未明确授予仓库作用域的情况下，无法访问私有仓库。”

如果 GitHub 返回带有速率限制标头的 403： 停止。显示重置时间，并告知用户添加 GITHUB_TOKEN。

---

第 4 步：在本地预处理——过滤、评分、检测被忽视的需求

不调用 API。纯 Python。必须在将任何内容交给 AI 之前运行。

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

raw = json.load(open("/tmp/ghd-raw-issues.json"))
target = open("/tmp/ghd-target.txt").read().strip()
now = datetime.now(tz=timezone.utc)

noise_patterns = re.compile(
    r"^(chore|deps|bump|renovate|dependabot|release|ci|build|revert)[\s:\[]",
    re.IGNORECASE
)
pr_title_patterns = re.compile(
    r"^(feat|fix|refactor|docs|test|style|perf|chore)(\(.+\))?:",
    re.IGNORECASE
)

filtered = []
noise_count = 0
noise_reasons = {}

for issue in raw:
    # Skip pull requests (GitHub Issues endpoint returns PRs too)
    if "pull_request" in issue:
        noise_count += 1
        noise_reasons["pull_request"] = noise_reasons.get("pull_request", 0) + 1
        continue

    title = issue.get("title", "")
    reactions = issue.get("reactions", {}).get("+1", 0)
    comments = issue.get("comments", 0)
    user_type = (issue.get("user") or {}).get("type", "User")

    # Skip bot-authored issues
    if user_type == "Bot":
        noise_count += 1
        noise_reasons["bot_author"] = noise_reasons.get("bot_author", 0) + 1
        continue

    # Skip bot-pattern titles
    if noise_patterns.match(title):
        noise_count += 1
        noise_reasons["bot_title"] = noise_reasons.get("bot_title", 0) + 1
        continue

    # Skip PR-as-issue titles
    if pr_title_patterns.match(title):
        noise_count += 1
        noise_reasons["pr_as_issue"] = noise_reasons.get("pr_as_issue", 0) + 1
        continue

    # Skip zero-signal issues
    if reactions == 0 and comments == 0:
        noise_count += 1
        noise_reasons["zero_signal"] = noise_reasons.get("zero_signal", 0) + 1
        continue

    # Compute demand score
    demand_score = (reactions * 2) + (comments * 0.5)

    # Detect ignored demand
    created_at = issue.get("created_at", "")
    if created_at:
        created = datetime.fromisoformat(created_at.replace("Z", "+00:00"))
        age_days = (now - created).days
    else:
        age_days = 0

    labels = [l.get("name", "").lower() for l in issue.get("labels", [])]
    has_planned_label = any(
        kw in label for label in labels
        for kw in ["in-progress", "planned", "roadmap", "wip", "in progress"]
    )

    ignored_demand = (
        reactions >= 10 and
        age_days >= 180 and
        not has_planned_label
    )

    filtered.append({
        "number": issue["number"],
        "title": title,
        "url": issue.get("html_url", f"https://github.com/{target}/issues/{issue['number']}"),
        "reactions_plus1": reactions,
        "comments": comments,
        "demand_score": demand_score,
        "age_days": age_days,
        "labels": labels,
        "ignored_demand": ignored_demand,
        "body_snippet": (issue.get("body") or "")[:300]
    })

# Sort by demand score descending
filtered.sort(key=lambda x: x["demand_score"], reverse=True)

print(f"Raw issues: {len(raw)}")
print(f"Noise filtered: {noise_count} ({', '.join(f'{k}: {v}' for k, v in noise_reasons.items())})")
print(f"Issues for analysis: {len(filtered)}")

if len(filtered) < 10:
    print(f"ERROR: Only {len(filtered)} issues remain after filtering. This repo has too few engaged issues for reliable clustering.")
    print("Try a larger repo or a repo with more community engagement.")
    import sys; sys.exit(1)

# Ignored demand summary
ignored = [i for i in filtered if i["ignored_demand"]]
print(f"Ignored demand issues: {len(ignored)}")

json.dump(filtered, open("/tmp/ghd-filtered-issues.json", "w"), indent=2)
print("Pre-processing complete.")
PYEOF

如果过滤后剩余的 issue 少于 10 个： 停止。准确告知用户找到和过滤掉的 issue 数量，并说明仓库规模太小，无法可靠地进行需求聚类。

---

第 5 步：聚类 Issues

打印过滤后的 issue 以供分析：

python3 << 'PYEOF'
import json
filtered = json.load(open("/tmp/ghd-filtered-issues.json"))
target = open("/tmp/ghd-target.txt").read().strip()

issue_list = filtered[:150]
print(f"Repo: {target}")
print(f"Issues to cluster: {len(issue_list)}")
print()
for i in issue_list:
    labels_str = f" [{', '.join(i['labels'][:3])}]" if i['labels'] else ""
    print(f"#{i['number']} [score:{round(i['demand_score'],1)} reactions:{i['reactions_plus1']}] {i['title']}{labels_str}")
PYEOF

将上方打印的每个 issue 归入以下 6 个类别之一： feature_gap、bug_pattern、ux_complaint、performance、integration_missing、docs_missing

规则：

• 每个 issue 必须且只能归入一个类别
• 使用标题中的用户原话提取一句痛点陈述，不要改写
• 识别 5-8 个聚类主题：用 3-6 个词的短语概括所有 issue 中占主导地位的投诉模式
• 不要使用破折号。不要使用营销语言。

将分析写入 /tmp/ghd-clusters.json，并采用以下精确结构：

{
  "classified_issues": [
    {"number": 123, "category": "feature_gap", "pain_statement": "Users need X which does not exist yet"}
  ],
  "cluster_themes": [
    {"theme_name": "Missing export options", "category": "feature_gap", "issue_numbers": [123, 456, 789]}
  ],
  "category_counts": {"feature_gap": 5, "bug_pattern": 3, "ux_complaint": 4, "performance": 2, "integration_missing": 6, "docs_missing": 1}
}

写入文件后，使用以下命令确认：

python3 -c "
import json
d = json.load(open('/tmp/ghd-clusters.json'))
print(f'Classified: {len(d[\"classified_issues\"])} issues, {len(d[\"cluster_themes\"])} themes')
print('Categories:', d['category_counts'])
"

---

步骤 6：消息简报

计算每个集群的总需求评分，并打印排名前 3 的集群：

python3 << 'PYEOF'
import json

filtered = json.load(open("/tmp/ghd-filtered-issues.json"))
clusters = json.load(open("/tmp/ghd-clusters.json"))
target = open("/tmp/ghd-target.txt").read().strip()

demand_by_issue = {i["number"]: i["demand_score"] for i in filtered}
issue_titles = {i["number"]: i["title"] for i in filtered}
issue_reactions = {i["number"]: i["reactions_plus1"] for i in filtered}

enriched_themes = []
for theme in clusters.get("cluster_themes", []):
    issue_nums = theme.get("issue_numbers", [])
    total_demand = sum(demand_by_issue.get(n, 0) for n in issue_nums)
    top_issues = sorted(issue_nums, key=lambda n: demand_by_issue.get(n, 0), reverse=True)[:3]
    enriched_themes.append({
        "theme_name": theme["theme_name"],
        "category": theme["category"],
        "issue_count": len(issue_nums),
        "total_demand_score": round(total_demand, 1),
        "top_issues": [
            {"number": n, "title": issue_titles.get(n, ""), "reactions": issue_reactions.get(n, 0)}
            for n in top_issues
        ]
    })

enriched_themes.sort(key=lambda x: x["total_demand_score"], reverse=True)
json.dump(enriched_themes, open("/tmp/ghd-enriched-themes.json", "w"), indent=2)

print(f"Top 3 clusters for messaging brief (repo: {target}):")
for t in enriched_themes[:3]:
    print(f"\n  {t['theme_name']} ({t['category']}) -- total demand: {t['total_demand_score']}")
    for ti in t["top_issues"]:
        print(f"    #{ti['number']}: \"{ti['title']}\" ({ti['reactions']} reactions)")
PYEOF

根据上面打印的排名前 3 的集群，生成一份 GTM 消息简报。

规则：

• 每个定位角度都必须注明其来源的具体集群
• 每个外联切入点都必须用引号引用一条原样的 issue 标题
• 标题必须包含数字或明确命名的痛点，不得使用泛泛的表述
• 不得使用 em dash。不得使用以下禁用词：powerful、robust、seamless、innovative、game-changing、streamline、leverage、transform

将简报写入 /tmp/ghd-brief.json，并严格采用以下结构：

{
  "positioning_angles": [
    {
      "angle_name": "3-5 word label",
      "cluster_source": "theme_name from cluster",
      "positioning_statement": "2-3 sentences on what your product does that this competitor does not",
      "evidence": "verbatim issue title that best illustrates this gap"
    }
  ],
  "outreach_hooks": [
    {
      "hook_type": "pain quote hook",
      "hook_text": "2-3 sentences quoting a verbatim issue title in quotes",
      "best_for": "audience this hook works for"
    }
  ],
  "cluster_headlines": [
    {
      "theme_name": "from the cluster",
      "headline": "specific headline with a number or named pain",
      "sub_copy": "1 sentence expanding the headline"
    }
  ]
}

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

python3 -c "
import json
d = json.load(open('/tmp/ghd-brief.json'))
print('Positioning angles:', len(d.get('positioning_angles', [])))
print('Outreach hooks:', len(d.get('outreach_hooks', [])))
print('Cluster headlines:', len(d.get('cluster_headlines', [])))
"

---

步骤 7：自我 QA

提交前运行。核验证据。消除违规内容。检查输出完整性。

python3 << 'PYEOF'
import json

filtered = json.load(open("/tmp/ghd-filtered-issues.json"))
clusters = json.load(open("/tmp/ghd-clusters.json"))
themes = json.load(open("/tmp/ghd-enriched-themes.json"))
brief = json.load(open("/tmp/ghd-brief.json"))
target = open("/tmp/ghd-target.txt").read().strip()

failures = []
real_titles = {i["number"]: i["title"] for i in filtered}

# Verify: classified_issues only reference real issue numbers
real_numbers = set(real_titles.keys())
hallucinated = [
    c["number"] for c in clusters.get("classified_issues", [])
    if c["number"] not in real_numbers
]
if hallucinated:
    failures.append(f"Removed {len(hallucinated)} hallucinated issue numbers from classified_issues: {hallucinated[:5]}")
    clusters["classified_issues"] = [
        c for c in clusters.get("classified_issues", [])
        if c["number"] in real_numbers
    ]

# Verify: top-10 list is sorted by demand_score descending
top10 = filtered[:10]
for i, issue in enumerate(top10):
    if i > 0 and issue["demand_score"] > top10[i-1]["demand_score"]:
        failures.append("Top-10 list was not sorted by demand_score -- re-sorted.")
        filtered.sort(key=lambda x: x["demand_score"], reverse=True)
        top10 = filtered[:10]
        break

# Verify: ignored demand issues meet all 3 criteria
ignored = [i for i in filtered if i["ignored_demand"]]
for issue in ignored:
    if issue["reactions_plus1"] < 10 or issue["age_days"] < 180:
        issue["ignored_demand"] = False
        failures.append(f"Removed issue #{issue['number']} from ignored demand -- did not meet all 3 criteria")

# Check messaging brief counts
if len(brief.get("positioning_angles", [])) != 3:
    failures.append(f"Expected 3 positioning angles, got {len(brief.get('positioning_angles', []))}")
if len(brief.get("outreach_hooks", [])) != 3:
    failures.append(f"Expected 3 outreach hooks, got {len(brief.get('outreach_hooks', []))}")
if len(brief.get("cluster_headlines", [])) != 3:
    failures.append(f"Expected 3 cluster headlines, got {len(brief.get('cluster_headlines', []))}")

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

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

# Build final output bundle
output = {
    "repo": target,
    "issues_analyzed": len(filtered),
    "clusters": clusters,
    "enriched_themes": themes,
    "filtered_issues": filtered,
    "messaging_brief": brief,
    "data_quality_flags": failures
}

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

---

步骤 8：保存并呈现输出

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

output = json.load(open("/tmp/ghd-output.json"))
target = output["repo"]
repo_slug = target.replace("/", "-")
date_str = datetime.now(tz=timezone.utc).strftime("%Y-%m-%d")

filtered = output["filtered_issues"]
themes = output["enriched_themes"]
clusters = output["clusters"]
brief = output["messaging_brief"]
flags = output["data_quality_flags"]

ignored = [i for i in filtered if i.get("ignored_demand")]
top10 = filtered[:10]

# Build category summary from cluster data
category_counts = clusters.get("category_counts", {})

lines = [
    f"## Demand Gap Report: {target}",
    f"Issues analyzed: {output['issues_analyzed']} | Date: {date_str}",
    "",
    "---",
    "",
    "### Demand Gap Leaderboard",
    "",
    "| Rank | Theme | Category | Issues | Total Demand Score | Top Issue Reactions |",
    "|---|---|---|---|---|---|",
]

for i, theme in enumerate(themes[:8], 1):
    top_reactions = theme["top_issues"][0]["reactions"] if theme["top_issues"] else 0
    lines.append(
        f"| {i} | {theme['theme_name']} | {theme['category']} | "
        f"{theme['issue_count']} | {theme['total_demand_score']} | {top_reactions} |"
    )

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

if ignored:
    lines += [
        "### Ignored Demand (High Reactions, No Maintainer Response)",
        "",
        "These issues have 10+ reactions, are 6+ months old, and have no planned/in-progress label.",
        "This is your opportunity window.",
        "",
    ]
    for issue in ignored[:10]:
        lines.append(
            f"- [{issue['title']}]({issue['url']}) -- "
            f"{issue['reactions_plus1']} reactions, {issue['age_days']} days old"
        )
    lines += ["", "---", ""]

lines += [
    "### Top 10 Highest-Demand Issues",
    "",
    "| Rank | Issue | Reactions | Comments | Demand Score | Link |",
    "|---|---|---|---|---|---|",
]
for i, issue in enumerate(top10, 1):
    short_title = issue["title"][:70] + ("..." if len(issue["title"]) > 70 else "")
    lines.append(
        f"| {i} | {short_title} | {issue['reactions_plus1']} | "
        f"{issue['comments']} | {round(issue['demand_score'], 1)} | "
        f"[#{issue['number']}]({issue['url']}) |"
    )

lines += ["", "---", "", "### Cluster Deep Dives", ""]

for theme in themes[:3]:
    lines.append(f"#### {theme['theme_name']}")
    lines.append(f"Category: {theme['category']} | Issues: {theme['issue_count']} | Total demand score: {theme['total_demand_score']}")
    lines.append("")
    lines.append("Top issues in this cluster:")
    for ti in theme["top_issues"]:
        lines.append(f"- \"{ti['title']}\" -- {ti['reactions']} reactions")
    lines.append("")

lines += ["---", "", "### Messaging Brief", ""]

for angle in brief.get("positioning_angles", []):
    lines.append(f"**{angle.get('angle_name', 'Angle')}**")
    lines.append(angle.get("positioning_statement", ""))
    lines.append(f"Evidence: \"{angle.get('evidence', '')}\"")
    lines.append("")

lines += ["---", "", "### GTM Angles", ""]

for hook in brief.get("outreach_hooks", []):
    lines.append(f"**{hook.get('hook_type', 'Hook')}**")
    lines.append(hook.get("hook_text", ""))
    lines.append(f"Best for: {hook.get('best_for', '')}")
    lines.append("")

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

output_path = f"docs/demand-signals/{repo_slug}-{date_str}.md"
import os
os.makedirs("docs/demand-signals", exist_ok=True)
open(output_path, "w").write("\n".join(lines))
print(f"Saved to: {output_path}")

# Print to console
print("\n" + "\n".join(lines))
PYEOF

清理临时文件：

rm -f /tmp/ghd-target.txt /tmp/ghd-raw-issues.json /tmp/ghd-filtered-issues.json \
      /tmp/ghd-cluster-request.json /tmp/ghd-clusters.json /tmp/ghd-enriched-themes.json \
      /tmp/ghd-brief-request.json /tmp/ghd-brief.json /tmp/ghd-output.json