···
+
import cross, media_util, util, database
+
from util import LOGGER
+
import requests, websockets
+
from typing import Callable, Any
+
URL = re.compile(r"(?:https?://|mailto:|localhost\b)[\w\-\._~:/\?#\[\]@!\$&'\(\)\*\+,;=%]+")
+
MD_INLINE_LINK = re.compile(r"\[([^\]]+)\]\(([^\)]+)\)")
+
MD_AUTOLINK = re.compile(r"<((?:https?://[^\s>]+|mailto:[^\s>]+))>")
+
HASHTAG = re.compile(r'(?<!\w)\#([\w]+)')
+
FEDIVERSE_HANDLE = re.compile(r'(?<![\w@])@([\w-]+)(?:@([\w\.-]+\.[\w\.-]+))?')
+
def get_image_common(mime: str):
+
if mime.startswith('image/'):
+
if mime == 'image/gif':
+
elif mime.startswith('video/'):
+
elif mime.startswith('audio/'):
+
def tokenize_note(note: dict) -> list[cross.Token]:
+
text: str = note.get('text', '')
+
mention_handles: dict = util.safe_get(note, 'mentionHandles', {})
+
handles: list[str] = []
+
for key, value in mention_handles.items():
+
tags: list[str] = util.safe_get(note, 'tags', [])
+
tokens: list[cross.Token] = []
+
tokens.append(cross.TextToken(''.join(buffer)))
+
md_inline = MD_INLINE_LINK.match(text, index)
+
label = md_inline.group(1)
+
href = md_inline.group(2)
+
tokens.append(cross.LinkToken(href, label))
+
index = md_inline.end()
+
md_auto = MD_AUTOLINK.match(text, index)
+
href = md_auto.group(1)
+
tokens.append(cross.LinkToken(href, href))
+
tag = HASHTAG.match(text, index)
+
tag_text = tag.group(1)
+
if tag_text.lower() in tags:
+
tokens.append(cross.TagToken(tag_text))
+
handle = FEDIVERSE_HANDLE.match(text, index)
+
handle_text = handle.group(0)
+
if handle_text.strip() in handles:
+
tokens.append(cross.MentionToken(handle_text, '')) # TODO misskey doesn't provide a uri
+
url = URL.match(text, index)
+
tokens.append(cross.LinkToken(href, href))
+
buffer.append(text[index])
+
class MisskeyPost(cross.Post):
+
def __init__(self, note: dict) -> None:
+
media_attachments: list[cross.MediaAttachment] = []
+
for attachment in note.get('files', []):
+
media_attachments.append(MisskeyAttachment(attachment))
+
sensitive |= attachment.get('isSensitive', False)
+
self.sensitive = sensitive
+
self.media_attachments = media_attachments
+
self.tokens = tokenize_note(self.note)
+
def get_tokens(self) -> list[cross.Token]:
+
def get_parent_id(self) -> str | None:
+
return self.note.get('replyId')
+
def get_attachments(self) -> list[cross.MediaAttachment]:
+
return self.media_attachments
+
def get_id(self) -> str:
+
def get_cw(self) -> str:
+
return util.safe_get(self.note, 'cw', '')
+
def get_languages(self) -> list[str]:
+
def is_sensitive(self) -> bool:
+
class MisskeyAttachment(cross.MediaAttachment):
+
def __init__(self, attachment: dict) -> None:
+
self.attachment = attachment
+
def create_meta(self, bytes: bytes) -> cross.MediaMeta:
+
if get_image_common(self.attachment['type']):
+
o_meta = media_util.get_media_meta(bytes)
+
return cross.MediaMeta(o_meta['width'], o_meta['height'], o_meta.get('duration', -1))
+
return cross.MediaMeta(-1, -1, -1)
+
def get_url(self) -> str:
+
return self.attachment.get('url', '')
+
def get_type(self) -> str | None:
+
return get_image_common(self.attachment['type'])
+
def get_alt(self) -> str:
+
return util.safe_get(self.attachment, 'comment', '')
+
class MisskeyInput(cross.Input):
+
def __init__(self, settings: dict, db: cross.DataBaseWorker) -> None:
+
self.options = settings.get('options', {})
+
self.token = util.get_or_envvar(settings, 'token')
+
instance: str = util.get_or_envvar(settings, 'instance')
+
service = instance[:-1] if instance.endswith('/') else instance
+
LOGGER.info("Verifying %s credentails...", service)
+
responce = requests.post(f"{instance}/api/i", json={ 'i': self.token }, headers={
+
"Content-Type": "application/json"
+
if responce.status_code != 200:
+
LOGGER.error("Failed to validate user credentials!")
+
responce.raise_for_status()
+
super().__init__(service, responce.json()["id"], settings, db)
+
def _on_note(self, outputs: list[cross.Output], note: dict):
+
if note['userId'] != self.user_id:
+
if note.get('renoteId') or note.get('poll'):
+
# TODO polls not supported on bsky. maybe 3rd party? skip for now
+
# we don't handle reblogs. possible with bridgy(?) and self
+
LOGGER.info("Skipping '%s'! Renote or poll..", note['id'])
+
reply_id: str | None = note.get('replyId')
+
if note.get('reply', {}).get('userId') != self.user_id:
+
LOGGER.info("Skipping '%s'! Reply to other user..", note['id'])
+
if note.get('visibility') not in self.options.get('allowed_visibility', []):
+
LOGGER.info("Skipping '%s'! '%s' visibility..", note['id'], note.get('visibility'))
+
parent_post = database.find_post(self.db, reply_id, self.user_id, self.service)
+
LOGGER.info("Skipping '%s' as parent post was not found in db!", note['id'])
+
root_id = parent_post['id']
+
if parent_post['root_id']:
+
root_id = parent_post['root_id']
+
LOGGER.info("Crossposting '%s'...", note['id'])
+
if root_id and parent_id:
+
cross_post = MisskeyPost(note)
+
output.accept_post(cross_post)
+
def _on_delete(self, outputs: list[cross.Output], note: dict):
+
def _on_message(self, outputs: list[cross.Output], data: dict):
+
if data['type'] == 'channel':
+
type: str = data['body']['type']
+
if type == 'note' or type == 'reply':
+
note_body = data['body']['body']
+
self._on_note(outputs, note_body)
+
async def _send_keepalive(self, ws: websockets.WebSocketClientProtocol):
+
await asyncio.sleep(120)
+
LOGGER.debug("Sent keepalive h..")
+
LOGGER.info("WebSocket is closed, stopping keepalive task.")
+
LOGGER.error(f"Error sending keepalive: {e}")
+
async def _listen_for_messages(
+
ws: websockets.WebSocketClientProtocol,
+
submit: Callable[[Callable[[], Any]], Any],
+
outputs: list[cross.Output]):
+
# TODO listen to deletes somehow
+
if False and data['type'] == 'channel':
+
payload_type = data['body']['type']
+
if payload_type == 'reply' or payload_type == 'note':
+
user_id = data['body']['body']['userId']
+
if self.user_id == user_id:
+
note_id = data['body']['body']['id']
+
await ws.send(json.dumps({
+
LOGGER.info('Subscribed to note %s updates.', note_id)
+
submit(lambda: self._on_message(outputs, data))
+
async def _subscribe_to_home(self, ws: websockets.WebSocketClientProtocol):
+
home_message = json.dumps({
+
"channel": "homeTimeline",
+
"id": str(uuid.uuid4())
+
await ws.send(home_message)
+
LOGGER.info("Subscribed to 'homeTimeline' channel...")
+
async def listen(self, outputs: list[cross.Output], submit: Callable[[Callable[[], Any]], Any]):
+
streaming: str = f"wss://{self.service.split("://", 1)[1]}"
+
url: str = f"{streaming}/streaming?i={self.token}"
+
async for ws in websockets.connect(url, extra_headers={"User-Agent": "XPost/0.0.2"}):
+
LOGGER.info("Listening to %s...", streaming)
+
await self._subscribe_to_home(ws)
+
keepalive = asyncio.create_task(self._send_keepalive(ws))
+
listen = asyncio.create_task(self._listen_for_messages(ws, submit, outputs))
+
await asyncio.gather(keepalive, listen)
+
except websockets.ConnectionClosedError as e:
+
LOGGER.error(e, stack_info=True, exc_info=True)
+
LOGGER.info("Reconnecting to %s...", streaming)
zenfyr.dev