1

dingMessage.py

+import base64
+import hashlib
+import hmac
+import json
+import requests
+import time
+import urllib
+
+def send_messages(content):
+    webhook = 'https://oapi.dingtalk.com/robot/send?access_token='
+    headers = {"Content-Type": "application/json; charset=utf-8"}
+    token = 'fb8da62f0af606820fe8bdfdf8ea261d468f09b57c22faaa454d45855ea82eef'
+    secret = 'SECf2cc843306d667575a491650c777d8aa21788c4c60f3f0949d5d0b61c86e5aef'
+    data_info = {
+        "msgtype": "text",
+        "text": {
+            "content": content
+        },
+        "at": {'atMobiles': '15726940779'}
+    }
+    timestamp = str(round(time.time() * 1000))
+    # 密钥进行编码
+    secret_enc = secret.encode('utf-8')
+    # 把timestamp+"\n"+密钥当做签名字符串
+    string_to_sign = '{}\n{}'.format(timestamp, secret)
+    # 对签名进行编码
+    string_to_sign_enc = string_to_sign.encode('utf-8')
+    # 使用HmacSHA256算法计算签名，然后进行Base64 encode
+    hmac_code = hmac.new(secret_enc, string_to_sign_enc, digestmod=hashlib.sha256).digest()
+    sign = urllib.parse.quote_plus(base64.b64encode(hmac_code))
+    # 拼接完整的url，用于鉴权
+    fullUrl = webhook + token + "&timestamp=" + timestamp + "&sign=" + sign
+    # 发送钉钉消息
+    return requests.post(url=fullUrl, data=json.dumps(data_info), headers=headers).json()
\ No newline at end of file

ios.py

+from airtest.core.api import *
+
+
+from poco.drivers.ios import iosPoco
+
+
+
+class IOS():
+    def __init__(self, device):
+        self.device = connect_device(device)
+        self.Poco =  iosPoco()
+
+    def start_app(self):
+        return self.device.start_app()
\ No newline at end of file

log.py

+import logging
+
+import sys
+import time
+
+class LOG:
+    # 第一步，创建一个logger
+    logger = logging.getLogger()
+    logger.setLevel(logging.INFO)  # Log等级总开关
+    # 第二步，创建一个handler，用于写入日志文件
+    rq = time.strftime('%Y%m%d', time.localtime(time.time()))
+    log_path = str(sys.path[0]) + '/logs/'
+    log_name = log_path + rq + '.log'
+    logfile = log_name
+    fh = logging.FileHandler(logfile, mode='w')
+    fh.setLevel(logging.DEBUG)  # 输出到file的log等级的开关
+    # 第三步，定义handler的输出格式
+    formatter = logging.Formatter("%(asctime)s - %(filename)s[line:%(lineno)d] - %(levelname)s: %(message)s")
+    fh.setFormatter(formatter)
+    # 第四步，将logger添加到handler里面
+    logger.addHandler(fh)
+
+    @classmethod
+    def debug(cls, msg):
+        logging.debug(msg)
+
+    @classmethod
+    def info(cls, msg):
+        logging.info(msg)
+
+    @classmethod
+    def warning(cls, msg):
+        logging.warning(msg)
+
+    @classmethod
+    def error(cls, msg):
+        logging.error(msg)
+
+    @classmethod
+    def critical(cls, msg):
+        logging.critical(msg)
\ No newline at end of file

main.py

+# This is a sample Python script.
+
+# Press ⌃R to execute it or replace it with your code.
+# Press Double ⇧ to search everywhere for classes, files, tool windows, actions, and settings.
+
+
+def print_hi(name):
+    # Use a breakpoint in the code line below to debug your script.
+    print(f'Hi, {name}')  # Press ⌘F8 to toggle the breakpoint.
+
+
+# Press the green button in the gutter to run the script.
+if __name__ == '__main__':
+    print_hi('PyCharm')
+
+# See PyCharm help at https://www.jetbrains.com/help/pycharm/

manager.py

+from time import sleep
+
+import requests
+
+from log import LOG
+
+
+class Zk:
+    def __init__(self):
+        self.host = 'http://zhongkui.tuia.cn'
+        # 获取有效可用的域名,ios支付宝域名巡查使用
+        self.valid_list_url = '/domain/queryValidDomainList?platformType=3'
+        # 上报被封域名列表,ios支付宝域名巡查使用
+        self.block_list_url = '/domain/receiveBlockedDomain'
+
+        self.session = requests.session()
+
+    def get_valid_list(self):
+        LOG.info('开始获取巡查列表')
+        r = self.session.get(url=self.host + self.valid_list_url)
+        if r.status_code == 200 and r.json().get('success') == True:
+            valid_list = r.json().get('data').get('domains')
+            if valid_list:
+                LOG.info('获取巡查列表结果：'+str(valid_list))
+                return valid_list
+        else:
+            LOG.error('获取巡查列表异常，repose：'+r.text)
+
+    def post_block_list(self, b_list):
+        LOG.info('上报被封域名列表：'+str(b_list))
+        data = {'platformType': 3, 'domains': b_list}
+        r = self.session.post(url=self.host+self.block_list_url,json=data)
+        if r.status_code == 200 and r.json().get('success') == True:
+            LOG.info('上报成功：'+ r.text)
+        else:
+            LOG.error('上报失败，1.5S后重新上报：' + r.text)
+
+
+

requirements.txt

+airtest==1.1.8
+pocoui==1.0.81
+requests==2.25.0
+poco==0.98.1

ssp.py

+from time import sleep
+
+import requests
+
+from log import LOG
+
+
+class SSP:
+    def __init__(self):
+        self.host = 'https://ssp-manager.tuia.cn'
+        # 获取待巡查列表
+        self.valid_list_url = '/wchatDomain/wechatScan/patrolListing'
+        # 上报被封域名列表,ios支付宝域名巡查使用
+        self.block_list_url = '/wchatDomain/wechatScan/patrolResulting?shieldDomain='
+
+        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):
+        LOG.info('SSP上报被封域名：'+b_url)
+
+        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') == "成功":
+            LOG.info('上报成功：'+ r.text)
+            LOG.info(r.url)
+        else:
+            LOG.error('上报失败，1.5S后重新上报：' + r.text)
+
+
+if __name__ == '__main__':
+    ssp = SSP()
+    ssp.get_valid_list()
+
+
+
+

st.sh

+cd /Users/a1/iOS-Tagent-master
+# 获取设备的UDID，用到了之前的 libimobiledevice
+UDID=$(idevice_id -l | head -n1)
+# 真机运行测试
+nohup xcodebuild -project WebDriverAgent.xcodeproj -scheme WebDriverAgentRunner -destination "id=$UDID" test >/dev/null 2>&1 &
+
+cd /Users/yaozheng/PycharmProjects/iosInspection
+nohup python3 starter.py >/dev/null 2>&1 &
+
+ps -ef | grep starter.py | grep -v grep | awk '{print $2}' | xargs kill -9
+

starter.py

+import urllib.parse
+import urllib
+from airtest.core.api import *
+from poco.drivers.ios import iosPoco
+
+from dingMessage import send_messages
+from log import LOG
+
+from ssp import SSP
+
+
+
+# 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 start():
+
+    device.stop_app('com.tencent.xin')
+    LOG.info('打开微信')
+    device.start_app('com.tencent.xin')
+    check_wx()
+    t1 = time.time()
+    LOG.info('打开扫一扫，选择二维码')
+    poco("快捷操作").click()
+    poco("扫一扫").click()
+    touch(Template(r"tpl1644909001064.png", record_pos=(0.396, 0.696), resolution=(828, 1792)))
+    poco("照片14,拍摄时间,2022年2月15日 下午2:21").click()
+    poco(name="完成").click()
+    LOG.info('扫码完成，开始进行巡查')
+    links = poco(type="Link")
+    for link in links:
+        link.click()
+        LOG.info('开始巡查：'+link.get_name())
+        sleep(1)
+        if poco(name="已停止访问该网页").exists():
+            print("上报被封域名")
+            ssp.post_block_list(link.get_name())
+        poco(name="返回").click()
+    t2 = time.time()
+
+    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():
+    LOG.info('判断微信是否异常')
+    while poco(name="下一步").exists():
+        LOG.info('微信异常，点击下一步')
+        poco(name="下一步").click()
+    if poco(name="进入微信").exists():
+        LOG.info('点击进入微信')
+        poco(name="进入微信").click()
+
+if __name__ == '__main__':
+
+    ssp = SSP()
+    valid_list = []
+    try:
+         valid_list.extend(ssp.get_valid_list())
+    except:
+        send_messages('获取待巡查列表异常！')
+        LOG.error('获取待巡查列表异常！')
+    if valid_list:
+        LOG.info('开始连接设备')
+        device = connect_device('ios:///http://127.0.0.1:8100')
+        LOG.info('连接成功')
+        poco = iosPoco()
+        while True:
+            try:
+                start()
+            except:
+                LOG.error('域名巡查脚本报错')
+                # re_device()
+                sleep(240)
+    else:
+        LOG.info('巡查列表为空！')