mastobot/app/routers/taloustaito.py

import traceback
import json
import requests
import traceback
import json
import feedparser
import requests
import logging

from datetime import datetime
from time import mktime
from typing import Annotated
from fastapi import Depends, APIRouter
from bs4 import BeautifulSoup

from settings.defaults import Settings, get_settings

router = APIRouter()
logger = logging.getLogger(__name__)

@router.get("/taloustaito", summary="Taloustaito RSS")
async def update(settings: Annotated[Settings, Depends(get_settings)]):

    feed_url = settings.feeds['taloustaito']['url']
    
    mastodon_server = settings.mastodon_server
    mastodon_aid = settings.feeds['taloustaito']['account_id']
    mastodon_token = str(settings.feeds['taloustaito']['token'])
    mastodon_get_statuses_url=f'{mastodon_server}/api/v1/accounts/{mastodon_aid}/statuses'
    mastodon_post_statuses_url=f'{mastodon_server}/api/v1/statuses'

    try:
        last_status = load_last_status(mastodon_get_statuses_url, mastodon_token)
        if last_status:
            last_status_timestamp=datetime.fromisoformat(last_status['created_at'])
            new_entries=load_feed_rss(feed_url, last_status_timestamp)
            logger.info(f'Found {len(new_entries)} new entries since {last_status_timestamp}')
        else:
            new_entries=load_feed_rss(feed_url, None)
            new_entries.sort(key=lambda e: e['published_parsed'], reverse=True)
            new_entries = new_entries[:3]
            logger.info(f'Found {len(new_entries)} entries')

        if (len(new_entries) == 0):
            return {
                "status": 200,
                "body": {
                    "posted_entries": 0,
                    "successful": True
                }
            }

        posted_entries=list(map(lambda x: post_rss_entry_to_mastodon(mastodon_post_statuses_url, mastodon_token, x), new_entries))

        return {
            "status": 200,
            "body": {
                "posted_entries": len(posted_entries),
                "successful": True
            }
        }
    except Exception as e:
        msg = ''.join(traceback.format_exception_only(e))
        logger.error(msg)
        return {
            "status": 501,
            "body": {
                "posted_entries": 0,
                "message": msg,
                "successful": False
            }
        }


def load_last_status(url, token):
    response=requests.get(url + '?limit=1', headers={ 'Authorization' : f'Bearer {token}' })
    if response.status_code != 200:
        raise Exception('Failed to contact Mastodon', response.text)
  
    document = json.loads(response.text)
    if len(document) > 0:
        return document[0]
    else:
        return None

def post_rss_entry_to_mastodon(url, token, entry):

    title = entry.title
    description = BeautifulSoup(entry.summary, features="html.parser").get_text()
    link = entry.link
    message = f"{title}\n\n{description}\n\n{link}"

    headers = {
        'Authorization': f'Bearer {token}',
        'Content-type': 'application/x-www-form-urlencoded',
        'User-Agent': 'Serverless Feed'
    }

    params = {
        'status': message,
        'language': 'fi',
        'visibility': 'public'
    }

    response = requests.post(url, data=params, headers=headers)
    if response.status_code != 200:
        logger.error('Failed to post message', response)
    return response


def load_feed_rss(url, since):
    feed=feedparser.parse(url)
    return [entry for entry in feed.entries if since == None or datetime.fromtimestamp(mktime(entry.published_parsed)) > since.replace(tzinfo=datetime.fromtimestamp(mktime(entry.published_parsed)).tzinfo)]