fetcher: store & edit comment (573c0fb8) · Commits · fediverse.events / Canvas Atlas · GitLab

tools/lemmy_fetcher.py

+66 −7

Original line number	Diff line number	Diff line
		@@ -47,6 +47,9 @@

		JWT = ""

		COMMENT_MARKER_PREFIX = "<!-- lemmy_comment_id: " + LEMMY_HOST + "/"
		COMMENT_MARKER_SUFFIX = " -->"

		def login():
		global JWT
		login_req = requests.post(ENDPOINTS["login"], json={
		@@ -193,27 +196,83 @@ def process_post(self, post_body, make_pr):
		pr_url = self.forge.make_pr_between_branches(new_branch_name, "main", FetcherMessages.git_mr_title(post_body["name"]), FetcherMessages.git_mr_description(str(post_id)))
		else:
		print("branch pushed, PR already exists")
		self.send_comment(post_id, FetcherMessages.SuccessfulSubmission(pr_url))
		self.send_comment(post_id, FetcherMessages.SuccessfulSubmission(pr_url), mr_branch=new_branch_name)
		else:
		self.git.restore(EDITIONS[self.edition]["entry_folder"])
		self.git.checkout("main")
		self.read_ids.add(str(post_id))
		self.mark_post_as_processed(post_id)

		def send_comment(self, post_id, comment):
		@staticmethod
		def _parse_comment_id_from_mr_description(description):
		"""Extract comment_id from MR description HTML comment marker, or None."""
		marker_start = description.find(COMMENT_MARKER_PREFIX)
		if marker_start == -1:
		return None
		value_start = marker_start + len(COMMENT_MARKER_PREFIX)
		value_end = description.find(COMMENT_MARKER_SUFFIX, value_start)
		if value_end == -1:
		return None
		try:
		return int(description[value_start:value_end])
		except ValueError:
		return None

		def _store_comment_id_in_mr_description(self, mr_branch, comment_id):
		"""Append the comment_id marker to the MR description."""
		current_desc = self.forge.get_mr_description(mr_branch)
		marker = "{}{}{}".format(
		COMMENT_MARKER_PREFIX, comment_id, COMMENT_MARKER_SUFFIX
		)
		# Remove any existing marker first
		new_desc = current_desc
		while COMMENT_MARKER_PREFIX in new_desc:
		start = new_desc.find(COMMENT_MARKER_PREFIX)
		end = new_desc.find(COMMENT_MARKER_SUFFIX, start)
		if end == -1:
		break
		new_desc = new_desc[:start] + new_desc[end + len(COMMENT_MARKER_SUFFIX):]
		# Append the marker
		new_desc = new_desc.rstrip() + "\n\n" + marker
		self.forge.set_mr_description(mr_branch, new_desc)

		def send_comment(self, post_id, comment, mr_branch=None):
		post_id_str = str(post_id)

		if self.dry_run:
		print("DRY RUN: would comment on post " + str(post_id) + ": " + comment)
		print("DRY RUN: would comment on post " + post_id_str + ": " + comment)
		return

		# If we have an MR, check for an existing comment to edit
		if mr_branch:
		mr_desc = self.forge.get_mr_description(mr_branch)
		existing_comment_id = self._parse_comment_id_from_mr_description(mr_desc)
		if existing_comment_id is not None:
		# Edit the existing comment on Lemmy
		req = requests.put(ENDPOINTS["comment"], json={
		"comment_id": existing_comment_id,
		"content": comment,
		}, headers={
		"Authorization": "Bearer " + JWT
		})
		req.raise_for_status()
		print("Edited comment https://toast.ooo/post/" + post_id_str + "/" + str(existing_comment_id))
		return

		# No existing comment — create a new one
		req = requests.post(ENDPOINTS["comment"], json={
		"content": comment,
		"post_id": post_id,
		}, headers={
		"Authorization": "Bearer " + JWT
		})
		assert req.status_code == 200
		req.raise_for_status()
		comment_id = req.json()["comment_view"]["comment"]["id"]
		print("Created comment https://toast.ooo/post/" + str(post_id) + "/" + str(comment_id))
		print("Created comment https://toast.ooo/post/" + post_id_str + "/" + str(comment_id))

		# Persist the comment_id in the MR description
		if mr_branch:
		self._store_comment_id_in_mr_description(mr_branch, comment_id)

		def crawl_latest(self):
		req = requests.get(ENDPOINTS["list_posts"], params={

tools/lib/gitlab_forge.py

+33 −5

Original line number	Diff line number	Diff line
		@@ -23,15 +23,43 @@ def from_env(self):
		def mr_branch(self, post_id: int):
		return "lemmy-" + str(post_id)

		def does_pr_already_exist(self, from_branch):
		url = "https://{}/api/v4/projects/{}/merge_requests?source_branch={}&target_branch={}".format(self.domain, self.project_id, from_branch, "main")
		def get_mr_by_branch(self, branch_name):
		url = "https://{}/api/v4/projects/{}/merge_requests?source_branch={}&target_branch={}".format(self.domain, self.project_id, branch_name, "main")
		req = requests.get(url)
		req.raise_for_status()
		results = req.json()
		if len(results) == 0:
		return None
		return results[0]

		if len(req.json()) == 0:
		def does_pr_already_exist(self, from_branch):
		mr = self.get_mr_by_branch(from_branch)
		if mr is None:
		return False
		else:
		return req.json()[0]["web_url"]
		return mr["web_url"]

		def get_mr_description(self, branch_name):
		mr = self.get_mr_by_branch(branch_name)
		if mr is None:
		return ""
		return mr.get("description", "")

		def set_mr_description(self, branch_name, description):
		mr = self.get_mr_by_branch(branch_name)
		if mr is None:
		raise Exception("No MR found for branch {}".format(branch_name))

		if self.dry_run:
		print("DRY RUN: would update MR {} description".format(mr["web_url"]))
		return

		url = "https://{}/api/v4/projects/{}/merge_requests/{}".format(
		self.domain, self.project_id, mr["iid"]
		)
		req = requests.put(url, json={"description": description}, headers={
		"PRIVATE-TOKEN": self.token
		})
		req.raise_for_status()

		def make_pr_between_branches(self, from_branch, to_branch, title, body):
		if self.dry_run: