nojhan/forthlift
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity

feat(mastodon): firs working version

Browse source
This commit is contained in:
Johann Dreo 2026-04-12 23:12:51 +02:00
commit c764b7b0ba
1 changed files with 56 additions and 49 deletions
Download patch file Download diff file Expand all files Collapse all files

105
src/forthlift/forthlift.py
View file

@ -87,7 +87,7 @@ class consume:
def __call__(self, stream): def __call__(self, stream):
current = "" current = ""
for item in stream: for item in stream:
if re.match(self.mark, item[0]): if re.match(self.mark, item.strip()):
yield current yield current
if self.skip: if self.skip:
current = "" current = ""
@ -251,7 +251,7 @@ class lift:
class mastodon(Lift): class mastodon(Lift):
def __init__(self, dryrun = 'dry'): def __init__(self, dryrun = 'nodry'):
self._scopes = ['read', 'write'] self._scopes = ['read', 'write']
if dryrun == 'dry': if dryrun == 'dry':
@ -275,23 +275,28 @@ class lift:
if self.needs_init(): if self.needs_init():
self.init() self.init()
else:
logger.debug(
"│ │ ├ Mastodon lifter logged in: " \
f"@{self.config['mastodon']['account']}" \
f"@{self.config['mastodon']['instance']}" \
)
logger.debug("│ │ └OK,")
self.masto = mastodon.Mastodon( self.masto = mastodon.Mastodon(
api_base_url = self.config["instance"], api_base_url = self.config["mastodon"]["instance"],
# client_id = self.config["client_id"], # client_id = self.config["mastodon"]["client_id"],
client_secret = self.config["client_secret"], client_secret = self.config["mastodon"]["client_secret"],
access_token = self.config["token"], access_token = self.config["mastodon"]["token"],
user_agent = f"{self.name}:{self.version}" user_agent = f"{self.name}:{self.version}"
) )
logger.debug(self.config)
def needs_init(self): def needs_init(self):
if not self.config["instance"] \ if not self.config["mastodon"]["instance"] \
or not self.config["account"] \ or not self.config["mastodon"]["account"] \
or not self.config["client_id"] \ or not self.config["mastodon"]["client_id"] \
or not self.config["client_secret"] \ or not self.config["mastodon"]["client_secret"] \
or not self.config["token"]: or not self.config["mastodon"]["token"]:
return True return True
else: else:
return False return False
@ -300,8 +305,8 @@ class lift:
if not self.needs_init(): if not self.needs_init():
print(f"The {__name__} operator has already been initialized") print(f"The {__name__} operator has already been initialized")
print("Current configuration is:") print("Current configuration is:")
print("\tinstance:", self.config["instance"]) print("\tinstance:", self.config["mastodon"]["instance"])
print("\tuser:", self.config["account"]) print("\tuser:", self.config["mastodon"]["account"])
ans = input("Do you want to re-init it? (yes/no): ").strip() ans = input("Do you want to re-init it? (yes/no): ").strip()
if ans[0].lower() == 'n': if ans[0].lower() == 'n':
@ -315,7 +320,7 @@ class lift:
print("For example: https://social.antigene.org") print("For example: https://social.antigene.org")
# instance = input("URL: ").strip() # instance = input("URL: ").strip()
instance = "https://social.antigene.org" instance = "https://social.antigene.org"
self.config["instance"] = instance self.config["mastodon"]["instance"] = instance
self.register_app() self.register_app()
self.oauth() self.oauth()
@ -326,42 +331,47 @@ class lift:
toml.dump(config, fd) toml.dump(config, fd)
def load_config(self): def load_config(self):
logger.debug(f"Load config from: {self.config_path}") logger.debug(f"│ │ ├ Load config from: {self.config_path}")
config = { config = {
"instance": None, "mastodon": {
"account": None, "instance": None,
"client_id": None, "account": None,
"client_secret": None, "client_id": None,
"token": None "client_secret": None,
"token": None,
}
} }
local_config = toml.load(self.config_path) local_config = toml.load(self.config_path)
config.update(local_config) config.update(local_config)
logger.debug(config)
for k,v in config["mastodon"].items():
logger.debug(f"│ │ │ ├ {k}: {v}")
logger.debug(f"│ │ │ └OK")
return config return config
def register_app(self): def register_app(self):
logger.debug(f"Register {self.name} on {self.config['instance']}") logger.debug(f"Register {self.name} on {self.config["mastodon"]['instance']}")
client_id, client_secret = mastodon.Mastodon.create_app( client_id, client_secret = mastodon.Mastodon.create_app(
self.name, self.name,
scopes = self._scopes, scopes = self._scopes,
api_base_url = self.config["instance"], api_base_url = self.config['mastodon']["instance"],
website = "https://nojhan.net/git/nojhan/forthlift", website = "https://nojhan.net/git/nojhan/forthlift",
user_agent = f"{self.name}:{self.version}", user_agent = f"{self.name}:{self.version}",
) )
self.config["client_id"] = client_id self.config["mastodon"]["client_id"] = client_id
logger.debug(f"ID: {client_id}") logger.debug(f"ID: {client_id}")
self.config["client_secret"] = client_secret self.config["mastodon"]["client_secret"] = client_secret
logger.debug(f"Secret: {client_secret}") logger.debug(f"Secret: {client_secret}")
self.save_config(self.config) self.save_config(self.config)
def oauth(self): def oauth(self):
logger.debug(f"OAuth to: {self.config['instance']}") logger.debug(f"OAuth to: {self.config['mastodon']['instance']}")
oauth = mastodon.Mastodon( oauth = mastodon.Mastodon(
client_id = self.config["client_id"], client_id = self.config['mastodon']["client_id"],
client_secret = self.config["client_secret"], client_secret = self.config['mastodon']["client_secret"],
api_base_url = self.config["instance"], api_base_url = self.config['mastodon']["instance"],
) )
oauth_url = oauth.auth_request_url(scopes = self._scopes) oauth_url = oauth.auth_request_url(scopes = self._scopes)
logger.debug(f"Opening web page: {oauth_url}") logger.debug(f"Opening web page: {oauth_url}")
@ -375,11 +385,11 @@ class lift:
code = oauth_code, code = oauth_code,
scopes = self._scopes, scopes = self._scopes,
) )
self.config["token"] = token self.config['mastodon']["token"] = token
logger.debug(f"Token: {token}") logger.debug(f"Token: {token}")
account = oauth.me().acct account = oauth.me().acct
self.config["account"] = account self.config['mastodon']["account"] = account
logger.debug(f"Account: {account}") logger.debug(f"Account: {account}")
self.save_config(self.config) self.save_config(self.config)
@ -387,32 +397,28 @@ class lift:
if self.dry_run: if self.dry_run:
print(item) print(item)
else: else:
print(item) if prev_status:
logger.debug(item)
return self.masto.status_reply(to_status = prev_status, status = item)
else:
logger.debug(item)
return self.masto.status_post(item)
def __call__(self, items): def __call__(self, items):
n = 0 n = 0
first_item = next(items, None) first_item = next(items, None)
if first_item == None: if first_item == None:
logger.error("No item to post") logger.error("│ │ │ ├ No item to post")
return return
else: else:
logger.debug(f"Post {n+1}") logger.debug(f"│ │ │ ├ Post #{n+1}")
n += 1 n += 1
# prev_status = self.masto.status_post( prev_status = self.post(first_item)
# status = first_item,
# )
# logger.debug(prev_status)
self.post(first_item)
for item in items: for item in items:
logger.debug(f"Post {n+1}") logger.debug(f"│ │ │ ├ Post #{n+1}")
n += 1 n += 1
# prev_status = self.masto.status_reply( prev_status = self.post(item)
# to_status = prev_status, logger.debug(f"│ │ │ └OK, posted {n} items")
# status = item,
# )
# logger.debug(prev_status)
self.post(item)
logger.debug(f"Posted {n} items")
class Forthlifter: class Forthlifter:
def __init__(self, def __init__(self,
@ -464,6 +470,7 @@ class Forthlifter:
logger.debug(f"│ │ ├ {type(lifter).__name__}") logger.debug(f"│ │ ├ {type(lifter).__name__}")
# Call # Call
lifter(items) lifter(items)
logger.debug(f"│ │ └OK")
def __call__(self): def __call__(self):
logger.debug("├ call") logger.debug("├ call")