251 lines
11 KiB
Python
251 lines
11 KiB
Python
## Copyright © 2026 Olaf Kolkman
|
|
## SPDX-License-Identifier: GPL-3.0-or-later
|
|
|
|
import json
|
|
import logging
|
|
from datetime import datetime, timezone
|
|
from urllib.error import HTTPError, URLError
|
|
from urllib.parse import urlencode
|
|
from urllib.request import Request
|
|
|
|
from backend.app.plugins.base import BasePlugin
|
|
from backend.app.services.secret_store import decrypt_secret
|
|
from backend.app.services.mastodon_security import open_no_redirect, validate_public_instance
|
|
|
|
DEFAULT_POST_PREFIX = 'From my #LinkLog: '
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class DefaultFrontendPlugin(BasePlugin):
|
|
"""Accept events for the built-in frontend plugin."""
|
|
|
|
name = 'default_frontend'
|
|
version = '1.0.0'
|
|
|
|
def handle_event(self, event):
|
|
"""Acknowledge an event without performing an external side effect."""
|
|
return {'status': 'accepted', 'plugin': self.name, 'event': event.get('type')}
|
|
|
|
|
|
class MastodonPlugin(BasePlugin):
|
|
"""Publish and remove LinkLog entries through a Mastodon instance."""
|
|
|
|
name = 'mastodon'
|
|
version = '1.0.0'
|
|
enabled = False
|
|
config = {}
|
|
|
|
def initialize(self, config=None):
|
|
"""Store the administrator-provided defaults for later event handling."""
|
|
self.config = config or {}
|
|
return True
|
|
|
|
def handle_event(self, event):
|
|
"""Publish a link event, returning a structured status result.
|
|
|
|
Per-user plugin settings override global settings. The access token is
|
|
decrypted only for the duration of the outbound request, and the
|
|
instance is validated before any network connection is opened.
|
|
"""
|
|
if not event.get('post_to_mastodon', True):
|
|
return {'status': 'skipped', 'plugin': self.name, 'reason': 'not_requested'}
|
|
|
|
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']))
|
|
config['access_token'] = decrypt_secret(config.get('access_token', ''))
|
|
|
|
try:
|
|
instance = validate_public_instance(str(config.get('instance', '')))
|
|
except ValueError as error:
|
|
return {'status': 'failed', 'plugin': self.name, 'reason': str(error)}
|
|
access_token = str(config.get('access_token', '')).strip()
|
|
if not instance or not access_token:
|
|
logger.debug(
|
|
'Mastodon post skipped: instance_configured=%s token_configured=%s user_id=%s',
|
|
bool(instance), bool(access_token), user_id,
|
|
)
|
|
return {'status': 'skipped', 'plugin': self.name, 'reason': 'not_configured'}
|
|
|
|
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).strip()
|
|
title = str(event.get('title') or '').strip()
|
|
status_parts = [post_prefix.strip()]
|
|
if title:
|
|
status_parts.append(title)
|
|
if event.get('comment'):
|
|
status_parts.append(event['comment'])
|
|
if title:
|
|
timestamp_value = event.get('timestamp') or event.get('created_at')
|
|
url = str(event.get('url', '') or '').strip()
|
|
if timestamp_value:
|
|
try:
|
|
parsed = datetime.fromisoformat(str(timestamp_value).replace('Z', '+00:00'))
|
|
if parsed.tzinfo is None:
|
|
parsed = parsed.replace(tzinfo=timezone.utc)
|
|
formatted = parsed.astimezone(timezone.utc).strftime('%Y %B %d - %H:%M')
|
|
status_parts.append(f'Logged on {formatted} UTC from: {url}')
|
|
except ValueError:
|
|
status_parts.append(f'Logged from: {url}')
|
|
else:
|
|
status_parts.append(f'Logged from: {url}')
|
|
if event.get('tags'):
|
|
status_parts.append(' '.join(event['tags']))
|
|
post_body = '\n\n'.join(status_parts)
|
|
endpoint = f'{instance}/api/v1/statuses'
|
|
logger.debug(
|
|
'Posting link to Mastodon: endpoint=%s user_id=%s event_id=%s body_length=%d',
|
|
endpoint, user_id, event.get('id'), len(post_body),
|
|
)
|
|
|
|
try:
|
|
request = Request(
|
|
endpoint,
|
|
data=urlencode({'status': post_body}).encode('utf-8'),
|
|
headers={
|
|
'Authorization': f'Bearer {access_token}',
|
|
'Content-Type': 'application/x-www-form-urlencoded',
|
|
'Accept': 'application/json',
|
|
'User-Agent': 'LinkLog/1.0',
|
|
},
|
|
method='POST',
|
|
)
|
|
with open_no_redirect(request, timeout=5) as response:
|
|
response_body = response.read().decode('utf-8')
|
|
logger.debug(
|
|
'Mastodon post response: endpoint=%s status=%s body_length=%d',
|
|
endpoint, response.status, len(response_body),
|
|
)
|
|
response_data = json.loads(response_body)
|
|
logger.info('Mastodon post succeeded: instance=%s user_id=%s post_id=%s', instance, user_id, response_data.get('id'))
|
|
return {
|
|
'status': 'posted',
|
|
'plugin': self.name,
|
|
'post_id': response_data.get('id'),
|
|
}
|
|
except HTTPError as error:
|
|
response_body = error.read().decode('utf-8', errors='replace')
|
|
logger.warning(
|
|
'Mastodon post failed: endpoint=%s user_id=%s status=%s response=%s',
|
|
endpoint, user_id, error.code, response_body[:500],
|
|
)
|
|
return {
|
|
'status': 'failed',
|
|
'plugin': self.name,
|
|
'reason': f'HTTP {error.code}: {response_body[:500]}',
|
|
}
|
|
except (URLError, TimeoutError, OSError, ValueError) as error:
|
|
logger.exception('Mastodon post failed: endpoint=%s user_id=%s error=%s', endpoint, user_id, error)
|
|
return {
|
|
'status': 'failed',
|
|
'plugin': self.name,
|
|
'reason': str(error),
|
|
}
|
|
|
|
def delete_posts(self, event):
|
|
"""Delete all Mastodon statuses recorded for a link 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']))
|
|
config['access_token'] = decrypt_secret(config.get('access_token', ''))
|
|
|
|
try:
|
|
instance = validate_public_instance(str(config.get('instance', '')))
|
|
except ValueError as error:
|
|
return {'status': 'failed', 'plugin': self.name, 'reason': str(error)}
|
|
access_token = str(config.get('access_token', '')).strip()
|
|
post_ids = event.get('mastodon_post_ids') or []
|
|
if not post_ids and event.get('mastodon_post_id'):
|
|
post_ids = [event['mastodon_post_id']]
|
|
if not instance or not access_token:
|
|
return {'status': 'failed', 'plugin': self.name, 'reason': 'Mastodon is not configured'}
|
|
|
|
try:
|
|
for post_id in post_ids:
|
|
request = Request(
|
|
f'{instance}/api/v1/statuses/{post_id}',
|
|
headers={'Authorization': f'Bearer {access_token}', 'User-Agent': 'LinkLog/1.0'},
|
|
method='DELETE',
|
|
)
|
|
with open_no_redirect(request, timeout=5) as response:
|
|
response.read()
|
|
return {'status': 'deleted', 'plugin': self.name, 'count': len(post_ids)}
|
|
except HTTPError as error:
|
|
response_body = error.read().decode('utf-8', errors='replace')
|
|
return {'status': 'failed', 'plugin': self.name, 'reason': f'HTTP {error.code}: {response_body[:500]}'}
|
|
except (URLError, TimeoutError, OSError) as error:
|
|
return {'status': 'failed', 'plugin': self.name, 'reason': str(error)}
|
|
|
|
|
|
class PluginManager:
|
|
"""Load enabled plugins and route LinkLog events to them."""
|
|
|
|
def __init__(self):
|
|
"""Create the built-in plugin registry."""
|
|
self.plugins = [DefaultFrontendPlugin(), MastodonPlugin()]
|
|
|
|
def refresh_from_db(self):
|
|
"""Refresh enabled flags and configuration from the plugin table."""
|
|
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):
|
|
"""Send an event to every currently enabled plugin."""
|
|
self.refresh_from_db()
|
|
results = []
|
|
for plugin in self.plugins:
|
|
if plugin.enabled:
|
|
result = plugin.handle_event(event)
|
|
results.append(result)
|
|
logger.debug('Plugin dispatch result: plugin=%s event_id=%s result=%s', plugin.name, event.get('id'), result)
|
|
return results
|
|
|
|
def post_to_mastodon(self, event):
|
|
"""Publish one event through Mastodon when that plugin is enabled."""
|
|
self.refresh_from_db()
|
|
plugin = next(plugin for plugin in self.plugins if plugin.name == 'mastodon')
|
|
if not plugin.enabled:
|
|
return {'status': 'skipped', 'plugin': 'mastodon', 'reason': 'disabled'}
|
|
return plugin.handle_event(event)
|
|
|
|
def delete_mastodon_posts(self, event):
|
|
"""Delete the Mastodon statuses associated with an event."""
|
|
self.refresh_from_db()
|
|
plugin = next(plugin for plugin in self.plugins if plugin.name == 'mastodon')
|
|
return plugin.delete_posts(event)
|
|
|
|
|
|
plugin_manager = PluginManager()
|