|
1
|
""" |
|
2
|
CODEOWNERS integration — map ownership to Person and Domain nodes. |
|
3
|
|
|
4
|
Parses GitHub/GitLab CODEOWNERS files and creates Person nodes with |
|
5
|
ASSIGNED_TO edges linking file paths to owners. |
|
6
|
|
|
7
|
Usage: |
|
8
|
from navegador.codeowners import CodeownersIngester |
|
9
|
|
|
10
|
ingester = CodeownersIngester(store) |
|
11
|
stats = ingester.ingest("/path/to/repo") |
|
12
|
""" |
|
13
|
|
|
14
|
from __future__ import annotations |
|
15
|
|
|
16
|
import logging |
|
17
|
from pathlib import Path |
|
18
|
from typing import Any |
|
19
|
|
|
20
|
from navegador.graph.schema import EdgeType, NodeLabel |
|
21
|
from navegador.graph.store import GraphStore |
|
22
|
|
|
23
|
logger = logging.getLogger(__name__) |
|
24
|
|
|
25
|
# CODEOWNERS files can live in the root, .github/, or .gitlab/ |
|
26
|
_CODEOWNERS_CANDIDATES = [ |
|
27
|
"CODEOWNERS", |
|
28
|
".github/CODEOWNERS", |
|
29
|
".gitlab/CODEOWNERS", |
|
30
|
"docs/CODEOWNERS", |
|
31
|
] |
|
32
|
|
|
33
|
|
|
34
|
class CodeownersIngester: |
|
35
|
""" |
|
36
|
Parse a CODEOWNERS file and write Person + File nodes into the graph. |
|
37
|
|
|
38
|
Each owner mentioned in CODEOWNERS becomes a Person node. |
|
39
|
Each pattern entry creates an ASSIGNED_TO edge from the File node (keyed |
|
40
|
by the pattern string) to the Person node. |
|
41
|
""" |
|
42
|
|
|
43
|
def __init__(self, store: GraphStore) -> None: |
|
44
|
self.store = store |
|
45
|
|
|
46
|
# ── Public API ──────────────────────────────────────────────────────────── |
|
47
|
|
|
48
|
def ingest(self, repo_path: str | Path) -> dict[str, Any]: |
|
49
|
""" |
|
50
|
Parse the CODEOWNERS file found inside *repo_path* and populate the |
|
51
|
graph. |
|
52
|
|
|
53
|
Returns a stats dict with keys: owners, patterns, edges. |
|
54
|
""" |
|
55
|
repo_path = Path(repo_path) |
|
56
|
codeowners_path = self._find_codeowners(repo_path) |
|
57
|
if codeowners_path is None: |
|
58
|
logger.warning("CodeownersIngester: no CODEOWNERS file found in %s", repo_path) |
|
59
|
return {"owners": 0, "patterns": 0, "edges": 0} |
|
60
|
|
|
61
|
entries = self._parse_codeowners(codeowners_path) |
|
62
|
owners_seen: set[str] = set() |
|
63
|
edges = 0 |
|
64
|
|
|
65
|
for pattern, owners in entries: |
|
66
|
# Ensure a File node exists for this pattern |
|
67
|
self.store.create_node( |
|
68
|
NodeLabel.File, |
|
69
|
{ |
|
70
|
"name": pattern, |
|
71
|
"path": pattern, |
|
72
|
"file_path": pattern, |
|
73
|
"language": "", |
|
74
|
"size": 0, |
|
75
|
"line_count": 0, |
|
76
|
"content_hash": "", |
|
77
|
}, |
|
78
|
) |
|
79
|
|
|
80
|
for owner in owners: |
|
81
|
display_name = owner.lstrip("@") |
|
82
|
email = owner if "@" in owner and not owner.startswith("@") else "" |
|
83
|
|
|
84
|
if display_name not in owners_seen: |
|
85
|
self.store.create_node( |
|
86
|
NodeLabel.Person, |
|
87
|
{ |
|
88
|
"name": display_name, |
|
89
|
"email": email, |
|
90
|
"role": "owner", |
|
91
|
"team": "", |
|
92
|
"file_path": "", |
|
93
|
}, |
|
94
|
) |
|
95
|
owners_seen.add(display_name) |
|
96
|
|
|
97
|
# File pattern -ASSIGNED_TO-> Person |
|
98
|
self.store.create_edge( |
|
99
|
NodeLabel.File, |
|
100
|
{"name": pattern}, |
|
101
|
EdgeType.ASSIGNED_TO, |
|
102
|
NodeLabel.Person, |
|
103
|
{"name": display_name}, |
|
104
|
) |
|
105
|
edges += 1 |
|
106
|
|
|
107
|
stats = { |
|
108
|
"owners": len(owners_seen), |
|
109
|
"patterns": len(entries), |
|
110
|
"edges": edges, |
|
111
|
} |
|
112
|
logger.info("CodeownersIngester: %s", stats) |
|
113
|
return stats |
|
114
|
|
|
115
|
# ── Parsing ─────────────────────────────────────────────────────────────── |
|
116
|
|
|
117
|
def _parse_codeowners(self, path: Path) -> list[tuple[str, list[str]]]: |
|
118
|
""" |
|
119
|
Parse a CODEOWNERS file at *path*. |
|
120
|
|
|
121
|
Returns a list of (pattern, [owner, ...]) tuples. Comment lines and |
|
122
|
blank lines are ignored. |
|
123
|
""" |
|
124
|
entries: list[tuple[str, list[str]]] = [] |
|
125
|
text = path.read_text(encoding="utf-8", errors="replace") |
|
126
|
|
|
127
|
for line in text.splitlines(): |
|
128
|
line = line.strip() |
|
129
|
if not line or line.startswith("#"): |
|
130
|
continue |
|
131
|
parts = line.split() |
|
132
|
if len(parts) < 2: |
|
133
|
continue |
|
134
|
pattern = parts[0] |
|
135
|
owners = parts[1:] |
|
136
|
entries.append((pattern, owners)) |
|
137
|
|
|
138
|
return entries |
|
139
|
|
|
140
|
# ── Helpers ─────────────────────────────────────────────────────────────── |
|
141
|
|
|
142
|
def _find_codeowners(self, repo_path: Path) -> Path | None: |
|
143
|
for candidate in _CODEOWNERS_CANDIDATES: |
|
144
|
p = repo_path / candidate |
|
145
|
if p.exists(): |
|
146
|
return p |
|
147
|
return None |
|
148
|
|