mirror of
https://github.com/Mistake35/Cedar-Django.git
synced 2026-07-18 00:21:14 +10:00
266 lines
8.5 KiB
Python
266 lines
8.5 KiB
Python
import urllib.request, urllib.error
|
|
# requests is only used for get_mii which is not being used currently
|
|
#import requests
|
|
from lxml import etree
|
|
from random import choice
|
|
import json
|
|
import time
|
|
import os.path
|
|
import random
|
|
from PIL import Image, ExifTags, ImageFile
|
|
from datetime import datetime
|
|
from binascii import crc32
|
|
from math import floor
|
|
from hashlib import md5, sha1
|
|
import io
|
|
from uuid import uuid4
|
|
import imghdr
|
|
import base64
|
|
from closedverse import settings
|
|
import re
|
|
from os import remove, rename
|
|
|
|
def HumanTime(date, full=False):
|
|
now = time.time()
|
|
time_difference = now - date
|
|
time_units = {86400: 'day', 3600: 'hour', 60: 'minute'}
|
|
if time_difference >= 259200 or full:
|
|
return datetime.fromtimestamp(date).strftime('%m/%d/%Y %I:%M %p')
|
|
elif time_difference <= 59:
|
|
return 'Less than a minute ago'
|
|
else:
|
|
for unit, unit_name in time_units.items():
|
|
if time_difference < unit:
|
|
continue
|
|
else:
|
|
number_of_units = floor(time_difference / unit)
|
|
if number_of_units > 1:
|
|
unit_name += 's'
|
|
return f'{number_of_units} {unit_name} ago'
|
|
|
|
# the current source as of now uses AJAX to get mii data
|
|
def get_mii(id):
|
|
# Using AccountWS
|
|
dmca = {
|
|
'X-Nintendo-Client-ID': 'a2efa818a34fa16b8afbc8a74eba3eda',
|
|
'X-Nintendo-Client-Secret': 'c91cdb5658bd4954ade78533a339cf9a',
|
|
}
|
|
|
|
# Perform the first request to get pid
|
|
url_pid = 'https://accountws.nintendo.net/v1/api/admin/mapped_ids?input_type=user_id&output_type=pid&input=' + id
|
|
request_pid = urllib.request.Request(url_pid, headers=dmca)
|
|
with urllib.request.urlopen(request_pid) as nnid_response:
|
|
nnid_content = nnid_response.read()
|
|
nnid_dec = etree.fromstring(nnid_content)
|
|
pid = nnid_dec[0][1].text
|
|
if not pid:
|
|
return False
|
|
|
|
# Perform the second request to get mii information
|
|
url_mii = 'https://accountws.nintendo.net/v1/api/miis?pids=' + pid
|
|
request_mii = urllib.request.Request(url_mii, headers=dmca)
|
|
with urllib.request.urlopen(request_mii) as mii_response:
|
|
mii_content = mii_response.read()
|
|
try:
|
|
mii_dec = etree.fromstring(mii_content)
|
|
except:
|
|
return False
|
|
|
|
try:
|
|
miihash = mii_dec[0][2][0][0].text.split('.net/')[1].split('_')[0]
|
|
except IndexError:
|
|
miihash = None
|
|
screenname = mii_dec[0][3].text
|
|
nnid = mii_dec[0][6].text
|
|
|
|
return [miihash, screenname, nnid]
|
|
|
|
def recaptcha_verify(request, key):
|
|
if not request.POST.get('g-recaptcha-response'):
|
|
return False
|
|
re_request = urllib.request.urlopen('https://www.google.com/recaptcha/api/siteverify?secret={0}&response={1}'.format(key, request.POST['g-recaptcha-response']))
|
|
jsond = json.loads(re_request.read().decode())
|
|
if not jsond['success']:
|
|
return False
|
|
return True
|
|
|
|
def video_upload(video):
|
|
hash = sha1()
|
|
for chunk in video.chunks():
|
|
hash.update(chunk)
|
|
imhash = hash.hexdigest()
|
|
# only either webm or mp4
|
|
extension = video.name[-4:]
|
|
if extension == '.mp4':
|
|
fname = imhash + '.mp4'
|
|
elif extension == 'webm':
|
|
fname = imhash + '.webm'
|
|
else:
|
|
return 1
|
|
# simply only write image if the hashed path doesn't already exist
|
|
if not os.path.exists(settings.MEDIA_ROOT + fname):
|
|
with open(settings.MEDIA_ROOT + fname, "wb+") as destination:
|
|
for chunk in video.chunks():
|
|
destination.write(chunk)
|
|
return settings.MEDIA_URL + fname
|
|
|
|
ImageFile.LOAD_TRUNCATED_IMAGES = True
|
|
def image_upload(img, stream=False, drawing=False, avatar=False):
|
|
if stream:
|
|
decodedimg = img.read()
|
|
else:
|
|
try:
|
|
decodedimg = base64.b64decode(img)
|
|
except ValueError:
|
|
return 1
|
|
if stream:
|
|
if not 'image' in img.content_type:
|
|
return 1
|
|
# upload svg?
|
|
#if 'svg' in mime:
|
|
#
|
|
try:
|
|
im = Image.open(io.BytesIO(decodedimg))
|
|
# OSError is probably from invalid images, SyntaxError probably from unsupported images
|
|
except (OSError, SyntaxError):
|
|
return 1
|
|
# Taken from https://coderwall.com/p/nax6gg/fix-jpeg-s-unexpectedly-rotating-when-saved-with-pil
|
|
if hasattr(im, '_getexif'):
|
|
orientation = 0x0112
|
|
exif = im._getexif()
|
|
if exif is not None:
|
|
orientation = exif.get(orientation)
|
|
rotations = {
|
|
3: Image.ROTATE_180,
|
|
6: Image.ROTATE_270,
|
|
8: Image.ROTATE_90
|
|
}
|
|
if orientation in rotations:
|
|
im = im.transpose(rotations[orientation])
|
|
if avatar:
|
|
# crop 1:1
|
|
width, height = im.size
|
|
min_dimension = min(width, height)
|
|
crop_size = min_dimension
|
|
left = (width - crop_size) // 2
|
|
top = (height - crop_size) // 2
|
|
right = left + crop_size
|
|
bottom = top + crop_size
|
|
im = im.crop((left, top, right, bottom))
|
|
im.thumbnail((256, 256))
|
|
else:
|
|
im.thumbnail((1280, 1280))
|
|
|
|
# Let's check the aspect ratio and see if it's crazy
|
|
# IF this is a drawing
|
|
if drawing and ((im.size[0] / im.size[1]) < 0.30):
|
|
return 1
|
|
|
|
# I know some people have aneurysms when they see people actually using SHA1 in the real world, for anything in general.
|
|
# Yes, we are really using it. Sorry if that offends you. It's just fast and I don't feel I need anything more random, since we are talking about IMAGES.
|
|
if stream:
|
|
hash = sha1()
|
|
for chunk in img.chunks():
|
|
hash.update(chunk)
|
|
imhash = hash.hexdigest()
|
|
else:
|
|
imhash = sha1(im.tobytes()).hexdigest()
|
|
# File saving target
|
|
target = 'png'
|
|
if stream:
|
|
# If we have a stream and either a JPEG or a WEBP, save them as those since those are a bit better than plain PNG
|
|
if 'jpeg' in img.content_type:
|
|
target = 'jpeg'
|
|
im = im.convert('RGB')
|
|
elif 'webp' in img.content_type:
|
|
target = 'webp'
|
|
floc = imhash + '.' + target
|
|
# If the file exists, just use it, that's what hashes are for.
|
|
if not os.path.exists(settings.MEDIA_ROOT + floc):
|
|
im.save(settings.MEDIA_ROOT + floc, target, optimize=True)
|
|
return settings.MEDIA_URL + floc
|
|
|
|
# Todo: Put this into post/comment delete thingy method
|
|
def image_rm(image_url):
|
|
if settings.IMAGE_DELETE_SETTING:
|
|
if settings.MEDIA_URL in image_url:
|
|
sysfile = image_url.split(settings.MEDIA_URL)[1]
|
|
sysloc = settings.MEDIA_ROOT + sysfile
|
|
if settings.IMAGE_DELETE_SETTING > 1:
|
|
try:
|
|
remove(sysloc)
|
|
except:
|
|
return False
|
|
else:
|
|
return True
|
|
# The RM'd directory to move it to
|
|
rmloc = sysloc.replace(settings.MEDIA_ROOT, settings.MEDIA_ROOT + 'rm/')
|
|
try:
|
|
rename(sysloc, rmloc)
|
|
except:
|
|
return False
|
|
else:
|
|
return True
|
|
else:
|
|
return False
|
|
|
|
def get_gravatar(email):
|
|
try:
|
|
page = urllib.request.urlopen('https://gravatar.com/avatar/'+ md5(email.encode('utf-8').lower()).hexdigest() +'?d=404&s=128')
|
|
except:
|
|
return False
|
|
return page.geturl()
|
|
|
|
def filterchars(str=""):
|
|
# If string is blank, None, any other object, etc, make it whitespace so it's detected by isspace.
|
|
if not str:
|
|
str = " "
|
|
# Forbid chars in this list, currently: Right-left override, largest Unicode character.
|
|
# Now restricting everything in https://www.reddit.com/r/Unicode/comments/5qa7e7/widestlongest_unicode_characters_list/
|
|
forbid = ["\u202e", "\ufdfd", "\u01c4", "\u0601", "\u2031", "\u0bb9", "\u0bf8", "\u0bf5", "\ua9c4", "\u102a", "\ua9c5", "\u2e3b", "\ud808", "\ude19", "\ud809", "\udc2b", "\ud808", "\udf04", "\ud808", "\ude1f", "\ud808", "\udf7c", "\ud808", "\udc4e", "\ud808", "\udc31", "\ud808", "\udf27", "\ud808", "\udd43", "\ud808", "\ude13", "\ud808", "\udf59", "\ud808", "\ude8e", "\ud808", "\udd21", "\ud808", "\udd4c", "\ud808", "\udc4f", "\ud808", "\udc30", "\ud809", "\udc2a", "\ud809", "\udc29", "\ud808", "\ude19", "\ud809", "\udc2b"]
|
|
for char in forbid:
|
|
if char in str:
|
|
str = str.replace(char, " ")
|
|
if str.isspace():
|
|
return 'None'
|
|
return str
|
|
|
|
""" Not using getipintel anymore
|
|
def getipintel(addr):
|
|
# My router's IP prefix is 192.168.1.*, so this works in debug
|
|
if settings.ipintel_email and not '192.168' in addr:
|
|
try:
|
|
site = urllib.request.urlopen('https://check.getipintel.net/check.php?ip={0}&contact={1}&flags=f'
|
|
.format(addr, settings.ipintel_email))
|
|
except:
|
|
return 0
|
|
return float(site.read().decode())
|
|
else:
|
|
return 0
|
|
"""
|
|
|
|
# not ideal, switch this to use a real cache when caching for everything else is implemented (never?)
|
|
iphub_cache = dict()
|
|
|
|
# Now using iphub
|
|
def iphub(addr, want_asn=False):
|
|
# hack to exclude my private network at the time (security flaw?)
|
|
if settings.IPHUB_KEY and not '192.168' in addr:
|
|
if addr in iphub_cache:
|
|
get_r = iphub_cache[addr]
|
|
#print('getting ip ' + addr + ' from cache 😌')
|
|
else:
|
|
#print('GETTING IP ' + addr + ' FROM IPHUB😤😤😤😤😤😤')
|
|
req = urllib.request.Request('http://v2.api.iphub.info/ip/' + addr, headers={'X-Key': settings.IPHUB_KEY})
|
|
response = urllib.request.urlopen(req)
|
|
data = response.read().decode()
|
|
get_r = json.loads(data)
|
|
iphub_cache[addr] = get_r
|
|
if want_asn:
|
|
return get_r.get('asn', '0')
|
|
if get_r.get('block', 0) == 1:
|
|
return True
|
|
# should just return falsey when returning nothing anyway?
|
|
#else:
|
|
# return False
|