(
summary: dict[str, Any],
ndjson_runs: list[dict[str, Any]],
consumed_ndjson_ids: set[str],
)
| 65 | |
| 66 | |
| 67 | def find_matching_ndjson( |
| 68 | summary: dict[str, Any], |
| 69 | ndjson_runs: list[dict[str, Any]], |
| 70 | consumed_ndjson_ids: set[str], |
| 71 | ) -> dict[str, Any] | None: |
| 72 | summary_source = summary.get("source") or "" |
| 73 | summary_parent = Path(summary_source).parent.as_posix() |
| 74 | summary_time = parse_iso(summary.get("collectedAt")) |
| 75 | |
| 76 | best_match: dict[str, Any] | None = None |
| 77 | best_score: tuple[float, float] | None = None |
| 78 | |
| 79 | for candidate in ndjson_runs: |
| 80 | candidate_id = candidate.get("id") |
| 81 | if not candidate_id or candidate_id in consumed_ndjson_ids: |
| 82 | continue |
| 83 | |
| 84 | candidate_source = candidate.get("source") or "" |
| 85 | candidate_parent = Path(candidate_source).parent.as_posix() |
| 86 | if candidate_parent != summary_parent: |
| 87 | continue |
| 88 | |
| 89 | candidate_time = parse_iso(candidate.get("collectedAt")) |
| 90 | time_delta = abs((summary_time - candidate_time).total_seconds()) |
| 91 | if time_delta > 120: |
| 92 | continue |
| 93 | |
| 94 | # Prefer the closest file in time, then the largest series count. |
| 95 | score = (time_delta, -len(candidate.get("series") or [])) |
| 96 | if best_score is None or score < best_score: |
| 97 | best_score = score |
| 98 | best_match = candidate |
| 99 | |
| 100 | return best_match |
| 101 | |
| 102 | |
| 103 | def merge_summary_with_ndjson(summary: dict[str, Any], ndjson_run: dict[str, Any]) -> dict[str, Any]: |
no test coverage detected