119 lines
4.1 KiB
Python
119 lines
4.1 KiB
Python
## Copyright © 2026 Olaf Kolkman
|
|
## SPDX-License-Identifier: GPL-3.0-or-later
|
|
|
|
import json
|
|
from urllib.error import HTTPError, URLError
|
|
from urllib.request import Request, urlopen
|
|
|
|
from backend.app.plugins.base import BasePlugin
|
|
|
|
DEFAULT_POST_PREFIX = 'From my #LinkLog: '
|
|
|
|
|
|
class DefaultFrontendPlugin(BasePlugin):
|
|
name = 'default_frontend'
|
|
version = '1.0.0'
|
|
|
|
def handle_event(self, event):
|
|
return {'status': 'accepted', 'plugin': self.name, 'event': event.get('type')}
|
|
|
|
|
|
class MastodonPlugin(BasePlugin):
|
|
name = 'mastodon'
|
|
version = '1.0.0'
|
|
enabled = False
|
|
config = {}
|
|
|
|
def initialize(self, config=None):
|
|
self.config = config or {}
|
|
return True
|
|
|
|
def handle_event(self, event):
|
|
config = dict(self.config)
|
|
user_id = event.get('user_id')
|
|
if user_id:
|
|
from backend.app.database import get_connection
|
|
|
|
with get_connection() as conn:
|
|
row = conn.execute(
|
|
'''
|
|
SELECT config FROM user_plugin_config
|
|
WHERE user_id = ? AND plugin_name = ?
|
|
''',
|
|
(user_id, self.name),
|
|
).fetchone()
|
|
if row and row['config']:
|
|
config.update(json.loads(row['config']))
|
|
|
|
instance = str(config.get('instance', '')).strip().rstrip('/')
|
|
if instance and '://' not in instance:
|
|
instance = f'https://{instance}'
|
|
access_token = str(config.get('access_token', '')).strip()
|
|
if not instance or not access_token:
|
|
return {'status': 'skipped', 'plugin': self.name, 'reason': 'not_configured'}
|
|
|
|
status_parts = [event.get('title') or event.get('url', '')]
|
|
if event.get('comment'):
|
|
status_parts.append(event['comment'])
|
|
post_prefix = config.get('post_prefix')
|
|
if post_prefix is None and config.get('hashtag'):
|
|
post_prefix = f'#{str(config["hashtag"]).strip().lstrip("#")} '
|
|
post_prefix = str(post_prefix if post_prefix is not None else DEFAULT_POST_PREFIX)
|
|
status = f'{post_prefix}{event.get("url", "")}'.strip()
|
|
if event.get('tags'):
|
|
status = f'{status} {" ".join(event["tags"])}'
|
|
status_parts.append(status)
|
|
|
|
try:
|
|
request = Request(
|
|
f'{instance}/api/v1/statuses',
|
|
data=json.dumps({'status': '\n'.join(status_parts)}).encode('utf-8'),
|
|
headers={
|
|
'Authorization': f'Bearer {access_token}',
|
|
'Content-Type': 'application/json',
|
|
},
|
|
method='POST',
|
|
)
|
|
with urlopen(request, timeout=5) as response:
|
|
response_data = json.loads(response.read().decode('utf-8'))
|
|
return {
|
|
'status': 'posted',
|
|
'plugin': self.name,
|
|
'post_id': response_data.get('id'),
|
|
}
|
|
except (HTTPError, URLError, TimeoutError, OSError, ValueError) as error:
|
|
return {
|
|
'status': 'failed',
|
|
'plugin': self.name,
|
|
'reason': str(error),
|
|
}
|
|
|
|
|
|
class PluginManager:
|
|
def __init__(self):
|
|
self.plugins = [DefaultFrontendPlugin(), MastodonPlugin()]
|
|
|
|
def refresh_from_db(self):
|
|
from backend.app.database import get_connection
|
|
|
|
with get_connection() as conn:
|
|
rows = conn.execute('SELECT name, enabled, config FROM plugins').fetchall()
|
|
|
|
enabled_names = {row['name'] for row in rows if row['enabled']}
|
|
for plugin in self.plugins:
|
|
plugin.enabled = plugin.name in enabled_names
|
|
row = next((item for item in rows if item['name'] == plugin.name), None)
|
|
plugin.initialize(json.loads(row['config']) if row and row['config'] else {})
|
|
return enabled_names
|
|
|
|
def dispatch(self, event):
|
|
self.refresh_from_db()
|
|
results = []
|
|
for plugin in self.plugins:
|
|
if plugin.enabled:
|
|
results.append(plugin.handle_event(event))
|
|
return results
|
|
|
|
|
|
plugin_manager = PluginManager()
|