fetcher: track errors in an issue to prevent duplicate comments (a4526fc5) · Commits · fediverse.events / Canvas Atlas · GitLab

tools/lemmy_fetcher.py

+122 −17

Original line number	Diff line number	Diff line
		import json
		import re
		from datetime import datetime
		import subprocess
		import requests
		@@ -13,8 +14,7 @@
		# cached in gitlab-ci
		STATE_FOLDER = {
		"ROOT": ".fetcher/",
		"processed": ".fetcher/processed_posts",
		"failed": ".fetcher/failed_posts"
		"processed": ".fetcher/processed_posts"
		}

		LEMMY_HOST = "toast.ooo"
		@@ -50,6 +50,14 @@
		COMMENT_MARKER_PREFIX = "<!-- lemmy_comment_id: " + LEMMY_HOST + "/"
		COMMENT_MARKER_SUFFIX = " -->"

		TRACKING_ROW_RE = re.compile(
		r"\\| <!-- pid:(\d+) bh:(\S+) cid:([^ ]*) -->(.+?)"
		r" \\| ([^\|]*)"
		r" \\| ([^\|]*)"
		r" \\| (.+?) \\|"
		)
		TRACKING_TABLE_HEADER = "\| Post \| URL \| Comment \| Error \|\n\|------\|-----\|---------\|-------\|"

		def login():
		global JWT
		login_req = requests.post(ENDPOINTS["login"], json={
		@@ -83,28 +91,118 @@ def __init__(self, edition, dry_run=False):
		for filename in os.listdir(STATE_FOLDER["processed"]):
		self.read_ids.add(filename)

		self.failed_ids_hashes = set()
		for filename in os.listdir(STATE_FOLDER["failed"]):
		self.failed_ids_hashes.add(filename)
		self.tracking_issue_iid = self._ensure_tracking_issue()
		self.tracking_rows = {}
		if self.tracking_issue_iid is not None and self.tracking_issue_iid != -1:
		desc = self.forge.get_issue_description(self.tracking_issue_iid)
		self.tracking_rows = self._parse_tracking_rows(desc)

		def prepare_dirs(self):
		os.makedirs(STATE_FOLDER["ROOT"], exist_ok=True)
		os.makedirs(STATE_FOLDER["processed"], exist_ok=True)
		os.makedirs(STATE_FOLDER["failed"], exist_ok=True)

		@staticmethod
		def _parse_tracking_rows(description):
		"""Parse tracking issue body into dict: {post_id: {body_hash, comment_id, title, post_url, comment_url, error}}"""
		rows = {}
		for match in TRACKING_ROW_RE.finditer(description):
		pid = int(match.group(1))
		body_hash = match.group(2)
		cid_str = match.group(3)
		title = match.group(4).strip()
		post_url = match.group(5).strip()
		comment_url = match.group(6).strip()
		error = match.group(7).strip()
		rows[pid] = {
		"body_hash": body_hash,
		"comment_id": int(cid_str) if cid_str else None,
		"title": title,
		"post_url": post_url,
		"comment_url": comment_url,
		"error": error,
		}
		return rows

		@staticmethod
		def _build_tracking_body(rows_dict):
		"""Build the full issue body from a rows dict (same shape as _parse_tracking_rows output)."""
		lines = [
		"# Error Tracking\n\n",
		"Posts that failed JSON parsing. The bot checks this issue before commenting.\n\n",
		TRACKING_TABLE_HEADER + "\n",
		]
		for pid in sorted(rows_dict.keys()):
		r = rows_dict[pid]
		cid_str = str(r["comment_id"]) if r["comment_id"] is not None else ""
		comment_url = r.get("comment_url", "")
		row = "\| <!-- pid:{} bh:{} cid:{} -->{} \| {} \| {} \| {} \|".format(
		pid,
		r["body_hash"],
		cid_str,
		r["title"],
		r.get("post_url", "https://toast.ooo/post/{}".format(pid)),
		comment_url,
		r["error"],
		)
		lines.append(row + "\n")
		return "".join(lines)

		def _ensure_tracking_issue(self):
		"""Find or create the tracking issue. Returns iid, or -1 on dry_run."""
		if self.dry_run:
		return -1
		return self.forge.find_or_create_tracking_issue()

		def is_post_failed(self, post_id, body):
		entry_key = str(post_id) + "-" + self.hash_body(body)
		return entry_key in self.failed_ids_hashes
		body_hash = self.hash_body(body)
		if post_id in self.tracking_rows:
		return self.tracking_rows[post_id]["body_hash"] == body_hash
		return False

		def mark_post_as_failed(self, post_id, body):
		def mark_post_as_failed(self, post_id, body, error="", title=""):
		if self.dry_run:
		print("DRY RUN: would mark post as processed: " + str(post_id))
		print("DRY RUN: would add post {} to tracking issue".format(post_id))
		return

		entry_key = str(post_id) + "-" + self.hash_body(body)
		self.failed_ids_hashes.add(entry_key)
		with open(STATE_FOLDER["failed"] + "/" + entry_key, "w") as f:
		f.write(body)
		body_hash = self.hash_body(body)
		existing = self.tracking_rows.get(post_id)

		if existing and existing["body_hash"] == body_hash:
		# Already tracked with same body — no update needed
		return

		# Upsert the row
		self.tracking_rows[post_id] = {
		"body_hash": body_hash,
		"comment_id": existing["comment_id"] if existing else None,
		"title": title,
		"post_url": "https://toast.ooo/post/{}".format(post_id),
		"comment_url": existing.get("comment_url", "") if existing else "",
		"error": error,
		}
		new_body = self._build_tracking_body(self.tracking_rows)
		self.forge.set_issue_description(self.tracking_issue_iid, new_body)

		def _set_comment_id_in_tracking(self, post_id, comment_id):
		"""After creating a comment for a failed post, store the comment_id in the tracking row."""
		if post_id not in self.tracking_rows:
		return
		self.tracking_rows[post_id]["comment_id"] = comment_id
		self.tracking_rows[post_id]["comment_url"] = "https://toast.ooo/post/{}/{}".format(post_id, comment_id)
		new_body = self._build_tracking_body(self.tracking_rows)
		self.forge.set_issue_description(self.tracking_issue_iid, new_body)

		def _remove_from_tracking_issue(self, post_id):
		"""Remove a post from the tracking issue after successful processing."""
		if post_id not in self.tracking_rows:
		return
		if self.dry_run:
		print("DRY RUN: would remove post {} from tracking issue".format(post_id))
		return
		del self.tracking_rows[post_id]
		new_body = self._build_tracking_body(self.tracking_rows)
		self.forge.set_issue_description(self.tracking_issue_iid, new_body)
		print("Removed post {} from tracking issue".format(post_id))

		def mark_post_as_processed(self, post_id):
		if self.dry_run:
		@@ -160,7 +258,9 @@ def process_post(self, post_body, make_pr):
		print("Post already failed with same body: " + str(post_id))
		return
		else:
		self.mark_post_as_failed(post_id, post_body_main_source)
		post_title = post_body.get("name", "Unknown")
		self.mark_post_as_failed(post_id, post_body_main_source,
		error=str(ex), title=post_title)
		self.send_comment(post_id, FetcherMessages.JsonError(str(ex)))
		return

		@@ -197,6 +297,8 @@ def process_post(self, post_body, make_pr):
		else:
		print("branch pushed, PR already exists")
		self.send_comment(post_id, FetcherMessages.SuccessfulSubmission(pr_url), mr_branch=new_branch_name)
		# If this post was previously in the error tracking issue, remove it
		self._remove_from_tracking_issue(post_id)
		else:
		self.git.restore(EDITIONS[self.edition]["entry_folder"])
		self.git.checkout("main")
		@@ -273,6 +375,9 @@ def send_comment(self, post_id, comment, mr_branch=None):
		# Persist the comment_id in the MR description
		if mr_branch:
		self._store_comment_id_in_mr_description(mr_branch, comment_id)
		else:
		# JSON error path — store comment_id in tracking issue
		self._set_comment_id_in_tracking(post_id, comment_id)

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

tools/lib/gitlab_forge.py

+75 −3

Original line number	Diff line number	Diff line
		import requests
		from urllib.parse import quote

		from lib.forge_base import ForgeBase
		import os
		@@ -61,6 +62,77 @@ def set_mr_description(self, branch_name, description):
		})
		req.raise_for_status()

		TRACKING_ISSUE_TITLE = "🤖 Lemmy Fetcher Error Tracking"

		def find_or_create_tracking_issue(self):
		"""Find the tracking issue by title, create if not found. Returns issue_iid."""
		url = "https://{}/api/v4/projects/{}/issues?search={}&per_page=50".format(
		self.domain, self.project_id, quote(self.TRACKING_ISSUE_TITLE)
		)
		req = requests.get(url, headers={"PRIVATE-TOKEN": self.token})
		req.raise_for_status()
		for issue in req.json():
		if issue.get("title") == self.TRACKING_ISSUE_TITLE:
		print("Found tracking issue #{}".format(issue["iid"]))
		return issue["iid"]

		# Create it
		initial_body = (
		"# Error Tracking\n\n"
		"Posts that failed JSON parsing. The bot checks this issue before commenting.\n\n"
		"\| Post \| URL \| Comment \| Error \|\n"
		"\|------\|-----\|---------\|-------\|\n"
		)
		if self.dry_run:
		print("DRY RUN: would create tracking issue '{}'".format(self.TRACKING_ISSUE_TITLE))
		return -1

		url = "https://{}/api/v4/projects/{}/issues".format(self.domain, self.project_id)
		req = requests.post(url, json={
		"title": self.TRACKING_ISSUE_TITLE,
		"description": initial_body,
		}, headers={"PRIVATE-TOKEN": self.token})
		req.raise_for_status()
		iid = req.json()["iid"]
		print("Created tracking issue #{}".format(iid))
		return iid

		def get_issue_description(self, iid):
		"""Get the description (body) of an issue."""
		if self.dry_run and iid == -1:
		return ""
		url = "https://{}/api/v4/projects/{}/issues/{}".format(
		self.domain, self.project_id, iid
		)
		req = requests.get(url, headers={"PRIVATE-TOKEN": self.token})
		req.raise_for_status()
		return req.json().get("description", "")

		def set_issue_description(self, iid, description):
		"""Update the description (body) of an issue."""
		if self.dry_run:
		print("DRY RUN: would update tracking issue #{} description".format(iid))
		return
		url = "https://{}/api/v4/projects/{}/issues/{}".format(
		self.domain, self.project_id, iid
		)
		req = requests.put(url, json={"description": description}, headers={
		"PRIVATE-TOKEN": self.token
		})
		req.raise_for_status()

		def reopen_issue(self, iid):
		"""Ensure the tracking issue is open (not closed)."""
		if self.dry_run:
		return
		url = "https://{}/api/v4/projects/{}/issues/{}".format(
		self.domain, self.project_id, iid
		)
		req = requests.put(url, json={"state_event": "reopen"}, 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:
		print("DRY RUN: would create mr " + from_branch + " onto " + to_branch + " ; " + title + " ; " + body)