Run post-build steps based on *postprocess* level. When *full_rebuild* is False and *changed_files* are available, uses incremental flow/community detection for faster updates. Returns a list of warning strings (empty on success).
(
store: Any,
build_result: dict[str, Any],
postprocess: str,
full_rebuild: bool = False,
changed_files: list[str] | None = None,
)
| 14 | |
| 15 | |
| 16 | def _run_postprocess( |
| 17 | store: Any, |
| 18 | build_result: dict[str, Any], |
| 19 | postprocess: str, |
| 20 | full_rebuild: bool = False, |
| 21 | changed_files: list[str] | None = None, |
| 22 | ) -> list[str]: |
| 23 | """Run post-build steps based on *postprocess* level. |
| 24 | |
| 25 | When *full_rebuild* is False and *changed_files* are available, |
| 26 | uses incremental flow/community detection for faster updates. |
| 27 | |
| 28 | Returns a list of warning strings (empty on success). |
| 29 | """ |
| 30 | warnings: list[str] = [] |
| 31 | build_result["postprocess_level"] = postprocess |
| 32 | |
| 33 | if postprocess == "none": |
| 34 | return warnings |
| 35 | |
| 36 | # -- Signatures + FTS (fast, always run unless "none") -- |
| 37 | try: |
| 38 | rows = store.get_nodes_without_signature() |
| 39 | for row in rows: |
| 40 | node_id, name, kind, params, ret = ( |
| 41 | row[0], |
| 42 | row[1], |
| 43 | row[2], |
| 44 | row[3], |
| 45 | row[4], |
| 46 | ) |
| 47 | if kind in ("Function", "Test"): |
| 48 | sig = f"def {name}({params or ''})" |
| 49 | if ret: |
| 50 | sig += f" -> {ret}" |
| 51 | elif kind == "Class": |
| 52 | sig = f"class {name}" |
| 53 | else: |
| 54 | sig = name |
| 55 | store.update_node_signature(node_id, sig[:512]) |
| 56 | store.commit() |
| 57 | build_result["signatures_updated"] = True |
| 58 | except (sqlite3.OperationalError, TypeError, KeyError) as e: |
| 59 | logger.warning("Signature computation failed: %s", e) |
| 60 | warnings.append(f"Signature computation failed: {type(e).__name__}: {e}") |
| 61 | |
| 62 | try: |
| 63 | from code_review_graph.search import rebuild_fts_index |
| 64 | |
| 65 | fts_count = rebuild_fts_index(store) |
| 66 | build_result["fts_indexed"] = fts_count |
| 67 | build_result["fts_rebuilt"] = True |
| 68 | except (sqlite3.OperationalError, ImportError) as e: |
| 69 | logger.warning("FTS index rebuild failed: %s", e) |
| 70 | warnings.append(f"FTS index rebuild failed: {type(e).__name__}: {e}") |
| 71 | |
| 72 | if postprocess == "minimal": |
| 73 | return warnings |
no test coverage detected