Mirrors/catask
1
0
Fork 0
mirror of https://codeberg.org/catask-org/catask.git synced 2025-04-19 13:23:41 -05:00
Code Issues Projects Releases Packages Wiki Activity Actions

revamp some fediverse crosspost logic and add bluesky crosspost

Browse source
This commit is contained in:
mystie 2025-03-28 19:51:41 +03:00
parent dd27fc8e09
commit 76b5c58b50
1 changed files with 76 additions and 12 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

88
functions.py
View file

@ -7,7 +7,6 @@ from pathlib import Path
from mistune import HTMLRenderer, escape from mistune import HTMLRenderer, escape
from PIL import Image from PIL import Image
from psycopg.rows import dict_row from psycopg.rows import dict_row
from mastodon import Mastodon
import traceback import traceback
import base64 import base64
import time import time
@ -313,7 +312,29 @@ def addAnswer(question_id: int, answer: str, cw: str) -> dict:
return jsonify({'message': _('Answer added successfully!')}), 201 return jsonify({'message': _('Answer added successfully!')}), 201
if cfg['crosspost']['fediverse']['enabled']:
from mastodon import Mastodon
app.logger.debug("Initializing Mastodon.py client...")
fedi_client = Mastodon(
client_id = cfg['crosspost']['fediverse']['client']['id'],
client_secret = cfg['crosspost']['fediverse']['client']['secret'],
api_base_url = 'https://' + cfg['crosspost']['fediverse']['instance'],
access_token = cfg['crosspost']['fediverse']['token']
)
if cfg['crosspost']['bluesky']['enabled']:
from atproto import Client, client_utils
app.logger.debug("[postOnBluesky] Initializing ATProto client...")
bsky_client = Client()
bsky_client.login(cfg['crosspost']['bluesky']['handle'], cfg['crosspost']['bluesky']['appPassword'])
def postOnFediverse(question_id: int, answer: str, cw: str) -> None: def postOnFediverse(question_id: int, answer: str, cw: str) -> None:
global fedi_client
# reloading config file # reloading config file
cfg = loadJSON(const.configFile) cfg = loadJSON(const.configFile)
@ -325,23 +346,66 @@ def postOnFediverse(question_id: int, answer: str, cw: str) -> None:
cursor.execute("SELECT id, content FROM questions WHERE id=%s", (question_id,)) cursor.execute("SELECT id, content FROM questions WHERE id=%s", (question_id,))
question = cursor.fetchone() question = cursor.fetchone()
app.logger.debug("[postOnFediverse] Initializing Mastodon client...") warning = f"{cfg['crosspost']['fediverse']['cw']}"
if question.get('cw') and not cw:
warning += f", {question['cw']}"
elif question.get('cw') and cw and (question['cw'] != cw):
warning += f", {question['cw']}, {cw}"
elif (question.get('cw') and cw) and question['cw'] == cw:
warning += f", {cw}"
elif cw and not question.get('cw'):
warning += f", {cw}"
client = Mastodon(
client_id = cfg['crosspost']['fediverse']['client']['id'],
client_secret = cfg['crosspost']['fediverse']['client']['secret'],
api_base_url = 'https://' + cfg['crosspost']['fediverse']['instance'],
access_token = cfg['crosspost']['fediverse']['token']
)
if cw:
warning = f"{cfg['crosspost']['fediverse']['cw']}, {cw}"
else: else:
warning = cfg['crosspost']['fediverse']['cw'] warning = ""
post_content = f"{question['content']} — {answer} {cfg['instance']['fullBaseUrl']}/q/{question['id']}/" post_content = f"{question['content']} — {answer} {cfg['instance']['fullBaseUrl']}/q/{question['id']}/"
client.status_post(post_content, visibility=cfg['crosspost']['fediverse']['visibility'], spoiler_text=warning) fedi_client.status_post(post_content, visibility=cfg['crosspost']['fediverse']['visibility'], spoiler_text=warning)
app.logger.debug(f"[postOnFediverse] Made Fediverse post: {post_content}") app.logger.debug(f"[postOnFediverse] Made Fediverse post: {post_content}")
def postOnBluesky(question_id: int, answer: str, cw: str) -> None:
global bsky_client
# reloading config file
cfg = loadJSON(const.configFile)
conn = connectToDb()
cursor = conn.cursor()
app.logger.debug(f"[postOnBluesky] Grabbing question {question_id} from database...")
cursor.execute("SELECT id, content, cw FROM questions WHERE id=%s", (question_id,))
question = cursor.fetchone()
if question.get('cw') and not cw:
warning = f"[CW: {question['cw']}]\n\n"
elif question.get('cw') and cw and (question['cw'] != cw):
warning = f"[CW: {question['cw']}, {cw}]\n\n"
elif question['cw'] == cw:
warning = f"[CW: {cw}]\n\n"
elif cw and not question.get('cw'):
warning = f"[CW: {cw}]\n\n"
else:
warning = ""
# warning = f"{ '[CW:' if cw or question.get('cw') else '' }{ ' ' + question['cw'] if question.get('cw') else '' }{ ', ' + cw if cw else '' }{']\n\n' if cw or question.get('cw') else '' }"
text_builder = client_utils.TextBuilder()
text_builder.text(f"{warning}{question['content']} — {answer} ")
text_builder.link(f"{cfg['instance']['fullBaseUrl']}/q/{question['id']}/", f"{cfg['instance']['fullBaseUrl']}/q/{question['id']}/")
bsky_client.send_post(text_builder)
app.logger.debug(f"[postOnBluesky] Made Bluesky post: {text_builder.build_text()}")
def ntfySend(cw, return_val, from_who, question) -> None: def ntfySend(cw, return_val, from_who, question) -> None:
app.logger.debug("[CatAsk/functions/ntfySend] started ntfy flow") app.logger.debug("[CatAsk/functions/ntfySend] started ntfy flow")
ntfy_cw = f" [CW: {cw}]" if cw else "" ntfy_cw = f" [CW: {cw}]" if cw else ""