def main()

in bsp_server/scip_sync_util/scip_sync.py [0:0]

• 108 lines of code
• 20 McCabe index (conditional complexity)

def main():
    parser = argparse.ArgumentParser(description="Sync scip index for given targets")
    parser.add_argument(
        "--cwd",
        type=str,
        required=True,
        help="current working dir where to execute bazel command",
    )

    parser.add_argument(
        "targets",
        type=str,
        nargs="*",
        help="targets to generate scip index",
    )

    parser.add_argument(
        "--filepath",
        type=str,
        help="files to generate scip index for, the target which includes this file is indexed",
    )

    parser.add_argument(
        "--depth",
        default=1,
        type=int,
        help="Dependency graph depth for the index generation",
    )

    sync_stats = SyncStats()

    index_start = datetime.now()

    args = parser.parse_args()
    targets = args.targets
    cwd = args.cwd
    filepath = args.filepath
    excludes = set()

    # Case 1: No initial files or targets - rewrite the workspace
    if not targets and not filepath:
        targets, excludes = fetch_targets_from_bazelproject(cwd)

    # Case 2: Filepath provided - check if it's in workspace
    if filepath:
        file_manifest = None
        try:
            # Remove cwd prefix from filepath for workspace lookup
            relative_filepath = filepath.replace(cwd + "/", "")
            file_manifest = scip_workspace.get_manifest_for_file(
                relative_filepath, os.path.join(cwd, SCIP_INDEX_DIR)
            )

            # If file is already in workspace, no need to update
            if file_manifest:
                file_index = incremental.index_file(
                    cwd, relative_filepath, file_manifest
                )
                scip_utils.old_copy_index(
                    {file_index}, os.path.join(cwd, SCIP_INDEX_DIR)
                )
                return
            else:
                print(f"File {relative_filepath} not found in workspace")
                return
        except Exception as e:
            print(f"Error checking file in workspace: {e}")
            return

        # File not in workspace or error occurred, add its target to the list
        file_target = scip_utils.get_containing_bazel_target(
            cwd, filepath, SUPPORTED_RULES
        )
        if file_target:
            targets.append(file_target)
        else:
            print(f"Could not find target for {filepath}")
            return

    if len(targets) == 0:
        print("No targets to sync ...")
        return

    # Get all deps
    print(f"Syncing deps for targets: {list(targets)}")
    buildable_scip_targets = get_dependency_graph(
        cwd=cwd, targets=targets, depth=args.depth, exclude_targets=excludes
    )

    if len(buildable_scip_targets) == 0:
        print("Found no targets to sync ...")
        return

    # Update stats
    sync_stats.total_scip_targets_identified = len(buildable_scip_targets)
    sync_stats.total_will_build_scip_target = len(buildable_scip_targets)
    sync_stats.index_target_extraction_time_sec = (
        datetime.now() - index_start
    ).total_seconds()

    # Generate SCIP indexes
    sync_scip(cwd, buildable_scip_targets, sync_stats)

    # Process target outputs and update workspace
    target_to_output = scip_utils.get_mnemonic_output(
        cwd,
        f"{ScipMnemonics.INDEX_OUTPUT_MNEMONIC.value}|{ScipMnemonics.JAVA_TARGET_MANIFEST_MNEMONIC.value}|{ScipMnemonics.UNPACKED_JAVA_SOURCES_MNEMONIC.value}",
        buildable_scip_targets,
    )

    # Collect indexes and update workspace
    index_target_map = {}
    for target, target_mnemonics in target_to_output.items():
        # Collect indexes
        index_list = target_mnemonics.get(ScipMnemonics.INDEX_OUTPUT_MNEMONIC.value, [])
        for index in index_list:
            if os.path.exists(os.path.join(cwd, index)):
                index_target_map.setdefault(target, []).append(os.path.join(cwd, index))

    workspace = scip_workspace.populate_workspace(cwd, target_to_output)
    scip_workspace.write_workspace(workspace, os.path.join(cwd, SCIP_INDEX_DIR))

    # Calculate stats and copy indexes
    relevant_index = list(
        set(index_target_map.keys()).intersection(set(buildable_scip_targets))
    )
    sync_stats.passed_index_cnt = len(relevant_index)
    sync_stats.failed_index_cnt = (
        sync_stats.total_will_build_scip_target - sync_stats.passed_index_cnt
    )

    # Copy indexes to SCIP directory
    index_to_copy = set()
    for idx in relevant_index:
        index_to_copy.update(index_target_map[idx])

    start_copy_index = datetime.now()
    scip_utils.copy_index(index_to_copy, os.path.join(cwd, SCIP_INDEX_DIR))
    sync_stats.copy_index_time_sec = (datetime.now() - start_copy_index).total_seconds()

    print(f"--- Sync Stats ---")
    sync_stats.total_duration_sec = (datetime.now() - index_start).total_seconds()
    pprint(asdict(sync_stats), indent=2, sort_dicts=False)