Setting the file. One moment. def top_edges_for_file
— line 183
This file
- Number
- 32.4
- Position
- 4 of 7
- Type
- Python
- Size
- 18 KB
- Lines
- 483
scripts/query_ownership.py
Python·483 lines·18 KB
import
Iterable
13
14
15def parse_args() -> argparse.Namespace:
16 parser = argparse.ArgumentParser(
17 description="Query ownership-map outputs with bounded JSON results."
18 )
19 parser.add_argument(
20 "--data-dir",
21 default="ownership-map-out",
22 help="Directory containing people.csv, files.csv, edges.csv",
23 )
24
25 subparsers = parser.add_subparsers(dest="command", required=True)
26
27 people = subparsers.add_parser("people", help="List people")
28 people.add_argument("--limit", type=int, default=20)
29 people.add_argument("--sort", default="touches")
30 people.add_argument("--email-contains", default=None)
31 people.add_argument("--min-touches", type=int, default=0)
32 people.add_argument("--min-sensitive", type=float, default=0.0)
33
34 files = subparsers.add_parser("files", help="List files")
35 files.add_argument("--limit", type=int, default=20)
36 files.add_argument("--sort", default="sensitivity_score")
37 files.add_argument("--path-contains", default=None)
38 files.add_argument("--tag", default=None)
39 files.add_argument("--bus-factor-max", type=int, default=None)
40 files.add_argument("--sensitivity-min", type=float, default=0.0)
41
42 person = subparsers.add_parser("person", help="Show person details and top files")
43 person.add_argument("--person", required=True, help="Exact email or substring")
44 person.add_argument("--limit", type=int, default=20)
45 person.add_argument("--sort", default="touches")
46
47 file_cmd = subparsers.add_parser("file", help="Show file details and top people")
48 file_cmd.add_argument("--file", required=True, help="Exact path or substring")
49 file_cmd.add_argument("--limit", type=int, default=20)
50 file_cmd.add_argument("--sort", default="touches")
51
52 cochange = subparsers.add_parser("cochange", help="List co-change neighbors for a file")
53 cochange.add_argument("--file", required=True, help="Exact path or substring")
54 cochange.add_argument("--limit", type=int, default=20)
55 cochange.add_argument("--sort", default="jaccard")
56 cochange.add_argument("--min-jaccard", type=float, default=0.0)
57 cochange.add_argument("--min-count", type=int, default=1)
58
59 tag = subparsers.add_parser("tag", help="Show top people/files for a sensitive tag")
60 tag.add_argument("--tag", required=True)
61 tag.add_argument("--limit", type=int, default=20)
62
63 summary = subparsers.add_parser("summary", help="Show summary.json sections")
64 summary.add_argument("--section", default=None)
65
66 communities = subparsers.add_parser("communities", help="List communities")
67 communities.add_argument("--limit", type=int, default=10)
68 communities.add_argument("--id", type=int, default=None)
69
70 community = subparsers.add_parser("community", help="Show community maintainers")
71 community.add_argument("--id", type=int, required=True)
72 community.add_argument("--include-files", action="store_true")
73 community.add_argument("--file-limit", type=int, default=50)
74
75 return parser.parse_args()
76
77
78def to_int(value: str) -> int:
79 try:
80 return int(value)
81 except (TypeError, ValueError):
82 return 0
83
84
85def to_float(value: str) -> float:
86 try:
87 return float(value)
88 except (TypeError, ValueError):
89 return 0.0
90
91
92def read_csv(path: Path) -> Iterable[dict[str, str]]:
93 with path.open("r", encoding="utf-8") as handle:
94 reader = csv.DictReader(handle)
95 yield from reader
96
97
98def load_people(data_dir: Path) -> list[dict[str, object]]:
99 people_path = data_dir / "people.csv"
100 people = []
101 for row in read_csv(people_path):
102 person = dict(row)
103 person["touches"] = to_int(row.get("touches", "0"))
104 person["commit_count"] = to_int(row.get("commit_count", "0"))
105 person["sensitive_touches"] = to_float(row.get("sensitive_touches", "0"))
106 people.append(person)
107 return people
108
109
110def load_files(data_dir: Path) -> list[dict[str, object]]:
111 files_path = data_dir / "files.csv"
112 files = []
113 for row in read_csv(files_path):
114 file_entry = dict(row)
115 file_entry["touches"] = to_int(row.get("touches", "0"))
116 file_entry["commit_count"] = to_int(row.get("commit_count", "0"))
117 file_entry["bus_factor"] = to_int(row.get("bus_factor", "0"))
118 file_entry["sensitivity_score"] = to_float(row.get("sensitivity_score", "0"))
119 tags = row.get("sensitivity_tags", "")
120 file_entry["sensitivity_tags"] = [tag for tag in tags.split(";") if tag]
121 files.append(file_entry)
122 return files
123
124
125def load_summary(data_dir: Path) -> dict[str, object]:
126 summary_path = data_dir / "summary.json"
127 with summary_path.open("r", encoding="utf-8") as handle:
128 return json.load(handle)
129
130
131def load_communities(data_dir: Path) -> list[dict[str, object]]:
132 communities_path = data_dir / "communities.json"
133 if not communities_path.exists():
134 raise FileNotFoundError("communities.json not found; rerun build with --communities")
135 with communities_path.open("r", encoding="utf-8") as handle:
136 return json.load(handle)
137
138
139def load_cochange_edges(data_dir: Path) -> Iterable[dict[str, object]]:
140 edges_path = data_dir / "cochange_edges.csv"
141 if not edges_path.exists():
142 raise FileNotFoundError("cochange_edges.csv not found; rerun build without --no-cochange")
143 for row in read_csv(edges_path):
144 yield {
145 "file_a": row.get("file_a"),
146 "file_b": row.get("file_b"),
147 "cochange_count": to_int(row.get("cochange_count", "0")),
148 "jaccard": to_float(row.get("jaccard", "0")),
149 }
150
151
152def select_single(records: list[dict[str, object]], key: str, query: str) -> dict[str, object]:
153 exact = [record for record in records if str(record.get(key, "")) == query]
154 if exact:
155 return exact[0]
156 contains = [record for record in records if query in str(record.get(key, ""))]
157 if len(contains) == 1:
158 return contains[0]
159 if not contains:
160 raise ValueError(f"No match for {query}")
161 candidates = [str(record.get(key, "")) for record in contains[:10]]
162 raise ValueError(f"Multiple matches for {query}: {', '.join(candidates)}")
163
164
165def top_edges_for_person(data_dir: Path, person_id: str) -> list[dict[str, object]]:
166 edges_path = data_dir / "edges.csv"
167 results = []
168 for row in read_csv(edges_path):
169 if row.get("person_id") != person_id:
170 continue
171 results.append(
172 {
173 "file_id": row.get("file_id"),
174 "touches": to_int(row.get("touches", "0")),
175 "recency_weight": to_float(row.get("recency_weight", "0")),
176 "sensitive_weight": to_float(row.get("sensitive_weight", "0")),
177 "last_seen": row.get("last_seen"),
178 }
179 )
180 return results
181
182
183def top_edges_for_file(data_dir: Path, file_id: str) -> list[dict[str, object]]:
184 edges_path = data_dir / "edges.csv"
185 results = []
186 for row in read_csv(edges_path):
187 if row.get("file_id") != file_id:
188 continue
189 results.append(
190 {
191 "person_id": row.get("person_id"),
192 "touches": to_int(row.get("touches", "0")),
193 "recency_weight": to_float(row.get("recency_weight", "0")),
194 "sensitive_weight": to_float(row.get("sensitive_weight", "0")),
195 "last_seen": row.get("last_seen"),
196 }
197 )
198 return results
199
200
201def sort_records(records: list[dict[str, object]], key: str) -> list[dict[str, object]]:
202 return sorted(records, key=lambda item: item.get(key, 0), reverse=True)
203
204
205def handle_people(args: argparse.Namespace, data_dir: Path) -> None:
206 people = load_people(data_dir)
207 if args.email_contains:
208 people = [p for p in people if args.email_contains in p.get("email", "")]
209 people = [p for p in people if p["touches"] >= args.min_touches]
210 people = [p for p in people if p["sensitive_touches"] >= args.min_sensitive]
211 people = sort_records(people, args.sort)[: args.limit]
212 payload = [
213 {
214 "person_id": p.get("person_id"),
215 "name": p.get("name"),
216 "email": p.get("email"),
217 "touches": p.get("touches"),
218 "commit_count": p.get("commit_count"),
219 "sensitive_touches": p.get("sensitive_touches"),
220 "primary_tz_offset": p.get("primary_tz_offset"),
221 }
222 for p in people
223 ]
224 print(json.dumps(payload, indent=2))
225
226
227def handle_files(args: argparse.Namespace, data_dir: Path) -> None:
228 files = load_files(data_dir)
229 if args.path_contains:
230 files = [f for f in files if args.path_contains in f.get("path", "")]
231 if args.tag:
232 files = [f for f in files if args.tag in f.get("sensitivity_tags", [])]
233 if args.bus_factor_max is not None:
234 files = [f for f in files if f["bus_factor"] <= args.bus_factor_max]
235 files = [f for f in files if f["sensitivity_score"] >= args.sensitivity_min]
236 files = sort_records(files, args.sort)[: args.limit]
237 payload = [
238 {
239 "file_id": f.get("file_id"),
240 "path": f.get("path"),
241 "touches": f.get("touches"),
242 "bus_factor": f.get("bus_factor"),
243 "sensitivity_score": f.get("sensitivity_score"),
244 "sensitivity_tags": f.get("sensitivity_tags"),
245 "last_seen": f.get("last_seen"),
246 }
247 for f in files
248 ]
249 print(json.dumps(payload, indent=2))
250
251
252def handle_person(args: argparse.Namespace, data_dir: Path) -> None:
253 people = load_people(data_dir)
254 person = select_single(people, "person_id", args.person)
255 files = load_files(data_dir)
256 file_map = {f["file_id"]: f for f in files}
257 edges = top_edges_for_person(data_dir, person["person_id"])
258 edges = sort_records(edges, args.sort)[: args.limit]
259 payload = {
260 "person": {
261 "person_id": person.get("person_id"),
262 "name": person.get("name"),
263 "email": person.get("email"),
264 "touches": person.get("touches"),
265 "commit_count": person.get("commit_count"),
266 "sensitive_touches": person.get("sensitive_touches"),
267 "primary_tz_offset": person.get("primary_tz_offset"),
268 "timezone_offsets": person.get("timezone_offsets"),
269 },
270 "top_files": [
271 {
272 "file_id": edge.get("file_id"),
273 "path": file_map.get(edge.get("file_id"), {}).get("path"),
274 "touches": edge.get("touches"),
275 "recency_weight": edge.get("recency_weight"),
276 "sensitive_weight": edge.get("sensitive_weight"),
277 "last_seen": edge.get("last_seen"),
278 "sensitivity_tags": file_map.get(edge.get("file_id"), {}).get("sensitivity_tags"),
279 }
280 for edge in edges
281 ],
282 }
283 print(json.dumps(payload, indent=2))
284
285
286def handle_file(args: argparse.Namespace, data_dir: Path) -> None:
287 files = load_files(data_dir)
288 file_entry = select_single(files, "file_id", args.file)
289 people = load_people(data_dir)
290 people_map = {p["person_id"]: p for p in people}
291 edges = top_edges_for_file(data_dir, file_entry["file_id"])
292 edges = sort_records(edges, args.sort)[: args.limit]
293 payload = {
294 "file": {
295 "file_id": file_entry.get("file_id"),
296 "path": file_entry.get("path"),
297 "touches": file_entry.get("touches"),
298 "bus_factor": file_entry.get("bus_factor"),
299 "sensitivity_score": file_entry.get("sensitivity_score"),
300 "sensitivity_tags": file_entry.get("sensitivity_tags"),
301 "last_seen": file_entry.get("last_seen"),
302 },
303 "top_people": [
304 {
305 "person_id": edge.get("person_id"),
306 "name": people_map.get(edge.get("person_id"), {}).get("name"),
307 "email": people_map.get(edge.get("person_id"), {}).get("email"),
308 "touches": edge.get("touches"),
309 "recency_weight": edge.get("recency_weight"),
310 "sensitive_weight": edge.get("sensitive_weight"),
311 "primary_tz_offset": people_map.get(edge.get("person_id"), {}).get(
312 "primary_tz_offset"
313 ),
314 }
315 for edge in edges
316 ],
317 }
318 print(json.dumps(payload, indent=2))
319
320
321def handle_cochange(args: argparse.Namespace, data_dir: Path) -> None:
322 files = load_files(data_dir)
323 file_entry = select_single(files, "file_id", args.file)
324
325 neighbors = []
326 for row in load_cochange_edges(data_dir):
327 file_a = row.get("file_a")
328 file_b = row.get("file_b")
329 if file_a == file_entry["file_id"]:
330 other = file_b
331 elif file_b == file_entry["file_id"]:
332 other = file_a
333 else:
334 continue
335
336 if row["cochange_count"] < args.min_count:
337 continue
338 if row["jaccard"] < args.min_jaccard:
339 continue
340
341 neighbors.append(
342 {
343 "file_id": other,
344 "path": other,
345 "cochange_count": row["cochange_count"],
346 "jaccard": row["jaccard"],
347 }
348 )
349
350 neighbors = sort_records(neighbors, args.sort)[: args.limit]
351 payload = {
352 "file": {
353 "file_id": file_entry.get("file_id"),
354 "path": file_entry.get("path"),
355 },
356 "neighbors": neighbors,
357 }
358 print(json.dumps(payload, indent=2))
359
360
361def handle_tag(args: argparse.Namespace, data_dir: Path) -> None:
362 files = load_files(data_dir)
363 tagged_files = [f for f in files if args.tag in f.get("sensitivity_tags", [])]
364 tagged_ids = {f["file_id"] for f in tagged_files}
365
366 person_touch = defaultdict(int)
367 edges_path = data_dir / "edges.csv"
368 for row in read_csv(edges_path):
369 if row.get("file_id") not in tagged_ids:
370 continue
371 person_touch[row.get("person_id")] += to_int(row.get("touches", "0"))
372
373 people = load_people(data_dir)
374 people_map = {p["person_id"]: p for p in people}
375 top_people = [
376 {
377 "person_id": person_id,
378 "name": people_map.get(person_id, {}).get("name"),
379 "email": people_map.get(person_id, {}).get("email"),
380 "touches": touches,
381 }
382 for person_id, touches in person_touch.items()
383 ]
384 top_people = sorted(top_people, key=lambda item: item.get("touches", 0), reverse=True)[
385 : args.limit
386 ]
387
388 top_files = sorted(tagged_files, key=lambda item: item.get("touches", 0), reverse=True)[
389 : args.limit
390 ]
391
392 payload = {
393 "tag": args.tag,
394 "top_people": top_people,
395 "top_files": [
396 {
397 "file_id": entry.get("file_id"),
398 "path": entry.get("path"),
399 "touches": entry.get("touches"),
400 "bus_factor": entry.get("bus_factor"),
401 }
402 for entry in top_files
403 ],
404 }
405 print(json.dumps(payload, indent=2))
406
407
408def handle_summary(args: argparse.Namespace, data_dir: Path) -> None:
409 summary = load_summary(data_dir)
410 if args.section:
411 if args.section not in summary:
412 raise ValueError(f"Section not found: {args.section}")
413 payload = summary[args.section]
414 else:
415 payload = summary
416 print(json.dumps(payload, indent=2))
417
418
419def handle_communities(args: argparse.Namespace, data_dir: Path) -> None:
420 communities = load_communities(data_dir)
421 if args.id is not None:
422 matches = [entry for entry in communities if entry.get("id") == args.id]
423 if not matches:
424 raise ValueError(f"Community id not found: {args.id}")
425 payload = matches[0]
426 else:
427 payload = sorted(communities, key=lambda item: item.get("size", 0), reverse=True)[
428 : args.limit
429 ]
430 print(json.dumps(payload, indent=2))
431
432
433def handle_community(args: argparse.Namespace, data_dir: Path) -> None:
434 communities = load_communities(data_dir)
435 matches = [entry for entry in communities if entry.get("id") == args.id]
436 if not matches:
437 raise ValueError(f"Community id not found: {args.id}")
438 entry = dict(matches[0])
439 files = entry.pop("files", [])
440 payload = entry
441 if args.include_files:
442 payload["files"] = files[: args.file_limit]
443 payload["files_truncated"] = len(files) > args.file_limit
444 print(json.dumps(payload, indent=2))
445
446
447def main() -> int:
448 args = parse_args()
449 data_dir = Path(args.data_dir)
450 if not data_dir.exists():
451 print(f"Data directory not found: {data_dir}", file=sys.stderr)
452 return 1
453
454 try:
455 if args.command == "people":
456 handle_people(args, data_dir)
457 elif args.command == "files":
458 handle_files(args, data_dir)
459 elif args.command == "person":
460 handle_person(args, data_dir)
461 elif args.command == "file":
462 handle_file(args, data_dir)
463 elif args.command == "cochange":
464 handle_cochange(args, data_dir)
465 elif args.command == "tag":
466 handle_tag(args, data_dir)
467 elif args.command == "summary":
468 handle_summary(args, data_dir)
469 elif args.command == "communities":
470 handle_communities(args, data_dir)
471 elif args.command == "community":
472 handle_community(args, data_dir)
473 else:
474 raise ValueError(f"Unknown command: {args.command}")
475 except (FileNotFoundError, ValueError) as exc:
476 print(str(exc), file=sys.stderr)
477 return 2
478
479 return 0
480
481
482if __name__ == "__main__":
483 raise SystemExit(main())