Merge branch 'fix/upload-avatar' into 'main' (4ff61d93) · Commits · to be continuous... / tools / gitlab-cp

gitlab_cp/sync.py

+80 −117

Original line number	Diff line number	Diff line
		@@ -147,42 +147,6 @@ class Synchronizer:
		else self.dest_sync_path + "/" + self.rel_path(src_path)
		)

		# can't be implemented: there is no simple way to GET the avatar image from a non-public project
		# def look_same_resources(self, url1: str, url2: str) -> bool:
		# resp1: HTTPResponse = urlopen(
		# Request(
		# url1,
		# headers=(
		# {
		# "Authorization": f"Basic: {base64.b64encode(bytes('token:'+self.src_client.private_token, 'utf-8')).decode()}"
		# }
		# if self.src_client.private_token
		# else {}
		# ),
		# method="HEAD",
		# ),
		# context=None if self.src_client.ssl_verify else INSECURE_SSL_CTX,
		# )
		# resp2: HTTPResponse = urlopen(
		# Request(
		# url2,
		# headers=(
		# {
		# "Authorization": f"Basic: {base64.b64encode(bytes('token:'+self.dest_client.private_token, 'utf-8')).decode()}"
		# }
		# if self.dest_client.private_token
		# else {}
		# ),
		# method="HEAD",
		# ),
		# context=None if self.dest_client.ssl_verify else INSECURE_SSL_CTX,
		# )
		# return (
		# resp1.headers.get_content_type() == resp2.headers.get_content_type()
		# and resp1.headers.get("content-length")
		# == resp2.headers.get("content-length")
		# )

		# Synchronize a project release
		def sync_releases(
		self,
		@@ -388,6 +352,78 @@ class Synchronizer:
		f" - git repository: {AnsiColors.HYELLOW}updated{AnsiColors.RESET} ({src_default_branch} on #{src_latest_commit.get_id() if src_latest_commit else 'n/a'})"
		)

		# Synchronize a project/group avatar
		def sync_avatar(
		self,
		src_avatar: Project \| Group,
		dest_avatar: Project \| Group,
		) -> None:
		assert all(any(isinstance(resource, type) for type in [Project, Group]) for resource in [src_avatar, dest_avatar])

		# Python lib `python-gitlab` doesn't support avatar endpoint (02/12/2024):
		# > We write our own requests calls.
		src_web_url = src_avatar.web_url
		src_avatar_url = src_avatar.avatar_url
		src_avatar_resource = "projects" if isinstance(src_avatar, Project) else "groups"
		src_avatar_api = f"{self.src_client.api_url}/{src_avatar_resource}/{src_avatar.id}/avatar"
		src_avatar_api_headers = {"PRIVATE-TOKEN": self.src_client.private_token} if self.src_client.private_token is not None else {}
		dest_web_url = dest_avatar.web_url
		dest_avatar_url = dest_avatar.avatar_url
		dest_avatar_resource = "projects" if isinstance(dest_avatar, Project) else "groups"
		dest_avatar_api = f"{self.dest_client.api_url}/{dest_avatar_resource}/{dest_avatar.id}/avatar"
		dest_avatar_api_headers = {"PRIVATE-TOKEN": self.dest_client.private_token} if self.dest_client.private_token is not None else {}

		# Up-to-date if no source avatar url or implicit avatar url ('/path/to/proj/-/avatar')
		if src_avatar_url is None or src_avatar_url == f"{src_web_url}/-/avatar":
		# image up-to-date
		print(
		f" - avatar image: {AnsiColors.HGRAY}up-to-date{AnsiColors.RESET} ({src_avatar_url} / {dest_avatar_url})"
		)
		return

		# Check whether avatar is up-to-date (if update not forced by user)
		if not self.force_update_avatar:
		# fetch avatar size
		try:
		src_avatar_size = requests.head(url=src_avatar_api, headers=src_avatar_api_headers, verify=self.ssl_verify).headers['Content-Length']
		dest_avatar_size = requests.head(url=dest_avatar_api, headers=dest_avatar_api_headers, verify=self.ssl_verify).headers['Content-Length']
		# compare avatar size
		if src_avatar_size == dest_avatar_size:
		# image up-to-date
		print(
		f" - avatar image: {AnsiColors.HGRAY}up-to-date{AnsiColors.RESET} ({src_avatar_url} / {dest_avatar_url}) size: {src_avatar_size}"
		)
		return
		except GitlabUpdateError as ge:
		print(
		f" - avatar image: fetch {AnsiColors.HYELLOW}failed{AnsiColors.RESET} ({src_avatar_url}) - assume update needed",
		ge,
		)
		# don't rethrow - continue anyway
		self.handle_warn(ge)

		# Update needed but dry run
		if self.dry_run:
		print(
		f" - avatar image: {AnsiColors.HYELLOW}update needed{AnsiColors.RESET} ({src_avatar_url} / {dest_avatar_url})"
		)
		return

		# Update avatar
		try:
		dest_avatar.avatar = requests.get(url=src_avatar_api, headers=dest_avatar_api_headers, verify=self.ssl_verify).content
		dest_avatar.save()
		print(
		f" - avatar image: {AnsiColors.HYELLOW}updated{AnsiColors.RESET} ({src_avatar_url})"
		)
		except GitlabUpdateError as ge:
		print(
		f" - avatar image: update {AnsiColors.HYELLOW}failed{AnsiColors.RESET} ({src_avatar_url})",
		ge,
		)
		self.handle_warn(ge)
		return

		# Synchronizes a GitLab project
		# $1: source project JSON
		# $2: destination parent group ID (number)
		@@ -489,53 +525,13 @@ class Synchronizer:
		self.handle_error(ge)
		return

		# set/update avatar url
		src_avatar_url = src_project.avatar_url
		src_web_url = src_project.web_url
		dest_avatar_url = dest_project.avatar_url
		dest_web_url = dest_project.web_url
		if (src_avatar_url is None or src_avatar_url == f"{src_web_url}/-/avatar") or (
		dest_avatar_url is not None
		and dest_avatar_url != f"{dest_web_url}/-/avatar"
		and not self.force_update_avatar
		):
		# image up-to-date
		print(
		f" - avatar image: {AnsiColors.HGRAY}up-to-date{AnsiColors.RESET} ({src_avatar_url} / {dest_avatar_url})"
		)
		elif self.dry_run:
		print(
		f" - avatar image: {AnsiColors.HYELLOW}update needed{AnsiColors.RESET} ({src_avatar_url} / {dest_avatar_url})"
		)
		elif (
		src_project.attributes.get("visibility", GlVisibility.public)
		!= GlVisibility.public
		):
		# no way to download the avatar image of a non-public project
		print(
		f"- avatar image: {AnsiColors.HGRAY}update needed{AnsiColors.HRED} skipped due to src project not public ({src_avatar_url} / {dest_avatar_url})"
		)
		else:
		try:
		dest_project.avatar = requests.get(
		src_avatar_url, verify=self.ssl_verify
		).content
		dest_project.save()
		print(
		f" - avatar image: {AnsiColors.HYELLOW}updated{AnsiColors.RESET} ({src_avatar_url})"
		)
		except GitlabUpdateError as ge:
		print(
		f" - avatar image: update {AnsiColors.HYELLOW}failed{AnsiColors.RESET} ({src_avatar_url})",
		ge,
		)
		self.handle_warn(ge)
		# don't rethrow - continue anyway
		# 2: sync Avatar
		self.sync_avatar(src_project, dest_project)

		# 2: sync Git repository
		# 3: sync Git repository
		self.sync_git_repo(src_project, dest_project)

		# 3: sync Releases
		# 4: sync Releases
		self.sync_releases(src_project, dest_project)

		# Synchronizes recursively a GitLab group
		@@ -664,43 +660,10 @@ class Synchronizer:
		self.handle_warn(ge)
		# continue anyway

		# set/update avatar url
		src_avatar_url = src_group.avatar_url
		dest_avatar_url = dest_group.avatar_url
		if src_avatar_url is None or (
		dest_avatar_url is not None and not self.force_update_avatar
		):
		# image up-to-date
		print(
		f"- avatar image: {AnsiColors.HGRAY}up-to-date{AnsiColors.RESET} ({src_avatar_url} / {dest_avatar_url})"
		)
		elif self.dry_run:
		print(
		f"- avatar image: {AnsiColors.HYELLOW}update needed{AnsiColors.RESET} ({src_avatar_url} / {dest_avatar_url})"
		)
		elif src_group.visibility != GlVisibility.public:
		# no way to download the avatar image of a non-public group
		print(
		f"- avatar image: {AnsiColors.HRED}update needed{AnsiColors.RESET} skipped due to src group not public ({src_avatar_url} / {dest_avatar_url})"
		)
		else:
		try:
		dest_group.avatar = requests.get(
		src_avatar_url, verify=self.ssl_verify
		).content
		dest_group.save()
		print(
		f"- avatar image: {AnsiColors.HYELLOW}updated{AnsiColors.RESET} ({src_avatar_url})"
		)
		except GitlabUpdateError as ge:
		print(
		f"- avatar image: update {AnsiColors.HYELLOW}failed{AnsiColors.RESET} ({src_avatar_url})",
		ge,
		)
		# don't rethrow - continue anyway
		self.handle_warn(ge)
		# 2: sync Avatar
		self.sync_avatar(src_group, dest_group)

		# 2: sync sub-projects
		# 3: sync sub-projects
		subprojects = src_group.projects.list(all=True)
		print(f"- sync {len(subprojects)} sub projects...")
		for src_project in subprojects:
		@@ -717,7 +680,7 @@ class Synchronizer:
		self.src_client.projects.get(src_project.get_id()), dest_group
		)

		# 3: sync sub-groups
		# 4: sync sub-groups
		subgroups = src_group.subgroups.list(all=True)
		print(f"- sync {len(subgroups)} sub groups...")
		for src_subgroup in subgroups: