mirror of
https://github.com/simple-login/app.git
synced 2024-11-17 01:18:29 +01:00
3e2c120a73
- add twilio lib - create phone listing, reservation page - add twilio callback to receive messages
128 lines
3.9 KiB
Python
128 lines
3.9 KiB
Python
from typing import Dict
|
|
|
|
import arrow
|
|
from flask import render_template, request, flash, redirect, url_for
|
|
from flask_login import login_required, current_user
|
|
from sqlalchemy import func
|
|
|
|
from app.db import Session
|
|
from app.models import PhoneCountry, PhoneNumber, PhoneReservation
|
|
from app.phone.base import phone_bp
|
|
|
|
|
|
@phone_bp.route("/", methods=["GET", "POST"])
|
|
@login_required
|
|
def index():
|
|
if not current_user.can_use_phone:
|
|
flash("You can't use this page", "error")
|
|
return redirect(url_for("dashboard.index"))
|
|
|
|
countries = available_countries()
|
|
|
|
now = arrow.now()
|
|
reservations = PhoneReservation.filter(
|
|
PhoneReservation.user_id == current_user.id,
|
|
PhoneReservation.start < now,
|
|
PhoneReservation.end > now,
|
|
).all()
|
|
|
|
past_reservations = PhoneReservation.filter(
|
|
PhoneReservation.user_id == current_user.id,
|
|
PhoneReservation.end <= now,
|
|
).all()
|
|
|
|
if request.method == "POST":
|
|
nb_minute = int(request.form.get("minute"))
|
|
|
|
if current_user.phone_quota < nb_minute:
|
|
flash(
|
|
f"You don't have enough phone quota. Current quota is {current_user.phone_quota}",
|
|
"error",
|
|
)
|
|
return redirect(request.url)
|
|
|
|
country_id = request.form.get("country")
|
|
country = PhoneCountry.get(country_id)
|
|
|
|
# get the first phone number available
|
|
now = arrow.now()
|
|
busy_phone_number_subquery = (
|
|
Session.query(PhoneReservation.number_id)
|
|
.filter(PhoneReservation.start < now, PhoneReservation.end > now)
|
|
.subquery()
|
|
)
|
|
|
|
phone_number = (
|
|
Session.query(PhoneNumber)
|
|
.filter(
|
|
PhoneNumber.country_id == country.id,
|
|
PhoneNumber.id.notin_(busy_phone_number_subquery),
|
|
)
|
|
.first()
|
|
)
|
|
|
|
if phone_number:
|
|
phone_reservation = PhoneReservation.create(
|
|
number_id=phone_number.id,
|
|
start=arrow.now(),
|
|
end=arrow.now().shift(minutes=nb_minute),
|
|
user_id=current_user.id,
|
|
)
|
|
|
|
current_user.phone_quota -= nb_minute
|
|
Session.commit()
|
|
|
|
return redirect(
|
|
url_for("phone.reservation_route", reservation_id=phone_reservation.id)
|
|
)
|
|
else:
|
|
flash(
|
|
f"No phone number available for {country.name} during {nb_minute} minutes"
|
|
)
|
|
|
|
return render_template(
|
|
"phone/index.html",
|
|
countries=countries,
|
|
reservations=reservations,
|
|
past_reservations=past_reservations,
|
|
)
|
|
|
|
|
|
def available_countries() -> [PhoneCountry]:
|
|
now = arrow.now()
|
|
|
|
phone_count_by_countries: Dict[PhoneCountry, int] = dict()
|
|
for country, count in (
|
|
Session.query(PhoneCountry, func.count(PhoneNumber.id))
|
|
.join(PhoneNumber, PhoneNumber.country_id == PhoneCountry.id)
|
|
.filter(PhoneNumber.active.is_(True))
|
|
.group_by(PhoneCountry)
|
|
.all()
|
|
):
|
|
phone_count_by_countries[country] = count
|
|
|
|
busy_phone_count_by_countries: Dict[PhoneCountry, int] = dict()
|
|
for country, count in (
|
|
Session.query(PhoneCountry, func.count(PhoneNumber.id))
|
|
.join(PhoneNumber, PhoneNumber.country_id == PhoneCountry.id)
|
|
.join(PhoneReservation, PhoneReservation.number_id == PhoneNumber.id)
|
|
.filter(PhoneReservation.start < now, PhoneReservation.end > now)
|
|
.group_by(PhoneCountry)
|
|
.all()
|
|
):
|
|
busy_phone_count_by_countries[country] = count
|
|
|
|
ret = []
|
|
for country in phone_count_by_countries:
|
|
if (
|
|
country not in busy_phone_count_by_countries
|
|
or phone_count_by_countries[country]
|
|
> busy_phone_count_by_countries[country]
|
|
):
|
|
ret.append(country)
|
|
|
|
return ret
|
|
|
|
|
|
def available_numbers() -> [PhoneNumber]:
|
|
Session.query(PhoneReservation).filter(PhoneReservation.start)
|