mirror of
https://github.com/zulip/zulip.git
synced 2026-06-24 21:08:25 +08:00
dcbb2a78ca
JsonableError has two major benefits over json_error: * It can be raised from anywhere in the codebase, rather than being a return value, which is much more convenient for refactoring, as one doesn't potentially need to change error handling style when extracting a bit of view code to a function. * It is guaranteed to contain the `code` property, which is helpful for API consistency. Various stragglers are not updated because JsonableError requires subclassing in order to specify custom data or HTTP status codes.
290 lines
13 KiB
Python
290 lines
13 KiB
Python
import urllib
|
|
from datetime import timedelta
|
|
from decimal import Decimal
|
|
from typing import Any, Dict, List, Optional
|
|
from urllib.parse import urlencode
|
|
|
|
from django.conf import settings
|
|
from django.core.exceptions import ValidationError
|
|
from django.core.validators import URLValidator
|
|
from django.http import HttpRequest, HttpResponse, HttpResponseRedirect
|
|
from django.shortcuts import render
|
|
from django.urls import reverse
|
|
from django.utils.timesince import timesince
|
|
from django.utils.timezone import now as timezone_now
|
|
from django.utils.translation import gettext as _
|
|
|
|
from confirmation.models import Confirmation, _properties, confirmation_url
|
|
from confirmation.settings import STATUS_ACTIVE
|
|
from zerver.decorator import require_server_admin
|
|
from zerver.forms import check_subdomain_available
|
|
from zerver.lib.actions import (
|
|
do_change_plan_type,
|
|
do_change_realm_subdomain,
|
|
do_deactivate_realm,
|
|
do_scrub_realm,
|
|
do_send_realm_reactivation_email,
|
|
)
|
|
from zerver.lib.exceptions import JsonableError
|
|
from zerver.lib.realm_icon import realm_icon_url
|
|
from zerver.lib.subdomains import get_subdomain_from_hostname
|
|
from zerver.models import MultiuseInvite, PreregistrationUser, Realm, UserProfile, get_realm
|
|
from zerver.views.invite import get_invitee_emails_set
|
|
|
|
if settings.BILLING_ENABLED:
|
|
from corporate.lib.stripe import (
|
|
approve_sponsorship,
|
|
attach_discount_to_realm,
|
|
downgrade_at_the_end_of_billing_cycle,
|
|
downgrade_now_without_creating_additional_invoices,
|
|
get_current_plan_by_realm,
|
|
get_customer_by_realm,
|
|
get_discount_for_realm,
|
|
get_latest_seat_count,
|
|
make_end_of_cycle_updates_if_needed,
|
|
update_billing_method_of_current_plan,
|
|
update_sponsorship_status,
|
|
void_all_open_invoices,
|
|
)
|
|
|
|
|
|
def get_plan_name(plan_type: int) -> str:
|
|
return ["", "self hosted", "limited", "standard", "open source"][plan_type]
|
|
|
|
|
|
def get_confirmations(
|
|
types: List[int], object_ids: List[int], hostname: Optional[str] = None
|
|
) -> List[Dict[str, Any]]:
|
|
lowest_datetime = timezone_now() - timedelta(days=30)
|
|
confirmations = Confirmation.objects.filter(
|
|
type__in=types, object_id__in=object_ids, date_sent__gte=lowest_datetime
|
|
)
|
|
confirmation_dicts = []
|
|
for confirmation in confirmations:
|
|
realm = confirmation.realm
|
|
content_object = confirmation.content_object
|
|
|
|
type = confirmation.type
|
|
days_to_activate = _properties[type].validity_in_days
|
|
expiry_date = confirmation.date_sent + timedelta(days=days_to_activate)
|
|
|
|
if hasattr(content_object, "status"):
|
|
if content_object.status == STATUS_ACTIVE:
|
|
link_status = "Link has been clicked"
|
|
else:
|
|
link_status = "Link has never been clicked"
|
|
else:
|
|
link_status = ""
|
|
|
|
now = timezone_now()
|
|
if now < expiry_date:
|
|
expires_in = timesince(now, expiry_date)
|
|
else:
|
|
expires_in = "Expired"
|
|
|
|
url = confirmation_url(confirmation.confirmation_key, realm, type)
|
|
confirmation_dicts.append(
|
|
{
|
|
"object": confirmation.content_object,
|
|
"url": url,
|
|
"type": type,
|
|
"link_status": link_status,
|
|
"expires_in": expires_in,
|
|
}
|
|
)
|
|
return confirmation_dicts
|
|
|
|
|
|
@require_server_admin
|
|
def support(request: HttpRequest) -> HttpResponse:
|
|
context: Dict[str, Any] = {}
|
|
|
|
if "success_message" in request.session:
|
|
context["success_message"] = request.session["success_message"]
|
|
del request.session["success_message"]
|
|
|
|
if settings.BILLING_ENABLED and request.method == "POST":
|
|
# We check that request.POST only has two keys in it: The
|
|
# realm_id and a field to change.
|
|
keys = set(request.POST.keys())
|
|
if "csrfmiddlewaretoken" in keys:
|
|
keys.remove("csrfmiddlewaretoken")
|
|
if len(keys) != 2:
|
|
raise JsonableError(_("Invalid parameters"))
|
|
|
|
realm_id = request.POST.get("realm_id")
|
|
realm = Realm.objects.get(id=realm_id)
|
|
|
|
if request.POST.get("plan_type", None) is not None:
|
|
new_plan_type = int(request.POST.get("plan_type"))
|
|
current_plan_type = realm.plan_type
|
|
do_change_plan_type(realm, new_plan_type, acting_user=request.user)
|
|
msg = f"Plan type of {realm.string_id} changed from {get_plan_name(current_plan_type)} to {get_plan_name(new_plan_type)} "
|
|
context["success_message"] = msg
|
|
elif request.POST.get("discount", None) is not None:
|
|
new_discount = Decimal(request.POST.get("discount"))
|
|
current_discount = get_discount_for_realm(realm) or 0
|
|
attach_discount_to_realm(realm, new_discount, acting_user=request.user)
|
|
context[
|
|
"success_message"
|
|
] = f"Discount of {realm.string_id} changed to {new_discount}% from {current_discount}%."
|
|
elif request.POST.get("new_subdomain", None) is not None:
|
|
new_subdomain = request.POST.get("new_subdomain")
|
|
old_subdomain = realm.string_id
|
|
try:
|
|
check_subdomain_available(new_subdomain)
|
|
except ValidationError as error:
|
|
context["error_message"] = error.message
|
|
else:
|
|
do_change_realm_subdomain(realm, new_subdomain, acting_user=request.user)
|
|
request.session[
|
|
"success_message"
|
|
] = f"Subdomain changed from {old_subdomain} to {new_subdomain}"
|
|
return HttpResponseRedirect(
|
|
reverse("support") + "?" + urlencode({"q": new_subdomain})
|
|
)
|
|
elif request.POST.get("status", None) is not None:
|
|
status = request.POST.get("status")
|
|
if status == "active":
|
|
do_send_realm_reactivation_email(realm, acting_user=request.user)
|
|
context[
|
|
"success_message"
|
|
] = f"Realm reactivation email sent to admins of {realm.string_id}."
|
|
elif status == "deactivated":
|
|
do_deactivate_realm(realm, acting_user=request.user)
|
|
context["success_message"] = f"{realm.string_id} deactivated."
|
|
elif request.POST.get("billing_method", None) is not None:
|
|
billing_method = request.POST.get("billing_method")
|
|
if billing_method == "send_invoice":
|
|
update_billing_method_of_current_plan(
|
|
realm, charge_automatically=False, acting_user=request.user
|
|
)
|
|
context[
|
|
"success_message"
|
|
] = f"Billing method of {realm.string_id} updated to pay by invoice."
|
|
elif billing_method == "charge_automatically":
|
|
update_billing_method_of_current_plan(
|
|
realm, charge_automatically=True, acting_user=request.user
|
|
)
|
|
context[
|
|
"success_message"
|
|
] = f"Billing method of {realm.string_id} updated to charge automatically."
|
|
elif request.POST.get("sponsorship_pending", None) is not None:
|
|
sponsorship_pending = request.POST.get("sponsorship_pending")
|
|
if sponsorship_pending == "true":
|
|
update_sponsorship_status(realm, True, acting_user=request.user)
|
|
context["success_message"] = f"{realm.string_id} marked as pending sponsorship."
|
|
elif sponsorship_pending == "false":
|
|
update_sponsorship_status(realm, False, acting_user=request.user)
|
|
context["success_message"] = f"{realm.string_id} is no longer pending sponsorship."
|
|
elif request.POST.get("approve_sponsorship") is not None:
|
|
if request.POST.get("approve_sponsorship") == "approve_sponsorship":
|
|
approve_sponsorship(realm, acting_user=request.user)
|
|
context["success_message"] = f"Sponsorship approved for {realm.string_id}"
|
|
elif request.POST.get("downgrade_method", None) is not None:
|
|
downgrade_method = request.POST.get("downgrade_method")
|
|
if downgrade_method == "downgrade_at_billing_cycle_end":
|
|
downgrade_at_the_end_of_billing_cycle(realm)
|
|
context[
|
|
"success_message"
|
|
] = f"{realm.string_id} marked for downgrade at the end of billing cycle"
|
|
elif downgrade_method == "downgrade_now_without_additional_licenses":
|
|
downgrade_now_without_creating_additional_invoices(realm)
|
|
context[
|
|
"success_message"
|
|
] = f"{realm.string_id} downgraded without creating additional invoices"
|
|
elif downgrade_method == "downgrade_now_void_open_invoices":
|
|
downgrade_now_without_creating_additional_invoices(realm)
|
|
voided_invoices_count = void_all_open_invoices(realm)
|
|
context[
|
|
"success_message"
|
|
] = f"{realm.string_id} downgraded and voided {voided_invoices_count} open invoices"
|
|
elif request.POST.get("scrub_realm", None) is not None:
|
|
if request.POST.get("scrub_realm") == "scrub_realm":
|
|
do_scrub_realm(realm, acting_user=request.user)
|
|
context["success_message"] = f"{realm.string_id} scrubbed."
|
|
|
|
query = request.GET.get("q", None)
|
|
if query:
|
|
key_words = get_invitee_emails_set(query)
|
|
|
|
users = set(UserProfile.objects.filter(delivery_email__in=key_words))
|
|
realms = set(Realm.objects.filter(string_id__in=key_words))
|
|
|
|
for key_word in key_words:
|
|
try:
|
|
URLValidator()(key_word)
|
|
parse_result = urllib.parse.urlparse(key_word)
|
|
hostname = parse_result.hostname
|
|
assert hostname is not None
|
|
if parse_result.port:
|
|
hostname = f"{hostname}:{parse_result.port}"
|
|
subdomain = get_subdomain_from_hostname(hostname)
|
|
try:
|
|
realms.add(get_realm(subdomain))
|
|
except Realm.DoesNotExist:
|
|
pass
|
|
except ValidationError:
|
|
users.update(UserProfile.objects.filter(full_name__iexact=key_word))
|
|
|
|
for realm in realms:
|
|
realm.customer = get_customer_by_realm(realm)
|
|
|
|
current_plan = get_current_plan_by_realm(realm)
|
|
if current_plan is not None:
|
|
new_plan, last_ledger_entry = make_end_of_cycle_updates_if_needed(
|
|
current_plan, timezone_now()
|
|
)
|
|
if last_ledger_entry is not None:
|
|
if new_plan is not None:
|
|
realm.current_plan = new_plan
|
|
else:
|
|
realm.current_plan = current_plan
|
|
realm.current_plan.licenses = last_ledger_entry.licenses
|
|
realm.current_plan.licenses_used = get_latest_seat_count(realm)
|
|
|
|
# full_names can have , in them
|
|
users.update(UserProfile.objects.filter(full_name__iexact=query))
|
|
|
|
context["users"] = users
|
|
context["realms"] = realms
|
|
|
|
confirmations: List[Dict[str, Any]] = []
|
|
|
|
preregistration_users = PreregistrationUser.objects.filter(email__in=key_words)
|
|
confirmations += get_confirmations(
|
|
[Confirmation.USER_REGISTRATION, Confirmation.INVITATION, Confirmation.REALM_CREATION],
|
|
preregistration_users,
|
|
hostname=request.get_host(),
|
|
)
|
|
|
|
multiuse_invites = MultiuseInvite.objects.filter(realm__in=realms)
|
|
confirmations += get_confirmations([Confirmation.MULTIUSE_INVITE], multiuse_invites)
|
|
|
|
confirmations += get_confirmations(
|
|
[Confirmation.REALM_REACTIVATION], [realm.id for realm in realms]
|
|
)
|
|
|
|
context["confirmations"] = confirmations
|
|
|
|
def get_realm_owner_emails_as_string(realm: Realm) -> str:
|
|
return ", ".join(
|
|
realm.get_human_owner_users()
|
|
.order_by("delivery_email")
|
|
.values_list("delivery_email", flat=True)
|
|
)
|
|
|
|
def get_realm_admin_emails_as_string(realm: Realm) -> str:
|
|
return ", ".join(
|
|
realm.get_human_admin_users(include_realm_owners=False)
|
|
.order_by("delivery_email")
|
|
.values_list("delivery_email", flat=True)
|
|
)
|
|
|
|
context["get_realm_owner_emails_as_string"] = get_realm_owner_emails_as_string
|
|
context["get_realm_admin_emails_as_string"] = get_realm_admin_emails_as_string
|
|
context["get_discount_for_realm"] = get_discount_for_realm
|
|
context["realm_icon_url"] = realm_icon_url
|
|
context["Confirmation"] = Confirmation
|
|
return render(request, "analytics/support.html", context=context)
|