diff options
Diffstat (limited to '')
| -rw-r--r-- | .agents/skills/obsidian-literature-workflow/scripts/build_literature_canvas.py | 644 | ||||
| -rwxr-xr-x | .agents/skills/obsidian-literature-workflow/scripts/build_literature_graph.py | 271 |
2 files changed, 915 insertions, 0 deletions
diff --git a/.agents/skills/obsidian-literature-workflow/scripts/build_literature_canvas.py b/.agents/skills/obsidian-literature-workflow/scripts/build_literature_canvas.py new file mode 100644 index 0000000..c1fc27d --- /dev/null +++ b/.agents/skills/obsidian-literature-workflow/scripts/build_literature_canvas.py @@ -0,0 +1,644 @@ +#!/usr/bin/env python3 +from __future__ import annotations + +import argparse +import hashlib +import json +import re +import sys +from dataclasses import dataclass +from pathlib import Path +from typing import Any + +WIKILINK_RE = re.compile(r"\[\[([^\]|#]+)(?:#[^\]|]+)?(?:\|[^\]]+)?\]\]") +LIST_KEYS = { + "authors", + "keywords", + "concepts", + "methods", + "related_papers", + "linked_knowledge", + "paper_relationships", + "linked_papers", + "argument_claims", + "argument_methods", + "argument_gaps", + "linked_claims", + "linked_methods", + "linked_gaps", + "claim_method_links", + "method_gap_links", +} +PAPER_LIMIT = 48 +KNOWLEDGE_LIMIT = 18 +KNOWLEDGE_COLUMNS = 3 +NODE_WIDTH = 300 +NODE_HEIGHT = 180 +ARG_NODE_WIDTH = 280 +ARG_NODE_HEIGHT = 120 +X_GAP = 80 +Y_GAP = 70 +GROUP_X_GAP = 180 +GROUP_Y_GAP = 180 +ARG_GROUP_X_GAP = 200 +PAPER_RELATION_LIMIT = 2 +PAPER_CLAIM_LIMIT = 1 +PAPER_METHOD_LIMIT = 1 +PAPER_RELATION_LIMIT = 1 +SUBFIELD_ORDER = ( + "speech-transfer-constraints", + "alignment-and-domain-adaptation", + "geometry-and-representation", + "subject-aware-adaptation", + "other", +) +SUBFIELD_LABELS = { + "speech-transfer-constraints": "Speech-specific transfer constraints", + "alignment-and-domain-adaptation": "Alignment and domain adaptation", + "geometry-and-representation": "Geometry and representation", + "subject-aware-adaptation": "Subject-aware adaptation", + "other": "Other papers", +} +SUBFIELD_COLORS = { + "speech-transfer-constraints": "6", + "alignment-and-domain-adaptation": "2", + "geometry-and-representation": "5", + "subject-aware-adaptation": "3", + "other": "1", +} +SUBFIELD_COLUMNS = 2 + + +@dataclass(frozen=True) +class NoteRecord: + title: str + note_relpath: str + vault_relpath: str + file_name: str + related_papers: tuple[str, ...] + linked_knowledge: tuple[str, ...] + paper_relationships: tuple[tuple[str, str], ...] + linked_papers: tuple[str, ...] + claim_refs: tuple[str, ...] + method_refs: tuple[str, ...] + gap_refs: tuple[str, ...] + claim_method_links: tuple[tuple[str, str, str], ...] + method_gap_links: tuple[tuple[str, str, str], ...] + wikilinks: tuple[str, ...] + subfield: str + canvas_visibility: str + + +def load_project_kb_module() -> Any: + script_dir = Path(__file__).resolve().parents[2] / "obsidian-project-memory" / "scripts" + sys.path.insert(0, str(script_dir)) + import project_kb # type: ignore + + return project_kb + + +def parse_args() -> argparse.Namespace: + parser = argparse.ArgumentParser( + description="Build a default literature canvas for the bound Obsidian project." + ) + parser.add_argument("--cwd", default=".") + parser.add_argument("--project-id", default="") + parser.add_argument("--output", default="Maps/literature.canvas") + return parser.parse_args() + + +def parse_frontmatter(text: str) -> dict[str, Any]: + if not text.startswith("---\n"): + return {} + end = text.find("\n---\n", 4) + if end == -1: + return {} + lines = text[4:end].splitlines() + data: dict[str, Any] = {} + current_key: str | None = None + for raw_line in lines: + if not raw_line.strip(): + continue + if raw_line.startswith(" - ") or raw_line.startswith("- "): + if current_key and current_key in LIST_KEYS: + data.setdefault(current_key, []).append( + raw_line.split("- ", 1)[1].strip().strip('"') + ) + continue + if ":" not in raw_line: + current_key = None + continue + key, value = raw_line.split(":", 1) + key = key.strip() + value = value.strip() + current_key = key + if not value: + if key in LIST_KEYS: + data[key] = [] + else: + data[key] = "" + continue + data[key] = value.strip('"') + return data + + +def body_without_frontmatter(text: str) -> str: + if not text.startswith("---\n"): + return text + end = text.find("\n---\n", 4) + if end == -1: + return text + return text[end + 5 :] + + +def normalize_note_target(value: str) -> str: + cleaned = value.strip() + if cleaned.endswith(".md"): + cleaned = cleaned[:-3] + if cleaned.startswith("./"): + cleaned = cleaned[2:] + return cleaned + + +def parse_relationship_entries(values: list[str]) -> tuple[tuple[str, str], ...]: + parsed: list[tuple[str, str]] = [] + for raw in values: + cleaned = raw.strip() + if not cleaned: + continue + if "::" in cleaned: + target, label = cleaned.split("::", 1) + else: + target, label = cleaned, "related" + parsed.append((normalize_note_target(target), label.strip() or "related")) + return tuple(parsed) + + +def parse_triple_entries(values: list[str]) -> tuple[tuple[str, str, str], ...]: + parsed: list[tuple[str, str, str]] = [] + for raw in values: + cleaned = raw.strip() + if not cleaned: + continue + parts = [part.strip() for part in cleaned.split("::")] + if len(parts) == 2: + parsed.append((parts[0], parts[1], "relates")) + elif len(parts) >= 3: + parsed.append((parts[0], parts[1], parts[2] or "relates")) + return tuple(parsed) + + +def parse_plain_labels(values: list[str]) -> tuple[str, ...]: + labels = [] + for raw in values: + cleaned = raw.strip() + if cleaned and cleaned not in labels: + labels.append(cleaned) + return tuple(labels) + + +def extract_wikilinks(text: str) -> tuple[str, ...]: + seen: list[str] = [] + for match in WIKILINK_RE.findall(text): + target = normalize_note_target(match) + if target not in seen: + seen.append(target) + return tuple(seen) + + +def note_id(prefix: str, value: str) -> str: + digest = hashlib.sha1(f"{prefix}:{value}".encode("utf-8")).hexdigest() + return digest[:16] + + +def file_node(note: NoteRecord, x: int, y: int, color: str) -> dict[str, Any]: + return { + "id": note_id("node", note.note_relpath), + "type": "file", + "x": x, + "y": y, + "width": NODE_WIDTH, + "height": NODE_HEIGHT, + "file": note.vault_relpath, + "color": color, + } + + +def text_node(node_key: str, text: str, x: int, y: int, width: int, height: int) -> dict[str, Any]: + return { + "id": note_id("text", node_key), + "type": "text", + "x": x, + "y": y, + "width": width, + "height": height, + "text": text, + "color": "5", + } + + +def group_node(node_key: str, label: str, x: int, y: int, width: int, height: int, color: str) -> dict[str, Any]: + return { + "id": note_id("group", node_key), + "type": "group", + "x": x, + "y": y, + "width": width, + "height": height, + "label": label, + "color": color, + } + + +def edge_id(source: str, target: str, label: str) -> str: + digest = hashlib.sha1(f"{source}:{target}:{label}".encode("utf-8")).hexdigest() + return digest[:16] + + +def make_edge(from_node: str, to_node: str, label: str) -> dict[str, Any]: + return { + "id": edge_id(from_node, to_node, label), + "fromNode": from_node, + "fromSide": "bottom", + "toNode": to_node, + "toSide": "top", + "toEnd": "arrow", + "label": label, + } + + +def collect_notes(folder: Path, folder_name: str, vault_root: Path, limit: int) -> list[NoteRecord]: + notes: list[NoteRecord] = [] + if not folder.exists(): + return notes + for path in sorted(folder.glob("*.md"))[:limit]: + text = path.read_text(encoding="utf-8") + frontmatter = parse_frontmatter(text) + body = body_without_frontmatter(text) + title = str(frontmatter.get("title") or path.stem.replace("-", " ")) + notes.append( + NoteRecord( + title=title, + note_relpath=f"{folder_name}/{path.name}", + vault_relpath=str(path.relative_to(vault_root)).replace("\\", "/"), + file_name=path.name, + related_papers=tuple( + normalize_note_target(v) for v in frontmatter.get("related_papers", []) + ), + linked_knowledge=tuple( + normalize_note_target(v) for v in frontmatter.get("linked_knowledge", []) + ), + paper_relationships=parse_relationship_entries( + list(frontmatter.get("paper_relationships", [])) + ), + linked_papers=tuple( + normalize_note_target(v) for v in frontmatter.get("linked_papers", []) + ), + claim_refs=parse_plain_labels( + list(frontmatter.get("argument_claims", [])) + + list(frontmatter.get("linked_claims", [])) + ), + method_refs=parse_plain_labels( + list(frontmatter.get("argument_methods", [])) + + list(frontmatter.get("linked_methods", [])) + ), + gap_refs=parse_plain_labels( + list(frontmatter.get("argument_gaps", [])) + + list(frontmatter.get("linked_gaps", [])) + ), + claim_method_links=parse_triple_entries( + list(frontmatter.get("claim_method_links", [])) + ), + method_gap_links=parse_triple_entries( + list(frontmatter.get("method_gap_links", [])) + ), + wikilinks=extract_wikilinks(body), + subfield=str(frontmatter.get("subfield") or "other"), + canvas_visibility=str(frontmatter.get("canvas_visibility") or "show"), + ) + ) + return notes + + +def looks_like_literature_knowledge(note: NoteRecord) -> bool: + key = note.note_relpath.lower() + if any(token in key for token in ("literature", "paper", "method", "survey", "gap", "related")): + return True + return any(target.startswith("Papers/") for target in note.wikilinks) + + +def select_knowledge_notes( + all_knowledge: list[NoteRecord], papers: list[NoteRecord] +) -> list[NoteRecord]: + requested = set() + for note in papers: + requested.update(note.linked_knowledge) + requested.update(target for target in note.wikilinks if target.startswith("Knowledge/")) + + selected: list[NoteRecord] = [] + seen: set[str] = set() + for note in all_knowledge: + note_key = note.note_relpath[:-3] + if note_key in requested: + selected.append(note) + seen.add(note_key) + + if selected: + return selected[:KNOWLEDGE_LIMIT] + + for note in all_knowledge: + note_key = note.note_relpath[:-3] + if note_key in seen: + continue + if looks_like_literature_knowledge(note): + selected.append(note) + seen.add(note_key) + if len(selected) >= KNOWLEDGE_LIMIT: + break + + return selected[:KNOWLEDGE_LIMIT] + + +def layout_grid(index: int, columns: int, x0: int, y0: int) -> tuple[int, int]: + col = index % columns + row = index // columns + x = x0 + col * (NODE_WIDTH + X_GAP) + y = y0 + row * (NODE_HEIGHT + Y_GAP) + return x, y + + +def compute_group_dimensions(item_count: int, columns: int) -> tuple[int, int]: + rows = max(1, (max(item_count, 1) - 1) // columns + 1) + width = columns * NODE_WIDTH + (columns - 1) * X_GAP + 120 + height = rows * NODE_HEIGHT + (rows - 1) * Y_GAP + 140 + return width, height + + +def compute_argument_group_dimensions(item_count: int, columns: int) -> tuple[int, int]: + rows = max(1, (max(item_count, 1) - 1) // columns + 1) + width = columns * ARG_NODE_WIDTH + (columns - 1) * X_GAP + 120 + height = rows * ARG_NODE_HEIGHT + (rows - 1) * Y_GAP + 140 + return width, height + + +def argument_node(kind: str, label: str, x: int, y: int) -> dict[str, Any]: + return { + "id": note_id(kind, label), + "type": "text", + "x": x, + "y": y, + "width": ARG_NODE_WIDTH, + "height": ARG_NODE_HEIGHT, + "text": label, + "color": {"claim": "4", "method": "2", "gap": "6"}.get(kind, "5"), + } + + +def render_canvas(project_title: str, papers: list[NoteRecord], knowledge: list[NoteRecord]) -> dict[str, Any]: + nodes: list[dict[str, Any]] = [] + edges: list[dict[str, Any]] = [] + node_lookup: dict[str, str] = {} + claim_lookup: dict[str, str] = {} + method_lookup: dict[str, str] = {} + gap_lookup: dict[str, str] = {} + + knowledge_rows = max(1, (max(len(knowledge), 1) - 1) // KNOWLEDGE_COLUMNS + 1) + knowledge_group_width = KNOWLEDGE_COLUMNS * NODE_WIDTH + (KNOWLEDGE_COLUMNS - 1) * X_GAP + 120 + knowledge_group_height = knowledge_rows * NODE_HEIGHT + (knowledge_rows - 1) * Y_GAP + 140 + knowledge_y = 220 + argument_y = knowledge_y + knowledge_group_height + 180 + + nodes.append( + text_node( + "literature-summary", + ( + f"# Literature Canvas\\n\\n" + f"Project: {project_title}\\n\\n" + f"- This is an argument map with `paper + claim + method + gap`.\\n" + f"- The graph is intentionally thinned: each paper only keeps its most important claim / method links.\\n" + f"- Papers are grouped by subfield and spaced out to reduce crossing lines." + ), + 0, + 0, + 1080, + 150, + ) + ) + nodes.append(group_node("knowledge-group", "Knowledge", -40, knowledge_y - 40, knowledge_group_width, knowledge_group_height, "4")) + + for idx, note in enumerate(knowledge): + x, y = layout_grid(idx, KNOWLEDGE_COLUMNS, 20, knowledge_y) + node = file_node(note, x, y, "4") + nodes.append(node) + node_lookup[note.note_relpath[:-3]] = node["id"] + + claim_labels: list[str] = [] + method_labels: list[str] = [] + gap_labels: list[str] = [] + for note in (*papers, *knowledge): + for label in note.claim_refs: + if label not in claim_labels: + claim_labels.append(label) + for label in note.method_refs: + if label not in method_labels: + method_labels.append(label) + for label in note.gap_refs: + if label not in gap_labels: + gap_labels.append(label) + + claim_group_width, claim_group_height = compute_argument_group_dimensions(len(claim_labels), 2) + method_group_width, method_group_height = compute_argument_group_dimensions(len(method_labels), 2) + gap_group_width, gap_group_height = compute_argument_group_dimensions(len(gap_labels), 2) + argument_row_height = max(claim_group_height, method_group_height, gap_group_height) + + argument_groups = [ + ("claim", "Claims", claim_labels, -40, argument_y, claim_group_width, claim_group_height, "4"), + ("method", "Methods", method_labels, -40 + claim_group_width + ARG_GROUP_X_GAP, argument_y, method_group_width, method_group_height, "2"), + ("gap", "Gaps", gap_labels, -40 + claim_group_width + ARG_GROUP_X_GAP + method_group_width + ARG_GROUP_X_GAP, argument_y, gap_group_width, gap_group_height, "6"), + ] + for kind, label, labels, group_x, group_y, group_width, group_height, color in argument_groups: + nodes.append(group_node(f"{kind}-group", label, group_x, group_y, group_width, group_height, color)) + for idx, arg_label in enumerate(labels): + x, y = layout_grid(idx, 2, group_x + 60, group_y + 40) + node = argument_node(kind, arg_label, x, y) + nodes.append(node) + if kind == "claim": + claim_lookup[arg_label] = node["id"] + elif kind == "method": + method_lookup[arg_label] = node["id"] + else: + gap_lookup[arg_label] = node["id"] + + paper_y = argument_y + argument_row_height + 180 + + grouped_papers: dict[str, list[NoteRecord]] = {} + for note in papers: + grouped_papers.setdefault(note.subfield or "other", []).append(note) + + active_groups = [key for key in SUBFIELD_ORDER if grouped_papers.get(key)] + if not active_groups: + active_groups = ["other"] + grouped_papers["other"] = list(papers) + + group_layout_meta: list[tuple[str, int, int, int, int]] = [] + for idx, subfield in enumerate(active_groups): + notes_in_group = grouped_papers[subfield] + group_width, group_height = compute_group_dimensions(len(notes_in_group), 2) + col = idx % SUBFIELD_COLUMNS + row = idx // SUBFIELD_COLUMNS + group_x = -40 + col * (group_width + GROUP_X_GAP) + group_y = paper_y + row * (group_height + GROUP_Y_GAP) + group_layout_meta.append((subfield, group_x, group_y, group_width, group_height)) + nodes.append( + group_node( + f"paper-group-{subfield}", + SUBFIELD_LABELS.get(subfield, subfield.replace("-", " ").title()), + group_x, + group_y, + group_width, + group_height, + SUBFIELD_COLORS.get(subfield, "2"), + ) + ) + + for note_idx, note in enumerate(notes_in_group): + x, y = layout_grid(note_idx, 2, group_x + 60, group_y + 40) + node = file_node(note, x, y, SUBFIELD_COLORS.get(subfield, "2")) + nodes.append(node) + node_lookup[note.note_relpath[:-3]] = node["id"] + + seen_edges: set[tuple[str, str, str]] = set() + for note in papers: + source = node_lookup.get(note.note_relpath[:-3]) + if not source: + continue + explicit_relationships = list(note.paper_relationships) + if not explicit_relationships: + explicit_relationships = [(target, "related") for target in note.related_papers] + for target, label in explicit_relationships[:PAPER_RELATION_LIMIT]: + target_node = node_lookup.get(target) + if not target_node: + continue + marker = (source, target_node, label) + if marker not in seen_edges: + edges.append(make_edge(source, target_node, label)) + seen_edges.add(marker) + for claim in note.claim_refs[:PAPER_CLAIM_LIMIT]: + target_node = claim_lookup.get(claim) + if not target_node: + continue + marker = (source, target_node, "supports") + if marker not in seen_edges: + edges.append(make_edge(source, target_node, "supports")) + seen_edges.add(marker) + for method in note.method_refs[:PAPER_METHOD_LIMIT]: + target_node = method_lookup.get(method) + if not target_node: + continue + marker = (source, target_node, "uses") + if marker not in seen_edges: + edges.append(make_edge(source, target_node, "uses")) + seen_edges.add(marker) + for note in knowledge: + source = node_lookup.get(note.note_relpath[:-3]) + if not source: + continue + for target in note.linked_papers: + target_node = node_lookup.get(target) + if not target_node: + continue + marker = (source, target_node, "summarizes") + if marker not in seen_edges: + edges.append(make_edge(source, target_node, "summarizes")) + seen_edges.add(marker) + for claim in note.claim_refs: + target_node = claim_lookup.get(claim) + if not target_node: + continue + marker = (source, target_node, "summarizes") + if marker not in seen_edges: + edges.append(make_edge(source, target_node, "summarizes")) + seen_edges.add(marker) + for method in note.method_refs: + target_node = method_lookup.get(method) + if not target_node: + continue + marker = (source, target_node, "summarizes") + if marker not in seen_edges: + edges.append(make_edge(source, target_node, "summarizes")) + seen_edges.add(marker) + for gap in note.gap_refs: + target_node = gap_lookup.get(gap) + if not target_node: + continue + marker = (source, target_node, "summarizes") + if marker not in seen_edges: + edges.append(make_edge(source, target_node, "summarizes")) + seen_edges.add(marker) + for claim, method, label in note.claim_method_links: + source_node = claim_lookup.get(claim) + target_node = method_lookup.get(method) + if not source_node or not target_node: + continue + marker = (source_node, target_node, label) + if marker not in seen_edges: + edges.append(make_edge(source_node, target_node, label)) + seen_edges.add(marker) + for method, gap, label in note.method_gap_links: + source_node = method_lookup.get(method) + target_node = gap_lookup.get(gap) + if not source_node or not target_node: + continue + marker = (source_node, target_node, label) + if marker not in seen_edges: + edges.append(make_edge(source_node, target_node, label)) + seen_edges.add(marker) + for target in note.wikilinks: + if not target.startswith("Knowledge/"): + continue + target_node = node_lookup.get(target) + if not target_node: + continue + marker = (source, target_node, "relates") + if marker not in seen_edges: + edges.append(make_edge(source, target_node, "relates")) + seen_edges.add(marker) + + return {"nodes": nodes, "edges": edges} + + +def main() -> None: + args = parse_args() + project_kb = load_project_kb_module() + repo_root = project_kb.find_repo_root(Path(args.cwd).resolve()) + binding = project_kb.resolve_binding(repo_root, args.project_id or None) + + papers = [ + note + for note in collect_notes(binding.project_root / "Papers", "Papers", binding.vault_path, PAPER_LIMIT) + if note.canvas_visibility != "hidden" + ] + all_knowledge = collect_notes( + binding.project_root / "Knowledge", "Knowledge", binding.vault_path, KNOWLEDGE_LIMIT * 3 + ) + knowledge = select_knowledge_notes(all_knowledge, papers) + canvas = render_canvas(project_kb.titleize_slug(binding.project_id), papers, knowledge) + + output_path = binding.project_root / args.output + output_path.parent.mkdir(parents=True, exist_ok=True) + output_path.write_text(json.dumps(canvas, ensure_ascii=False, indent=2) + "\n", encoding="utf-8") + print( + json.dumps( + { + "project_id": binding.project_id, + "output": str(output_path), + "paper_count": len(papers), + "knowledge_count": len(knowledge), + }, + ensure_ascii=False, + ) + ) + + +if __name__ == "__main__": + main() diff --git a/.agents/skills/obsidian-literature-workflow/scripts/build_literature_graph.py b/.agents/skills/obsidian-literature-workflow/scripts/build_literature_graph.py new file mode 100755 index 0000000..cfa1ec1 --- /dev/null +++ b/.agents/skills/obsidian-literature-workflow/scripts/build_literature_graph.py @@ -0,0 +1,271 @@ +#!/usr/bin/env python3 +from __future__ import annotations + +import argparse +import re +import sys +from dataclasses import dataclass +from pathlib import Path +from typing import Any + +PAPER_LIMIT = 48 +EDGE_LIMIT = 160 +WIKILINK_RE = re.compile(r"\[\[([^\]|#]+)(?:#[^\]|]+)?(?:\|[^\]]+)?\]\]") +LIST_KEYS = { + 'authors', 'keywords', 'concepts', 'methods', + 'related_papers', 'linked_experiments', 'linked_results' +} + + +@dataclass(frozen=True) +class PaperNote: + title: str + note_relpath: str + file_name: str + concepts: tuple[str, ...] + methods: tuple[str, ...] + related_papers: tuple[str, ...] + linked_experiments: tuple[str, ...] + linked_results: tuple[str, ...] + wikilinks: tuple[str, ...] + + +def load_project_kb_module() -> Any: + script_dir = Path(__file__).resolve().parents[2] / 'obsidian-project-memory' / 'scripts' + sys.path.insert(0, str(script_dir)) + import project_kb # type: ignore + return project_kb + + +def parse_args() -> argparse.Namespace: + parser = argparse.ArgumentParser(description='Build a literature graph note for the bound Obsidian project.') + parser.add_argument('--cwd', default='.') + parser.add_argument('--project-id', default='') + parser.add_argument('--output', default='Knowledge/Literature-Map.md') + return parser.parse_args() + + +def parse_frontmatter(text: str) -> dict[str, Any]: + if not text.startswith('---\n'): + return {} + end = text.find('\n---\n', 4) + if end == -1: + return {} + lines = text[4:end].splitlines() + data: dict[str, Any] = {} + current_key: str | None = None + for raw_line in lines: + if not raw_line.strip(): + continue + if raw_line.startswith(' - ') or raw_line.startswith('- '): + if current_key and current_key in LIST_KEYS: + data.setdefault(current_key, []).append(raw_line.split('- ', 1)[1].strip().strip('"')) + continue + if ':' not in raw_line: + current_key = None + continue + key, value = raw_line.split(':', 1) + key = key.strip() + value = value.strip() + current_key = key + if not value: + if key in LIST_KEYS: + data[key] = [] + else: + data[key] = '' + continue + data[key] = value.strip('"') + return data + + +def body_without_frontmatter(text: str) -> str: + if not text.startswith('---\n'): + return text + end = text.find('\n---\n', 4) + if end == -1: + return text + return text[end + 5:] + + +def normalize_note_target(value: str) -> str: + cleaned = value.strip() + if cleaned.endswith('.md'): + cleaned = cleaned[:-3] + if cleaned.startswith('./'): + cleaned = cleaned[2:] + return cleaned + + +def mermaid_id(prefix: str, value: str) -> str: + slug = re.sub(r'[^a-zA-Z0-9]+', '_', value).strip('_').lower() + if not slug: + slug = 'node' + return f'{prefix}_{slug[:48]}' + + +def extract_wikilinks(text: str) -> tuple[str, ...]: + seen: list[str] = [] + for match in WIKILINK_RE.findall(text): + target = normalize_note_target(match) + if target not in seen: + seen.append(target) + return tuple(seen) + + +def collect_paper_notes(project_root: Path) -> list[PaperNote]: + papers_dir = project_root / 'Papers' + notes: list[PaperNote] = [] + if not papers_dir.exists(): + return notes + for path in sorted(papers_dir.glob('*.md'))[:PAPER_LIMIT]: + text = path.read_text(encoding='utf-8') + frontmatter = parse_frontmatter(text) + body = body_without_frontmatter(text) + title = str(frontmatter.get('title') or path.stem.replace('-', ' ')) + notes.append( + PaperNote( + title=title, + note_relpath=f'Papers/{path.name}', + file_name=path.name, + concepts=tuple(frontmatter.get('concepts', [])), + methods=tuple(frontmatter.get('methods', [])), + related_papers=tuple(normalize_note_target(v) for v in frontmatter.get('related_papers', [])), + linked_experiments=tuple(normalize_note_target(v) for v in frontmatter.get('linked_experiments', [])), + linked_results=tuple(normalize_note_target(v) for v in frontmatter.get('linked_results', [])), + wikilinks=extract_wikilinks(body), + ) + ) + return notes + + +def render_mermaid(notes: list[PaperNote]) -> str: + lines: list[str] = ['graph LR'] + paper_index = {note.note_relpath[:-3]: note for note in notes} + seen_edges: set[tuple[str, str, str]] = set() + + for note in notes: + paper_node = mermaid_id('paper', note.file_name) + lines.append(f' {paper_node}["{note.title}"]') + + for concept in note.concepts: + concept_node = mermaid_id('concept', concept) + lines.append(f' {concept_node}(("{concept}"))') + edge = (paper_node, concept_node, 'concept') + if edge not in seen_edges and len(seen_edges) < EDGE_LIMIT: + lines.append(f' {paper_node} -->|concept| {concept_node}') + seen_edges.add(edge) + + for method in note.methods: + method_node = mermaid_id('method', method) + lines.append(f' {method_node}{{"{method}"}}') + edge = (paper_node, method_node, 'method') + if edge not in seen_edges and len(seen_edges) < EDGE_LIMIT: + lines.append(f' {paper_node} -->|method| {method_node}') + seen_edges.add(edge) + + related_targets = set(note.related_papers) + related_targets.update(target for target in note.wikilinks if target.startswith('Papers/')) + for target in sorted(related_targets): + target_note = paper_index.get(target) + if not target_note: + continue + target_node = mermaid_id('paper', target_note.file_name) + edge = (paper_node, target_node, 'related') + if edge not in seen_edges and len(seen_edges) < EDGE_LIMIT: + lines.append(f' {paper_node} -->|related| {target_node}') + seen_edges.add(edge) + + for experiment in note.linked_experiments: + exp_node = mermaid_id('experiment', experiment) + label = experiment.split('/')[-1] + lines.append(f' {exp_node}["{label}"]') + edge = (paper_node, exp_node, 'experiment') + if edge not in seen_edges and len(seen_edges) < EDGE_LIMIT: + lines.append(f' {paper_node} -->|experiment| {exp_node}') + seen_edges.add(edge) + + for result in note.linked_results: + result_node = mermaid_id('result', result) + label = result.split('/')[-1] + lines.append(f' {result_node}["{label}"]') + edge = (paper_node, result_node, 'result') + if edge not in seen_edges and len(seen_edges) < EDGE_LIMIT: + lines.append(f' {paper_node} -->|result| {result_node}') + seen_edges.add(edge) + + deduped: list[str] = [] + seen_lines: set[str] = set() + for line in lines: + if line not in seen_lines: + deduped.append(line) + seen_lines.add(line) + return '\n'.join(deduped) + + +def render_map_note(project_id: str, notes: list[PaperNote], mermaid_graph: str, updated: str) -> str: + paper_bullets = '\n'.join(f'- [[{note.note_relpath[:-3]}]]' for note in notes) or '- No paper notes yet.' + concept_counts: dict[str, int] = {} + method_counts: dict[str, int] = {} + for note in notes: + for concept in note.concepts: + concept_counts[concept] = concept_counts.get(concept, 0) + 1 + for method in note.methods: + method_counts[method] = method_counts.get(method, 0) + 1 + concept_lines = '\n'.join( + f'- {name} ({count})' for name, count in sorted(concept_counts.items(), key=lambda kv: (-kv[1], kv[0]))[:12] + ) or '- No concept clusters recorded yet.' + method_lines = '\n'.join( + f'- {name} ({count})' for name, count in sorted(method_counts.items(), key=lambda kv: (-kv[1], kv[0]))[:12] + ) or '- No method clusters recorded yet.' + + return f'''--- +type: knowledge +title: Literature Map +project: {project_id} +updated: {updated} +--- + +# Literature Map + +## Purpose +- Provide a lightweight literature knowledge map for the current project. +- Help navigate paper notes, concept clusters, and downstream experiment/result hooks. + +## Paper notes included +{paper_bullets} + +## Concept clusters +{concept_lines} + +## Method clusters +{method_lines} + +## Graph +```mermaid +{mermaid_graph} +``` + +## How to use +- Open the linked paper notes above for detailed reading notes. +- Use Obsidian backlinks/local graph on top of these wikilinks. +- Refresh this note after a batch Zotero ingestion or major paper-note update. +''' + + +def main() -> None: + args = parse_args() + project_kb = load_project_kb_module() + repo_root = project_kb.find_repo_root(Path(args.cwd).resolve()) + binding = project_kb.resolve_binding(repo_root, args.project_id or None) + notes = collect_paper_notes(binding.project_root) + output_rel = args.output if args.output.endswith('.md') else f'{args.output}.md' + output_path = binding.project_root / output_rel + mermaid_graph = render_mermaid(notes) + content = render_map_note(binding.project_id, notes, mermaid_graph, project_kb.now_iso()) + output_path.parent.mkdir(parents=True, exist_ok=True) + output_path.write_text(content.rstrip() + '\n', encoding='utf-8') + print(output_path) + + +if __name__ == '__main__': + main() |