initial commit
This commit is contained in:
commit
4eef9374f0
4
.gitignore
vendored
Normal file
4
.gitignore
vendored
Normal file
|
|
@ -0,0 +1,4 @@
|
|||
*.secret
|
||||
**/__pycache__
|
||||
**/*.egg-info
|
||||
build/
|
||||
1
blastodon/__init__.py
Normal file
1
blastodon/__init__.py
Normal file
|
|
@ -0,0 +1 @@
|
|||
from blastodon.auth import cli
|
||||
7
blastodon/__main__.py
Normal file
7
blastodon/__main__.py
Normal file
|
|
@ -0,0 +1,7 @@
|
|||
from blastodon.client import Client
|
||||
|
||||
client = Client.init_cli()
|
||||
|
||||
post_text = input(' enter a status to post: ')
|
||||
|
||||
client.send_text_post(post_text)
|
||||
0
blastodon/auth/__init__.py
Normal file
0
blastodon/auth/__init__.py
Normal file
81
blastodon/auth/cli.py
Normal file
81
blastodon/auth/cli.py
Normal file
|
|
@ -0,0 +1,81 @@
|
|||
from pathlib import Path
|
||||
from sqlite3 import Connection as SQLite3
|
||||
from getpass import getpass
|
||||
|
||||
from mastodon import Mastodon
|
||||
from atproto import Client as BskyClient
|
||||
|
||||
|
||||
def auth_mastodon() -> Mastodon:
|
||||
metadata_file = Path.cwd() / 'mastodon_client.secret'
|
||||
scopes = [
|
||||
"read", "write", # "follow", "push"
|
||||
]
|
||||
# For first run of app instance
|
||||
if not metadata_file.exists():
|
||||
instance = input(' enter your mastodon instance: https://')
|
||||
api_base_url = f'https://{instance}'
|
||||
Mastodon.create_app('blastodon',
|
||||
api_base_url=instance,
|
||||
scopes=scopes,
|
||||
to_file=metadata_file)
|
||||
login_file = _mastodon_find_single_user_login()
|
||||
if login_file is None:
|
||||
username = input(' enter your mastodon username: ')
|
||||
login_file = Path.cwd() / f'mastodon.{username}.secret'
|
||||
# for first run for user
|
||||
if login_file.exists():
|
||||
return Mastodon(client_id=metadata_file, access_token=login_file)
|
||||
|
||||
mastodon = Mastodon(client_id=metadata_file)
|
||||
print('Visit', mastodon.auth_request_url(scopes=scopes), 'to log in')
|
||||
oauth_token = input(' Paste the OAuth token shown to you on login: ')
|
||||
mastodon.log_in(code=oauth_token, to_file=login_file, scopes=scopes)
|
||||
return mastodon
|
||||
|
||||
|
||||
def auth_bsky() -> BskyClient:
|
||||
session_path = _bsky_find_single_user_login()
|
||||
username = None
|
||||
if session_path is None:
|
||||
username = input(' enter your bsky username: @')
|
||||
session_path = Path.cwd() / f'bsky_{username}_session.secret'
|
||||
client = BskyClient()
|
||||
if session_path.exists():
|
||||
with open(session_path) as file:
|
||||
client.login(session_string=file.read())
|
||||
return client
|
||||
client.login(
|
||||
# username is definitely not None here because if session is not None that means
|
||||
# the file exists and the function has already returned with session restoration
|
||||
login=username,
|
||||
password=getpass(' password: ')
|
||||
)
|
||||
with open(session_path, mode='w') as file:
|
||||
file.write(client.export_session_string())
|
||||
return client
|
||||
|
||||
|
||||
def _mastodon_find_single_user_login() -> Path | None:
|
||||
found_file = None
|
||||
for file in Path.cwd().iterdir():
|
||||
if file.name.startswith('login.') and file.name.endswith('.secret'):
|
||||
print('found secret file', file.name)
|
||||
if found_file is None:
|
||||
found_file = file
|
||||
else:
|
||||
# More than one login found, skip to prompt user for name
|
||||
return None
|
||||
return found_file
|
||||
|
||||
|
||||
def _bsky_find_single_user_login() -> Path | None:
|
||||
found_file = None
|
||||
for file in Path.cwd().iterdir():
|
||||
if file.name.startswith('bsky-') and file.name.endswith('-jwt.secret'):
|
||||
if found_file is None:
|
||||
found_file = file
|
||||
else:
|
||||
# More than one login found, skip to prompt user for name
|
||||
return None
|
||||
return found_file
|
||||
3
blastodon/client/__init__.py
Normal file
3
blastodon/client/__init__.py
Normal file
|
|
@ -0,0 +1,3 @@
|
|||
from blastodon.client.retreived_post import RetreivedPost
|
||||
from blastodon.client.created_post import CreatedPost
|
||||
from blastodon.client.client import Client
|
||||
39
blastodon/client/client.py
Normal file
39
blastodon/client/client.py
Normal file
|
|
@ -0,0 +1,39 @@
|
|||
from typing import Self
|
||||
|
||||
from atproto import Client as BskyClient
|
||||
from mastodon import Mastodon, MastodonError
|
||||
|
||||
from blastodon import auth
|
||||
from blastodon.client import RetreivedPost, CreatedPost
|
||||
|
||||
class Client:
|
||||
"""A wrapper around both Mastodon and Bsky clients"""
|
||||
mastodon: Mastodon
|
||||
bsky: BskyClient
|
||||
|
||||
def __init__(self, mastodon: Mastodon, bsky: BskyClient):
|
||||
self.mastodon = mastodon
|
||||
self.bsky = bsky
|
||||
|
||||
@classmethod
|
||||
def init_cli(cls) -> Self:
|
||||
return cls(
|
||||
mastodon=auth.cli.auth_mastodon(),
|
||||
bsky=auth.cli.auth_bsky()
|
||||
)
|
||||
|
||||
def most_recent_posts(self) -> RetreivedPost:
|
||||
return RetreivedPost.most_recent(client=self)
|
||||
|
||||
def send_text_post(self, text: str):
|
||||
bsky_post = self.bsky.send_post(text=text)
|
||||
try:
|
||||
mastodon_post = self.mastodon.status_post(status=text)
|
||||
except MastodonError as err:
|
||||
print('error posting to mastodon after posting to bsky. Deleting bsky post.')
|
||||
self.bsky.delete_post(bsky_post.uri)
|
||||
raise err
|
||||
|
||||
return CreatedPost(bsky=bsky_post, mastodon=mastodon_post)
|
||||
|
||||
|
||||
8
blastodon/client/created_post.py
Normal file
8
blastodon/client/created_post.py
Normal file
|
|
@ -0,0 +1,8 @@
|
|||
from dataclasses import dataclass
|
||||
from atproto_client.models import AppBskyFeedPost
|
||||
from mastodon.statuses import Status
|
||||
|
||||
@dataclass
|
||||
class CreatedPost:
|
||||
bsky: AppBskyFeedPost.CreateRecordResponse
|
||||
mastodon: Status
|
||||
23
blastodon/client/retreived_post.py
Normal file
23
blastodon/client/retreived_post.py
Normal file
|
|
@ -0,0 +1,23 @@
|
|||
from dataclasses import dataclass
|
||||
from typing import Self
|
||||
|
||||
from mastodon.statuses import Status as MastodonStatus
|
||||
from atproto_client.models import AppBskyFeedDefs
|
||||
|
||||
@dataclass
|
||||
class RetreivedPost:
|
||||
mastodon: MastodonStatus
|
||||
bsky: AppBskyFeedDefs.FeedViewPost
|
||||
|
||||
@classmethod
|
||||
def most_recent(cls, client: 'Client') -> Self:
|
||||
logged_in_user = client.mastodon.account_verify_credentials()
|
||||
mpost = client.mastodon.account_statuses(
|
||||
id=logged_in_user.id,
|
||||
exclude_replies=True,
|
||||
exclude_reblogs=True,
|
||||
limit=1
|
||||
)[0]
|
||||
logged_in_user = client.bsky.me.did
|
||||
bpost = client.bsky.get_author_feed(actor=logged_in_user).feed[0]
|
||||
return cls(mastodon=mpost, bsky=bpost)
|
||||
22
pyproject.toml
Normal file
22
pyproject.toml
Normal file
|
|
@ -0,0 +1,22 @@
|
|||
[build-system]
|
||||
requires = ["setuptools", "setuptools-scm"]
|
||||
build-backend = "setuptools.build_meta"
|
||||
|
||||
[project]
|
||||
name = "blastodon"
|
||||
authors = [
|
||||
{name = "D. Scott Boggs", email = "scott@techwork.zone"}
|
||||
]
|
||||
description = "Bluesky and Mastdon Cross-Poster"
|
||||
readme = "README.md"
|
||||
requires-python = ">=3.11" # Self type
|
||||
license = "AGPL-3.0-only"
|
||||
dependencies = [
|
||||
"atproto",
|
||||
"mastodon.py",
|
||||
]
|
||||
dynamic = ["version"]
|
||||
|
||||
|
||||
[project.scripts]
|
||||
# Put scripts here
|
||||
Loading…
Reference in a new issue