forked from Gitlink/gitlink-cli
637 lines
25 KiB
Python
637 lines
25 KiB
Python
"""visual.py — S6 科研成果可视化沉淀。
|
||
|
||
把一个科研仓库的「成果」沉淀成一张可交互的可视化报告：开发时间线（commit/issue/pr 周粒度
|
||
趋势）、贡献者×周热力图、语言占比饼图、里程碑甘特。同时从 README/提交里抽取论文引用
|
||
（arXiv / DOI）并按目录对仓库产物做分类，便于科研工作者一眼看清「成果产出节奏 + 引用源头」。
|
||
|
||
数据全部经 gitlink-cli 获取（commit via Raw API、issue/pr/milestone/repo +list、contributors、
|
||
languages、readme、tree）。算法纯函数化（按周分桶 / 热力矩阵 / 论文链接抽取 / 产物分类），
|
||
便于离线单测。
|
||
|
||
用法：
|
||
python visual.py --owner mindspore-Ecosystem --repo mindspore --weeks 26 --out ./out
|
||
python visual.py --owner O --repo R # 仅打印 JSON
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
import json
|
||
import os
|
||
import re
|
||
import sys
|
||
from datetime import datetime, timezone
|
||
from typing import Any
|
||
|
||
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
|
||
import collect as c # noqa: E402
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 时间解析工具
|
||
# ---------------------------------------------------------------------------
|
||
|
||
def _to_timestamp(value: Any) -> float:
|
||
"""把 GitLink 多种时间表示统一成 epoch 秒。
|
||
|
||
支持整数秒（pr_created_unix）、ISO 字符串（created_at / timestamp 字符串）。
|
||
无法解析返回 0.0（最远古时间，会被周分桶丢弃到「太早」一端）。
|
||
"""
|
||
if value is None:
|
||
return 0.0
|
||
# 整数秒（commit.timestamp 形如 "1719500000" 也可走这里）
|
||
if isinstance(value, (int, float)):
|
||
f = float(value)
|
||
# 毫秒级时间戳兜底（13 位）
|
||
return f / 1000.0 if f > 1e12 else f
|
||
s = str(value).strip()
|
||
if not s:
|
||
return 0.0
|
||
# 纯数字字符串
|
||
if re.fullmatch(r"\d+(\.\d+)?", s):
|
||
f = float(s)
|
||
return f / 1000.0 if f > 1e12 else f
|
||
# ISO 8601（兼容带/不带 Z、带毫秒、带时区偏移）
|
||
txt = s.replace("Z", "+00:00")
|
||
fmts = ("%Y-%m-%dT%H:%M:%S%z",
|
||
"%Y-%m-%dT%H:%M:%S.%f%z",
|
||
"%Y-%m-%d %H:%M:%S",
|
||
"%Y-%m-%d")
|
||
for fmt in fmts:
|
||
try:
|
||
dt = datetime.strptime(txt, fmt)
|
||
if dt.tzinfo is None:
|
||
dt = dt.replace(tzinfo=timezone.utc)
|
||
return dt.timestamp()
|
||
except ValueError:
|
||
continue
|
||
return 0.0
|
||
|
||
|
||
def _commit_time(commit: dict) -> float:
|
||
"""提交对象取时间戳：优先 timestamp（字符串），退而 author.committed_unix。"""
|
||
ts = commit.get("timestamp")
|
||
if ts is not None:
|
||
return _to_timestamp(ts)
|
||
auth = commit.get("author") or {}
|
||
if isinstance(auth, dict):
|
||
for k in ("committed_unix", "committed_at", "time", "date"):
|
||
if auth.get(k) is not None:
|
||
return _to_timestamp(auth.get(k))
|
||
return 0.0
|
||
|
||
|
||
def _issue_time(issue: dict) -> float:
|
||
return _to_timestamp(issue.get("created_at"))
|
||
|
||
|
||
def _pr_time(pr: dict) -> float:
|
||
return _to_timestamp(pr.get("pr_created_unix") or pr.get("created_at"))
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 算法 1：按周分桶
|
||
# ---------------------------------------------------------------------------
|
||
|
||
def bin_weekly(items: list, weeks: int, time_getter) -> dict[str, list]:
|
||
"""把带时间戳的对象按「最近 weeks 周」分桶（含本周在内的 weeks 个连续周桶）。
|
||
|
||
Args:
|
||
items: 待分桶对象列表。
|
||
weeks: 保留最近多少个周桶。
|
||
time_getter: 从单个对象取 epoch 秒的函数。
|
||
|
||
返回 ``{"labels": [...], "counts": [...]}``：
|
||
- labels[i] 形如 "2026-W13"（ISO 周标签），从最近一周倒序到最老一周。
|
||
- counts[i] 为该周命中数；时间非法或越界（早于窗口左端）的对象不计入。
|
||
"""
|
||
weeks = max(1, int(weeks))
|
||
now = datetime.now(timezone.utc)
|
||
# 右端边界对齐到「下周一 00:00 UTC」（不含），使窗口包含当前周在内共 weeks 周。
|
||
# 若右端用「本周一」，当前周会被整体排除，最近一周的数据被吞掉。
|
||
today = now.replace(hour=0, minute=0, second=0, microsecond=0)
|
||
# Monday=0 .. Sunday=6；toordinal - weekday = 本周一，+7 = 下周一
|
||
next_monday = today.fromordinal(today.toordinal() - today.weekday() + 7).replace(tzinfo=timezone.utc)
|
||
end_ts = next_monday.timestamp()
|
||
start_ts = end_ts - weeks * 7 * 86400
|
||
|
||
counts = [0] * weeks
|
||
iso_labels: list[str] = []
|
||
for i in range(weeks):
|
||
# 桶 i 的周一 = end - (weeks-1-i) 周
|
||
bucket_monday_ts = start_ts + i * 7 * 86400
|
||
bucket_monday = datetime.fromtimestamp(bucket_monday_ts, tz=timezone.utc)
|
||
iso_year, iso_week, _ = bucket_monday.isocalendar()
|
||
iso_labels.append(f"{iso_year}-W{iso_week:02d}")
|
||
|
||
for item in items:
|
||
ts = time_getter(item)
|
||
if ts <= 0:
|
||
continue
|
||
if ts < start_ts or ts >= end_ts:
|
||
continue
|
||
offset = ts - start_ts
|
||
idx = int(offset // (7 * 86400))
|
||
if 0 <= idx < weeks:
|
||
counts[idx] += 1
|
||
|
||
return {"labels": iso_labels, "counts": counts}
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 算法 2：贡献者 × 周热力矩阵
|
||
# ---------------------------------------------------------------------------
|
||
|
||
def contribution_heatmap(contributors: list, commits: list, weeks: int,
|
||
top_users: int = 12) -> dict[str, list]:
|
||
"""构建 top 贡献者 × 周桶的提交数矩阵。
|
||
|
||
Args:
|
||
contributors: GitLink contributors[]（用于排序与展示名）。
|
||
commits: GitLink commits[]（含 author.login）。
|
||
weeks: 周桶数（与 bin_weekly 同口径）。
|
||
top_users: 矩阵最多保留多少个贡献者（按贡献数 desc）。
|
||
|
||
返回 ``{"users": [login...], "weeks": [label...], "matrix": [[cnt...]]}``：
|
||
matrix[user_i][week_j] = 该用户在该周的提交数。无 contributor 信息时也按 commit 作者聚合。
|
||
"""
|
||
weeks = max(1, int(weeks))
|
||
# 用 bin_weekly 的同口径周标签（取贡献数排序后的 login 列表）
|
||
window = bin_weekly(commits, weeks, _commit_time)
|
||
week_labels = window["labels"]
|
||
now = datetime.now(timezone.utc)
|
||
end = now.replace(hour=0, minute=0, second=0, microsecond=0)
|
||
end = end.fromordinal(end.toordinal() - end.weekday()).replace(tzinfo=timezone.utc)
|
||
start_ts = end.timestamp() - weeks * 7 * 86400
|
||
|
||
# 候选用户顺序：contributors（按 contributions desc）+ 提交里出现但不在 contributors 的作者
|
||
ordered: list[str] = []
|
||
seen: set[str] = set()
|
||
for contrib in contributors or []:
|
||
login = c.login_of(contrib) or ""
|
||
if login and login not in seen:
|
||
ordered.append(login)
|
||
seen.add(login)
|
||
for cm in commits or []:
|
||
auth = cm.get("author") or {}
|
||
login = c.login_of(auth) if isinstance(auth, dict) else ""
|
||
if login and login not in seen:
|
||
ordered.append(login)
|
||
seen.add(login)
|
||
|
||
users = ordered[:top_users]
|
||
matrix = [[0] * weeks for _ in users]
|
||
user_idx = {u: i for i, u in enumerate(users)}
|
||
|
||
for cm in commits or []:
|
||
ts = _commit_time(cm)
|
||
if ts <= 0 or ts < start_ts or ts >= end.timestamp():
|
||
continue
|
||
auth = cm.get("author") or {}
|
||
login = c.login_of(auth) if isinstance(auth, dict) else ""
|
||
if not login or login not in user_idx:
|
||
continue
|
||
offset = ts - start_ts
|
||
idx = int(offset // (7 * 86400))
|
||
if 0 <= idx < weeks:
|
||
matrix[user_idx[login]][idx] += 1
|
||
|
||
return {"users": users, "weeks": week_labels, "matrix": matrix}
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 算法 3：论文引用链接抽取
|
||
# ---------------------------------------------------------------------------
|
||
|
||
# arXiv: arxiv.org/abs/2401.00012 / arxiv.org/pdf/... / arxiv:2401.00012
|
||
_ARXIV_RE = re.compile(
|
||
r"(?:https?://)?(?:www\.)?arxiv\.org/(?:abs|pdf)/(\d{4}\.\d{4,5})(?:v\d+)?(?:\.pdf)?",
|
||
re.IGNORECASE,
|
||
)
|
||
_ARXIV_BARE_RE = re.compile(r"\barXiv:\s*(\d{4}\.\d{4,5})", re.IGNORECASE)
|
||
# DOI: doi.org/10.xxxx/... 或裸 10.xxxx/...（论文里的 DOI 形式）
|
||
_DOI_URL_RE = re.compile(
|
||
r"(?:https?://)?(?:dx\.)?doi\.org/(10\.\d{4,9}/[^\s)\"'<>]+)", re.IGNORECASE,
|
||
)
|
||
_DOI_BARE_RE = re.compile(
|
||
r"\b(10\.\d{4,9}/[^\s)\"'<>]+)", re.IGNORECASE,
|
||
)
|
||
|
||
# 在 DOI 字符串里清掉常见尾部分隔符（避免吃进句号、逗号）
|
||
_TRAILING_PUNCT = ".,;:)\"'>]"
|
||
|
||
|
||
def _clean_doi(doi: str) -> str:
|
||
return doi.rstrip(_TRAILING_PUNCT)
|
||
|
||
|
||
def _snippet(text: str, pos: int, span: int = 60) -> str:
|
||
"""以匹配位置为中心截一段上下文。"""
|
||
a = max(0, pos - span // 2)
|
||
b = min(len(text), pos + span // 2)
|
||
frag = text[a:b].replace("\n", " ").strip()
|
||
return ("…" + frag) if a > 0 else frag
|
||
|
||
|
||
def extract_paper_links(text: str) -> list[dict[str, str]]:
|
||
"""从文本里抽取 arXiv 与 DOI 论文引用链接。
|
||
|
||
返回 ``[{"source_text_snippet": str, "target": url, "type": "arxiv"|"doi"}]``，
|
||
按 (出现位置, 类型优先 arxiv) 排序，去重（同一 arxiv id / doi 只保留首次）。
|
||
"""
|
||
if not text:
|
||
return []
|
||
out: list[dict[str, str]] = []
|
||
seen_arxiv: set[str] = set()
|
||
seen_doi: set[str] = set()
|
||
hits: list[tuple[int, dict[str, str]]] = []
|
||
|
||
for m in _ARXIV_RE.finditer(text):
|
||
aid = m.group(1)
|
||
if aid in seen_arxiv:
|
||
continue
|
||
seen_arxiv.add(aid)
|
||
hits.append((m.start(), {
|
||
"source_text_snippet": _snippet(text, m.start()),
|
||
"target": f"https://arxiv.org/abs/{aid}",
|
||
"type": "arxiv",
|
||
}))
|
||
for m in _ARXIV_BARE_RE.finditer(text):
|
||
aid = m.group(1)
|
||
if aid in seen_arxiv:
|
||
continue
|
||
seen_arxiv.add(aid)
|
||
hits.append((m.start(), {
|
||
"source_text_snippet": _snippet(text, m.start()),
|
||
"target": f"https://arxiv.org/abs/{aid}",
|
||
"type": "arxiv",
|
||
}))
|
||
for m in _DOI_URL_RE.finditer(text):
|
||
doi = _clean_doi(m.group(1))
|
||
if doi.lower() in seen_doi:
|
||
continue
|
||
seen_doi.add(doi.lower())
|
||
hits.append((m.start(), {
|
||
"source_text_snippet": _snippet(text, m.start()),
|
||
"target": f"https://doi.org/{doi}",
|
||
"type": "doi",
|
||
}))
|
||
for m in _DOI_BARE_RE.finditer(text):
|
||
doi = _clean_doi(m.group(1))
|
||
if doi.lower() in seen_doi:
|
||
continue
|
||
seen_doi.add(doi.lower())
|
||
hits.append((m.start(), {
|
||
"source_text_snippet": _snippet(text, m.start()),
|
||
"target": f"https://doi.org/{doi}",
|
||
"type": "doi",
|
||
}))
|
||
|
||
hits.sort(key=lambda x: (x[0], 0 if x[1]["type"] == "arxiv" else 1))
|
||
return [h[1] for h in hits]
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 算法 4：仓库产物分类
|
||
# ---------------------------------------------------------------------------
|
||
|
||
def classify_artifacts(tree_entries: list) -> list[dict[str, Any]]:
|
||
"""按路径把仓库文件/目录归入科研产物类别。
|
||
|
||
规则（按优先级，先匹配先归类）：
|
||
- path 含 ``benchmark/`` 段 → benchmark
|
||
- path 含 ``model/`` 段或 *.ckpt/*.safetensors/*.onnx → model
|
||
- path 含 ``data/`` 段或 *.csv/*.parquet → dataset
|
||
- *.pdf / *.ipynb / paper 关键词 → paper
|
||
|
||
每个产物 ``{"path", "category", "name"}``。tree_entries 既可能是文件列表
|
||
（含 path/name/type）也可能是目录项；本函数尽力取 path/name 字段。
|
||
"""
|
||
out: list[dict[str, Any]] = []
|
||
seen: set[str] = set()
|
||
for entry in tree_entries or []:
|
||
if not isinstance(entry, dict):
|
||
continue
|
||
path = entry.get("path") or entry.get("name") or ""
|
||
if not path:
|
||
continue
|
||
norm = path.replace("\\", "/").lower()
|
||
name = norm.rsplit("/", 1)[-1]
|
||
category = None
|
||
# benchmark（必须含 benchmark 目录段，避免误把文件名含词的归入）
|
||
if "/benchmark/" in norm or norm.startswith("benchmark/"):
|
||
category = "benchmark"
|
||
elif "/model/" in norm or norm.startswith("model/") or name.endswith(
|
||
(".ckpt", ".safetensors", ".onnx", ".pb", ".h5", ".pt")):
|
||
category = "model"
|
||
elif "/data/" in norm or norm.startswith("data/") or name.endswith(
|
||
(".csv", ".parquet", ".npy", ".npz", ".hdf5", ".h5")):
|
||
# .h5 已先被 model 吃掉，这里主要 csv/parquet/npy
|
||
category = "dataset"
|
||
elif (name.endswith(".pdf") or name.endswith(".ipynb")
|
||
or "paper" in norm or "arxiv" in norm):
|
||
category = "paper"
|
||
if category and path not in seen:
|
||
seen.add(path)
|
||
out.append({"path": path, "category": category,
|
||
"name": name or path.rsplit("/", 1)[-1]})
|
||
return out
|
||
|
||
|
||
def artifact_summary(artifacts: list[dict[str, Any]]) -> dict[str, int]:
|
||
"""统计各类产物数量，返回 {paper: n, dataset: n, model: n, benchmark: n}。"""
|
||
summary: dict[str, int] = {"paper": 0, "dataset": 0, "model": 0, "benchmark": 0}
|
||
for a in artifacts or []:
|
||
cat = a.get("category")
|
||
if cat in summary:
|
||
summary[cat] += 1
|
||
return summary
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 数据采集（取数层，主流程调用；单测不触达）
|
||
# ---------------------------------------------------------------------------
|
||
|
||
def collect(owner: str, repo: str, weeks: int) -> dict[str, Any]:
|
||
"""从 GitLink 取本场景所需的全部数据。"""
|
||
# commits 取够 ~weeks 周（每周按 30 条粗估，上限 max_pages=10）
|
||
cm_pages = max(2, min(10, (weeks // 3) + 1))
|
||
commits = c.commits(owner, repo, ref="master", max_pages=cm_pages, page_size=100)
|
||
issues = c.issues_all(owner, repo, max_pages=10, page_size=50)
|
||
pullreqs = c.prs_all(owner, repo, max_pages=10, page_size=50)
|
||
milestones = c.milestones(owner, repo, state="all")
|
||
langs = c.languages(owner, repo)
|
||
contribs = c.contributors(owner, repo)
|
||
readme = c.readme(owner, repo)
|
||
tree = c.tree(owner, repo)
|
||
return {
|
||
"info": c.repo_info(owner, repo),
|
||
"commits": commits,
|
||
"issues": issues,
|
||
"prs": pullreqs,
|
||
"milestones": milestones,
|
||
"languages": langs,
|
||
"contributors": contribs,
|
||
"readme": readme,
|
||
"tree": tree,
|
||
}
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 主算法：组装结果 dict
|
||
# ---------------------------------------------------------------------------
|
||
|
||
def run(owner: str, repo: str, weeks: int, raw: dict[str, Any] | None = None) -> dict[str, Any]:
|
||
"""主入口：取数（或复用传入的 raw）→ 算法 → 结果 dict。"""
|
||
if raw is None:
|
||
raw = collect(owner, repo, weeks)
|
||
|
||
commits = raw.get("commits") or []
|
||
issues = raw.get("issues") or []
|
||
prs = raw.get("prs") or []
|
||
milestones = raw.get("milestones") or []
|
||
langs = raw.get("languages") or {}
|
||
contribs = raw.get("contributors") or []
|
||
readme = raw.get("readme") or ""
|
||
tree = raw.get("tree") or []
|
||
|
||
commits_ts = bin_weekly(commits, weeks, _commit_time)
|
||
issues_ts = bin_weekly(issues, weeks, _issue_time)
|
||
prs_ts = bin_weekly(prs, weeks, _pr_time)
|
||
heatmap = contribution_heatmap(contribs, commits, weeks)
|
||
|
||
# 合并 readme + 提交信息作为论文链接抽取语料
|
||
corpus_parts = [readme]
|
||
for cm in commits[:50]:
|
||
msg = cm.get("message") or ""
|
||
if isinstance(msg, str):
|
||
corpus_parts.append(msg)
|
||
paper_links = extract_paper_links("\n".join(corpus_parts))
|
||
|
||
artifacts = classify_artifacts(tree)
|
||
art_summary = artifact_summary(artifacts)
|
||
|
||
# 里程碑甘特数据：取有 due_on 的，转成 [start, end, title]
|
||
gantt: list[dict[str, Any]] = []
|
||
for ms in milestones:
|
||
if not isinstance(ms, dict):
|
||
continue
|
||
title = ms.get("name") or ms.get("title") or ""
|
||
due = _to_timestamp(ms.get("due_on") or ms.get("effective_date"))
|
||
start = _to_timestamp(ms.get("start_date"))
|
||
if due > 0:
|
||
gantt.append({
|
||
"title": title,
|
||
"start": start if start > 0 else due - 14 * 86400,
|
||
"due": due,
|
||
})
|
||
|
||
return {
|
||
"scenario": "S6_research_visualization",
|
||
"repo": f"{owner}/{repo}",
|
||
"weeks": weeks,
|
||
"timeline": {
|
||
"labels": commits_ts["labels"],
|
||
"commits": commits_ts["counts"],
|
||
"issues": issues_ts["counts"],
|
||
"prs": prs_ts["counts"],
|
||
},
|
||
"heatmap": heatmap,
|
||
"languages": langs,
|
||
"milestones": gantt,
|
||
"paper_links": paper_links,
|
||
"artifacts": artifacts,
|
||
"artifact_summary": art_summary,
|
||
"meta": {
|
||
"commit_count": len(commits),
|
||
"issue_count": len(issues),
|
||
"pr_count": len(prs),
|
||
"milestone_count": len(milestones),
|
||
"contributor_count": len(contribs),
|
||
},
|
||
}
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 渲染：Markdown 摘要报告
|
||
# ---------------------------------------------------------------------------
|
||
|
||
def render_report(result: dict[str, Any]) -> str:
|
||
repo = result["repo"]
|
||
tl = result["timeline"]
|
||
weeks = result["weeks"]
|
||
meta = result["meta"]
|
||
art = result["artifact_summary"]
|
||
lines = [
|
||
f"# 科研成果可视化沉淀报告 — {repo}\n",
|
||
f"> 场景 S6 · 子赛题四「应用 GitLink 辅助科研」\n",
|
||
f"## 一、活跃度概览（最近 {weeks} 周）\n",
|
||
f"- 提交数: **{meta['commit_count']}**（窗口内峰值 "
|
||
f"{max(tl['commits']) if tl['commits'] else 0} 提交/周）",
|
||
f"- 新增 Issue: **{meta['issue_count']}**，新增 PR: **{meta['pr_count']}**",
|
||
f"- 贡献者: **{meta['contributor_count']}**，里程碑: **{meta['milestone_count']}**\n",
|
||
"## 二、开发节奏（最近 8 周快照）\n",
|
||
"| 周 | commits | issues | prs |",
|
||
"|----|---------|--------|-----|",
|
||
]
|
||
tail = tl["labels"][-8:]
|
||
for i, label in enumerate(tail):
|
||
idx = len(tl["labels"]) - len(tail) + i
|
||
lines.append(f"| {label} | {tl['commits'][idx]} | {tl['issues'][idx]} | {tl['prs'][idx]} |")
|
||
|
||
lines += ["\n## 三、核心贡献者热力（贡献者 × 周提交数）\n",
|
||
"| 贡献者 | 窗口内提交 |",
|
||
"|--------|-----------|"]
|
||
hm = result["heatmap"]
|
||
for i, user in enumerate(hm["users"][:10]):
|
||
total = sum(hm["matrix"][i])
|
||
lines.append(f"| `{user}` | {total} |")
|
||
|
||
lines += ["\n## 四、科研产物分类\n",
|
||
f"- 论文/笔记 (paper): **{art['paper']}**",
|
||
f"- 数据集 (dataset): **{art['dataset']}**",
|
||
f"- 模型 (model): **{art['model']}**",
|
||
f"- 基准 (benchmark): **{art['benchmark']}**\n"]
|
||
if result["paper_links"]:
|
||
lines += ["## 五、抽取到的论文引用\n",
|
||
"| 类型 | 链接 |",
|
||
"|------|------|"]
|
||
for p in result["paper_links"][:15]:
|
||
lines.append(f"| {p['type']} | {p['target']} |")
|
||
else:
|
||
lines.append("## 五、抽取到的论文引用\n\n_未在 README/提交信息中发现 arXiv 或 DOI 引用_\n")
|
||
|
||
lines.append(f"\n_交互可视化见 visual.html（或原始数据 visual.json）_\n")
|
||
return "\n".join(lines)
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 渲染：交互 HTML（plotly 多子图）+ JSON bundle
|
||
# ---------------------------------------------------------------------------
|
||
|
||
def render_html(result: dict[str, Any]) -> str | None:
|
||
"""构建单一交互 HTML（多子图）。无 plotly 时返回 None（调用方降级）。"""
|
||
try:
|
||
import plotly.graph_objects as go
|
||
from plotly.subplots import make_subplots
|
||
except ImportError:
|
||
return None
|
||
|
||
tl = result["timeline"]
|
||
hm = result["heatmap"]
|
||
langs = result["languages"] or {}
|
||
|
||
# 行布局：timeline(跨整行) | heatmap | pie | gantt
|
||
fig = make_subplots(
|
||
rows=4, cols=2,
|
||
specs=[
|
||
[{"colspan": 2}, None],
|
||
[{"colspan": 2}, None],
|
||
[{"type": "domain"}, {"type": "scatter"}],
|
||
[{"colspan": 2}, None],
|
||
],
|
||
row_heights=[0.25, 0.30, 0.25, 0.20],
|
||
vertical_spacing=0.08,
|
||
subplot_titles=("开发时间线（周粒度）", "贡献者 × 周提交热力图",
|
||
"语言占比", "里程碑甘特（due_date）"),
|
||
)
|
||
|
||
# 1) 时间线折线
|
||
fig.add_trace(go.Scatter(x=tl["labels"], y=tl["commits"], name="commits",
|
||
mode="lines+markers", line=dict(color="#636efa")), row=1, col=1)
|
||
fig.add_trace(go.Scatter(x=tl["labels"], y=tl["issues"], name="issues",
|
||
mode="lines+markers", line=dict(color="#ef553b")), row=1, col=1)
|
||
fig.add_trace(go.Scatter(x=tl["labels"], y=tl["prs"], name="prs",
|
||
mode="lines+markers", line=dict(color="#00cc96")), row=1, col=1)
|
||
|
||
# 2) 贡献热力图
|
||
if hm["users"]:
|
||
fig.add_trace(go.Heatmap(
|
||
z=hm["matrix"], x=hm["weeks"], y=hm["users"],
|
||
colorscale="Blues", name="contributions",
|
||
colorbar=dict(title="提交数", len=0.25, y=0.78),
|
||
), row=2, col=1)
|
||
|
||
# 3a) 语言饼图
|
||
if langs:
|
||
labels = list(langs.keys())
|
||
values = []
|
||
for v in langs.values():
|
||
# GitLink languages 形如 {"Python":"99.7%"}；剥离 %
|
||
s = str(v).strip().rstrip("%")
|
||
try:
|
||
values.append(float(s))
|
||
except ValueError:
|
||
values.append(0.0)
|
||
fig.add_trace(go.Pie(labels=labels, values=values, name="languages",
|
||
textinfo="label+percent"), row=3, col=1)
|
||
|
||
# 3b) 里程碑甘特（用散点的水平线段近似）
|
||
gantt = result["milestones"]
|
||
for g in gantt[:20]:
|
||
title = g["title"] or "(milestone)"
|
||
start = g["start"]
|
||
due = g["due"]
|
||
fig.add_trace(go.Scatter(
|
||
x=[start, due], y=[title, title],
|
||
mode="lines+markers",
|
||
line=dict(color="#ffa15a", width=6),
|
||
marker=dict(size=8),
|
||
showlegend=False, hovertemplate=f"{title}<br>%{{x}}",
|
||
), row=3, col=2)
|
||
|
||
fig.update_layout(
|
||
title=f"科研成果可视化 — {result['repo']}（最近 {result['weeks']} 周）",
|
||
height=1100, width=1100,
|
||
legend=dict(orientation="h", y=1.02),
|
||
margin=dict(l=40, r=40, t=80, b=40),
|
||
)
|
||
fig.update_xaxes(row=1, col=1, tickangle=-45)
|
||
|
||
return fig.to_html(full_html=True, include_plotlyjs="cdn",
|
||
default_width="100%", default_height="1100px")
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 主入口
|
||
# ---------------------------------------------------------------------------
|
||
|
||
def main():
|
||
ap = argparse.ArgumentParser(description="S6 科研成果可视化沉淀")
|
||
ap.add_argument("--owner", required=True)
|
||
ap.add_argument("--repo", required=True)
|
||
ap.add_argument("--weeks", type=int, default=26, help="回溯多少周（默认 26）")
|
||
ap.add_argument("--out", help="输出目录（写 visual.html + visual.json + report.md）；"
|
||
"省略则打印 JSON")
|
||
args = ap.parse_args()
|
||
|
||
result = run(args.owner, args.repo, args.weeks)
|
||
|
||
if args.out:
|
||
os.makedirs(args.out, exist_ok=True)
|
||
# 原始数据 bundle（供前端二次开发）
|
||
with open(os.path.join(args.out, "visual.json"), "w", encoding="utf-8") as f:
|
||
json.dump(result, f, ensure_ascii=False, indent=2)
|
||
with open(os.path.join(args.out, "report.md"), "w", encoding="utf-8") as f:
|
||
f.write(render_report(result))
|
||
|
||
html = render_html(result)
|
||
if html is not None:
|
||
with open(os.path.join(args.out, "visual.html"), "w", encoding="utf-8") as f:
|
||
f.write(html)
|
||
print(f"✓ S6 可视化完成 → {args.out}/visual.html | visual.json | report.md")
|
||
else:
|
||
print(f"✓ S6 可视化完成（无 plotly，已降级）→ {args.out}/visual.json | report.md")
|
||
print(" 提示：pip install plotly 后可生成交互 HTML")
|
||
meta = result["meta"]
|
||
art = result["artifact_summary"]
|
||
print(f" commits={meta['commit_count']} issues={meta['issue_count']} "
|
||
f"prs={meta['pr_count']} 贡献者={meta['contributor_count']}")
|
||
print(f" 产物 paper={art['paper']} dataset={art['dataset']} "
|
||
f"model={art['model']} benchmark={art['benchmark']}")
|
||
print(f" 论文引用: {len(result['paper_links'])} 条")
|
||
else:
|
||
print(json.dumps(result, ensure_ascii=False, indent=2))
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|