def patch(self):
"""??????(???)"""
form = request.form
nickname = form.get('nickname', '').strip()
description = form.get('description', '').strip()
avatar = request.files.get('avatar', None)
if len(nickname):
current_user.nickname = nickname
if len(description):
current_user.description = description
if avatar is not None and avatar.content_type.startswith('image/'):
filename = '%s_%d.webp' % (current_user.id, time2stamp(datetime.now()))
url = current_app.static_folder + '/images/user/%s'
Image.open(avatar.stream).save(url % filename)
old = current_user.avatar
if old != 'MonkeyEye.webp':
os.remove(url % old)
current_user.avatar = filename
db.session.commit()
return {'message': '????????'}, 200
python类static_folder()的实例源码
def on_model_change(self, form, user, is_created):
avatar = form.avatar.data
# ?????????????
if avatar.content_type.startswith('image/'):
filename = '%s_%d.webp' % (user.id, time2stamp(datetime.now()))
url = current_app.static_folder + '/images/user/%s'
Image.open(avatar.stream).save(url % filename)
old = form.avatar.object_data
if old != 'MonkeyEye.webp':
os.remove(url % old)
user.avatar = filename
elif is_created:
user.avatar = 'MonkeyEye.webp'
else: # ???????????????
user.avatar = form.avatar.object_data
if is_created: # ???????????md5
user.password = MD5Twice(form.password.data)
user.payPassword = MD5Twice(form.payPassword.data)
def on_model_change(self, form, movie, is_created):
poster = form.poster.data
if is_created:
movie.id = UUID()
if poster.content_type.startswith('image/'):
filename = '%s.webp' % movie.id
url = '%s/images/poster/%s' % (current_app.static_folder, filename)
Image.open(poster.stream).save(url % filename)
movie.poster = filename
elif is_created:
raise ValidationError('Poster is required.')
else:
movie.poster = form.poster.object_data
if form.description.data.strip() == '':
movie.description = '????'
def static_from_root():
return send_from_directory(current_app.static_folder, request.path[1:])
def admin_files(chalid):
if request.method == 'GET':
files = Files.query.filter_by(chal=chalid).all()
json_data = {'files':[]}
for x in files:
json_data['files'].append({'id':x.id, 'file':x.location})
return jsonify(json_data)
if request.method == 'POST':
if request.form['method'] == "delete":
f = Files.query.filter_by(id=request.form['file']).first_or_404()
if os.path.exists(os.path.join(app.static_folder, 'uploads', f.location)): ## Some kind of os.path.isfile issue on Windows...
os.unlink(os.path.join(app.static_folder, 'uploads', f.location))
db.session.delete(f)
db.session.commit()
db.session.close()
return "1"
elif request.form['method'] == "upload":
files = request.files.getlist('files[]')
for f in files:
filename = secure_filename(f.filename)
if len(filename) <= 0:
continue
md5hash = hashlib.md5(os.urandom(64)).hexdigest()
if not os.path.exists(os.path.join(os.path.normpath(app.static_folder), 'uploads', md5hash)):
os.makedirs(os.path.join(os.path.normpath(app.static_folder), 'uploads', md5hash))
f.save(os.path.join(os.path.normpath(app.static_folder), 'uploads', md5hash, filename))
db_f = Files(chalid, os.path.join('static', 'uploads', md5hash, filename))
db.session.add(db_f)
db.session.commit()
db.session.close()
return redirect('/admin/chals')
def admin_create_chal():
files = request.files.getlist('files[]')
## TODO: Expand to support multiple flags
flags = [{'flag':request.form['key'], 'type':int(request.form['key_type[0]'])}]
# Create challenge
chal = Challenges(request.form['name'], request.form['desc'], request.form['value'], request.form['category'], flags)
db.session.add(chal)
db.session.commit()
for f in files:
filename = secure_filename(f.filename)
if len(filename) <= 0:
continue
md5hash = hashlib.md5(os.urandom(64)).hexdigest()
if not os.path.exists(os.path.join(os.path.normpath(app.static_folder), 'uploads', md5hash)):
os.makedirs(os.path.join(os.path.normpath(app.static_folder), 'uploads', md5hash))
f.save(os.path.join(os.path.normpath(app.static_folder), 'uploads', md5hash, filename))
db_f = Files(chal.id, os.path.join('static', 'uploads', md5hash, filename))
db.session.add(db_f)
db.session.commit()
db.session.close()
return redirect('/admin/chals')
def get(self, path):
static_folder = current_app.static_folder
img_folder = os.path.join(static_folder, 'img')
return send_from_directory(img_folder, path,
mimetype='image/png',
as_attachment=False)
def file_upload():
""" Uploads images dropped on the web editor's markdown box to static/images
and notifies editors by email
"""
upload_folder = 'images'
title = request.form['title']
files = request.files
uploadedFiles = []
if files:
for img_file in files.values():
filename = secure_filename(title + "_" + img_file.filename).lower()
dst_folder = os.path.join(current_app.static_folder, upload_folder)
if is_allowed_image_format(img_file):
try:
img_file.save(os.path.join(dst_folder, filename))
send_from_directory(dst_folder, filename)
uploadedFiles += [url_for("static", filename=os.path.join(upload_folder, filename))]
except Exception as e:
error_msg = u"ERROR during image upload: {}".format(str(e))
logger.error(error_msg)
return json.dumps({'error_msg': error_msg, 'success': False})
elif is_pdf(filename):
from PyPDF2 import PdfFileReader
try:
src_pdf = PdfFileReader(img_file)
filename = os.path.splitext(filename)[0]
num_pages = src_pdf.getNumPages()
for page_num in range(num_pages):
page_png = pdf_page_to_png(src_pdf, page_num)
page_name = u"{filename}_{page_num}.jpg".format(**locals())
page_png.save(filename=os.path.join(dst_folder, page_name))
uploadedFiles += [url_for("static", filename=os.path.join(upload_folder, page_name))]
except Exception as e:
error_msg = u"ERROR during pdf upload: {}".format(str(e))
logger.error(error_msg)
return json.dumps({'error_msg': error_msg, 'success': False})
return json.dumps({'links': uploadedFiles, 'success': True})
def after_model_delete(self, user):
# ???????????
if user.avatar != 'MonkeyEye.webp':
url = '%s/images/user/%s' % (current_app.static_folder, user.avatar)
os.remove(url)
def after_model_delete(self, movie):
# ????????????
url = '%s/images/poster/%s' % (current_app.static_folder, movie.poster)
os.remove(url)
