pyfedi/app/community/util.py

from datetime import datetime
from threading import Thread
from time import sleep
from typing import List
import requests
from PIL import Image, ImageOps
from flask import request, abort, g, current_app
from flask_login import current_user
from pillow_heif import register_heif_opener

from app import db, cache
from app.activitypub.util import find_actor_or_create, actor_json_to_model, post_json_to_model
from app.constants import POST_TYPE_ARTICLE, POST_TYPE_LINK, POST_TYPE_IMAGE
from app.models import Community, File, BannedInstances, PostReply, PostVote, Post, utcnow, CommunityMember
from app.utils import get_request, gibberish, markdown_to_html, domain_from_url, validate_image, allowlist_html, \
    html_to_markdown, is_image_url, ensure_directory_exists
from sqlalchemy import desc, text
import os
from opengraph_parse import parse_page


allowed_extensions = ['.gif', '.jpg', '.jpeg', '.png', '.webp', '.heic']


def search_for_community(address: str):
    if address.startswith('!'):
        name, server = address[1:].split('@')

        banned = BannedInstances.query.filter_by(domain=server).first()
        if banned:
            reason = f" Reason: {banned.reason}" if banned.reason is not None else ''
            raise Exception(f"{server} is blocked.{reason}")  # todo: create custom exception class hierarchy

        already_exists = Community.query.filter_by(ap_id=address[1:]).first()
        if already_exists:
            return already_exists

        # Look up the profile address of the community using WebFinger
        # todo: try, except block around every get_request
        webfinger_data = get_request(f"https://{server}/.well-known/webfinger",
                                     params={'resource': f"acct:{address[1:]}"})
        if webfinger_data.status_code == 200:
            webfinger_json = webfinger_data.json()
            for links in webfinger_json['links']:
                if 'rel' in links and links['rel'] == 'self':  # this contains the URL of the activitypub profile
                    type = links['type'] if 'type' in links else 'application/activity+json'
                    # retrieve the activitypub profile
                    community_data = get_request(links['href'], headers={'Accept': type})
                    # to see the structure of the json contained in community_data, do a GET to https://lemmy.world/c/technology with header Accept: application/activity+json
                    if community_data.status_code == 200:
                        community_json = community_data.json()
                        community_data.close()
                        if community_json['type'] == 'Group':
                            community = actor_json_to_model(community_json, address, server)
                            thr = Thread(target=retrieve_mods_and_backfill_thread, args=[community, current_app._get_current_object()])
                            thr.start()
                            return community
        return None


def retrieve_mods_and_backfill_thread(community: Community, app):
    with app.app_context():
        if community.ap_moderators_url:
            mods_request = get_request(community.ap_moderators_url, headers={'Accept': 'application/activity+json'})
            if mods_request.status_code == 200:
                mods_data = mods_request.json()
                mods_request.close()
                if mods_data and mods_data['type'] == 'OrderedCollection' and 'orderedItems' in mods_data:
                    for actor in mods_data['orderedItems']:
                        sleep(0.5)
                        user = find_actor_or_create(actor)
                        if user:
                            existing_membership = CommunityMember.query.filter_by(community_id=community.id, user_id=user.id).first()
                            if existing_membership:
                                existing_membership.is_moderator = True
                            else:
                                new_membership = CommunityMember(community_id=community.id, user_id=user.id, is_moderator=True)
                                db.session.add(new_membership)
                    db.session.commit()

        # only backfill nsfw if nsfw communities are allowed
        if (community.nsfw and not g.site.enable_nsfw) or (community.nsfl and not g.site.enable_nsfl):
            return

        # download 50 old posts
        if community.ap_public_url:
            outbox_request = get_request(community.ap_public_url + '/outbox', headers={'Accept': 'application/activity+json'})
            if outbox_request.status_code == 200:
                outbox_data = outbox_request.json()
                outbox_request.close()
                if outbox_data['type'] == 'OrderedCollection' and 'orderedItems' in outbox_data:
                    activities_processed = 0
                    for activity in outbox_data['orderedItems']:
                        user = find_actor_or_create(activity['object']['actor'])
                        if user:
                            post = post_json_to_model(activity['object']['object'], user, community)
                            post.ap_create_id = activity['object']['id']
                            post.ap_announce_id = activity['id']
                            db.session.commit()

                        activities_processed += 1
                        if activities_processed >= 50:
                            break
                    c = Community.query.get(community.id)
                    c.post_count = activities_processed
                    c.last_active = utcnow()
                    db.session.commit()


def community_url_exists(url) -> bool:
    community = Community.query.filter_by(ap_profile_id=url).first()
    return community is not None


def actor_to_community(actor) -> Community:
    actor = actor.strip()
    if '@' in actor:
        community = Community.query.filter_by(banned=False, ap_id=actor).first()
    else:
        community = Community.query.filter_by(name=actor, banned=False, ap_id=None).first()
    return community


@cache.memoize(timeout=50)
def opengraph_parse(url):
    try:
        return parse_page(url)
    except Exception as ex:
        return None


def url_to_thumbnail_file(filename) -> File:
    unused, file_extension = os.path.splitext(filename)
    response = requests.get(filename, timeout=5)
    if response.status_code == 200:
        new_filename = gibberish(15)
        directory = 'app/static/media/posts/' + new_filename[0:2] + '/' + new_filename[2:4]
        ensure_directory_exists(directory)
        final_place = os.path.join(directory, new_filename + file_extension)
        with open(final_place, 'wb') as f:
            f.write(response.content)
        with Image.open(final_place) as img:
            img = ImageOps.exif_transpose(img)
            img.thumbnail((150, 150))
            img.save(final_place)
            thumbnail_width = img.width
            thumbnail_height = img.height
        return File(file_name=new_filename + file_extension, thumbnail_width=thumbnail_width,
                    thumbnail_height=thumbnail_height, thumbnail_path=final_place,
                    source_url=filename)


def save_post(form, post: Post):
    post.nsfw = form.nsfw.data
    post.nsfl = form.nsfl.data
    post.notify_author = form.notify_author.data
    if form.type.data == '' or form.type.data == 'discussion':
        post.title = form.discussion_title.data
        post.body = form.discussion_body.data
        post.body_html = markdown_to_html(post.body)
        post.type = POST_TYPE_ARTICLE
    elif form.type.data == 'link':
        post.title = form.link_title.data
        url_changed = post.id is None or form.link_url.data != post.url
        post.url = form.link_url.data
        post.type = POST_TYPE_LINK
        domain = domain_from_url(form.link_url.data)
        domain.post_count += 1
        post.domain = domain

        if url_changed:
            if post.image_id:
                remove_old_file(post.image_id)
                post.image_id = None
            valid_extensions = {'.jpg', '.jpeg', '.png', '.gif', '.webp'}
            unused, file_extension = os.path.splitext(form.link_url.data)  # do not use _ here instead of 'unused'
            # this url is a link to an image - generate a thumbnail of it
            if file_extension in valid_extensions:
                file = url_to_thumbnail_file(form.link_url.data)
                if file:
                    post.image = file
                    db.session.add(file)
            else:
                # check opengraph tags on the page and make a thumbnail if an image is available in the og:image meta tag
                opengraph = opengraph_parse(form.link_url.data)
                if opengraph and opengraph.get('og:image', '') != '':
                    filename = opengraph.get('og:image')
                    valid_extensions = {'.jpg', '.jpeg', '.png', '.gif', '.webp'}
                    unused, file_extension = os.path.splitext(filename)
                    if file_extension.lower() in valid_extensions:
                        file = url_to_thumbnail_file(filename)
                        if file:
                            file.alt_text = opengraph.get('og:title')
                            post.image = file
                            db.session.add(file)

    elif form.type.data == 'image':
        post.title = form.image_title.data
        post.type = POST_TYPE_IMAGE
        uploaded_file = request.files['image_file']
        if uploaded_file and uploaded_file.filename != '':
            if post.image_id:
                remove_old_file(post.image_id)
                post.image_id = None

            # check if this is an allowed type of file
            file_ext = os.path.splitext(uploaded_file.filename)[1]
            if file_ext.lower() not in allowed_extensions or file_ext != validate_image(
                    uploaded_file.stream):
                abort(400)
            new_filename = gibberish(15)

            # set up the storage directory
            directory = 'app/static/media/posts/' + new_filename[0:2] + '/' + new_filename[2:4]
            ensure_directory_exists(directory)

            # save the file
            final_place = os.path.join(directory, new_filename + file_ext)
            final_place_thumbnail = os.path.join(directory, new_filename + '_thumbnail.webp')
            uploaded_file.save(final_place)

            if file_ext.lower() == '.heic':
                register_heif_opener()

            # resize if necessary
            img = Image.open(final_place)
            img = ImageOps.exif_transpose(img)
            img_width = img.width
            img_height = img.height
            if img.width > 2000 or img.height > 2000:
                img.thumbnail((2000, 2000))
                img.save(final_place)
                img_width = img.width
                img_height = img.height
            # save a second, smaller, version as a thumbnail
            img.thumbnail((256, 256))
            img.save(final_place_thumbnail, format="WebP", quality=93)
            thumbnail_width = img.width
            thumbnail_height = img.height

            file = File(file_path=final_place, file_name=new_filename + file_ext, alt_text=form.image_title.data,
                        width=img_width, height=img_height, thumbnail_width=thumbnail_width,
                        thumbnail_height=thumbnail_height, thumbnail_path=final_place_thumbnail)
            post.image = file
            db.session.add(file)

    elif form.type.data == 'poll':
        ...
    else:
        raise Exception('invalid post type')
    if post.id is None:
        postvote = PostVote(user_id=current_user.id, author_id=current_user.id, post=post, effect=1.0)
        post.up_votes = 1
        post.score = 1
        db.session.add(postvote)
        db.session.add(post)
    g.site.last_active = utcnow()


def remove_old_file(file_id):
    remove_file = File.query.get(file_id)
    remove_file.delete_from_disk()


def save_icon_file(icon_file) -> File:
    # check if this is an allowed type of file
    file_ext = os.path.splitext(icon_file.filename)[1]
    if file_ext.lower() not in allowed_extensions or file_ext != validate_image(
            icon_file.stream):
        abort(400)
    new_filename = gibberish(15)

    # set up the storage directory
    directory = 'app/static/media/communities/' + new_filename[0:2] + '/' + new_filename[2:4]
    ensure_directory_exists(directory)

    # save the file
    final_place = os.path.join(directory, new_filename + file_ext)
    final_place_thumbnail = os.path.join(directory, new_filename + '_thumbnail.webp')
    icon_file.save(final_place)

    if file_ext.lower() == '.heic':
        register_heif_opener()

    # resize if necessary
    img = Image.open(final_place)
    img = ImageOps.exif_transpose(img)
    img_width = img.width
    img_height = img.height
    if img.width > 250 or img.height > 250:
        img.thumbnail((250, 250))
        img.save(final_place)
        img_width = img.width
        img_height = img.height
    # save a second, smaller, version as a thumbnail
    img.thumbnail((40, 40))
    img.save(final_place_thumbnail, format="WebP", quality=93)
    thumbnail_width = img.width
    thumbnail_height = img.height

    file = File(file_path=final_place, file_name=new_filename + file_ext, alt_text='community icon',
                width=img_width, height=img_height, thumbnail_width=thumbnail_width,
                thumbnail_height=thumbnail_height, thumbnail_path=final_place_thumbnail)
    db.session.add(file)
    return file


def save_banner_file(banner_file) -> File:
    # check if this is an allowed type of file
    file_ext = os.path.splitext(banner_file.filename)[1]
    if file_ext.lower() not in allowed_extensions or file_ext != validate_image(
            banner_file.stream):
        abort(400)
    new_filename = gibberish(15)

    # set up the storage directory
    directory = 'app/static/media/communities/' + new_filename[0:2] + '/' + new_filename[2:4]
    ensure_directory_exists(directory)

    # save the file
    final_place = os.path.join(directory, new_filename + file_ext)
    final_place_thumbnail = os.path.join(directory, new_filename + '_thumbnail.webp')
    banner_file.save(final_place)

    if file_ext.lower() == '.heic':
        register_heif_opener()

    # resize if necessary
    img = Image.open(final_place)
    img = ImageOps.exif_transpose(img)
    img_width = img.width
    img_height = img.height
    if img.width > 1600 or img.height > 600:
        img.thumbnail((1600, 600))
        img.save(final_place)
        img_width = img.width
        img_height = img.height

    # save a second, smaller, version as a thumbnail
    img.thumbnail((700, 500))
    img.save(final_place_thumbnail, format="WebP", quality=93)
    thumbnail_width = img.width
    thumbnail_height = img.height

    file = File(file_path=final_place, file_name=new_filename + file_ext, alt_text='community banner',
                width=img_width, height=img_height, thumbnail_width=thumbnail_width, thumbnail_height=thumbnail_height)
    db.session.add(file)
    return file