def mirror()

in build/fbcode_builder/getdeps/fetcher.py [0:0]

• 43 lines of code
• 7 McCabe index (conditional complexity)

    def mirror(self, fbsource_root, dest_root) -> ChangeStatus:
        self._minimize_roots()
        self._sort_mapping()

        change_status = ChangeStatus()

        # Record the full set of files that should be in the tree
        full_file_list = set()

        if sys.platform == "win32":
            # Let's not assume st_dev has a consistent value on Windows.
            def st_dev(path):
                return 1

        else:

            def st_dev(path):
                return os.lstat(path).st_dev

        for fbsource_subdir in self.roots:
            dir_to_mirror = os.path.join(fbsource_root, fbsource_subdir)
            root_dev = st_dev(dir_to_mirror)
            prefetch_dir_if_eden(dir_to_mirror)
            if not os.path.exists(dir_to_mirror):
                raise Exception(
                    "%s doesn't exist; check your sparse profile!" % dir_to_mirror
                )

            for root, dirs, files in os.walk(dir_to_mirror):
                dirs[:] = [d for d in dirs if root_dev == st_dev(os.path.join(root, d))]

                for src_file in files:
                    full_name = os.path.join(root, src_file)
                    rel_name = os.path.relpath(full_name, fbsource_root)
                    norm_name = rel_name.replace("\\", "/")

                    target_name = self._map_name(norm_name, dest_root)
                    if target_name:
                        full_file_list.add(target_name)
                        if copy_if_different(full_name, target_name):
                            change_status.record_change(target_name)

        # Compare the list of previously shipped files; if a file is
        # in the old list but not the new list then it has been
        # removed from the source and should be removed from the
        # destination.
        # Why don't we simply create this list by walking dest_root?
        # Some builds currently have to be in-source builds and
        # may legitimately need to keep some state in the source tree :-/
        installed_name = os.path.join(dest_root, ".shipit_shipped")
        if os.path.exists(installed_name):
            with open(installed_name, "rb") as f:
                for name in f.read().decode("utf-8").splitlines():
                    name = name.strip()
                    if name not in full_file_list:
                        print("Remove %s" % name)
                        os.unlink(name)
                        change_status.record_change(name)

        with open(installed_name, "wb") as f:
            for name in sorted(list(full_file_list)):
                f.write(("%s\n" % name).encode("utf-8"))

        return change_status