def mail_compose():
if request.method == 'POST':
content = request.form['content']
if content:
receiver = User.query.get(request.form['receiver'])
subject = request.form['subject']
mail = Mail(content=content, subject=subject, sender=g.user, receiver=receiver)
db.session.add(mail)
db.session.commit()
# generate automated Administrator response
if receiver.id == 1:
content = "I would be more than happy to help you with that. Unforunately, the person respsonsible for that is unavailable at the moment. We'll get back with you soon. Thanks."
mail = Mail(content=content, subject='RE:'+subject, sender=receiver, receiver=g.user)
db.session.add(mail)
db.session.commit()
flash('Mail sent.')
return redirect(url_for('ph_bp.mail'))
users = User.query.filter(User.id != g.user.id).order_by(User.username.asc()).all()
return render_template('mail_compose.html', users=users)
python类id()的实例源码
def api_messages(id=None):
if request.method == 'POST':
jsonobj = request.get_json(force=True)
message = jsonobj['message']
if message:
msg = Message(comment=message, user=g.user)
db.session.add(msg)
db.session.commit()
if request.method == 'DELETE':
message = Message.query.get(id)
if message and message.user == g.user:
db.session.delete(message)
db.session.commit()
messages = []
# add is_owner field to each message
for message in Message.query.order_by(Message.created.desc()).all():
is_owner = False
if message.user == g.user:
is_owner = True
message = message.serialize()
message['is_owner'] = is_owner
messages.append(message)
resp = jsonify(messages=messages)
resp.mimetype = 'text/html'
return resp
def login():
# redirect to home if already logged in
if session.get('user_id'):
return redirect(url_for('ph_bp.home'))
if request.method == 'POST':
token = md5(request.form['password']+session.get('nonce', '')).hexdigest()
if token == request.form['token']:
query = "SELECT * FROM users WHERE username='{}' AND password_hash='{}'"
username = request.form['username']
password_hash = xor_encrypt(request.form['password'], current_app.config['PW_ENC_KEY'])
user = db.session.execute(query.format(username, password_hash)).first()
if user and user['status'] == 1:
session['user_id'] = user.id
path = os.path.join(current_app.config['UPLOAD_FOLDER'], md5(str(user.id)).hexdigest())
if not os.path.exists(path):
os.makedirs(path)
session['upload_folder'] = path
session.rotate()
return redirect(request.args.get('next') or url_for('ph_bp.home'))
return redirect(url_for('ph_bp.login', error='Invalid username or password.'))
return redirect(url_for('ph_bp.login', error='Bot detected.'))
session['nonce'] = get_token(5)
return render_template('login.html')
def update_rank(event, context):
with application.app_context():
my_kwargs = event.get("kwargs")
date = datetime.datetime.utcnow().strftime('%Y/%m/%d')
for i in range(my_kwargs["start"], my_kwargs["end"]):
url = "https://www.acmicpc.net/ranklist/" + str(i)
soup = get_soup_from_url(url)
table = soup.find(id='ranklist')
trs = table.tbody.find_all('tr')
boj_ids = list()
boj_ranks = list()
for tr in trs:
tds = tr.find_all('td')
if int(tds[3].a.string.strip()) <= 19:
break
boj_ids.append(''.join(tds[1].find_all(text=True, recursive=True)).strip())
boj_ranks.append(int(tds[0].string))
api = request_koo_api("user", boj_ids)
koo_ranks = list(user["ranking"] for user in api)
for _ in range(len(boj_ids)):
boj_id = boj_ids[_]
boj_rank = boj_ranks[_]
if koo_ranks[_] == None:
koo_rank = 0
else:
koo_rank = koo_ranks[_] + 1
data = {date: [boj_rank, koo_rank]}
if not Ranking.query.filter_by(boj_id=boj_id).scalar():
ranking = Ranking(boj_id=boj_id, ranking=data)
db.session.add(ranking)
db.session.commit()
else:
user = Ranking.query.filter_by(boj_id=boj_id)
new_ranking = user.first().ranking
new_ranking.update(data)
user.first().ranking = new_ranking
db.session.commit()
print("{0} {1} {2}".format(boj_id, boj_rank, koo_rank))
return "OK"
def update_user():
if request.is_xhr:
user_id = request.args.get('id')
update_profile(user_id)
return "OK"
else:
abort(404)
def mark_fix(self, fix_id):
solution = self.db_session.query(Solution).filter(Solution.complete == True) \
.filter(Solution.id == fix_id).one_or_none()
solution.in_osm = True
self.db_session.commit()
def read_fix(self):
"""
Returns an array of dicts containing fixes from kort
"""
solutions = self.db_session.query(Solution).filter(Solution.complete == True) \
.filter(Solution.valid == True) \
.filter(Solution.in_osm == False)
kort_fixes = []
for s in solutions:
error = self.db_session.query(kort_errors).filter(kort_errors.errorId == s.error_id) \
.filter(kort_errors.osmId == s.osmId).one_or_none()
osm_type = self.db_session.query(error_type).filter(error_type.type == s.error_type).one_or_none()
if error and osm_type:
u = self.db_session.query(User).filter(User.id == s.user_id).one_or_none()
entry = {
'osm_id': s.osmId,
'osm_type': error.osmType,
'osm_tag': osm_type.osm_tag,
'answer': s.solution,
'error_type': s.error_type,
'username': u.username if u else '',
'user_id': s.user_id,
'fix_id': s.id,
'source': error.source
}
kort_fixes.append(entry)
return kort_fixes
def users(user_id):
form = AboutMeForm()
user = User.query.filter(User.id == user_id).first()
if user_id is not current_user.id:
return redirect(url_for('index'))
if not user:
flash("The user is not exist.")
redirect("/index")
all_item = user.all_item.all()
return render_template("user.html", form=form, user=user, all_item=all_item)
def delete_item(item_id, user_id):
try:
db.session.query(Monitor).filter_by(id=item_id).delete()
db.session.commit()
except:
flash("???????????")
return redirect(url_for("users", user_id=user_id))
return redirect(url_for("users", user_id=user_id))
def on_item(item_id, user_id):
try:
db.session.query(Monitor).filter_by(id=item_id).update({"status": True})
db.session.commit()
except:
flash("???????????")
return redirect(url_for("users", user_id=user_id))
return redirect(url_for("users", user_id=user_id))
def off_item(item_id, user_id):
try:
db.session.query(Monitor).filter_by(id=item_id).update({"status": False})
db.session.commit()
except:
flash("???????????")
return redirect(url_for("users", user_id=user_id))
return redirect(url_for("users", user_id=user_id))
def admin_tools_remove(id):
tool = Tool.query.get(id)
if tool:
db.session.delete(tool)
db.session.commit()
flash('Tool removed.')
else:
flash('Invalid tool ID.')
return redirect(url_for('ph_bp.admin_tools'))
def admin_users():
users = User.query.filter(User.id != g.user.id).order_by(User.username.asc()).all()
return render_template('admin_users.html', users=users)
def admin_users_modify(action, id):
user = User.query.get(id)
if user:
if user != g.user:
if action == 'promote':
user.role = 0
db.session.add(user)
db.session.commit()
flash('User promoted.')
elif action == 'demote':
user.role = 1
db.session.add(user)
db.session.commit()
flash('User demoted.')
elif action == 'enable':
user.status = 1
db.session.add(user)
db.session.commit()
flash('User enabled.')
elif action == 'disable':
user.status = 0
db.session.add(user)
db.session.commit()
flash('User disabled.')
else:
flash('Invalid user action.')
else:
flash('Self-modification denied.')
else:
flash('Invalid user ID.')
return redirect(url_for('ph_bp.admin_users'))
def mail_reply(id=0):
mail = Mail.query.filter(Mail.id == id).first()
return render_template('mail_reply.html', mail=mail)
def mail_delete(id):
mail = Mail.query.get(id)
if mail:
db.session.delete(mail)
db.session.commit()
flash('Mail deleted.')
else:
flash('Invalid mail ID.')
return redirect(url_for('ph_bp.mail'))
def messages_delete(id):
message = Message.query.get(id)
if message and message.user == g.user:
db.session.delete(message)
db.session.commit()
flash('Message deleted.')
else:
flash('Invalid message ID.')
return redirect(url_for('ph_bp.messages'))
def tools_info():
query = "SELECT * FROM tools WHERE id={}"
tid = request.form['tid']
try:
tools = db.session.execute(query.format(tid))
except exc.OperationalError:
tools = ()
return jsonify(tools=[dict(t) for t in tools])
def get_current_user():
"""Set g.user to the currently logged in user.
Called before each request, get_current_user sets the global g.user
variable to the currently logged in user. A currently logged in user is
determined by seeing if it exists in Flask's session dictionary.
If it is the first time the user is logging into this application it will
create the user and insert it into the database. If the user is not logged
in, None will be set to g.user.
"""
# Set the user in the session dictionary as a global g.user and bail out
# of this function early.
if session.get('user'):
g.user = session.get('user')
return
# Attempt to get the short term access token for the current user.
result = get_user_from_cookie(cookies=request.cookies, app_id=FB_APP_ID,
app_secret=FB_APP_SECRET)
# If there is no result, we assume the user is not logged in.
if result:
# Check to see if this user is already in our database.
user = User.query.filter(User.id == result['uid']).first()
if not user:
# Not an existing user so get info
graph = GraphAPI(result['access_token'])
profile = graph.get_object('me')
if 'link' not in profile:
profile['link'] = ""
# Create the user and insert it into the database
user = User(id=str(profile['id']), name=profile['name'],
profile_url=profile['link'],
access_token=result['access_token'])
db.session.add(user)
elif user.access_token != result['access_token']:
# If an existing user, update the access token
user.access_token = result['access_token']
# Add the user to the current session
session['user'] = dict(name=user.name, profile_url=user.profile_url,
id=user.id, access_token=user.access_token)
# Commit changes to the database and set the user as a global g.user
db.session.commit()
g.user = session.get('user', None)
def get_user():
submissions = []
accepted_submissions = []
ranking_date = []
boj_rank = []
koo_rank = []
user_dict = []
user_id = request.args.get("id")
acc_user_id = is_boj_user(user_id)
if acc_user_id:
if not User.query.filter_by(boj_id=acc_user_id).scalar():
user = User(boj_id=acc_user_id)
db.session.add(user)
db.session.commit()
else:
return render_template("index.html", id=user_id, err=True)
user = User.query.filter_by(boj_id=acc_user_id).first()
if user.update_time is None or (datetime.datetime.utcnow() - user.update_time).seconds > 600:
updated = False
else:
updated = True
two_weeks_ago = datetime.date.today() - datetime.timedelta(days=14)
submissions = Submission.query.filter_by(boj_id=acc_user_id).filter(Submission.datetime > two_weeks_ago).all()
accepted_submissions = AcceptedSubmission.query.filter_by(boj_id=acc_user_id).order_by(
AcceptedSubmission.datetime).all()
if Ranking.query.filter_by(boj_id=acc_user_id).scalar():
ranking_json = Ranking.query.filter_by(boj_id=acc_user_id).first().ranking
ranking_date = sorted(list(ranking_json.keys()))
ranking_values = [ranking_json[i] for i in ranking_date]
boj_rank = [i[0] for i in ranking_values]
koo_rank = [i[1] for i in ranking_values]
user_ids = [i.boj_id for i in User.query.order_by(User.update_time).all()][::-1]
user_dict = OrderedDict()
for i in user_ids:
user_dict[i] = None
return render_template("user.html", user=user, updated=updated, submissions=submissions,
accepted_submissions=accepted_submissions, ranking_date=ranking_date,
boj_rank=boj_rank, koo_rank=koo_rank, user_ids=json.dumps(user_dict))
def login():
# ?????????
if current_user.is_authenticated:
return redirect('index')
# ????
form = LoginForm()
if form.validate_on_submit():
user = User.login_check(request.form.get('user_name'))
user_name = request.form.get('user_name')
password = request.form.get('password')
# ????
try:
user_forpwd = User.query.filter(User.nickname == user_name).first()
print(type(user_forpwd), user_forpwd.password)
if not check_password_hash(user_forpwd.password, password):
flash('????????')
return redirect('/login')
except:
flash("??????????")
return redirect('/login')
if user:
login_user(user)
user.last_seen = datetime.datetime.now()
try:
db.session.add(user)
db.session.commit()
except:
flash("???????????")
return redirect('/login')
# flash(request.form.get('user_name'))
# flash('remember me? ' + str(request.form.get('remember_me')))
flash('????')
return redirect(url_for("users", user_id=current_user.id))
else:
flash('?????????')
return redirect('/login')
return render_template("login.html", title="Sign In", form=form)
