1
0
Fork 0

[WIP] Added automatic user_id retrieving and more

* Added safe-er way to unpack data from instances
   -> like, now we're passing only the fields that
      we really need, not just everything that was
      sent by out our instance. that fixes pleroma
      being weird and adding pleroma field to some
      of objects, which was causing KeyError to be
      raised. this should be fixed now but I'm not
      completely sure so that's why it's in unsafe
      branch of the project. oh wow also this comm
      message is looking funky. hahahahah. ok bai.
This commit is contained in:
Casey 2022-09-14 20:29:23 +03:00
parent 01a384161c
commit 84eb94447f
Signed by: hkc
GPG Key ID: F0F6CFE11CDB0960
3 changed files with 29 additions and 17 deletions

View File

@ -12,14 +12,11 @@ instance = mastodon.example.org
# ${instance}/settings/applications # ${instance}/settings/applications
token = blahblah token = blahblah
# Mastodon user ID. Used to filter out posts. Unfortunately, I can't find a way # Your user ID.
# to get it using token itself. GARGROOOOOOON!!!!! # Doesn't necessarily yours, it can be any user's ID, but that user should be
# Anyways, you could navigate to your profile ${instance}/@${username} and # on the list for crossposter to find it.
# look for your profile picture link. For example, for me it's # Setting it to "auto" will just grab yours instead. Don't worry about it
# https://mastodon.astrr.ru/system/accounts/avatars/107/914/495/779/447/227/original/9651ac2f47cb2993.jpg user = auto
# that part between "avarars" and "original" is the user ID. Grab it, remove
# all of the slashes and you should be left with, for example, this:
user = 107914495779447227
# Mastodon user list ID. AGAIN, UNFORTUNATELY, there is no way to reliably use # Mastodon user list ID. AGAIN, UNFORTUNATELY, there is no way to reliably use
# streaming API to get all of your posts. Using home timeline is unreliable and # streaming API to get all of your posts. Using home timeline is unreliable and

View File

@ -5,7 +5,12 @@ from mastoposter import execute_integrations, load_integrations_from
from mastoposter.integrations import FilteredIntegration from mastoposter.integrations import FilteredIntegration
from mastoposter.sources import websocket_source from mastoposter.sources import websocket_source
from typing import AsyncGenerator, Callable, List from typing import AsyncGenerator, Callable, List
from mastoposter.types import Status from mastoposter.types import Account, Status
from httpx import Client
WSOCK_TEMPLATE = "wss://{instance}/api/v1/streaming"
VERIFY_CREDS_TEMPLATE = "https://{instance}/api/v1/account/verify_credentials"
async def listen( async def listen(
@ -50,12 +55,22 @@ def main(config_path: str):
modules: List[FilteredIntegration] = load_integrations_from(conf) modules: List[FilteredIntegration] = load_integrations_from(conf)
user_id: str = conf["main"]["user"]
if user_id == "auto":
with Client() as c:
rq = c.get(
VERIFY_CREDS_TEMPLATE.format(**conf["main"]),
params={"access_token": conf["main"]["token"]},
)
account = Account.from_dict(rq.json())
user_id = account.id
url = "wss://{}/api/v1/streaming".format(conf["main"]["instance"]) url = "wss://{}/api/v1/streaming".format(conf["main"]["instance"])
run( run(
listen( listen(
websocket_source, websocket_source,
modules, modules,
conf["main"]["user"], user_id,
url=url, url=url,
reconnect=conf["main"].getboolean("auto_reconnect", False), reconnect=conf["main"].getboolean("auto_reconnect", False),
list=conf["main"]["list"], list=conf["main"]["list"],

View File

@ -1,4 +1,4 @@
from dataclasses import dataclass, field from dataclasses import dataclass, field, fields
from datetime import datetime from datetime import datetime
from typing import Any, Callable, Optional, List, Literal, TypeVar from typing import Any, Callable, Optional, List, Literal, TypeVar
@ -47,7 +47,7 @@ class Emoji:
@classmethod @classmethod
def from_dict(cls, data: dict) -> "Emoji": def from_dict(cls, data: dict) -> "Emoji":
return cls(**data) return cls(**{f.name: data[f.name] for f in fields(cls) if f in data})
@dataclass @dataclass
@ -122,7 +122,7 @@ class AttachmentMetaImage:
@classmethod @classmethod
def from_dict(cls, data: dict) -> "AttachmentMetaImage": def from_dict(cls, data: dict) -> "AttachmentMetaImage":
return cls( return cls(
**data, **{f.name: data[f.name] for f in fields(cls) if f in data},
original=cls.AttachmentMetaImageDimensions(**data["original"]), original=cls.AttachmentMetaImageDimensions(**data["original"]),
small=cls.AttachmentMetaImageDimensions(**data["small"]), small=cls.AttachmentMetaImageDimensions(**data["small"]),
focus=cls.Vec2F(**data["focus"]) focus=cls.Vec2F(**data["focus"])
@ -183,7 +183,7 @@ class Attachment:
@classmethod @classmethod
def from_dict(cls, data: dict) -> "Attachment": def from_dict(cls, data: dict) -> "Attachment":
return cls(**data) return cls(**{f.name: data[f.name] for f in fields(cls) if f in data})
@dataclass @dataclass
@ -194,7 +194,7 @@ class Application:
@classmethod @classmethod
def from_dict(cls, data: dict) -> "Application": def from_dict(cls, data: dict) -> "Application":
return cls(**data) return cls(**{f.name: data[f.name] for f in fields(cls) if f in data})
@dataclass @dataclass
@ -206,7 +206,7 @@ class Mention:
@classmethod @classmethod
def from_dict(cls, data: dict) -> "Mention": def from_dict(cls, data: dict) -> "Mention":
return cls(**data) return cls(**{f.name: data[f.name] for f in fields(cls) if f in data})
@dataclass @dataclass
@ -216,7 +216,7 @@ class Tag:
@classmethod @classmethod
def from_dict(cls, data: dict) -> "Tag": def from_dict(cls, data: dict) -> "Tag":
return cls(**data) return cls(**{f.name: data[f.name] for f in fields(cls) if f in data})
@dataclass @dataclass