ydb/build/plugins/lib/nots/package_manager/base/node_modules_bundler.py

import os
import tempfile
import logging

from .utils import build_nm_path
from .timeit import timeit

PEERS_DIR = ".peers"
PEERS_INDEX = "index"


@timeit
def bundle_node_modules(build_root, peers, node_modules_path, bundle_path):
    """
    Creates node_modules bundle.
    Bundle contains node_modules directory, peers' node_modules directories,
    and index file with the list of added peers (\\n delimited).
    :param build_root: arcadia build root
    :type build_root: str
    :param peers: list of peers (arcadia root related)
    :type peers: list of str
    :param node_modules_path: node_modules path
    :type node_modules_path: str
    :param bundle_path: tarball path
    :type bundle_path: str
    """
    import library.python.archive as archive

    paths_to_pack = []
    paths_to_pack.append((node_modules_path, "."))

    # Peers' node_modules.
    added_peers = []
    peers = set(peers)
    for p in peers:
        peer_nm_path = build_nm_path(os.path.join(build_root, p))
        peer_bundled_nm_path = build_nm_path(os.path.join(PEERS_DIR, p))
        if not os.path.isdir(peer_nm_path):
            continue
        paths_to_pack.append((peer_nm_path, peer_bundled_nm_path))
        added_peers.append(p)

    # Peers index.
    with tempfile.TemporaryDirectory() as temp_dir:
        peers_index_tmppath = os.path.join(temp_dir, PEERS_INDEX)
        peers_index_relpath = os.path.join(PEERS_DIR, PEERS_INDEX)
        with open(peers_index_tmppath, "w") as peers_index:
            peers_index.write("\n".join(added_peers))
        paths_to_pack.append((peers_index_tmppath, peers_index_relpath))

        archive.tar(set(paths_to_pack), bundle_path, compression_filter=None, compression_level=None, fixed_mtime=0)


def extract_node_modules(build_root, node_modules_path, bundle_path):
    """
    Extracts node_modules bundle.
    :param build_root: arcadia build root
    :type build_root: str
    :param node_modules_path: node_modules path
    :type node_modules_path: str
    :param bundle_path: tarball path
    :type bundle_path: str
    """
    import library.python.archive as archive

    os.makedirs(node_modules_path, exist_ok=True)
    archive.extract_tar(bundle_path, node_modules_path, fail_on_duplicates=False)

    with open(os.path.join(node_modules_path, PEERS_DIR, PEERS_INDEX)) as peers_file:
        peers = peers_file.read().split("\n")
        for p in peers:
            if not p:
                continue
            bundled_nm_path = build_nm_path(os.path.join(node_modules_path, PEERS_DIR, p))
            nm_path = build_nm_path(os.path.join(build_root, p))

            try:
                if os.path.exists(bundled_nm_path):
                    os.rename(bundled_nm_path, nm_path)
            except FileNotFoundError as e:
                parent_directory = os.path.dirname(nm_path)
                logging.error(f"bundled_nm exists={os.path.exists(bundled_nm_path)} : {bundled_nm_path}")
                logging.error(f"nm_path/.. exists={os.path.exists(parent_directory)} : {parent_directory}")
                raise e

    return True