gear-orders/web/api.py

import re
import requests
import json
from functools import wraps
from flask import Blueprint, jsonify, abort, request
from flask_login import login_required, current_user
from db.constants import TIMELINE_ORDERS_POOL_CREATED, TIMELINE_ORDERS_POOL_DELETED, TIMELINE_ORDERS_POOL_UPDATED
from db.models import database, OrdersPool, Order, OrderAddOn, MastodonServer
from db.queries import timeline_event_put, timeline_event_recent, user_can_orders_pools_details, user_can_orders_pools_edit, user_can_orders_pools_view, user_doms, user_get, domsubusers_list, orders_pool_list, orders_pool, mastodon_server_get, mastodon_server_put, user_has_doms, user_mastodon_server_set, user_preferences_set
from settings import MASTODON_OAUTH_CLIENT_NAME, MASTODON_OAUTH_REDIRECT_URI, MASTODON_OAUTH_SCOPES, MASTODON_OAUTH_CLIENT_WEBSITE
from util import time_sqlite

RE_MASTODON_ATTN_LIST = re.compile(r'^(@(\w+)(@([\w\.]+))? )*(@\w+)(@[\w\.]+)?$')

api = Blueprint('api', __name__)

@api.route("/me")
@login_required
def me():
  user = current_user.db_user
  has_doms = user_has_doms(user.id)

  result = {
    "username": user.telegram_username,
    "telegram_photo_url": user.telegram_photo_url,
    "mastodon_server": user.mastodon_server.name if user.mastodon_server else None,
    "mastodon_username": user.mastodon_username,
    "mastodon_attn_list": user.mastodon_attn_list,
    "mastodon_post_public": user.mastodon_post_public,
    "has_doms": has_doms
  }


  if not has_doms:
    result["verify_mastodon_alt_text"] = user.verify_mastodon_alt_text
    result["verify_mastodon_favorite"] = user.verify_mastodon_favorite
    result["verify_delay"] = user.verify_delay

  return jsonify(result)

@api.route("/profile", methods=["POST",])
@login_required
def profile():
  user = current_user.db_user

  attn_list = request.json['attn_list'] if 'attn_list' in request.json else user.mastodon_attn_list
  post_public = request.json['post_public'] if 'post_public' in request.json else user.mastodon_post_public

  if attn_list is not None and RE_MASTODON_ATTN_LIST.fullmatch(attn_list) is None:
    abort(500)

  user_preferences_set(
    user.id,
    post_public,
    attn_list
  )

  return ('', 204)

def maybe_json(str):
  try:
    return json.loads(str)
  except:
    return None

@api.route('/timeline')
@login_required
def timeline():
  subs = [dsu.sub for dsu in domsubusers_list(current_user.db_user)]
  subs.append(current_user.db_user)

  return jsonify([
    {
      "id": t.id,
      "updated_at": time_sqlite(t.updated_at).isoformat(),
      "type": t.type,
      "text": t.text,
      "extra": maybe_json(t.extra),
      "username": t.user.telegram_username,
      "actor_username": t.actor.telegram_username if t.actor is not None else None,
      "orders_pool": {
        "id": t.orders_pool.id if user_can_orders_pools_edit(current_user.db_user, t.user) else None,
        "name": t.orders_pool.name,
      } if (
        t.orders_pool is not None and
        user_can_orders_pools_view(current_user.db_user, t.user)
       ) else None,
      "order_status": t.order_status.text if t.order_status is not None else None,
    } for t in timeline_event_recent(current_user.db_user.id)
  ])

@api.route('/mastodon_oauth')
@login_required
def mastodon_oauth():
  server_name = request.args['server']

  try:
    server = mastodon_server_get(server_name)
  except MastodonServer.DoesNotExist:
    payload = {
      'client_name': MASTODON_OAUTH_CLIENT_NAME,
      'redirect_uris': MASTODON_OAUTH_REDIRECT_URI,
      'scopes': MASTODON_OAUTH_SCOPES,
      'website': MASTODON_OAUTH_CLIENT_WEBSITE
    }
    r = requests.post(f'https://{server_name}/api/v1/apps', data=payload)

    if r.ok:
      app = r.json()
      server = mastodon_server_put(
        server_name,
        app['client_id'],
        app['client_secret']
      )
    else:
      abort(500)
  
  user_mastodon_server_set(current_user.db_user.id, server)

  payload = {
    "client_id": server.client_id,
    "client_secret": server.client_secret,
    "redirect_uri": MASTODON_OAUTH_REDIRECT_URI,
    "grant_type": "client_credentials"
  }

  return jsonify({
    "url": f"https://{server_name}/oauth/authorize?client_id={server.client_id}&scope={MASTODON_OAUTH_SCOPES}&redirect_uri={MASTODON_OAUTH_REDIRECT_URI}&response_type=code"
  })

@api.route('/subs')
@login_required
def subs():
  return jsonify(
    [
      {
        "sub_username": dsu.sub.telegram_username,
        "telegram_photo_url": dsu.sub.telegram_photo_url
      }
      for dsu
      in domsubusers_list(current_user.db_user)
    ]
  )

def authorized_sub(func):
  @wraps(func)
  def wrapper(*args, **kwargs):
    try:
      sub = user_get(request.view_args['username'])
    except:
      abort(403)
      return
    
    if (sub.id != current_user.db_user.id and
        sub.telegram_username not in [dsu.sub.telegram_username for dsu in domsubusers_list(current_user.db_user)]):
      abort(403)
      return
    
    kwargs['sub'] = sub
    return func(*args, **kwargs)
  return wrapper

@api.route('/subs/<username>', methods=["GET", "POST"])
@login_required
@authorized_sub
def sub(username, sub):
  if request.method == "POST":
    if (user_has_doms(sub) and sub.id == current_user.db_user.id and
        ('permission_orders_pools_view' in request.json or
         'permission_orders_pools_details' in request.json or
         'permission_orders_pools_edit' in request.json)):
        abort(403)
        return
    
    if 'verify_mastodon_favorite' in request.json:
      sub.verify_mastodon_favorite = bool(request.json['verify_mastodon_favorite'])
    if 'verify_mastodon_alt_text' in request.json:
      sub.verify_mastodon_alt_text = bool(request.json['verify_mastodon_alt_text'])
    if 'verify_delay' in request.json and request.json['verify_delay'] is not None:
      sub.verify_delay = int(request.json["verify_delay"])

    if 'permission_orders_pools_view' in request.json:
      sub.permission_orders_pools_view = bool(request.json['permission_orders_pools_view'])
    if 'permission_orders_pools_details' in request.json:
      sub.permission_orders_pools_details = bool(request.json['permission_orders_pools_details'])
    if 'permission_orders_pools_edit' in request.json:
      sub.permission_orders_pools_edit = bool(request.json['permission_orders_pools_edit'])

    sub.save()

    return ('', 204)
  else:
    return jsonify({
      "username": sub.telegram_username,
      "mastodon_server": sub.mastodon_server.name if sub.mastodon_server is not None else None,
      "mastodon_username": sub.mastodon_username,
      "verify_mastodon_alt_text": sub.verify_mastodon_alt_text,
      "verify_mastodon_favorite": sub.verify_mastodon_favorite,
      "verify_delay": sub.verify_delay,
      "can_edit_permissions": current_user.db_user in user_doms(sub.id),
      "permission_orders_pools_view": sub.permission_orders_pools_view,
      "permission_orders_pools_details": sub.permission_orders_pools_details,
      "permission_orders_pools_edit": sub.permission_orders_pools_edit
    })

@api.route('/orders/')
@login_required
def my_order_sets():
  user = current_user.db_user
  result = {
    'permissions': {
      'can_view': user_can_orders_pools_view(user, user),
      'can_details': user_can_orders_pools_details(user, user),
      'can_edit': user_can_orders_pools_edit(user, user)
    }
  }
  
  if result['permissions']['can_details']:
    result['pools'] = [
      {
          'id': op.id,
          'name': op.name,
          'scheduled': op.scheduled,
          'time': op.time,
          'weekends': op.weekends,
          'weekdays': op.weekdays,
          'probability': op.probability,
          'punishment_pool_name': op.punishment_pool.name if op.punishment_pool is not None else None,
          'orders': [{
            'id': order.id,
            'name': order.name,
            'weight': order.weight
          } for order in op.orders]
        }
      for op
      in orders_pool_list(current_user.db_user) 
    ]
  elif result['permissions']['can_view']:
    result['pools'] = [
      {
          'id': op.id,
          'name': op.name,
          'scheduled': op.scheduled,
          'time': op.time,
          'weekends': op.weekends,
          'weekdays': op.weekdays,
          'probability': op.probability,
          'punishment_pool_name': op.punishment_pool.name if op.punishment_pool is not None else None,
        }
      for op
      in orders_pool_list(current_user.db_user) 
    ]

  return jsonify(result)

@api.route('/orders/<username>/sets')
@login_required
@authorized_sub
def sub_order_sets(username, sub):
  result = {
    'permissions': {
      'can_view': user_can_orders_pools_view(current_user.db_user, sub),
      'can_details': user_can_orders_pools_details(current_user.db_user, sub),
      'can_edit': user_can_orders_pools_edit(current_user.db_user, sub)
    }
  }

  if result['permissions']['can_details']:
    result['pools'] = [
      {
        'id': op.id,
        'name': op.name,
        'scheduled': op.scheduled,
        'time': op.time,
        'weekends': op.weekends,
        'weekdays': op.weekdays,
        'probability': op.probability,
        'punishment_pool_name': op.punishment_pool.name if op.punishment_pool is not None else None,
        'orders': [{
          'id': order.id,
          'name': order.name,
          'weight': order.weight
        } for order in op.orders]
      }
      for op
      in orders_pool_list(sub.id)
    ]
  elif result['permissions']['can_view']:
    result['pools'] = [
      {
        'id': op.id,
        'name': op.name,
        'scheduled': op.scheduled,
        'time': op.time,
        'weekends': op.weekends,
        'weekdays': op.weekdays,
        'probability': op.probability,
        'punishment_pool_name': op.punishment_pool.name if op.punishment_pool is not None else None,
      }
      for op
      in orders_pool_list(sub.id)
    ]

  return jsonify(result)

@api.route('/orders/<username>/sets/', methods=['POST'])
@login_required
@authorized_sub
def sub_order_set_create(username, sub):
  if not user_can_orders_pools_edit(current_user.db_user, sub):
    abort(403)
    return

  # Create new
  with database.atomic() as transaction:
    try:
      new_order_pool = OrdersPool.create(
        user=sub,
        name=request.json['name'],
        scheduled=request.json['scheduled'],
        probability=request.json['probability'] if 'probability' in request.json else None,
        time=request.json['time'] if 'time' in request.json else None,
        weekdays=request.json['weekdays'] if 'weekdays' in request.json else None,
        weekends=request.json['weekends'] if 'weekends' in request.json else None,
        confirm_delay=request.json['confirm_delay'] if 'confirm_delay' in request.json else None,
      )

      for order_to_create in request.json['orders']:
        new_order = Order.create(
          pool=new_order_pool,
          name=order_to_create['name'],
          weight=order_to_create['weight'],
          repeat=order_to_create['repeat']
        )

        for add_on_to_create in order_to_create['add_ons']:
          OrderAddOn.create(
            order=new_order,
            name=add_on_to_create['name'],
            probability=add_on_to_create['probability']
          )

      timeline_event_put(
        TIMELINE_ORDERS_POOL_CREATED,
        new_order_pool.name,
        sub,
        new_order_pool,
        actor=current_user.db_user
      )
    except Exception as e:
      print(e)
      transaction.rollback()
      abort(500)

  return jsonify(new_order_pool.to_dict())

@api.route('/orders/<username>/sets/<set_id>', methods = ['GET', 'POST', 'DELETE'])
@login_required
@authorized_sub
def sub_order_set(username, set_id, sub):
  op = orders_pool(sub.id, set_id)

  if request.method == 'POST':
    if not user_can_orders_pools_edit(current_user.db_user, sub):
        abort(403)
        return

    def update_add_ons(order, add_ons):
      for updated_add_on in add_ons:
        if isinstance(updated_add_on['id'], int):
          add_on_to_update = order.add_ons.where(OrderAddOn.id == updated_add_on['id']).get()

          if '_delete' in updated_add_on:
            add_on_to_update.delete_instance(recursive=True)
          else:
            add_on_to_update.name = updated_add_on['name']
            add_on_to_update.probability = updated_add_on['probability']

            add_on_to_update.save()
        else:
          OrderAddOn.create(
            order=order,
            name=updated_add_on['name'],
            probability=updated_add_on['probability']
          )

    with database.atomic() as transaction:
      try:
        op.name = request.json['name']

        if ('punishment_pool_id' in request.json and
            request.json['punishment_pool_id'] in [
              op.id for op in orders_pool_list(sub.id)
            ]):
          op.punishment_pool_id = request.json['punishment_pool_id']
        else:
          op.punishment_pool_id = None

        op.scheduled = request.json['scheduled']
        if op.scheduled:
          op.probability = request.json['probability']
          op.weekdays = request.json['weekdays']
          op.weekends = request.json['weekends']
          op.time = request.json['time']
          op.confirm_delay = request.json['confirm_delay']
        else:
          op.probability = None
          op.weekdays = None
          op.weekends = None
          op.time = None
          op.confirm_delay = None

        op.save()

        for updated_order in request.json['orders']:
          if isinstance(updated_order['id'], int):
            order_to_update = op.orders.where(Order.id == updated_order['id']).get()

            if '_delete' in updated_order:
              order_to_update.delete_instance(recursive=True)
            else:
              order_to_update.name = updated_order['name']
              order_to_update.weight = updated_order['weight']
              order_to_update.repeat = updated_order['repeat']

              order_to_update.save()

              update_add_ons(order_to_update, updated_order['add_ons'])
          else:
            order_to_update = Order.create(
              pool=op,
              name=updated_order['name'],
              weight=updated_order['weight'],
              repeat=updated_order['repeat']
            )

            update_add_ons(order_to_update, updated_order['add_ons'])

        timeline_event_put(
          TIMELINE_ORDERS_POOL_UPDATED,
          op.name,
          sub,
          op,
          actor=current_user.db_user
        )
      except Exception as e:
        print(e)
        transaction.rollback()
        abort(500)
  elif request.method == 'DELETE':
    try:
      if not user_can_orders_pools_edit(current_user.db_user, sub):
        abort(403)
        return

      op.delete_instance(recursive=True)

      timeline_event_put(
        TIMELINE_ORDERS_POOL_DELETED,
        op.name,
        sub,
        actor=current_user.db_user
      )

      return ('', 204)
    except Exception as e:
      print(e)
      abort(500)

  return jsonify(op.to_dict())