mastofeed/utils/feed_generator.py

import datetime, rfeed, requests, re

# error codes
USER_NOT_FOUND = 1
CANNOT_ACCESS_INSTANCE = 2
INVALID_HANDLE = 3


def parse_handle(user_handle: str) -> list[str] | None:
    # validate user's handle
    HANDLE_PATTERN = "@[a-zA-Z0-9_]+@[^\t\n\r\f\v]+"
    if re.match(HANDLE_PATTERN, user_handle) is None:
        return None
    
    # parse user's handle
    return user_handle.split("@")[1:]


def get_statuses_of_user(user_handle: str) -> list[dict] | str:
    parsed_handle = parse_handle(user_handle)
    if parsed_handle is None:
        return INVALID_HANDLE
    [username, instance] = parsed_handle

    try:
        account_lookup = requests.get(f"https://{instance}/api/v1/accounts/lookup?acct={username}")
    except requests.exceptions.RequestException as err:
        print(err)
        return CANNOT_ACCESS_INSTANCE

    if not account_lookup.ok:
        return USER_NOT_FOUND

    account_info = account_lookup.json()
    user_id = account_info["id"]

    try:
        statuses_get = requests.get(f"https://{instance}/api/v1/accounts/{user_id}/statuses?exclude_reblogs=true&exclude_replies=true")
    except requests.exceptions.RequestException as err:
        print(err)
        return CANNOT_ACCESS_INSTANCE

    if not statuses_get.ok:
        return USER_NOT_FOUND

    return statuses_get.json()


def generate_feed_of_user(user_handle: str) -> str | int:
    # get user's statuses
    statuses = get_statuses_of_user(user_handle)
    if type(statuses) is int:
        return statuses

    # generate rss feed
    status_items = []
    for status in statuses:
        content = status["content"]

        # get media of status
        for media in status["media_attachments"]:
            media_url = media["url"]
            media_description = media["description"]
            if media["type"] == "image":
                content += f"<img src='{media_url}' alt='{media_description if not media_description is None else ''}' />"
            elif media["type"] == "video" or media["type"] == "gifv":
                content += f"<video><source src='{media_url}'>{f'<p>{media_description}</p>' if not media_description is None else ''}</source></video>"
            elif media["type"] == "audio":
                content += f"<audio src='{media_url}' controls></audio>{f'<span>{media_description}</span>' if not media_description is None else ''}"
            else:
                content += f"<a href='{media_url}'>{media_description if not media_description is None else media_url}</a>"

        status_items.append(rfeed.Item(
            description=content,
            link=status["url"],
            guid=rfeed.Guid(status["url"]),
            pubDate=datetime.datetime.strptime(status["created_at"], "%Y-%m-%dT%H:%M:%S.%fZ")
        ))

    feed = rfeed.Feed(
        title=status["account"]["display_name"],
        link=status["account"]["url"],
        description=status["account"]["note"],
        lastBuildDate=datetime.datetime.now(),
        items=status_items
    )

    return feed.rss()