Skip to content

Instantly share code, notes, and snippets.

@shihongzhi
Created May 26, 2012 03:00
Show Gist options
  • Save shihongzhi/2791877 to your computer and use it in GitHub Desktop.
Save shihongzhi/2791877 to your computer and use it in GitHub Desktop.
一道面试题,具体见我的博客http://shihongzhi.github.com/。这里使用bjoern作为WSGI server,从reqeust的‘QUERY_STRING’中得到json数据。这里的stategys和actions变量保存了需要处理的策略和行为。limit_time为限制次数。 本程序有个问题,当数据特别大的时候,datas这个list会特别大,导致性能下降,可以考虑用队列替换list,定时删除那些过时的json数据
# --- coding:utf-8 ---
import strategy
import bjoern
from urlparse import parse_qsl
strategys = [strategy.strategy_highfreq, strategy.strategy_repetition]
actions = ['answer', 'comment', 'question']
limit_time = 10
datas = []
def wsgi_app(env, start_response):
start_response('200 ok', [])
data = dict(parse_qsl(env.get('QUERY_STRING', ''))) #string to dict
datas.append(data)
for s in strategys:
for action in actions:
s(action, datas, limit_time)
return ['hello\n']
if __name__ == '__main__':
bjoern.run(wsgi_app, '0.0.0.0', 8082)
@shihongzhi
Copy link
Author

strategy.py
主要是antispam的strategy 策略逻辑代码。根据题目要求,这里实现了两个strategy

# --- coding:utf-8 ---

def strategy_highfreq(action, data, limit):
    '''
    data is a list.The index varaible is in order to simulate stack
    '''
    time = int(data[-1]['time'])
    user = data[-1]['user']
    count = 1
    limit_time = time - 60
    index = len(data) - 2
    while index >= 0:
        if data[index]['time'] > limit_time:
            if data[index]['user'] == user and data[index]['action'] == action:
                count += 1
                if count >= limit:
                    print u'warning: %s, 频繁提问'%user
                    return True
        else:
            break
        index -= 1
    return False

def strategy_repetition(action, data, limit):
    '''
    data is a list.The index varaible is in order to simulate stack
    '''
    user = data[-1]['user']
    content = data[-1]['content']
    count = 1
    index = len(data) - 2
    while index >= 0:
        if data[index]['user'] == user and data[index]['action'] == action and data[index]['content'] == content:
            count += 1
            if count >= limit:
                print u'warning: %s, 频繁提问'%user
                return True
        index -= 1
    return False

@shihongzhi
Copy link
Author

测试用的客户端,给服务器发送json数据。
本程序支持参数配置,分别为ip和测试文件(默认是testdata.txt)

# --- coding:utf-8 ---
import httplib
import urllib
import ast
import sys
import argparse

def client(ip, filename):
    headers = {'Content-type': 'application/x-www-form-urlencoded',
               'Accept': 'text/plain'}
    with open(filename, 'rb') as f:
        for line in f:
            data = ast.literal_eval(line) #string to dict
            value = urllib.urlencode(data)
            try:
                conn = httplib.HTTPConnection(ip)
                conn.request('POST', '/?%s'%value, '', headers)
                response = conn.getresponse()
                print response.status, response.reason
                conn.close
            except httplib.NonConnected:
                print 'Can not connect to server'
                sys.exit()
            except httplib.HTTPException:
                print 'http exception'
                sys.exit()


def main():
    parser = argparse.ArgumentParser(description='This is a client to request json data to server')
    parser.add_argument('-i', '--ip', default='127.0.0.1:8082', type=str, help="the server's ip:port")
    parser.add_argument('-f', '--filename', default='testdata.txt', type=str, help="the file that contain the json data")
    args = parser.parse_args()
    client(args.ip, args.filename)

if __name__ == '__main__':
    main()

@axiaoxin
Copy link

这个通过笔试了?

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment