refactor some operations into utils.py

This commit is contained in:
stupidcomputer 2024-12-20 22:05:41 -06:00
parent 2fceb6679f
commit f8f642cead
4 changed files with 45 additions and 48 deletions

View File

View File

@ -34,7 +34,7 @@
<div class="btn-group"> <div class="btn-group">
<a href="{% url 'add_trusted_channel' org.id %}" class="btn btn-primary">Add new trusted channel</a> <a href="{% url 'add_trusted_channel' org.id %}" class="btn btn-primary">Add new trusted channel</a>
<a href="{{ scheme }}://{{ hostname }}{% url 'add_channel_to_organization' org.id %}" class="btn btn-primary">Add new follower channel</a> <a href="{{ scheme }}://{{ hostname }}{% url 'add_channel_to_organization' org.id %}" id="sharelink" class="btn btn-primary">Add new follower channel</a>
<button id="sharebutton" type="button" class="btn btn-primary">Copy share link</button> <button id="sharebutton" type="button" class="btn btn-primary">Copy share link</button>
<a href="{% url 'qr_code' org.id %}" class="btn btn-primary">View QR Code</a> <a href="{% url 'qr_code' org.id %}" class="btn btn-primary">View QR Code</a>
</div> </div>

16
tfbbridge/bridge/utils.py Normal file
View File

@ -0,0 +1,16 @@
import pickle
from .models import Organization
def to_pickle(data):
return pickle.dumps(data).hex()
def from_pickle(data):
return pickle.loads(bytes.fromhex(data))
# remove duplicate users
# https://stackoverflow.com/questions/9427163/remove-duplicate-dict-in-list-in-python
def remove_dup_dict_in_list(l):
return [i for n, i in enumerate(l) if i not in l[n + 1:]]
def get_org_by_id(org_id):
return Organization.objects.filter(id__exact=org_id)[0]

View File

@ -7,6 +7,7 @@ import pickle
import qrcode import qrcode
import json import json
import random import random
from .utils import to_pickle, from_pickle, remove_dup_dict_in_list, get_org_by_id
bot_name = "tfbbot" bot_name = "tfbbot"
auth_link = "https://oauth.groupme.com/oauth/authorize?client_id=KbStJZciPGivvLuTN9XX6aXf6NHVcxDSiVR0wRC0cqVDuDfG" auth_link = "https://oauth.groupme.com/oauth/authorize?client_id=KbStJZciPGivvLuTN9XX6aXf6NHVcxDSiVR0wRC0cqVDuDfG"
@ -15,10 +16,7 @@ def login(request):
return redirect(auth_link) return redirect(auth_link)
def add_organization_flow_select(request): def add_organization_flow_select(request):
user_group_data = pickle.loads( user_group_data = from_pickle(request.session["groupme_groups_pickle"])
bytes.fromhex(
request.session["groupme_groups_pickle"]
))
user_group_data = user_group_data["response"] user_group_data = user_group_data["response"]
if request.method == "GET": if request.method == "GET":
rendered_groups = [] rendered_groups = []
@ -41,9 +39,7 @@ def add_organization_flow_select(request):
break break
# remove duplicate users rendered_users = remove_dup_dict_in_list(rendered_users)
# https://stackoverflow.com/questions/9427163/remove-duplicate-dict-in-list-in-python
rendered_users = [i for n, i in enumerate(rendered_users) if i not in rendered_users[n + 1:]]
# sort users by name # sort users by name
rendered_users = sorted(rendered_users, key=lambda x: x["name"].lower()) rendered_users = sorted(rendered_users, key=lambda x: x["name"].lower())
@ -61,10 +57,7 @@ def add_organization_flow_select(request):
name = request.POST["name"] name = request.POST["name"]
users = [i[2:] for i in request.POST.keys() if i.startswith("u_")] users = [i[2:] for i in request.POST.keys() if i.startswith("u_")]
groups = [i[2:] for i in request.POST.keys() if i.startswith("g_")] groups = [i[2:] for i in request.POST.keys() if i.startswith("g_")]
group_name_mapping = pickle.loads( group_name_mapping = from_pickle(request.session["group_name_mapping"])
bytes.fromhex(
request.session["group_name_mapping"]
))
org = Organization() org = Organization()
org.owner = request.session["groupme_id"] org.owner = request.session["groupme_id"]
@ -105,7 +98,7 @@ def add_organization_flow_select(request):
group_data_collected.append(group_data) group_data_collected.append(group_data)
user_data_collected = [] user_data_collected = []
user_name_mappings = pickle.loads(bytes.fromhex(request.session["user_name_mapping"])) user_name_mappings = from_pickle(request.session["user_name_mapping"])
for user in users: for user in users:
try: try:
username = user_name_mappings[user] username = user_name_mappings[user]
@ -114,8 +107,8 @@ def add_organization_flow_select(request):
user_data = { "uid": user, "role": "trusted", "name": username, "number_messages": 0 } user_data = { "uid": user, "role": "trusted", "name": username, "number_messages": 0 }
user_data_collected.append(user_data) user_data_collected.append(user_data)
org.trusted_users = pickle.dumps(user_data_collected).hex() org.trusted_users = to_pickle(user_data_collected)
org.channel_info = pickle.dumps(group_data_collected).hex() org.channel_info = to_pickle(group_data_collected)
org.save() org.save()
return redirect(reverse("view_organization", kwargs={"org_id": org.id})) return redirect(reverse("view_organization", kwargs={"org_id": org.id}))
@ -135,20 +128,20 @@ def handle_oauth(request):
request.session["groupme_access_token"] request.session["groupme_access_token"]
)) ))
request.session["groupme_groups_pickle"] = pickle.dumps(r.json()).hex() request.session["groupme_groups_pickle"] = to_pickle(r.json())
request.session["logged_in"] = "yes" request.session["logged_in"] = "yes"
group_name_mapping = {} group_name_mapping = {}
for group in r.json()["response"]: for group in r.json()["response"]:
group_name_mapping[group["id"]] = group["name"] group_name_mapping[group["id"]] = group["name"]
request.session["group_name_mapping"] = pickle.dumps(group_name_mapping).hex() request.session["group_name_mapping"] = to_pickle(group_name_mapping)
user_name_mapping = {} user_name_mapping = {}
for group in r.json()["response"]: for group in r.json()["response"]:
for user in group["members"]: for user in group["members"]:
user_name_mapping[user["id"]] = user["name"] user_name_mapping[user["id"]] = user["name"]
user_name_mapping[data["response"]["id"]] = data["response"]["name"] user_name_mapping[data["response"]["id"]] = data["response"]["name"]
request.session["user_name_mapping"] = pickle.dumps(user_name_mapping).hex() request.session["user_name_mapping"] = to_pickle(user_name_mapping)
try: try:
if request.session["needs_redirect"]: if request.session["needs_redirect"]:
@ -186,13 +179,13 @@ def logout(request):
return redirect(index) return redirect(index)
def view_organization(request, org_id): def view_organization(request, org_id):
org = Organization.objects.filter(id__exact=org_id)[0] org = get_org_by_id(org_id)
scheme = 'https' if request.is_secure() else 'http' scheme = 'https' if request.is_secure() else 'http'
hostname = request.get_host() hostname = request.get_host()
if request.session["groupme_id"] == org.owner: if request.session["groupme_id"] == org.owner:
org_channel_info = pickle.loads(bytes.fromhex(org.channel_info)) org_channel_info = from_pickle(org.channel_info)
org_trusted_users = pickle.loads(bytes.fromhex(org.trusted_users)) org_trusted_users = from_pickle(org.trusted_users)
return render(request, "view_organization.html", { return render(request, "view_organization.html", {
"request": request, "request": request,
"org": org, "org": org,
@ -205,7 +198,7 @@ def view_organization(request, org_id):
pass # blow up pass # blow up
def qr_code(request, org_id): def qr_code(request, org_id):
org = Organization.objects.filter(id__exact=org_id)[0] org = get_org_by_id(org_id)
scheme = 'https' if request.is_secure() else 'http' scheme = 'https' if request.is_secure() else 'http'
hostname = request.get_host() hostname = request.get_host()
@ -237,13 +230,10 @@ def add_channel_to_organization(request, org_id, trusted=False):
request.session["needs_redirect"] = "add_channel_to_organization" request.session["needs_redirect"] = "add_channel_to_organization"
return redirect(reverse("login")) return redirect(reverse("login"))
org = Organization.objects.filter(id__exact=org_id)[0] org = get_org_by_id(org_id)
if request.method == "GET": if request.method == "GET":
user_group_data = pickle.loads( user_group_data = from_pickle(request.session["groupme_groups_pickle"])["response"]
bytes.fromhex(
request.session["groupme_groups_pickle"]
))["response"]
our_id = request.session["groupme_id"] our_id = request.session["groupme_id"]
groups_we_admin = [] groups_we_admin = []
@ -261,11 +251,8 @@ def add_channel_to_organization(request, org_id, trusted=False):
}) })
elif request.method == "POST": elif request.method == "POST":
group_name_mapping = pickle.loads( group_name_mapping = from_pickle(request.session["group_name_mapping"])
bytes.fromhex( org = get_org_by_id(org_id)
request.session["group_name_mapping"]
))
org = Organization.objects.filter(id__exact=org_id)[0]
groups = [i[2:] for i in request.POST.keys() if i.startswith("g_")] groups = [i[2:] for i in request.POST.keys() if i.startswith("g_")]
org_group_info = pickle.loads(bytes.fromhex(org.channel_info)) org_group_info = pickle.loads(bytes.fromhex(org.channel_info))
@ -311,7 +298,6 @@ def add_channel_to_organization(request, org_id, trusted=False):
headers={"Content-Type": "application/json"}, headers={"Content-Type": "application/json"},
) )
print(r)
data = r.json() data = r.json()
if not trusted: if not trusted:
@ -361,8 +347,8 @@ def delete_channel(request, org_id, chan):
return redirect(reverse('view_organization', args=[org.id])) return redirect(reverse('view_organization', args=[org.id]))
def demote_channel(request, org_id, chan): def demote_channel(request, org_id, chan):
org = Organization.objects.filter(id__exact=org_id)[0] org = get_org_by_id(org_id)
org_group_info = pickle.loads(bytes.fromhex(org.channel_info)) org_group_info = from_pickle(org.channel_info)
for index, channel in enumerate(org_group_info): for index, channel in enumerate(org_group_info):
print(index, channel) print(index, channel)
@ -370,18 +356,15 @@ def demote_channel(request, org_id, chan):
org_group_info[index]["role"] = "following" org_group_info[index]["role"] = "following"
break break
org.channel_info = pickle.dumps(org_group_info).hex() org.channel_info = to_pickle(org_group_info)
org.save() org.save()
return redirect(reverse('view_organization', args=[org.id])) return redirect(reverse('view_organization', args=[org.id]))
def add_trusted_user(request, org_id): def add_trusted_user(request, org_id):
org = Organization.objects.filter(id__exact=org_id)[0] org = get_org_by_id(org_id)
org_trusted_users = pickle.loads(bytes.fromhex(org.trusted_users)) org_trusted_users = from_pickle(org.trusted_users)
user_group_data = pickle.loads( user_group_data = from_pickle(request.session["groupme_groups_pickle"])
bytes.fromhex(
request.session["groupme_groups_pickle"]
))
user_group_data = user_group_data["response"] user_group_data = user_group_data["response"]
rendered_users = [] rendered_users = []
@ -398,9 +381,7 @@ def add_trusted_user(request, org_id):
break break
# remove duplicate users rendered_users = remove_dup_dict_in_list(rendered_users)
# https://stackoverflow.com/questions/9427163/remove-duplicate-dict-in-list-in-python
rendered_users = [i for n, i in enumerate(rendered_users) if i not in rendered_users[n + 1:]]
if request.method == "GET": if request.method == "GET":
return render(request, "add_user.html", { return render(request, "add_user.html", {
@ -427,8 +408,8 @@ def add_trusted_user(request, org_id):
return redirect(reverse('view_organization', args=[org.id])) return redirect(reverse('view_organization', args=[org.id]))
def remove_user(request, org_id, user): def remove_user(request, org_id, user):
org = Organization.objects.filter(id__exact=org_id)[0] org = get_org_by_id(org_id)
org_trusted_users = pickle.loads(bytes.fromhex(org.trusted_users)) org_trusted_users = from_pickle(org.trusted_users)
for index, tuser in enumerate(org_trusted_users): for index, tuser in enumerate(org_trusted_users):
print(index, tuser) print(index, tuser)
@ -437,7 +418,7 @@ def remove_user(request, org_id, user):
org_trusted_users.pop(index) org_trusted_users.pop(index)
break break
org.trusted_users = pickle.dumps(org_trusted_users).hex() org.trusted_users = to_pickle(org_trusted_users)
org.save() org.save()
return redirect(reverse('view_organization', args=[org.id])) return redirect(reverse('view_organization', args=[org.id]))