python类get()的实例源码

def update_password():
    username = request.forms.get('username', None)
    password = request.forms.get('password', None)
    new_pass = request.forms.get('new_pass', None)

    dict_for_return = dict(universal_POST_dict)
    if not verify_login(username, password):
        dict_for_return['verified'] = False
        dict_for_return['message_str'] = "login failed"
        json_for_return = json.dumps(dict_for_return)
        return json_for_return
    else:
        dbMgr = MmrzSyncDBManager("USERS")
        dbMgr.update_USERS_DB([username, new_pass])
        dbMgr.closeDB()

        dict_for_return['verified'] = True
        dict_for_return['message_str'] = "password updated"
        json_for_return = json.dumps(dict_for_return)
        return json_for_return

def do_register():
    """do register
    @param: url
    @param: name
    @param: description
    @param: creator
    """

    data = ChpplData()
    data.set_url(str(request.params.get("url")))
    data.set_name(str(request.params.get("name")))
    data.set_description(str(request.params.get("description")))
    data.set_creator(str(request.params.get("creator")))

    result = ChpplResult()
    result.set_data(data)

    if result.check_data() is "Success":
        result.execute_query("insert")

    msg_list = result.get_msg_list()

    return template('result', result=result.check_data(), msg_list=msg_list)

def to_delete():
    """do delete
    @param: package
    @param: confilm
    """

    data = ChpplData()
    data.set_package(str(request.params.get('package')))
    data.set_confilm(str(request.params.get('confilm')))

    result = ChpplResult()
    result.set_data(data)

    if result.confilm_data() is "Success":
        result.execute_query("delete")

    msg_list = result.get_msg_list()

    return template('result', result=result.confilm_data(), msg_list=msg_list)

def cal_remind_time(memTimes, types):
    curTime = int(time.time())

    remindTime = {
        0: curTime + (60 * 0), # back end import always 0 minute, request from @smilebin818
        1: curTime + (60 * 30), # 30 minutes
        2: curTime + (60 * 60 * 12), # 12 hours
        3: curTime + (60 * 60 * 24), # 1 day
        4: curTime + (60 * 60 * 24 * 2), # 2 days
        5: curTime + (60 * 60 * 24 * 4), # 4 days
        6: curTime + (60 * 60 * 24 * 7), # 7 days
        7: curTime + (60 * 60 * 24 * 15), # 15 days
    }.get(memTimes, curTime)

    if types == "int":
        remindTime = remindTime
    elif types ==  "str":
        remindTime = time.strftime('%Y-%m-%d %H:%M:%S +0800', time.localtime(remindTime))

    return remindTime

def is_state_cache_available():
    username = request.forms.get('username', None)
    password = request.forms.get('password', None)

    dict_for_return = dict(universal_POST_dict)
    if not verify_login(username, password):
        dict_for_return['mmrz_code'] = MMRZ_CODE_Universal_Verification_Fail
        json_for_return = json.dumps(dict_for_return)

        return json_for_return

    else:
        dbMgr = MongoDBManager()
        userData = dbMgr.query_memorize_state(username)
        dbMgr.closeDB()

        state_cached = userData.get('state_cached', False)
        dict_for_return['state_cached'] = state_cached
        json_for_return = json.dumps(dict_for_return)

        return json_for_return

def restore_remote_saved_state():
    username = request.forms.get('username', None)
    password = request.forms.get('password', None)

    dict_for_return = dict(universal_POST_dict)
    if not verify_login(username, password):
        dict_for_return['mmrz_code'] = MMRZ_CODE_Universal_Verification_Fail

        json_for_return = json.dumps(dict_for_return)
        return json_for_return
    else:
        dbMgr = MongoDBManager()
        userData = dbMgr.query_memorize_state(username)
        dbMgr.closeDB()

        dict_for_return['mmrz_code'] = MMRZ_CODE_Restore_State_OK
        dict_for_return['current_cursor'] = int(userData['current_cursor'])
        dict_for_return['max_size_this_turn'] = int(userData['max_size_this_turn'])
        dict_for_return['data'] = userData['data']
        json_for_return = json.dumps(dict_for_return)

        return json_for_return

def clear_state_cached_flag_and_eiginvalue():
    username = request.forms.get('username', None)
    password = request.forms.get('password', None)

    dict_for_return = dict(universal_POST_dict)
    if not verify_login(username, password):
        dict_for_return['mmrz_code'] = MMRZ_CODE_Universal_Verification_Fail

        json_for_return = json.dumps(dict_for_return)
        return json_for_return
    else:
        dbMgr = MongoDBManager()
        dbMgr.clear_state_cached_flag_and_eiginvalue(username)
        dbMgr.closeDB()

        dict_for_return['mmrz_code'] = MMRZ_CODE_Universal_OK

        json_for_return = json.dumps(dict_for_return)
        return json_for_return

def is_word_exist():
    username = request.forms.get('username', None)
    password = request.forms.get('password', None)

    # wordInfo = request.forms.get('wordInfo', None)

    word      = request.forms.get('word', None)
    pronounce = request.forms.get('pronounce', None)

    dict_for_return = dict(universal_POST_dict)
    if not verify_login(username, password):
        dict_for_return['mmrz_code'] = MMRZ_CODE_Universal_Verification_Fail
        dict_for_return['message_str'] = "/is_word_exist verify failed"

        json_for_return = json.dumps(dict_for_return)
        return json_for_return
    else:
        dbMgr = MmrzSyncDBManager(username)
        exist, wordID = dbMgr.is_word_exist(word, pronounce)

        dict_for_return['exist'] = exist
        dict_for_return['message_str'] = "/is_word_exist verify OK"

        json_for_return = json.dumps(dict_for_return)
        return json_for_return

def update_row():
    username = request.forms.get('username', None)
    password = request.forms.get('password', None)

    dict_for_return = dict(universal_POST_dict)
    if not verify_login(username, password):
        dict_for_return['verified'] = False
        dict_for_return['message_str'] = "login failed"
        json_for_return = json.dumps(dict_for_return)
        return json_for_return
    else:
        row = request.forms['row']
        update_whole_row = request.forms.get('update_whole_row', False)
        row = json.loads(row)
        dbMgr = MmrzSyncDBManager(username)
        dbMgr.createDB()
        dbMgr.updateDB(row, update_whole_row)
        dbMgr.closeDB()
        dict_for_return['verified'] = True
        dict_for_return['message_str'] = "Update row success"
        json_for_return = json.dumps(dict_for_return)
        return json_for_return

def get_ranking_info():
    period  = request.params.get('period', None)

    rank_info = []
    tikMgr = TikTimeDBManager()

    if period == "day":
        rank_info = tikMgr.getDailyRanking(int(time.time()))
    elif period == "week":
        rank_info = tikMgr.getWeeklyRanking(int(time.time()))
    elif period == "month":
        rank_info = tikMgr.getMonthlyRanking(int(time.time()))
    elif period == "year":
        rank_info = tikMgr.getYearlyRanking(int(time.time()))
    else:
        rank_info = []

    tikMgr.closeDB()

    # append 5 users in the end
    rank_info += [("None", 0) for i in range(5)]

    return json.dumps(rank_info)

def download_wordbook():
    username = request.params.get('username', None)
    password = request.params.get('password', None)

    dict_for_return = dict(universal_POST_dict)
    if not verify_login(username, password):
        dict_for_return['verified'] = False
        dict_for_return['message_str'] = "login failed"
        dict_for_return['wordbook'] = []
        json_for_return = json.dumps(dict_for_return)
        return json_for_return
    else:
        dbMgr = MmrzSyncDBManager(username)
        dbMgr.createDB()
        rows = dbMgr.readAllDB()
        dbMgr.closeDB()
        dict_for_return['verified'] = True
        dict_for_return['message_str'] = "Download success"
        dict_for_return['wordbook'] = rows
        json_for_return = json.dumps(dict_for_return)
        return json_for_return

# show serving IP (for debug)

def new_comment():
    data = request.json
    comments_dir = app._config['henet']['comments_dir']
    article_uuid = data['source_path']
    article_thread = ArticleThread(comments_dir, article_uuid)
    article_thread.add_comment(text=data['text'],
                               author=data['author'])

    article_thread.save()
    notifs = app._config['notifications']
    moderator = notifs.get('moderate_comment')
    if moderator is not None:
        app.send_email([moderator], u'Nouveau commentaire',
                       MODERATE_BODY)

    emit(EVENT_CREATED_COMMENT, article_uuid=article_uuid)
    return {'result': 'OK'}

def problem_list_handler():
    try:
        page = int(bottle.request.query.get('page', 1))
    except ValueError:
        page = 1
    count = model.count_public_problems()
    problems = model.get_public_problems(
        limit=FLAGS.pagination_items_per_page,
        skip=(page - 1) * FLAGS.pagination_items_per_page)
    team_display_name_map = handler_util.compute_team_display_name_map(
        problem['owner'] for problem in problems)
    pagination = handler_util.Pagination(
        page, FLAGS.pagination_items_per_page, count)
    template_dict = {
        'problems': problems,
        'team_display_name_map': team_display_name_map,
        'pagination': pagination,
    }
    return handler_util.render('problem_list.html', template_dict)

def admin_problem_list_handler():
    page = int(bottle.request.query.get('page', 1))
    count = model.count_all_problems_for_admin()
    problems = model.get_all_problems_for_admin(
        limit=FLAGS.pagination_items_per_page,
        skip=(page - 1) * FLAGS.pagination_items_per_page)
    team_display_name_map = handler_util.compute_team_display_name_map(
        problem['owner'] for problem in problems)
    pagination = handler_util.Pagination(
        page, FLAGS.pagination_items_per_page, count)
    template_dict = {
        'problems': problems,
        'team_display_name_map': team_display_name_map,
        'pagination': pagination,
    }
    return handler_util.render('admin/problem_list.html', template_dict)

def admin_solution_list_handler():
    page = int(bottle.request.query.get('page', 1))
    count = model.count_all_solutions_for_admin()
    solutions = model.get_all_solutions_for_admin(
        limit=FLAGS.pagination_items_per_page,
        skip=(page - 1) * FLAGS.pagination_items_per_page)
    team_display_name_map = handler_util.compute_team_display_name_map(
        solution['owner'] for solution in solutions)
    pagination = handler_util.Pagination(
        page, FLAGS.pagination_items_per_page, count)
    template_dict = {
        'solutions': solutions,
        'team_display_name_map': team_display_name_map,
        'pagination': pagination,
    }
    return handler_util.render('admin/solution_list.html', template_dict)

def ingest():
    fileId = request.query.get('file_id')
    filePath = request.query.get('file_path')
    toHost = request.query.get('to_host')
    ingestRate = request.query.get('ingest_rate')
    if (fileId == None):
        return 'No file id is provided'
    else:
        try:
            ngamsJobMWALib.fileIngested(fileId, filePath, toHost, ingestRate)
            msg = 'File %s is just ingested at %s on %s with a rate %s' % (fileId, filePath, toHost, ingestRate)
            logger.info(msg)
            return msg
        except Exception, err:
            logger.error(traceback.format_exc())
            return 'Exception (%s) when doing - File %s is just ingested at %s on %s' % (str(err), fileId, filePath, toHost)

def main():
    #FORMAT = "%(asctime)-15s %(message)s"
    FORMAT = "%(asctime)-15s - %(name)s - %(levelname)s - %(message)s"
    logging.basicConfig(filename='/tmp/NGAS_MWA/log/ngamsJobMAN.log', level=logging.DEBUG, format = FORMAT)
    logger.info('ngamsJobMAN Started.......')

    config = getConfig()
    if (not config):
        exit(1)

    # start the web server supported by bottle and paste
    run(host = config.get('Web Server', 'IpAddress'),
        server = 'paste',
        port = config.getint('Web Server', 'Port'),
        debug = config.getboolean('Web Server', 'Debug'))

    logger.info('ngamsJobMAN Shutdown.......')

def echo(ws):
    cs.add(ws)
    print(u'???'+str(ws)+u'???websocket')
    for _msg in cache['msg']:
        ws.send(_msg)
    while True:
        msg = ws.receive()
        if msg is not None:
            _dict = json.loads(msg)
            _type = _dict.get("action",EVENT_EMPTY)
            if _type in funcs:
                funcs[_type]((msg,_dict))
            else:
                empty_func(msg)
        else: break
    cs.remove(ws)
    print(u'???'+str(ws)+u'????')

def edit():
    #file_list = {
    #    'filename1': 'path1',
    #    'filename2': 'path2',
    #    'dirname1': {
    #        'filename3': 'path3',
    #        'dirname2': {
    #            'filename4': 'path4',
    #            'filename5': 'path5'
    #        }
    #    }
    #}
    file_list = make_file_tree(ROOT)
    file = request.GET.get('file')
    if file:
        with open(os.path.join(ROOT, file), 'r') as in_file:
            code = in_file.read()
        if file.split('.')[-1] in ['pyui', 'json']:
            code = json.dumps(json.loads(code), indent=4, separators=(',', ': '))
        output = template(os.path.join(IDE_REL_ROOT, 'main.tpl'), files = file_list, save_as = file, code = code)
    else:
        output = template(os.path.join(IDE_REL_ROOT, 'main.tpl'), files = file_list)
    return output

def set_sskey():
    u = current_user
    u.sskey = request.forms.get('sskey')
    u.write()

    import cron;
    cd = cron.start()
    if cd <= 0.5:
        msg = "Your Shadowsocks key is changed!"
    else:
        msg = "The Shadowsocks key will be changed in %.2f sec" % cd

    return get_home_content(msg)



# Website for Administrator

def do_login():
    username = request.forms.get('username')
    password = get_salted_password()

    current_user = user.get_by_username(username)
    logined = current_user and current_user.salted_password == password

    if logined:
        response.set_cookie('ssl_uid', str(current_user.id))
        response.set_cookie('ssl_pw', password)
        return redirect('/')

    return template('login', 
        username=username, 
        message='User not found.' if not current_user else 'Password is incorrect.', 
        salt=config.USER_SALT
    )

def process_signup():

    email = bottle.request.forms.get("email")
    username = bottle.request.forms.get("username")
    password = bottle.request.forms.get("password")
    verify = bottle.request.forms.get("verify")

    # set these up in case we have an error case
    errors = {'username': cgi.escape(username), 'email': cgi.escape(email)}
    if validate_signup(username, password, verify, email, errors):

        if not users.add_user(username, password, email):
            # this was a duplicate
            errors['username_error'] = "Username already in use. Please choose another"
            return bottle.template("signup", errors)

        session_id = sessions.start_session(username)
        print session_id
        bottle.response.set_cookie("session", session_id)
        bottle.redirect("/welcome")
    else:
        print "user did not validate"
        return bottle.template("signup", errors)

def process_signup():

    email = bottle.request.forms.get("email")
    username = bottle.request.forms.get("username")
    password = bottle.request.forms.get("password")
    verify = bottle.request.forms.get("verify")

    # set these up in case we have an error case
    errors = {'username': cgi.escape(username), 'email': cgi.escape(email)}
    if validate_signup(username, password, verify, email, errors):

        if not users.add_user(username, password, email):
            # this was a duplicate
            errors['username_error'] = ("Username already in use. " +
                                        "Please choose another")
            return bottle.template("signup", errors)

        session_id = sessions.start_session(username)
        print session_id
        bottle.response.set_cookie("session", session_id)
        bottle.redirect("/welcome")
    else:
        print "user did not validate"
        return bottle.template("signup", errors)

def examples():
    http_host = bottle.request.environ.get('HTTP_HOST', 'localhost:3350')
    configs = restful_rfcat.example_configs.get(http_host)
    if is_cli():
        bottle.response.content_type = 'text/plain'
        lines = []
        for software, software_configs in configs.items():
            for variant, config in software_configs.items():
                lines.append("# %s - %s" % (software, variant))
                lines.append('')
                lines.append(config)
                lines.append('')
        return '\n'.join(lines)
    else:
        page = markup.page()
        page.init(script=['app.js'], css=['style.css'])
        for software, software_configs in configs.items():
            page.h2(software)
            for variant, config in software_configs.items():
                page.h3('%s - %s' % (software, variant))
                page.pre(config)
        return str(page)

def check_credentials():
    user = request.forms.get('user', '')
    password = request.forms.get('password', '')
    if not pubsub.check_user(user, password):
        return show_main_page()
    token = secrets.token_bytes(32)
    logged_in_users.setdefault(token, user)
    response.set_cookie('token', token, max_age=60, secret=secret)
    return show_main_page(user)

def get_logged_in_user():
    token = request.get_cookie('token', secret=secret)
    if token is not None:
        return logged_in_users.get(token)
    return None

def post_message():
    user = get_logged_in_user()
    if user is None:
        return template('login', null=None)
    text = request.forms.get('text', '')
    if text:
        pubsub.post_message(user, text)
    return show_main_page(user)

def show_search():
    user = get_logged_in_user()
    phrase = request.query.get('phrase', '')
    posts = pubsub.search(phrase, limit=10)
    heading = f'Posts matching: {phrase}'
    return dict(user=user, posts=posts, heading=heading, comb=comb)

def is_redis_available():
    try:
        rs.get(None)  # getting None returns None or throws an exception
    except (redis.exceptions.ConnectionError,
            redis.exceptions.BusyLoadingError):
        return False
    return True

def welcome():
    if request.params.get('req_thing') == "version_info":
        return version_info()

    username = request.get_cookie('username')
    password = request.get_cookie('password')
    password = urllib.unquote(password) if password else None

    if verify_login(username, password):
        redirect('/memorize')

    else:
        return dict(universal_ROUTE_dict)