diff --git a/bluesky/input.py b/bluesky/input.py index ec69f3a..e8c743d 100644 --- a/bluesky/input.py +++ b/bluesky/input.py @@ -53,10 +53,7 @@ class BlueskyBaseInputService(BlueskyService, InputService, ABC): post_cid = cast(str, record["$xpost.strongRef"]["cid"]) if self._is_post_crossposted(self.url, self.did, post_uri): - self.log.info( - "Skipping '%s': already crossposted", - post_uri, - ) + self.log.info("Skipping '%s': already crossposted", post_uri) return parent_uri = cast( @@ -167,7 +164,7 @@ class BlueskyBaseInputService(BlueskyService, InputService, ABC): "service": self.url, "identifier": post_uri, "parent": parent["id"], - "root": parent["id"] if not parent["root"] else parent["root"], + "root": parent["root"] or parent["id"], "extra_data": json.dumps({"cid": post_cid}), } ) diff --git a/bluesky/output.py b/bluesky/output.py index f4d5829..ff9257b 100644 --- a/bluesky/output.py +++ b/bluesky/output.py @@ -195,48 +195,40 @@ class BlueskyOutputService(BlueskyService, OutputService): new_parent_id: int | None = None if post.parent_id: - parent = self._get_post(post.service, post.author, post.parent_id) - if not parent: - self.log.error( - "Skipping '%s': parent post not found in db", post.parent_id - ) - return - thread = self._find_mapped_thread( - parent["identifier"], - parent["service"], - parent["user"], + post.parent_id, + post.service, + post.author, self.url, self.did, ) if not thread: self.log.error( - "Skipping '%s': parent thread tuple not found in db", - post.parent_id, + "Skipping '%s': parent thread tuple not found in db", post.id ) return - root_uri, reply_uri, root_db_id, reply_db_id = thread + root_uri, reply_uri, new_root_id, new_parent_id = thread - root_post = self._get_post(self.url, self.did, root_uri) - reply_post = self._get_post(self.url, self.did, reply_uri) + root_post = self._get_post_by_id(new_root_id) + reply_post = self._get_post_by_id(new_parent_id) if not root_post or not reply_post: - self.log.error("Skipping '%s': failed to fetch parent posts from db") + self.log.error( + "Skipping '%s': failed to fetch parent posts from db", post.id + ) return root_cid = cid_from_json(root_post["extra_data"]) reply_cid = cid_from_json(reply_post["extra_data"]) if not root_cid or not reply_cid: - self.log.error("Skipping '%s': failed to parse CID from db") + self.log.error("Skipping '%s': failed to parse CID from db", post.id) return root_ref = StrongRef(uri=root_uri, cid=root_cid) reply_ref = StrongRef(uri=reply_uri, cid=reply_cid) reply_to = ReplyRef(root=root_ref, parent=reply_ref) - new_root_id = root_db_id - new_parent_id = reply_db_id labels_attachment = post.attachments.get(LabelsAttachment) spoiler: str | None = ( @@ -305,24 +297,26 @@ class BlueskyOutputService(BlueskyService, OutputService): quoted_uri: str | None = None if quote_attachment: if quote_attachment.quoted_user != post.author: - self.log.info("Skipping '%s': quoted other user") + self.log.info("Skipping '%s': quoted other user", post.id) return quoted_post = self._get_post( post.service, post.author, quote_attachment.quoted_id ) if not quoted_post: - self.log.error("Skipping '%s': quoted post not found in db!") + self.log.error("Skipping '%s': quoted post not found in db!", post.id) return quoted_mappings = self._get_mappings(quoted_post["id"], self.url, self.did) if not quoted_mappings: - self.log.error("Skipping '%s': failed to find mappings for quoted post") + self.log.error( + "Skipping '%s': failed to find mappings for quoted post", post.id + ) return quoted_cid = cid_from_json(quoted_mappings[0]["extra_data"]) if not quoted_cid: - self.log.error("Skipping '%s': failed to parse CID from db") + self.log.error("Skipping '%s': failed to parse CID from db", post.id) return quoted_uri = quoted_mappings[0]["identifier"] @@ -374,6 +368,7 @@ class BlueskyOutputService(BlueskyService, OutputService): created_records: list[tuple[str, str]] = [] post_root_ref: StrongRef | None = None previous_reply_ref: StrongRef | None = None + original_thread_root: StrongRef | None = reply_to.root if reply_to else None richtext_index = 0 @@ -389,9 +384,8 @@ class BlueskyOutputService(BlueskyService, OutputService): if i == 0: current_reply_to = reply_to elif previous_reply_ref and post_root_ref: - current_reply_to = ReplyRef( - root=post_root_ref, parent=previous_reply_ref - ) + root_ref = original_thread_root or post_root_ref + current_reply_to = ReplyRef(root=root_ref, parent=previous_reply_ref) embed: dict[str, Any] | None = None if i == 0 and quoted_uri and quoted_cid: @@ -496,8 +490,9 @@ class BlueskyOutputService(BlueskyService, OutputService): self.options.quote_gate, ) + all_records = created_records if new_root_id is None or new_parent_id is None: - self._insert_post( + new_root_id = self._insert_post( { "user": self.did, "service": self.url, @@ -509,16 +504,12 @@ class BlueskyOutputService(BlueskyService, OutputService): "crossposted": 1, } ) - new_post = self._get_post(self.url, self.did, created_records[0][0]) - if not new_post: - raise ValueError("Inserted post not found!") - new_root_id = new_post["id"] new_parent_id = new_root_id + created_records = created_records[1:] + self._insert_post_mapping(db_post["id"], new_parent_id) - self._insert_post_mapping(db_post["id"], new_parent_id) - - for uri, cid in created_records[1:]: - self._insert_post( + for uri, cid in created_records: + new_parent_id = self._insert_post( { "user": self.did, "service": self.url, @@ -530,16 +521,12 @@ class BlueskyOutputService(BlueskyService, OutputService): "crossposted": 1, } ) - reply_post = self._get_post(self.url, self.did, uri) - if not reply_post: - raise ValueError("Inserted reply post not found!") - new_parent_id = reply_post["id"] self._insert_post_mapping(db_post["id"], new_parent_id) self.log.info( "Post accepted successfully: %s -> %s", post.id, - [r[0] for r in created_records], + [r[0] for r in all_records], ) @override @@ -560,7 +547,7 @@ class BlueskyOutputService(BlueskyService, OutputService): db_repost = self._get_post(repost.service, repost.author, repost.id) db_reposted = self._get_post(reposted.service, reposted.author, reposted.id) if not db_repost or not db_reposted: - self.log.info("Skipping repost '%s': post not found in db") + self.log.info("Skipping repost '%s': post not found in db", repost.id) return mappings = self._get_mappings(db_reposted["id"], self.url, self.did) @@ -569,7 +556,9 @@ class BlueskyOutputService(BlueskyService, OutputService): cid = cid_from_json(mappings[0]["extra_data"]) if not cid: - self.log.exception("Skipping '%s': failed to parse CID from extra_data") + self.log.exception( + "Skipping repost '%s': failed to parse CID from extra_data", repost.id + ) return response = self._client.repost(mappings[0]["identifier"], cid) diff --git a/cross/service.py b/cross/service.py index d77f02a..efb8da0 100644 --- a/cross/service.py +++ b/cross/service.py @@ -60,7 +60,7 @@ class Service: WHERE m.original = ? AND p.service = ? AND p.user = ? - ORDER BY p.id; + ORDER BY p.id ASC; """, (original, service, user), ) @@ -68,7 +68,7 @@ class Service: def _find_mapped_thread( self, parent: str, iservice: str, iuser: str, oservice: str, ouser: str - ): + ) -> tuple[str, str, int, int] | None: reply_data = self._get_post(iservice, iuser, parent) if not reply_data: return None @@ -95,16 +95,18 @@ class Service: return ( root_identifier["identifier"], # real ids reply_identifier["identifier"], - reply_data["root"], # db ids - reply_data["id"], + root_identifier["id"], # db ids + reply_identifier["id"], ) - def _insert_post(self, post_data: dict[str, Any]): + def _insert_post(self, post_data: dict[str, Any]) -> int: values = [post_data.get(col) for col in columns] cursor = self.db.get_conn().cursor() _ = cursor.execute( - f"INSERT INTO posts ({column_names}) VALUES ({placeholders})", values + f"INSERT INTO posts ({column_names}) VALUES ({placeholders}) RETURNING id", + values, ) + return int(cast(sqlite3.Row, cursor.fetchone())["id"]) def _insert_post_mapping(self, original: int, mapped: int): cursor = self.db.get_conn().cursor() diff --git a/mastodon/output.py b/mastodon/output.py index 04d0e97..166c11f 100644 --- a/mastodon/output.py +++ b/mastodon/output.py @@ -250,7 +250,7 @@ class MastodonOutputService(MastodonService, OutputService): def accept_post(self, post: Post): db_post = self._get_post(post.service, post.author, post.id) if not db_post: - self.log.error("Skipping '%s': post not found in db") + self.log.error("Skipping '%s': post not found in db", post.id) return new_root_id: int | None = None @@ -259,10 +259,16 @@ class MastodonOutputService(MastodonService, OutputService): reply_ref: str | None = None if post.parent_id: thread = self._find_mapped_thread( - post.parent_id, post.service, post.author, self.url, self.user_id + post.parent_id, + post.service, + post.author, + self.url, + self.user_id, ) if not thread: - self.log.error("Skipping '%s': parent thread tuple not found in db") + self.log.error( + "Skipping '%s': parent thread tuple not found in db", post.id + ) return _, reply_ref, new_root_id, new_parent_id = thread @@ -270,12 +276,12 @@ class MastodonOutputService(MastodonService, OutputService): quote = post.attachments.get(QuoteAttachment) if quote: if quote.quoted_user != post.author: - self.log.info("Skipping '%s': quote of other user") + self.log.info("Skipping '%s': quote of other user", post.id) return quoted_post = self._get_post(post.service, post.author, quote.quoted_id) if not quoted_post: - self.log.error("Skipping '%s': quoted post not found in db") + self.log.error("Skipping '%s': quoted post not found in db", post.id) return quoted_mappings = self._get_mappings( @@ -283,7 +289,7 @@ class MastodonOutputService(MastodonService, OutputService): ) if not quoted_mappings: self.log.error( - "Skipping '%s': mappings for quoted post not found in db" + "Skipping '%s': mappings for quoted post not found in db", post.id ) return @@ -316,7 +322,7 @@ class MastodonOutputService(MastodonService, OutputService): raw_statuses = self._split_tokens_and_media(post_tokens, media_blobs) if not raw_statuses: - self.log.error("Skipping '%s': couldn't split post into statuses") + self.log.error("Skipping '%s': couldn't split post into statuses", post.id) return baked_statuses: list[tuple[str, list[str] | None]] = [] @@ -325,7 +331,9 @@ class MastodonOutputService(MastodonService, OutputService): if raw_media: media_ids = self._upload_media(raw_media) if not media_ids: - self.log.error("Skipping '%s': failed to upload attachments") + self.log.error( + "Skipping '%s': failed to upload attachments", post.id + ) return baked_statuses.append((status_text, media_ids)) @@ -369,8 +377,9 @@ class MastodonOutputService(MastodonService, OutputService): if i == 0: reply_ref = status_id + all_statuses = created_statuses if new_root_id is None or new_parent_id is None: - self._insert_post( + new_root_id = self._insert_post( { "user": self.user_id, "service": self.url, @@ -382,16 +391,12 @@ class MastodonOutputService(MastodonService, OutputService): "crossposted": 1, } ) - new_post = self._get_post(self.url, self.user_id, created_statuses[0]) - if not new_post: - raise ValueError("Inserted post not found!") - new_root_id = new_post["id"] new_parent_id = new_root_id + created_statuses = created_statuses[1:] + self._insert_post_mapping(db_post["id"], new_parent_id) - self._insert_post_mapping(db_post["id"], new_parent_id) - - for status_id in created_statuses[1:]: - self._insert_post( + for status_id in created_statuses: + new_parent_id = self._insert_post( { "user": self.user_id, "service": self.url, @@ -403,19 +408,15 @@ class MastodonOutputService(MastodonService, OutputService): "crossposted": 1, } ) - reply_post = self._get_post(self.url, self.user_id, status_id) - if not reply_post: - raise ValueError("Inserted reply post not found!") - new_parent_id = reply_post["id"] self._insert_post_mapping(db_post["id"], new_parent_id) - self.log.info("Post accepted successfully: %s -> %s", post.id, created_statuses) + self.log.info("Post accepted successfully: %s -> %s", post.id, all_statuses) @override def delete_post(self, post: PostRef): db_post = self._get_post(post.service, post.author, post.id) if not db_post: - self.log.warning("Skipping delete '%s': post not found in db: %s", post.id) + self.log.warning("Skipping delete '%s': post not found in db", post.id) return mappings = self._get_mappings(db_post["id"], self.url, self.user_id) @@ -434,12 +435,16 @@ class MastodonOutputService(MastodonService, OutputService): def accept_repost(self, repost: PostRef, reposted: PostRef): original = self._get_post(reposted.service, reposted.author, reposted.id) if not original: - self.log.info("Skipping repost '%s': reposted post not found in db") + self.log.info( + "Skipping repost '%s': reposted post not found in db", repost.id + ) return mappings = self._get_mappings(original["id"], self.url, self.user_id) if not mappings: - self.log.error("Skipping repost '%s': no mappings found for reposted post") + self.log.error( + "Skipping repost '%s': no mappings found for reposted post", repost.id + ) return response = self.http.post( @@ -466,9 +471,7 @@ class MastodonOutputService(MastodonService, OutputService): original_repost = self._get_post(repost.service, repost.author, repost.id) if not original_repost: - self.log.error( - "Skipping repost '%s': repost not found in db: %s", repost.id - ) + self.log.error("Skipping repost '%s': repost not found in db", repost.id) return self._insert_post_mapping(original_repost["id"], inserted["id"]) @@ -491,9 +494,7 @@ class MastodonOutputService(MastodonService, OutputService): return if not rmappings: self.log.warning( - "Skipping delete '%s': no mappings found for post", - repost.id, - db_repost["reposted"], + "Skipping delete '%s': no mappings found for post", repost.id ) return