Python models.User 模块，id() 实例源码

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)

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)