## 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): 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): 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): 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: 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: 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): 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): self.refresh_from_db() plugin = next(plugin for plugin in self.plugins if plugin.name == 'mastodon') return plugin.delete_posts(event) plugin_manager = PluginManager()