Files
aaPanel/class_v2/safeModelV2/freeipModel.py
T
Jack a24f2e23c6 Update to 7.25.0
1. Added Mail Marketing Automation trigger tasks
2. Added Mail Marketing Groups Import, Export, Merge
3. Added Mail Marketing Subscribers to support paste import
4. Added Mail Marketing Template Import Export Duplicate
5. Added Mail server add timed automatic reply
6. Added HTTPS Protection function for Website (disables automatic HTTP to HTTPS redirection when enabled)
7. Optimize Website Interface Button Integration
8. Optimize the response speed of WP Toolkit interface
2025-03-13 17:52:26 +08:00

99 lines
3.0 KiB
Python

#coding: utf-8
#-------------------------------------------------------------------
# aaPanel
#-------------------------------------------------------------------
# Copyright (c) 2015-2099 aaPanel(www.aapanel.com) All rights reserved.
#-------------------------------------------------------------------
# Author: cjxin <cjxin@aapanel.com>
#-------------------------------------------------------------------
# 免费IP库
#------------------------------
import os,re,json,time
from safeModelV2.base import safeBase
import public
class main(safeBase):
_sfile = '{}/data/free_ip_area.json'.format(public.get_panel_path())
def __init__(self):
try:
self.user_info = public.get_user_info()
except:
self.user_info = None
def get_ip_area(self,get):
"""
@获取IP地址所在地
@param get: dict/array
"""
ips = get['ips']
arrs,result = [],{}
for ip in ips:arrs.append(ip)
if len(arrs) > 0:
data = self.__get_cloud_ip_info(arrs)
for ip in data:
result[ip] = data[ip]
return result
def __get_cloud_ip_info(self,ips):
"""
@获取IP地址所在地
@得判断是否是我们的用户
@param ips:
"""
result = {}
try:
'''
@从云端获取IP地址所在地
@param data 是否是宝塔用户,如果不是则不返回
@param ips: IP地址
'''
data = {}
data['ip'] = ','.join(ips)
data['uid'] = self.user_info['uid']
# 与面板字段差异
data["serverid"]=self.user_info["server_id"]
#如果不是我们的用户,那么不返回数据
res = public.httpPost('https://wafapi2.aapanel.com/api/ip/info',data)
res = json.loads(res)
data = self.get_ip_area_cache()
for key in res:
info = res[key]
if public.is_local_ip(key):
res[key]['city']="Intranet"
if not res[key]['city']: continue
if not res[key]['city'].strip() and not res[key]['continent'].strip():
info = {'info':'Unknown IP'}
else:
info['info'] = '{} {} {} {}'.format(info['carrier'],info['country'],info['province'],info['city']).strip()
data[key] = info
result[key] = info
self.set_ip_area_cache(data)
except:
pass
return result
def get_ip_area_cache(self):
"""
@获取IP地址所在地
@param get:
"""
data = {}
try:
data = json.loads(public.readFile(self._sfile))
except:
public.writeFile(self._sfile,json.dumps({}))
return data
def set_ip_area_cache(self,data):
"""
@设置IP地址所在地
@param data:
"""
public.writeFile(self._sfile,json.dumps(data))
return True