Skip to content

W
wxScanck
Commit 436f5b8c authored by TesterYao's avatar TesterYao
Browse files
Options

1

parent 23e625e6
Show whitespace changes
Inline Side-by-side
Showing with 69 additions and 144 deletions
ssp.py
View file @ 436f5b8c
...@@ -8,40 +8,32 @@ from log import LOG ...@@ -8,40 +8,32 @@ from log import LOG
class SSP: class SSP:
def __init__(self): def __init__(self):
self.host = 'https://ssp-manager.tuia.cn' self.host = 'https://ssp-manager.tuia.cn'
# 获取待巡查列表 # ssp上报被封域名列表,微信扫码巡查使用
self.valid_list_url = '/wchatDomain/wechatScan/patrolListing'
# 上报被封域名列表,ios支付宝域名巡查使用
self.block_list_url = '/wchatDomain/wechatScan/patrolResulting?shieldDomain=' self.block_list_url = '/wchatDomain/wechatScan/patrolResulting?shieldDomain='
# 盘古上报被封域名列表,微信扫码巡查使用
self.block_pg_list_url = 'https://ssp-manager.tuia.cn/wchatDomain/wechatScan/sendDingDing'
self.session = requests.session() self.session = requests.session()
def get_valid_list(self):
LOG.info('开始获取巡查列表')
r = self.session.get(url=self.host + self.valid_list_url)
# print(r.text)
if r.status_code == 200 and r.json().get('desc') == "成功":
valid_list = r.json().get('data')
# print(valid_list)
if valid_list:
LOG.info('获取巡查列表结果：'+str(valid_list))
return valid_list
else:
LOG.error('获取巡查列表异常，repose：'+r.text)
def post_block_list(self, b_url): def post_block_list(self, b_url):
LOG.info('SSP上报被封域名：'+b_url) LOG.info('SSP上报被封域名：'+b_url)
r = self.session.get(url=self.host+self.block_list_url+b_url.split('https://')[1]+'&resulting=1') r = self.session.get(url=self.host+self.block_list_url+b_url.split('https://')[1]+'&resulting=1')
if r.status_code == 200 and r.json().get('desc') == "成功": if r.status_code == 200 and r.json().get('desc') == "成功":
LOG.info('上报成功：'+ r.text) LOG.info('上报成功：'+ r.text)
LOG.info(r.url)
else: else:
LOG.error('上报失败，1.5S后重新上报：' + r.text) LOG.error('上报失败，1.5S后重新上报：' + r.text)
def post_pg_block_list(self, b_url):
if __name__ == '__main__': LOG.info('盘古上报被封域名：'+b_url)
ssp = SSP() data = {
ssp.get_valid_list() 'domain':b_url,
'type':'1'
}
r = self.session.post(url=self.block_pg_list_url,json=data)
print(r.text)
if r.status_code == 200 and r.json().get('desc') == "成功":
LOG.info('上报成功：'+ r.text)
else:
LOG.error('上报失败，1.5S后重新上报：' + r.text)
......
starter.py
View file @ 436f5b8c
import urllib.parse
import urllib
from airtest.core.api import * from airtest.core.api import *
from poco.drivers.ios import iosPoco from poco.drivers.ios import iosPoco
from dingMessage import send_messages
from log import LOG from log import LOG
from ssp import SSP from ssp import SSP
def check_ex():
LOG.info('判断是否在微信首页')
if not poco("快捷操作").exists():
LOG.info('不在微信首页，重新打开微信')
device.stop_app('com.tencent.xin')
device.start_app('com.tencent.xin')
# def runner(url, channel,i=0):
# base_url = 'alipays://platformapi/startApp?appId=20000067&url='
# if url.startswith('http'):
# f_url = base_url + url
# else:
# f_url = base_url + 'http://' + url
# try:
# if device.is_locked():
# device.unlock()
# LOG.info('关闭浏览器')
# device.stop_app('com.apple.mobilesafari')
# # device.stop_app('com.ucweb.iphone.lowersion')
# LOG.info('关闭浏览器成功')
# sleep(3)
# device.start_app('com.apple.mobilesafari')
# # device.start_app('com.ucweb.iphone.lowersion')
# LOG.info('浏览器缓存重启完成,开始删除标签页，打开新的无痕窗口')
# sleep(1)
# poco("TabsButton").click()
# poco("closeTabButton").click()
# poco("TabViewDoneButton").click()
# LOG.info('无痕窗口打开完成，开始输入url')
# poco("URL").click()
# poco("URL")
# device.text(f_url)
# sleep(1)
# try:
# poco("打开").click()
# except:
# sleep(1)
# LOG.info('点击【打开】异常，开始输入url')
# poco("打开").click()
# sleep(2)
# if poco(name="已停止访问该网页").exists():
# LOG.info('巡查结束，域名被封：%s' % url)
# if isinstance(channel,SSP):
# ssp.post_block_list(url)
# return url
#
# else:
# LOG.info('巡查结束，域名正常：%s' % url.split('url=http://')[1])
# except:
# if i < 5:
# i += 1
# LOG.error('支付宝ios域名巡查,执行巡查时脚本出错,正在进行第' + str(i) + '次重试')
# device.snapshot()
# runner(url, i)
# else:
# raise Exception(print('重试5次后抛出异常'))
# def inspection(valid_list):
#
# block_list = []
# for xc in valid_list:
# l_xc = urllib.parse.unquote(xc)
#
# LOG.info('开始巡查，域名为：%s' % l_xc)
# block = runner(l_xc)
# if block:
# block_list.append(xc)
# return block_list
# 出现异常时，此方法重新运行测试脚本
# def re_device():
# LOG.error('开始运行WDA')
# send_messages('支付宝巡查任务异常退出，请及时处理！！！')
# os.system('/st.sh')
def check_wx():
LOG.info('判断微信是否异常')
while poco(name="下一步").exists():
LOG.info('微信异常，点击下一步')
poco(name="下一步").click()
if poco(name="进入微信").exists():
LOG.info('点击进入微信')
poco(name="进入微信").click()
def start():
device.stop_app('com.tencent.xin') def scan(photo):
LOG.info('打开微信') block_list = []
device.start_app('com.tencent.xin')
check_wx()
t1 = time.time()
LOG.info('打开扫一扫，选择二维码') LOG.info('打开扫一扫，选择二维码')
poco("快捷操作").click() poco("快捷操作").click()
poco("扫一扫").click() poco("扫一扫").click()
touch(Template(r"tpl1644909001064.png", record_pos=(0.396, 0.696), resolution=(828, 1792))) touch(Template(r"tpl1644909001064.png", record_pos=(0.396, 0.696), resolution=(828, 1792)))
poco("照片14,拍摄时间,2022年2月15日 下午2:21").click() poco(photo).click()
poco(name="完成").click() poco(name="完成").click()
LOG.info('扫码完成，开始进行巡查') LOG.info('扫码完成，开始进行巡查')
links = poco(type="Link") links = poco(type="Link")
for link in links: for link in links:
link.click() link.click()
LOG.info('开始巡查：'+link.get_name()) LOG.info('开始巡查：' + link.get_name())
sleep(1) sleep(1)
if poco(name="已停止访问该网页").exists(): if poco(name="已停止访问该网页").exists():
print("上报被封域名") block_list.append(link.get_name())
ssp.post_block_list(link.get_name())
poco(name="返回").click() poco(name="返回").click()
t2 = time.time() LOG.info('巡查结束，关闭扫码webview！')
poco(name="关闭").click()
return block_list
print("巡查结束，总耗时："+str(t2-t1))
# if block_list:
# try:
# # ssp.post_block_list(block_list)
# send_messages(ssp.__class__.__name__ + '上报微信扫码被封域名成功：' + str(block_list))
# except:
# send_messages('上报被封域名失败!')
# LOG.error('上报被封域名失败!')
def check_wx(): def check_ssp():
LOG.info('判断微信是否异常') LOG.info('巡查ssp')
while poco(name="下一步").exists(): b_list = scan("照片15,拍摄时间,2022年2月17日 下午4:47")
LOG.info('微信异常，点击下一步') if b_list:
poco(name="下一步").click() LOG.info('巡查ssp结束，被封列表：' + str(b_list))
if poco(name="进入微信").exists(): for domain in b_list:
LOG.info('点击进入微信') ssp.post_block_list(domain)
poco(name="进入微信").click()
def check_pg():
LOG.info('巡查盘古')
b_list = scan("照片16,拍摄时间,2022年2月17日 下午4:47")
if b_list:
LOG.info('巡查盘古结束，被封列表：'+str(b_list))
for domain in b_list:
ssp.post_block_list(domain)
if __name__ == '__main__': if __name__ == '__main__':
ssp = SSP() ssp = SSP()
valid_list = []
try:
valid_list.extend(ssp.get_valid_list())
except:
send_messages('获取待巡查列表异常！')
LOG.error('获取待巡查列表异常！')
if valid_list:
LOG.info('开始连接设备') LOG.info('开始连接设备')
device = connect_device('ios:///http://127.0.0.1:8100') device = connect_device('ios:///http://127.0.0.1:8100')
LOG.info('连接成功') LOG.info('连接成功')
poco = iosPoco() poco = iosPoco()
while True: while True:
try: try:
start() check_ex()
except: check_wx()
LOG.error('域名巡查脚本报错') check_ssp()
# re_device() check_pg()
sleep(240) except Exception as e:
else: print(e)
LOG.info('巡查列表为空！') LOG.error('域名巡查脚本报错,三分钟后重试！')
sleep(180)
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment