Changeset View
Changeset View
Standalone View
Standalone View
swh/loader/svn/replay.py
# Copyright (C) 2016-2022 The Software Heritage developers | # Copyright (C) 2016-2022 The Software Heritage developers | ||||
# See the AUTHORS file at the top-level directory of this distribution | # See the AUTHORS file at the top-level directory of this distribution | ||||
# License: GNU General Public License version 3, or any later version | # License: GNU General Public License version 3, or any later version | ||||
# See top-level LICENSE file for more information | # See top-level LICENSE file for more information | ||||
"""Remote Access client to svn server. | """Remote Access client to svn server. | ||||
""" | """ | ||||
from __future__ import annotations | from __future__ import annotations | ||||
import codecs | import codecs | ||||
from collections import defaultdict | from collections import defaultdict | ||||
from copy import copy | |||||
from dataclasses import dataclass, field | from dataclasses import dataclass, field | ||||
from distutils.dir_util import copy_tree | from distutils.dir_util import copy_tree | ||||
from itertools import chain | from itertools import chain | ||||
import logging | import logging | ||||
import os | import os | ||||
import shutil | import shutil | ||||
import tempfile | import tempfile | ||||
from typing import ( | from typing import ( | ||||
▲ Show 20 Lines • Show All 207 Lines • ▼ Show 20 Lines | ) -> DirEditor: | ||||
) | ) | ||||
path_bytes = os.fsencode(path) | path_bytes = os.fsencode(path) | ||||
fullpath = os.path.join(self.rootpath, path_bytes) | fullpath = os.path.join(self.rootpath, path_bytes) | ||||
os.makedirs(fullpath, exist_ok=True) | os.makedirs(fullpath, exist_ok=True) | ||||
if copyfrom_rev == -1: | if copyfrom_rev == -1: | ||||
if path_bytes and path_bytes not in self.directory: | if path_bytes and path_bytes not in self.directory: | ||||
self.dir_states[path_bytes] = DirState() | |||||
self.directory[path_bytes] = from_disk.Directory() | self.directory[path_bytes] = from_disk.Directory() | ||||
else: | else: | ||||
url = svn_urljoin(self.svnrepo.remote_url, copyfrom_path) | url = svn_urljoin(self.svnrepo.remote_url, copyfrom_path) | ||||
self.remove_child(path_bytes) | self.remove_child(path_bytes) | ||||
self.svnrepo.export( | self.svnrepo.export( | ||||
url, | url, | ||||
to=fullpath, | to=fullpath, | ||||
peg_rev=copyfrom_rev, | peg_rev=copyfrom_rev, | ||||
ignore_keywords=True, | ignore_keywords=True, | ||||
overwrite=True, | overwrite=True, | ||||
) | ) | ||||
self.directory[path_bytes] = from_disk.Directory.from_disk(path=fullpath) | self.directory[path_bytes] = from_disk.Directory.from_disk(path=fullpath) | ||||
assert copyfrom_path is not None | |||||
copyfrom_path_bytes = os.fsencode(copyfrom_path).lstrip(b"/") | |||||
copyfrom_fullpath = os.path.join(self.rootpath, copyfrom_path_bytes) | |||||
def _copy_dir_state(path: bytes, copied_path: bytes): | |||||
self.dir_states[path] = copy(self.dir_states[copied_path]) | |||||
for external_path in self.dir_states[path].externals_paths: | |||||
self.editor.external_paths[os.path.join(path, external_path)] += 1 | |||||
_copy_dir_state(path_bytes, copyfrom_path_bytes) | |||||
for root, dirs, _ in os.walk(fullpath): | |||||
for dir in dirs: | |||||
dir_fullpath = os.path.join(root, dir) | |||||
copied_dir_fullpath = dir_fullpath.replace( | |||||
fullpath, copyfrom_fullpath | |||||
) | |||||
dir_path = dir_fullpath.replace(self.rootpath, b"").lstrip(b"/") | |||||
copied_dir_path = copied_dir_fullpath.replace(self.rootpath, b"") | |||||
_copy_dir_state(dir_path, copied_dir_path.lstrip(b"/")) | |||||
return DirEditor( | return DirEditor( | ||||
self.directory, | self.directory, | ||||
self.rootpath, | self.rootpath, | ||||
path_bytes, | path_bytes, | ||||
self.dir_states, | self.dir_states, | ||||
svnrepo=self.svnrepo, | svnrepo=self.svnrepo, | ||||
) | ) | ||||
▲ Show 20 Lines • Show All 627 Lines • Show Last 20 Lines |