项目要求
项目简介
要求客户在前端输入一个IP地址，点击查询按钮，前端不刷新(ajax)生成IP查询结果
除此之外，要求前端在界面载入的时候，请求后端获得今日查询次数，后端也是一样，每查询一次需要将今日查询次数+1
后端要求:
- 使用web.py和MYSQL进行开发
- IP地址需要校验合法性(1.1.1.1-255.255.255.255)
- 单个API查询出错时直接忽略，全部API都出错返回出错
- 需要储存每日查询的次数，每天一条记录
IP查询的在线API地址
- 百度 接口1
- 百度 接口2
- ipip.net 接口1
- ipip.net 接口2(注意: 需要附带头为 User-Agent: BestTrace/Windows V3.7.3)
- ipip.net 接口3
- 纯真IP数据库 接口
- ip138 网页版(注意: 需要自行取出对应的数据 仅需要取出 本站数据 即可 方式随意，建议使用正则)
- 淘宝 接口
- 以上API要求: 要求异步处理和超时处理（超时最高2s，淘宝的接口定为1s超时）
项目API
判断IP地址是否合法
判断IP是否合法采用正则的方法
import re
def check_ip(ip_):
compile_ip = re.compile('^(1\d{2}|2[0-4]\d|25[0-5]|[1-9]\d|[1-9])\.(1\d{2}|2[0-4]\d|25[0-5]|[1-9]\d|\d)\.(1\d{2}|2[0-4]\d|25[0-5]|[1-9]\d|\d)\.(1\d{2}|2[0-4]\d|25[0-5]|[1-9]\d|\d)$')
if compile_ip.match(ip_):
return True
else:
return False
API查询IP的地址
由于项目最开始并没有要求做异步处理，所以最开始采用的是requests的方法调用的API进行的查询，后期发现requests并难以进行异步处理以及代码的复用率较低等问题，对代码进行了重构，并选择了可以进行异步处理的aiohttp的方法，查询API的调用时间得到了明显的提高
requests方法
以调用百度API1为例，使用requests方法
timeout = 2
URL = "http://opendata.baidu.com/api.php?query=" + ip_ + "&co=&resource_id=6006&t=1433920989928&ie=utf8&oe=utf-8&format=json"
try:
r = requests.get(URL, timeout=timeout)
except requests.RequestException as E:
msg = "访问超时"
aiohttp方法
aiohttp的方法，由于需要进行异步处理，直接调用封装好的aiohttp即可。
async def fetch_async(url,params,timeout,headers,API):
async with aiohttp.ClientSession() as session: #协程嵌套，只需要处理最外层协程即可fetch_async
try:
async with session.get(url,params=params,timeout=timeout,headers=headers) as resp:
reponse = await resp.text()#因为这里使用到了await关键字，实现异步，所有他上面的函数体需要声明为异步async
return API,reponse
except:
print(API)
return API,"API访问错误"
URL = ['http://opendata.baidu.com/api.php?co=&resource_id=6006&t=1433920989928&ie=utf8&oe=utf-8&format=json',
'https://sp0.baidu.com/8aQDcjqpAAV3otqbppnN2DJv/api.php?co=&resource_id=6006&t=1433920989928&ie=utf8&oe=utf-8&format=json',
'http://freeapi.ipip.net/' + ip_,
'https://clientapi.ipip.net/free/ip?lang=CN',
'https://btapi.ipip.net/host/info?host=Router&lang=CN',
'http://ip.cz88.net/data.php',
'http://www.ip138.com/ips138.asp?action=2',
'http://ip.taobao.com/service/getIpInfo.php',
]
API = ["baidu_1", "baidu_2", "ipip_1", "ipip_2", "ipip_3", "chunzhen", "ip138", "taobao"]
tasks = [fetch_async(URL[0], {"query": ip_}, 2, {}, API[0]),
fetch_async(URL[1], {"query": ip_}, 2, {}, API[1]),
fetch_async(URL[2], {}, 2, {}, API[2]),
fetch_async(URL[3], {"ip": ip_}, 2, {'User-Agent': 'BestTrace/Windows V3.7.3'}, API[3]),
fetch_async(URL[4], {"ip": ip_}, 2, {}, API[4]),
fetch_async(URL[5], {"ip": ip_}, 2, {}, API[5]),
fetch_async(URL[6], {"ip": ip_}, 2, {'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/78.0.3887.7 Safari/537.36'},API[6]),
fetch_async(URL[7], {"ip": ip_}, 1, {}, API[7]),
]
event_loop = asyncio.new_event_loop()
asyncio.set_event_loop(event_loop)
#event_loop = asyncio.get_event_loop()
results = event_loop.run_until_complete(asyncio.gather(*tasks))
对返回的数据处理
由于所给的API有多中形式，所以使用不同的方式处理
1. json格式
对于json格式的数据，直接提取其中的有关于地址详细信息的字典即可，以下以百度接口1为例，其返回的形式为
{
"status": "0",
"t": "1433920989928",
"set_cache_time": "",
"data": [
{
"location": "四川省成都市 教育网",
"titlecont": "IP地址查询",
"origip": "180.85.12.2",
.....
"role_id": 0,
"disp_type": 0
}
]
}
所以只需要提取 [‘data’] 的 **[‘location’]**部分就可以了
# 百度接口1
def baidu_API_1(r):
data = {"source": "opendata.baidu.com"}
status = 0
msg = ""
json_data = json.loads(r)
if json_data['status'] == '0':
res = {}
for line in json_data['data']:
res.update(line)
status = 1
data = {"source": "opendata.baidu.com","location": res['location']}
return {"status": status,"msg": msg,"data": data}
2. 文本形式
对于文本形式就只能根据特点，从文本中选取需要的部分，以下以纯真IP数据库为例
其获得的数据为 ShowIPAddr(‘180.85.12.2’,‘四川省成都市 西南交通大学’,‘Linux Chrome 66.0.3359.126’); 需要提取 ‘四川省成都市 西南交通大学’ 这部分，代码如下：
# 纯真IP数据库 接口
def chunzhen_API(r):
status = 1
msg = ""
Str = r[11:-2]
count = 0
str = ''
for ch in Str:
if count == 3:
str = str + ch
if ch == "\'":
count = count + 1
elif count == 4:
str = str[:-1]
break
data = {"source": "ip.cz88.net", "location": str}
return {"status": status, "msg": msg, "data": data}
3. 网页形式
对于网页形式的，就需要在其中爬取信息了，推荐使用正则来爬取。
例如.ip138中的中需要提取的信息为下列中的 本站数据： 这部分的数据
<tdalign="center">
<ulclass="ul1">
<li>本站数据：四川省成都市西南交通大学教育网</li>
<li>参考数据1：四川成都西南交通大学教育网</li>
<li>参考数据2：四川省成都市西南交通大学</li>
<li>兼容IPv6地址：: : B455: 0C02</li>
<li>映射IPv6地址：: : FFFF: B455: 0C02</li>
</ul>
</td>
所以采用正则表达式 r"(?<=<li>).*?(?=</li>)" 来匹配信息
# ip138 网页版
def ip138_API(r):
status = 1
msg = ""
text = r
result_list = re.findall(r"(?<=<li>).*?(?=</li>)", text)
str = result_list[0][5:]
data = {"source": "www.ip138.com", "location": str}
return {"status": status, "msg": msg, "data": data}
数据库记录IP查询次数
数据库使用pymysql对数据库进行操作
封装数据库类
首先对数据库的操作进行封装，封装成一个数据库类，这样在对数据库进行操作的时候代码就显得比较简洁了。
class db_Connect:
#创建数据库
flag = 0
def __init__(self):
try:
self.db = pymysql.Connect(
host=dbconfig['host'],
user=dbconfig['user'],
passwd=dbconfig['passwd'],
db=dbconfig['db'],
charset=dbconfig['charset'],
)
except BaseException as err:
self.flag = 1
print("数据库连接失败！")
print(err)
#关闭数据库
def db_Close(self):
self.db.close()
def db_Create(self):
sql = """CREATE TABLE IF NOT EXISTS `IP_num`(
`Data` char(255),
`num` int,
PRIMARY KEY ( `Data` )
)"""
self.cursor.execute(sql)
def db_Insert(self,sql,Time):
self.cursor = self.db.cursor()
try:
self.cursor.execute(sql,(Time))
self.db.commit()
except:
# 发生错误时回滚
self.flag = 1
self.db.rollback()
print("数据库插入失败！")
finally:
self.cursor.close()
def db_Updata(self,sql,Time):
self.cursor = self.db.cursor()
try:
self.cursor.execute(sql,(Time))
self.db.commit()
except:
# 发生错误时回滚
self.flag = 1
self.db.rollback()
print('Error: unable to Updata data')
finally:
self.cursor.close()
def db_Query(self,sql,Time):
self.cursor = self.db.cursor()
try:
self.cursor.execute(sql,(Time)) # 返回 查询数据 条数 可以根据 返回值 判定处理结果
num = self.cursor.rowcount
if num == 0:
sql = "INSERT INTO IP_num(Data,num) VALUES (%s, 0)"
self.db_Insert(sql,Time)
return 0
else:
data = self.cursor.fetchall() # 返回所有记录列表
for row in data:
return row[1]
except BaseException as err:
print(err)
self.flag = 1
print('Error: unable to Query data')
finally:
self.cursor.close()
查询API调用次数
查询API调用次数，相对简单。
def config_QUERY(Time):
# 连接数据库
db = db_Connect()
# 查询数据
sql = "SELECT * FROM IP_num WHERE Data = %s"
num = db.db_Query(sql,Time)
#关闭数据库
db.db_Close()
if db.flag == 0:
return {'status': 0, 'msg': "", 'num': num}
else:
return {'status': -1, 'msg': "操作数据库失败"}
API调用次数+1
API调用次数+1操作同样简单，具体操作代码注释已经解释清楚了。
def config_ADD(Time):
# 连接数据库
db = db_Connect()
# 查询数据
sql = "SELECT * FROM IP_num WHERE Data = %s"
db.db_Query(sql,Time)
# 更新数据
sql = "UPDATE IP_num SET num=num+1 WHERE Data = %s"
db.db_Updata(sql,Time)
#关闭数据库
db.db_Close()
if db.flag == 0:
return True
else:
return False
web.py框架的搭建
web的框架采用的是web.py，可能并不完整，但API文档所需要的后端都已经实现。
urls = (
#今日查询数量接口
"/api/todaynum", "todaynum",
#IP查询文档
"/api/ip", "ip",
)
app = web.application(urls, globals())
class todaynum:
def GET(self):
Time = time.strftime("%D")
JSONDATA = config.config_QUERY(Time)
return json.dumps(JSONDATA)
class ip:
def POST(self):
data = web.data()
ip_ = json.loads(data)
print(ip_)
if legitimacy.check_ip(ip_["ip"]) == False:
JSONDATA = {"status": -1, "msg": "ip不合法"}
return json.dumps(JSONDATA)
else:
Time = time.strftime("%D")
config_Data = config.config_ADD(Time)
if config_Data == True:
JSONDATA = IP_query.Query(ip_["ip"])
return json.dumps(JSONDATA)
else:
return config_Data
if __name__ == "__main__":
app.run()
参考
版权声明：本文为Taunt_原创文章，遵循CC 4.0 BY-SA版权协议，转载请附上原文出处链接和本声明。