commit ada6f7ddf74f5000e11973c6f513a6cf15dd6977 · zenfyr.dev/xpost

+2 -2

mastodon/input.py

···

       64
        
               outputs: list[OutputService],

     

       65
        
               submitter: Callable[[Callable[[], None]], None],

     

       66
        
           ):

     

       67
       -
               uri = f"{self.streaming_url}/api/v1/streaming?stream=user"

     

       68
        
       

     

       69
        
               async for ws in websockets.connect(

     

       70
       -
                   uri, additional_headers={"Authorization": f"Bearer {self.options.token}"}

     

       71
        
               ):

     

       72
        
                   try:

     

       73
        
                       LOGGER.info("Listening to %s...", self.streaming_url)

···

       64
        
               outputs: list[OutputService],

     

       65
        
               submitter: Callable[[Callable[[], None]], None],

     

       66
        
           ):

     

       67
       +
               url = f"{self.streaming_url}/api/v1/streaming?stream=user"

     

       68
        
       

     

       69
        
               async for ws in websockets.connect(

     

       70
       +
                   url, additional_headers={"Authorization": f"Bearer {self.options.token}"}

     

       71
        
               ):

     

       72
        
                   try:

     

       73
        
                       LOGGER.info("Listening to %s...", self.streaming_url)

+22

misskey/info.py

···

       1
       +
       from abc import ABC, abstractmethod

     

       2
       +
       from cross.service import Service

     

       3
       +
       

     

       4
       +
       import requests

     

       5
       +
       

     

       6
       +
       from util.util import LOGGER

     

       7
       +
       

     

       8
       +
       class MisskeyService(ABC, Service):

     

       9
       +
           def verify_credentials(self):

     

       10
       +
               responce = requests.post(

     

       11
       +
                   f"{self.url}/api/i",

     

       12
       +
                   json={"i": self._get_token()},

     

       13
       +
                   headers={"Content-Type": "application/json"},

     

       14
       +
               )

     

       15
       +
               if responce.status_code != 200:

     

       16
       +
                   LOGGER.error("Failed to validate user credentials!")

     

       17
       +
                   responce.raise_for_status()

     

       18
       +
               return dict(responce.json())

     

       19
       +
       

     

       20
       +
           @abstractmethod

     

       21
       +
           def _get_token(self) -> str:

     

       22
       +
               pass

+95

misskey/input.py

···

       1
       +
       import asyncio

     

       2
       +
       from dataclasses import dataclass, field

     

       3
       +
       import json

     

       4
       +
       from pathlib import Path

     

       5
       +
       import re

     

       6
       +
       from typing import Any, Callable, override

     

       7
       +
       import uuid

     

       8
       +
       

     

       9
       +
       import websockets

     

       10
       +
       

     

       11
       +
       from cross.service import InputService, OutputService

     

       12
       +
       from misskey.info import MisskeyService

     

       13
       +
       from util.util import LOGGER

     

       14
       +
       

     

       15
       +
       ALLOWED_VISIBILITY = ["public", "home"]

     

       16
       +
       

     

       17
       +
       

     

       18
       +
       @dataclass

     

       19
       +
       class MisskeyInputOptions:

     

       20
       +
           token: str

     

       21
       +
           instance: str

     

       22
       +
           allowed_visibility: list[str] = field(

     

       23
       +
               default_factory=lambda: ALLOWED_VISIBILITY.copy()

     

       24
       +
           )

     

       25
       +
           regex_filters: list[re.Pattern[str]] = field(default_factory=lambda: [])

     

       26
       +
       

     

       27
       +
           @classmethod

     

       28
       +
           def from_dict(cls, data: dict[str, Any]) -> "MisskeyInputOptions":

     

       29
       +
               data["instance"] = (

     

       30
       +
                   data["instance"][:-1]

     

       31
       +
                   if data["instance"].endswith("/")

     

       32
       +
                   else data["instance"]

     

       33
       +
               )

     

       34
       +
       

     

       35
       +
               if "allowed_visibility" in data:

     

       36
       +
                   for vis in data.get("allowed_visibility", []):

     

       37
       +
                       if vis not in ALLOWED_VISIBILITY:

     

       38
       +
                           raise ValueError(f"Invalid visibility option {vis}!")

     

       39
       +
       

     

       40
       +
               if "regex_filters" in data:

     

       41
       +
                   data["regex_filters"] = [re.compile(r) for r in data["regex_filters"]]

     

       42
       +
       

     

       43
       +
               return MisskeyInputOptions(**data)

     

       44
       +
       

     

       45
       +
       

     

       46
       +
       class MisskeyInputService(MisskeyService, InputService):

     

       47
       +
           def __init__(self, db: Path, options: MisskeyInputOptions) -> None:

     

       48
       +
               super().__init__(options.instance, db)

     

       49
       +
               self.options: MisskeyInputOptions = options

     

       50
       +
       

     

       51
       +
               LOGGER.info("Verifying %s credentails...", self.url)

     

       52
       +
               responce = self.verify_credentials()

     

       53
       +
               self.user_id: str = responce["id"]

     

       54
       +
       

     

       55
       +
           @override

     

       56
       +
           def _get_token(self) -> str:

     

       57
       +
               return self.options.token

     

       58
       +
       

     

       59
       +
           async def _subscribe_to_home(self, ws: websockets.ClientConnection) -> None:

     

       60
       +
               await ws.send(

     

       61
       +
                   json.dumps(

     

       62
       +
                       {

     

       63
       +
                           "type": "connect",

     

       64
       +
                           "body": {"channel": "homeTimeline", "id": str(uuid.uuid4())},

     

       65
       +
                       }

     

       66
       +
                   )

     

       67
       +
               )

     

       68
       +
               LOGGER.info("Subscribed to 'homeTimeline' channel...")

     

       69
       +
       

     

       70
       +
           @override

     

       71
       +
           async def listen(

     

       72
       +
               self,

     

       73
       +
               outputs: list[OutputService],

     

       74
       +
               submitter: Callable[[Callable[[], None]], None],

     

       75
       +
           ):

     

       76
       +
               streaming: str = f"{'wss' if self.url.startswith('https') else 'ws'}://{self.url.split('://', 1)[1]}"

     

       77
       +
               url: str = f"{streaming}/streaming?i={self.options.token}"

     

       78
       +
       

     

       79
       +
               async for ws in websockets.connect(url):

     

       80
       +
                   try:

     

       81
       +
                       LOGGER.info("Listening to %s...", streaming)

     

       82
       +
                       await self._subscribe_to_home(ws)

     

       83
       +
       

     

       84
       +
                       async def listen_for_messages():

     

       85
       +
                           async for msg in ws:

     

       86
       +
                               LOGGER.info(msg)  # TODO

     

       87
       +
       

     

       88
       +
                       #keepalive = asyncio.create_task(self._send_keepalive(ws))

     

       89
       +
                       listen = asyncio.create_task(listen_for_messages())

     

       90
       +
       

     

       91
       +
                       _ = await asyncio.gather(listen)

     

       92
       +
                   except websockets.ConnectionClosedError as e:

     

       93
       +
                       LOGGER.error(e, stack_info=True, exc_info=True)

     

       94
       +
                       LOGGER.info("Reconnecting to %s...", streaming)

     

       95
       +
                       continue