From a7238685ca67e880581f9d59bef2b4d775c4e65d Mon Sep 17 00:00:00 2001 From: zenfyr Date: Fri, 20 Feb 2026 18:19:57 +0000 Subject: [PATCH] chore: clean-up logging --- atproto/xrpc.py | 9 +++------ bluesky/info.py | 4 ++-- bluesky/input.py | 50 ++++++++++++++++++++++++-------------------------- bluesky/output.py | 101 +++++++++++++++++++++++++++++++++-------------------------------------------------------------------- main.py | 6 +++--- mastodon/input.py | 48 ++++++++++++++++++++++-------------------------- mastodon/output.py | 130 +++++++++++++++++++++++++++++++++++++--------------------------------------------------------------------------------------------- misskey/input.py | 42 +++++++++++++++++++++--------------------- registry.py | 4 ++-- util/util.py | 3 ++- 10 file(s) changed, 149 insertion(s)(+), 248 deletion(s)(-) diff --git a/atproto/xrpc.py b/atproto/xrpc.py --- a/atproto/xrpc.py +++ b/atproto/xrpc.py @@ -57,9 +57,8 @@ if not session: return None if session.is_access_token_expired(): if not session.is_refresh_token_expired(): - LOGGER.info("refreshing session for %s", session.did) + LOGGER.info("Refreshing session for '%s'", session.did) return self.refresh_session(session) - LOGGER.info("both tokens expired for %s, removing session", session.did) self.store.remove_session(did) raise ValueError( "Both access and refresh tokens expired. Please login again." @@ -93,7 +92,7 @@ ) session = Session.from_dict(response.json(), self.pds_url) self.store.set_session(session) - LOGGER.info("Created session for %s (%s)", session.handle, session.did) + LOGGER.info("Created session for '%s'", session.did) return session def refresh_session(self, session: Session) -> Session: @@ -115,9 +114,7 @@ raise ValueError(f"Refresh failed with status {response.status_code}") new_session = Session.from_dict(response.json(), self.pds_url) self.store.set_session(new_session) - LOGGER.info( - "Refreshed session for %s (%s)", new_session.handle, new_session.did - ) + LOGGER.info("Refreshed session for '%s'", new_session.did) return new_session def get_access_token(self, did: str) -> str | None: diff --git a/bluesky/info.py b/bluesky/info.py --- a/bluesky/info.py +++ b/bluesky/info.py @@ -40,12 +40,12 @@ if not did: if not handle: raise KeyError("No did: or atproto handle provided!") - self.log.info("Resolving ATP identity for %s...", handle) + self.log.info("Resolving ATP identity for '%s'...", handle) identity = resolve_identity(handle, self._store) self.did = identity.did if not pds: - self.log.info("Resolving PDS for %s...", self.did) + self.log.info("Resolving PDS for '%s'...", self.did) identity = resolve_identity(self.did, self._store) self.pds = identity.pds diff --git a/bluesky/input.py b/bluesky/input.py --- a/bluesky/input.py +++ b/bluesky/input.py @@ -52,11 +52,9 @@ def _on_post(self, record: dict[str, Any]): post_uri = cast(str, record["$xpost.strongRef"]["uri"]) post_cid = cast(str, record["$xpost.strongRef"]["cid"]) - self.log.info("Processing new post: %s", post_uri) - if self._is_post_crossposted(self.url, self.did, post_uri): self.log.info( - "Skipping %s, already crossposted", + "Skipping '%s': already crossposted", post_uri, ) return @@ -66,10 +64,15 @@ str, None if not record.get("reply") else record["reply"]["parent"]["uri"] ) parent = None if parent_uri: + did, _, _ = AtUri.record_uri(parent_uri) + if did != self.did: + self.log.info("Skipping '%s': reply to other user..", post_uri) + return + parent = self._get_post(self.url, self.did, parent_uri) if not parent: self.log.info( - "Skipping %s, parent %s not found in db", post_uri, parent_uri + "Skipping '%s': parent '%s' not found in db", post_uri, parent_uri ) return @@ -101,13 +104,13 @@ else embed["record"] ) did, collection, _ = AtUri.record_uri(rcrd["uri"]) if collection != "app.bsky.feed.post": - return f"Unhandled record collection {collection}" + return f"unhandled record collection '{collection}'" if did != self.did: return "" rquote = self._get_post(self.url, did, rcrd["uri"]) if not rquote: - return f"Quote {rcrd['uri']} not found in the db" + return f"quote '{rcrd['uri']}' not found in db" post.attachments.put( QuoteAttachment(quoted_id=rcrd["uri"], quoted_user=did) ) @@ -124,23 +127,23 @@ blob_cid = embed["video"]["ref"]["$link"] url = f"{self.pds}/xrpc/com.atproto.sync.getBlob?did={self.did}&cid={blob_cid}" blob_urls.append((url, blob_cid, embed.get("alt"))) case _: - self.log.warning(f"Unhandled embed type {embed['$type']}") + self.log.warning(f"unhandled embed type '{embed['$type']}'") return None if embed: fexit = handle_embeds(embed) if fexit is not None: - self.log.info("Skipping %s! %s", post_uri, fexit) + self.log.info("Skipping '%s': %s", post_uri, fexit) return if blob_urls: blobs: list[Blob] = [] for url, cid, alt in blob_urls: - self.log.info("Downloading %s...", cid) + self.log.info("Downloading '%s'...", cid) blob: Blob | None = download_blob(url, alt, client=self.http) if not blob: self.log.error( - "Skipping %s! Failed to download blob %s.", post_uri, cid + "Skipping '%s': failed to download blob '%s'", post_uri, cid ) return blobs.append(blob) @@ -178,8 +181,7 @@ "extra_data": json.dumps({"cid": post_cid}), } ) - self.log.info("Post stored in DB: %s", post_uri) - + self.log.info("Crossposting: '%s'", post_uri) for out in self.outputs: self.submitter(lambda: out.accept_post(post)) @@ -187,13 +189,11 @@ def _on_repost(self, record: dict[str, Any]): post_uri = cast(str, record["$xpost.strongRef"]["uri"]) post_cid = cast(str, record["$xpost.strongRef"]["cid"]) - self.log.info("Processing repost: %s", post_uri) - reposted_uri = cast(str, record["subject"]["uri"]) reposted = self._get_post(self.url, self.did, reposted_uri) if not reposted: self.log.info( - "Skipping repost '%s' as reposted post '%s' was not found in the db.", + "Skipping repost '%s': reposted post '%s' not found in db", post_uri, reposted_uri, ) @@ -209,31 +209,29 @@ "extra_data": json.dumps({"cid": post_cid}), } ) - self.log.info("Repost stored in DB: %s", post_uri) - repost_ref = PostRef(id=post_uri, author=self.did, service=self.url) reposted_ref = PostRef(id=reposted_uri, author=self.did, service=self.url) + + self.log.info("Crossposting: '%s'", post_uri) for out in self.outputs: self.submitter(lambda: out.accept_repost(repost_ref, reposted_ref)) def _on_delete_post(self, post_id: str, repost: bool): - self.log.info("Processing delete for %s (repost: %s)...", post_id, repost) post = self._get_post(self.url, self.did, post_id) if not post: - self.log.warning("Post not found in DB: %s", post_id) + self.log.warning("Skipping delete '%s': post not found in db", post_id) return post_ref = PostRef(id=post_id, author=self.did, service=self.url) if repost: - self.log.info("Deleting repost: %s", post_id) + self.log.info("Deleting repost: '%s'", post_id) for output in self.outputs: self.submitter(lambda: output.delete_repost(post_ref)) else: - self.log.info("Deleting post: %s", post_id) + self.log.info("Deleting post: '%s'", post_id) for output in self.outputs: self.submitter(lambda: output.delete_post(post_ref)) self.submitter(lambda: self._delete_post_by_id(post["id"])) - self.log.info("Delete successful for %s", post_id) class BlueskyJetstreamInputService(BlueskyBaseInputService): @@ -304,7 +302,7 @@ ping_timeout=10, close_timeout=5, ): try: - self.log.info("Listening to %s...", env.JETSTREAM_URL) + self.log.info("Listening to '%s'...", env.JETSTREAM_URL) async def listen_for_messages(): async for msg in ws: @@ -315,9 +313,9 @@ _ = await asyncio.gather(listen) except websockets.ConnectionClosedError as e: self.log.error(e, stack_info=True, exc_info=True) - self.log.info("Reconnecting to %s...", env.JETSTREAM_URL) + self.log.info("Reconnecting to '%s'...", env.JETSTREAM_URL) continue except TimeoutError as e: - self.log.error("Connection timeout: %s", e) - self.log.info("Reconnecting to %s...", env.JETSTREAM_URL) + self.log.error("Connection timeout: '%s'", e) + self.log.info("Reconnecting to '%s'...", env.JETSTREAM_URL) continue diff --git a/bluesky/output.py b/bluesky/output.py --- a/bluesky/output.py +++ b/bluesky/output.py @@ -56,7 +56,7 @@ def from_dict(cls, data: dict[str, Any]) -> "BlueskyOutputOptions": validate_and_transform(data) if "password" not in data: - raise KeyError("password is required for bluesky") + raise KeyError("'password' is required for bluesky") if "quote_gate" in data: data["quote_gate"] = bool(data["quote_gate"]) @@ -96,7 +96,7 @@ http, self.options.password, ) self.options.password = "" - self.log.info("Logged in as %s", self.did) + self.log.info("Logged in as '%s'", self.did) @override def get_identity_options(self) -> tuple[str | None, str | None, str | None]: @@ -184,12 +184,11 @@ return SelfLabels(values=[SelfLabel(val=label) for label in unique_labels]) @override def accept_post(self, post: Post): - self.log.info( - "Accepting post %s (author: %s, service: %s)...", - post.id, - post.author, - post.service, - ) + 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") + return + reply_to: ReplyRef | None = None new_root_id: int | None = None new_parent_id: int | None = None @@ -197,7 +196,9 @@ if post.parent_id: parent = self._get_post(post.service, post.author, post.parent_id) if not parent: - self.log.error("Parent post not found in DB: %s", post.parent_id) + self.log.error( + "Skipping '%s': parent post not found in db", post.parent_id + ) return thread = self._find_mapped_thread( @@ -209,7 +210,7 @@ self.did, ) if not thread: self.log.error( - "Failed to find thread tuple in the database for parent: %s", + "Skipping '%s': parent thread tuple not found in db", post.parent_id, ) return @@ -220,7 +221,7 @@ root_post = self._get_post(self.url, self.did, root_uri) reply_post = self._get_post(self.url, self.did, reply_uri) if not root_post or not reply_post: - self.log.error("Failed to fetch parent posts from database!") + self.log.error("Skipping '%s': failed to fetch parent posts from db") return try: @@ -233,7 +234,7 @@ reply_cid = ( json.loads(reply_cid_data).get("cid", "") if reply_cid_data else "" ) except (json.JSONDecodeError, AttributeError, KeyError): - self.log.error("Failed to parse CID from database!") + self.log.error("Skipping '%s': failed to parse CID from db") return root_ref = StrongRef(uri=root_uri, cid=root_cid) @@ -309,26 +310,30 @@ quoted_cid: str | None = None quoted_uri: str | None = None if quote_attachment: if quote_attachment.quoted_user != post.author: - self.log.info("Quoted other user, skipping quote!") + self.log.info("Skipping '%s': quoted other user") return quoted_post = self._get_post( post.service, post.author, quote_attachment.quoted_id ) if not quoted_post: - self.log.error("Failed to find quoted post in the database!") + self.log.error("Skipping '%s': quoted post not found in db!") else: quoted_mappings = self._get_mappings( quoted_post["id"], self.url, self.did ) if not quoted_mappings: - self.log.error("Failed to find mappings for quoted post!") + self.log.error( + "Skipping '%s': failed to find mappings for quoted post" + ) else: bluesky_quoted_post = self._get_post( self.url, self.did, quoted_mappings[0]["identifier"] ) if not bluesky_quoted_post: - self.log.error("Failed to find Bluesky quoted post!") + self.log.error( + "Skipping '%s': Failed to find Bluesky quoted post!" + ) else: quoted_cid_data = bluesky_quoted_post["extra_data"] quoted_cid = ( @@ -342,28 +347,26 @@ splitter = TokenSplitter(max_chars=300, max_link_len=30) token_blocks = splitter.split(tokens) if token_blocks is None: - self.log.error( - "Skipping '%s' as it contains links/tags that are too long!", post.id - ) + self.log.error("Skipping '%s': links/tags are too long", post.id) return for blob in supported_media: if blob.mime.startswith("image/") and len(blob.io) > 2_000_000: self.log.error( - "Skipping post '%s', image too large!", + "Skipping '%s': image too large", post.id, ) return if blob.mime.startswith("video/"): if blob.mime != "video/mp4" and not self.options.encode_videos: self.log.info( - "Video is not mp4, but encoding is disabled. Skipping '%s'...", + "Skipping '%s': video is not mp4, but encoding is disabled", post.id, ) return if len(blob.io) > 100_000_000: self.log.error( - "Skipping post '%s', video too large!", + "Skipping '%s': video too large", post.id, ) return @@ -378,7 +381,7 @@ for block in token_blocks: result = tokens_to_richtext(block) if result is None: self.log.error( - "Skipping '%s' as it contains invalid rich text types!", + "Skipping '%s': invalid rich text types", post.id, ) return @@ -479,7 +482,7 @@ duration = None if duration and duration > 180: self.log.info( - "Skipping post '%s', video too long (%.1f > 180s)!", + "Skipping '%s': video too long (%.1f > 180s)", post.id, duration, ) @@ -509,11 +512,6 @@ self.options.thread_gate, self.options.quote_gate, ) - db_post = self._get_post(post.service, post.author, post.id) - if not db_post: - self.log.error("Post not found in database!") - return - if new_root_id is None or new_parent_id is None: self._insert_post( { @@ -562,42 +560,23 @@ ) @override def delete_post(self, post: PostRef): - self.log.info( - "Deleting post %s (author: %s, service: %s)...", - post.id, - post.author, - post.service, - ) db_post = self._get_post(post.service, post.author, post.id) if not db_post: - self.log.warning( - "Post not found in DB: %s (author: %s, service: %s)", - post.id, - post.author, - post.service, - ) + self.log.warning("Skipping delete '%s': post not found in db", post.id) return mappings = self._get_mappings(db_post["id"], self.url, self.did) for mapping in mappings[::-1]: - self.log.info("Deleting '%s'...", mapping["identifier"]) self._client.delete_post(mapping["identifier"]) self._delete_post_by_id(mapping["id"]) - self.log.info("Post deleted successfully: %s", post.id) + self.log.info("Deleted '%s'", post.id) @override def accept_repost(self, repost: PostRef, reposted: PostRef): - self.log.info( - "Accepting repost %s of %s (author: %s, service: %s)...", - repost.id, - reposted.id, - repost.author, - repost.service, - ) 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("Post not found in db, skipping repost..") + self.log.info("Skipping repost '%s': post not found in db") return mappings = self._get_mappings(db_reposted["id"], self.url, self.did) @@ -607,7 +586,7 @@ try: cid = json.loads(mappings[0]["extra_data"])["cid"] except (json.JSONDecodeError, AttributeError, KeyError): - self.log.exception("Failed to parse CID from extra_data!") + self.log.exception("Skipping '%s': failed to parse CID from extra_data") return response = self._client.repost(mappings[0]["identifier"], cid) @@ -632,27 +611,13 @@ self.log.info("Repost accepted successfully: %s", repost.id) @override def delete_repost(self, repost: PostRef): - self.log.info( - "Deleting repost %s (author: %s, service: %s)...", - repost.id, - repost.author, - repost.service, - ) db_repost = self._get_post(repost.service, repost.author, repost.id) if not db_repost: - self.log.warning( - "Repost not found in DB: %s (author: %s, service: %s)", - repost.id, - repost.author, - repost.service, - ) + self.log.warning("Skipping delete '%s': repost not found in db", repost.id) return mappings = self._get_mappings(db_repost["id"], self.url, self.did) if mappings: - self.log.info("Deleting '%s'...", mappings[0]["identifier"]) self._client.delete_repost(mappings[0]["identifier"]) self._delete_post_by_id(mappings[0]["id"]) - self.log.info("Repost deleted successfully: %s", repost.id) - else: - self.log.error([mappings]) + self.log.info("Deleted %s", repost.id) diff --git a/main.py b/main.py --- a/main.py +++ b/main.py @@ -101,14 +101,14 @@ settings = json.load(f) read_env(settings) if "services" not in settings: - raise KeyError("No `services` specified in settings!") + raise KeyError("No 'services' specified in settings!") service_pairs: list[tuple[Any, list[Any]]] = [] for svc in settings["services"]: if "input" not in svc: - raise KeyError("Each service must have an `input` field!") + raise KeyError("Each service must have an 'input' field!") if "outputs" not in svc: - raise KeyError("Each service must have an `outputs` field!") + raise KeyError("Each service must have an 'outputs' field!") inp = create_input_service(db_pool, http_client, svc["input"]) outs = [ diff --git a/mastodon/input.py b/mastodon/input.py --- a/mastodon/input.py +++ b/mastodon/input.py @@ -56,11 +56,11 @@ ) -> None: super().__init__(options.instance, db, http) self.options: MastodonInputOptions = options - self.log.info("Verifying %s credentails...", self.url) + self.log.info("Verifying '%s' credentails...", self.url) response = self.verify_credentials() self.user_id: str = response["id"] - self.log.info("Getting %s configuration...", self.url) + self.log.info("Getting '%s' configuration...", self.url) response = self.fetch_instance_info() self.streaming_url: str = response["urls"]["streaming_api"] @@ -69,14 +69,12 @@ def _get_token(self) -> str: return self.options.token def _on_create_post(self, status: dict[str, Any]): - self.log.info("Processing new post: %s", status["id"]) - if status["account"]["id"] != self.user_id: return if status["visibility"] not in self.options.allowed_visibility: self.log.info( - "Skipping post with disallowed visibility: %s (%s)", + "Skipping '%s': disallowed visibility (%s)", status["id"], status["visibility"], ) @@ -84,7 +82,7 @@ return if self._is_post_crossposted(self.url, self.user_id, status["id"]): self.log.info( - "Skipping %s, already crossposted", + "Skipping '%s': already crossposted", status["id"], ) return @@ -97,7 +95,7 @@ self._on_reblog(status, reblog) return if status.get("poll"): - self.log.info("Skipping '%s'! Contains a poll..", status["id"]) + self.log.info("Skipping '%s': polls not supported", status["id"]) return quote: dict[str, Any] | None = status.get("quote") @@ -109,7 +107,9 @@ rquote = self._get_post(self.url, self.user_id, quote["id"]) if not rquote: self.log.info( - "Skipping %s, parent %s not found in db", status["id"], quote["id"] + "Skipping '%s': quoted post '%s' not found in db", + status["id"], + quote["id"], ) return @@ -123,7 +123,7 @@ if in_reply: parent = self._get_post(self.url, self.user_id, in_reply) if not parent: self.log.info( - "Skipping %s, parent %s not found in db", status["id"], in_reply + "Skipping '%s': parent '%s' not found in db", status["id"], in_reply ) return parser = StatusParser(status) @@ -154,13 +154,13 @@ post.attachments.put(LabelsAttachment(labels=[status["spoiler_text"]])) blobs: list[Blob] = [] for media in status.get("media_attachments", []): - self.log.info("Downloading %s...", media["url"]) + self.log.info("Downloading '%s'...", media["url"]) blob: Blob | None = download_blob( media["url"], media.get("alt"), client=self.http ) if not blob: self.log.error( - "Skipping %s! Failed to download media %s.", + "Skipping '%s': failed to download attachment '%s'", status["id"], media["url"], ) @@ -189,17 +189,15 @@ "identifier": status["id"], } ) - self.log.info("Post stored in DB: %s", status["id"]) - + self.log.info("Crossposting: '%s'", status["id"]) for out in self.outputs: self.submitter(lambda: out.accept_post(post)) def _on_reblog(self, status: dict[str, Any], reblog: dict[str, Any]): - self.log.info("Processing reblog: %s", status["id"]) reposted = self._get_post(self.url, self.user_id, reblog["id"]) if not reposted: self.log.info( - "Skipping repost '%s' as reposted post '%s' was not found in the db.", + "Skipping repost '%s': reposted post '%s' not found in db", status["id"], reblog["id"], ) @@ -213,32 +211,30 @@ "identifier": status["id"], "reposted": reposted["id"], } ) - - self.log.info("Reblog stored in DB: %s", status["id"]) repost_ref = PostRef(id=status["id"], author=self.user_id, service=self.url) reposted_ref = PostRef(id=reblog["id"], author=self.user_id, service=self.url) + + self.log.info("Crossposting: '%s'", status["id"]) for out in self.outputs: self.submitter(lambda: out.accept_repost(repost_ref, reposted_ref)) def _on_delete_post(self, status_id: str): - self.log.info("Processing delete for %s...", status_id) post = self._get_post(self.url, self.user_id, status_id) if not post: - self.log.warning("Post not found in DB: %s", status_id) + self.log.warning("Skipping delete '%s': post not found in db", status_id) return post_ref = PostRef(id=status_id, author=self.user_id, service=self.url) if post["reposted"]: - self.log.info("Deleting repost: %s", status_id) + self.log.info("Deleting repost: '%s'", status_id) for output in self.outputs: self.submitter(lambda: output.delete_repost(post_ref)) else: - self.log.info("Deleting post: %s", status_id) + self.log.info("Deleting post: '%s'", status_id) for output in self.outputs: self.submitter(lambda: output.delete_post(post_ref)) self.submitter(lambda: self._delete_post_by_id(post["id"])) - self.log.info("Delete processed successfully for %s", status_id) def _accept_msg(self, msg: websockets.Data) -> None: data: dict[str, Any] = cast(dict[str, Any], json.loads(msg)) @@ -262,7 +258,7 @@ ping_timeout=10, close_timeout=5, ): try: - self.log.info("Listening to %s...", self.streaming_url) + self.log.info("Listening to '%s'...", self.streaming_url) async def listen_for_messages(): async for msg in ws: @@ -273,9 +269,9 @@ _ = await asyncio.gather(listen) except websockets.ConnectionClosedError as e: self.log.error(e, stack_info=True, exc_info=True) - self.log.info("Reconnecting to %s...", self.streaming_url) + self.log.info("Reconnecting to '%s'...", self.streaming_url) continue except TimeoutError as e: - self.log.error("Connection timeout: %s", e) - self.log.info("Reconnecting to %s...", self.streaming_url) + self.log.error("Connection timeout: '%s'", e) + self.log.info("Reconnecting to '%s'...", self.streaming_url) continue diff --git a/mastodon/output.py b/mastodon/output.py --- a/mastodon/output.py +++ b/mastodon/output.py @@ -39,7 +39,7 @@ if ( "visibility" in data and data["visibility"] not in ALLOWED_POSTING_VISIBILITY ): - raise ValueError(f"Invalid visibility option {data['visibility']}!") + raise ValueError(f"Invalid visibility option '{data['visibility']}'!") return MastodonOutputOptions(**data) @@ -57,11 +57,11 @@ ) -> None: super().__init__(options.instance, db, http) self.options: MastodonOutputOptions = options - self.log.info("Verifying %s credentails...", self.url) + self.log.info("Verifying '%s' credentails...", self.url) response = self.verify_credentials() self.user_id: str = response["id"] - self.log.info("Getting %s configuration...", self.url) + self.log.info("Getting '%s' configuration...", self.url) response = self.fetch_instance_info() self.instance_info: InstanceInfo = InstanceInfo.from_api(response) @@ -209,19 +209,19 @@ ) if response.status_code == 200: self.log.info( - "Uploaded %s! (%s)", blob.name or "unknown", response.json()["id"] + "Uploaded '%s'! (%s)", blob.name or "unknown", response.json()["id"] ) uploads.append( MediaUploadResult(id=response.json()["id"], processed=True) ) elif response.status_code == 202: - self.log.info("Waiting for %s to process!", blob.name or "unknown") + self.log.info("Waiting for '%s' to process...", blob.name or "unknown") uploads.append( MediaUploadResult(id=response.json()["id"], processed=False) ) else: self.log.error( - "Failed to upload %s! %s", + "Failed to upload '%s'! %s", blob.name or "unknown", response.text, ) @@ -248,12 +248,11 @@ return [result.id for result in uploads] @override def accept_post(self, post: Post): - self.log.info( - "Accepting post %s (author: %s, service: %s)...", - post.id, - post.author, - post.service, - ) + 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") + return + new_root_id: int | None = None new_parent_id: int | None = None @@ -263,7 +262,7 @@ thread = self._find_mapped_thread( post.parent_id, post.service, post.author, self.url, self.user_id ) if not thread: - self.log.error("Failed to find thread tuple in the database!") + self.log.error("Skipping '%s': parent thread tuple not found in db") return _, reply_ref, new_root_id, new_parent_id = thread @@ -271,19 +270,21 @@ quoted_status_id: str | None = None quote = post.attachments.get(QuoteAttachment) if quote: if quote.quoted_user != post.author: - self.log.info("Quoted other user, skipping!") + self.log.info("Skipping '%s': quote of other user") return quoted_post = self._get_post(post.service, post.author, quote.quoted_id) if not quoted_post: - self.log.error("Failed to find quoted post in the database!") + self.log.error("Skipping '%s': quoted post not found in db") return quoted_mappings = self._get_mappings( quoted_post["id"], self.url, self.user_id ) if not quoted_mappings: - self.log.error("Failed to find mappings for quoted post!") + self.log.error( + "Skipping '%s': mappings for quoted post not found in db" + ) return quoted_status_id = quoted_mappings[-1]["identifier"] @@ -315,7 +316,7 @@ media_blobs = media_attachment.blobs if media_attachment else [] raw_statuses = self._split_tokens_and_media(post_tokens, media_blobs) if not raw_statuses: - self.log.error("Failed to split post into statuses!") + self.log.error("Skipping '%s': couldn't split post into statuses") return baked_statuses: list[tuple[str, list[str] | None]] = [] @@ -324,7 +325,7 @@ media_ids: list[str] | None = None if raw_media: media_ids = self._upload_media(raw_media) if not media_ids: - self.log.error("Failed to upload attachments!") + self.log.error("Skipping '%s': failed to upload attachments") return baked_statuses.append((status_text, media_ids)) @@ -360,27 +361,14 @@ "Content-Type": "application/json", }, json=payload, ) - - if response.status_code != 200: - self.log.error( - "Failed to post status! %s - %s", - response.status_code, - response.text, - ) - response.raise_for_status() + response.raise_for_status() status_id = response.json()["id"] - self.log.info("Created new status %s!", status_id) created_statuses.append(status_id) if i == 0: reply_ref = status_id - db_post = self._get_post(post.service, post.author, post.id) - if not db_post: - self.log.error("Post not found in database!") - return - if new_root_id is None or new_parent_id is None: self._insert_post( { @@ -425,64 +413,40 @@ self.log.info("Post accepted successfully: %s -> %s", post.id, created_statuses) @override def delete_post(self, post: PostRef): - self.log.info( - "Deleting post %s (author: %s, service: %s)...", - post.id, - post.author, - post.service, - ) db_post = self._get_post(post.service, post.author, post.id) if not db_post: - self.log.warning( - "Post not found in DB: %s (author: %s, service: %s)", - post.id, - post.author, - post.service, - ) + self.log.warning("Skipping delete '%s': post not found in db: %s", post.id) return mappings = self._get_mappings(db_post["id"], self.url, self.user_id) for mapping in mappings[::-1]: - self.log.info("Deleting '%s'...", mapping["identifier"]) - self.http.delete( + response = self.http.delete( f"{self.url}/api/v1/statuses/{mapping['identifier']}", headers={"Authorization": f"Bearer {self._get_token()}"}, ) + response.raise_for_status() + self._delete_post_by_id(mapping["id"]) - self.log.info("Post deleted successfully: %s", post.id) + self.log.info("Deleted '%s'", mapping["identifier"]) @override def accept_repost(self, repost: PostRef, reposted: PostRef): - self.log.info( - "Accepting repost %s of %s (author: %s, service: %s)...", - repost.id, - reposted.id, - repost.author, - repost.service, - ) original = self._get_post(reposted.service, reposted.author, reposted.id) if not original: - self.log.info("Post not found in db, skipping repost..") + self.log.info("Skipping repost '%s': reposted post not found in db") return mappings = self._get_mappings(original["id"], self.url, self.user_id) if not mappings: - self.log.error("No mappings found for reposted post!") + self.log.error("Skipping repost '%s': no mappings found for reposted post") return response = self.http.post( f"{self.url}/api/v1/statuses/{mappings[0]['identifier']}/reblog", headers={"Authorization": f"Bearer {self._get_token()}"}, ) - - if response.status_code != 200: - self.log.error( - "Failed to boost status! status_code: %s, msg: %s", - response.status_code, - response.content, - ) - return + response.raise_for_status() self._insert_post( { @@ -502,7 +466,9 @@ raise ValueError("Inserted post not found!") original_repost = self._get_post(repost.service, repost.author, repost.id) if not original_repost: - self.log.error("original repost not found in DB: %s", repost.id) + self.log.error( + "Skipping repost '%s': repost not found in db: %s", repost.id + ) return self._insert_post_mapping(original_repost["id"], inserted["id"]) @@ -510,54 +476,32 @@ self.log.info("Repost accepted successfully: %s", repost.id) @override def delete_repost(self, repost: PostRef): - self.log.info( - "Deleting repost %s (author: %s, service: %s)...", - repost.id, - repost.author, - repost.service, - ) db_repost = self._get_post(repost.service, repost.author, repost.id) if not db_repost: - self.log.warning( - "Repost not found in DB: %s (author: %s, service: %s)", - repost.id, - repost.author, - repost.service, - ) + self.log.warning("Skipping delete '%s': repost not found in db", repost.id) return mappings = self._get_mappings(db_repost["id"], self.url, self.user_id) rmappings = self._get_mappings(db_repost["reposted"], self.url, self.user_id) if not mappings: - self.log.warning("No mappings found for repost %s", repost.id) + self.log.warning( + "Skipping delete '%s': no mappings found for repost", repost.id + ) return if not rmappings: self.log.warning( - "No mappings found for original post %s (reposted_id=%s)", + "Skipping delete '%s': no mappings found for post", repost.id, db_repost["reposted"], ) return - self.log.info( - "Removing '%s' Repost of '%s'...", - mappings[0]["identifier"], - rmappings[0]["identifier"], - ) - response = self.http.post( f"{self.url}/api/v1/statuses/{rmappings[0]['identifier']}/unreblog", headers={"Authorization": f"Bearer {self._get_token()}"}, ) - - if response.status_code != 200: - self.log.error( - "Failed to unreblog! status_code: %s, msg: %s", - response.status_code, - response.text, - ) - return + response.raise_for_status() self._delete_post_by_id(mappings[0]["id"]) self.log.info("Repost deleted successfully: %s", repost.id) diff --git a/misskey/input.py b/misskey/input.py --- a/misskey/input.py +++ b/misskey/input.py @@ -58,7 +58,7 @@ ) -> None: super().__init__(options.instance, db, http) self.options: MisskeyInputOptions = options - self.log.info("Verifying %s credentails...", self.url) + self.log.info("Verifying '%s' credentails...", self.url) response = self.verify_credentials() self.user_id: str = response["id"] @@ -67,14 +67,12 @@ def _get_token(self) -> str: return self.options.token def _on_note(self, note: dict[str, Any]): - self.log.info("Processing new note: %s", note["id"]) - if note["userId"] != self.user_id: return if note["visibility"] not in self.options.allowed_visibility: self.log.info( - "Skipping note with disallowed visibility: %s (%s)", + "Skipping '%s': disallowed visibility (%s)", note["id"], note["visibility"], ) @@ -82,13 +80,13 @@ return if self._is_post_crossposted(self.url, self.user_id, note["id"]): self.log.info( - "Skipping %s, already crossposted", + "Skipping '%s': already crossposted", note["id"], ) return if note.get("poll"): - self.log.info("Skipping '%s'! Contains a poll..", note["id"]) + self.log.info("Skipping '%s': polls not supported", note["id"]) return renote: dict[str, Any] | None = note.get("renote") @@ -103,13 +101,15 @@ rrenote = self._get_post(self.url, self.user_id, renote["id"]) if not rrenote: self.log.info( - "Skipping %s, quote %s not found in db", note["id"], renote["id"] + "Skipping '%s': quoted post '%s' not found in db", + note["id"], + renote["id"], ) return reply: dict[str, Any] | None = note.get("reply") if reply and reply.get("userId") != self.user_id: - self.log.info("Skipping '%s'! Reply to other user..", note["id"]) + self.log.info("Skipping '%s': Reply to other user..", note["id"]) return parent = None @@ -117,7 +117,9 @@ if reply: parent = self._get_post(self.url, self.user_id, reply["id"]) if not parent: self.log.info( - "Skipping %s, parent %s not found in db", note["id"], reply["id"] + "Skipping '%s': parent '%s' not found in db", + note["id"], + reply["id"], ) return @@ -151,13 +153,13 @@ post.attachments.put(LabelsAttachment(labels=[note["cw"]])) blobs: list[Blob] = [] for media in note.get("files", []): - self.log.info("Downloading %s...", media["url"]) + self.log.info("Downloading '%s'...", media["url"]) blob: Blob | None = download_blob( media["url"], media.get("comment", ""), client=self.http ) if not blob: self.log.error( - "Skipping %s! Failed to download media %s.", + "Skipping '%s': failed to download media '%s'.", note["id"], media["url"], ) @@ -186,17 +188,15 @@ "identifier": note["id"], } ) - self.log.info("Note stored in DB: %s", note["id"]) - + self.log.info("Crossposting: '%s'", note["id"]) for out in self.outputs: self.submitter(lambda: out.accept_post(post)) def _on_renote(self, note: dict[str, Any], renote: dict[str, Any]): - self.log.info("Processing renote: %s", note["id"]) reposted = self._get_post(self.url, self.user_id, renote["id"]) if not reposted: self.log.info( - "Skipping repost '%s' as reposted post '%s' was not found in the db.", + "Skipping repost '%s': reposted post '%s' not found in db", note["id"], renote["id"], ) @@ -210,11 +210,11 @@ "identifier": note["id"], "reposted": reposted["id"], } ) - - self.log.info("Renote stored in DB: %s", note["id"]) repost_ref = PostRef(id=note["id"], author=self.user_id, service=self.url) reposted_ref = PostRef(id=renote["id"], author=self.user_id, service=self.url) + + self.log.info("Crossposting: '%s'", note["id"]) for out in self.outputs: self.submitter(lambda: out.accept_repost(repost_ref, reposted_ref)) @@ -250,7 +250,7 @@ ping_timeout=10, close_timeout=5, ): try: - self.log.info("Listening to %s...", streaming) + self.log.info("Listening to '%s'...", streaming) await self._subscribe_to_home(ws) async def listen_for_messages(): @@ -262,9 +262,9 @@ _ = await asyncio.gather(listen) except websockets.ConnectionClosedError as e: self.log.error(e, stack_info=True, exc_info=True) - self.log.info("Reconnecting to %s...", streaming) + self.log.info("Reconnecting to '%s'...", streaming) continue except TimeoutError as e: - self.log.error("Connection timeout: %s", e) - self.log.info("Reconnecting to %s...", streaming) + self.log.error("Connection timeout: '%s'", e) + self.log.info("Reconnecting to '%s'...", streaming) continue diff --git a/registry.py b/registry.py --- a/registry.py +++ b/registry.py @@ -19,7 +19,7 @@ def create_input_service( db: DatabasePool, http: httpx.Client, data: dict[str, Any] ) -> InputService: if "type" not in data: - raise ValueError("No `type` field in input data!") + raise ValueError("No 'type' field in input data!") type: str = str(data["type"]) del data["type"] @@ -33,7 +33,7 @@ def create_output_service( db: DatabasePool, http: httpx.Client, data: dict[str, Any] ) -> OutputService: if "type" not in data: - raise ValueError("No `type` field in input data!") + raise ValueError("No 'type' field in input data!") type: str = str(data["type"]) del data["type"] diff --git a/util/util.py b/util/util.py --- a/util/util.py +++ b/util/util.py @@ -17,9 +17,10 @@ def normalize_service_url(url: str) -> str: if not url.startswith("https://") and not url.startswith("http://"): - raise ValueError(f"Invalid service url {url}! Only http/https are supported.") + raise ValueError(f"Invalid service url {url}! Must start with http/https!") return url[:-1] if url.endswith("/") else url + def _read_env(data: Any) -> None: match data: -- tangled.sh