背景

因为公司有台 MacBookPro 笔记本跑一些自动化脚本,但是公司的网络经常会自动断开,平均每两天会自动断开一次,只能手动重新连接网络.

解决目标

主要想通过自动脚本自动连接网络并且形成网络巡检工具.

核心难点是,如何通过命令链接网络.

Mac OSX 命令行下控制 Wifi 命令

Mac 命令行下查看当前 Wifi 网络设备名称

networksetup -listallhardwareports

可以看出相关网卡和蓝牙信息.

启动 Wifi

networksetup -setairportpower en0 on

关闭 Wifi

networksetup -setairportpower en0 off

加入 Wifi

networksetup -setairportnetwork en0  WIFI_SSID_I_WANT_TO_JOIN  WIFI_PASSWORD

WIFI_SSID_I_WANT_TO_JOIN 是 WIFI 的名字.
WIFI_PASSWORD 是 WIFI 的密码.

执行命令后,WIFI 就会自动恢复链接,大概 5 秒左右.

定时监控脚本

脚本逻辑是: 请求百度共 5 次并且每次请求间隔 2 秒,如果出现连接异常,则重新连接网络.

import socket
import fcntl
import struct
import os
import time
import socket
import subprocess
import random
import traceback
import requests
from retrying import retry
from datetime import datetime, timedelta
from apscheduler.schedulers.blocking import BlockingScheduler

request_count = 0
wait_wifi_time = 60
request_url = 'https://www.baidu.com'
wifi_name = 'xxxxxx'
wifi_password = ''xxxxxx'

def get_host_ip():
    """
    查询本机ip地址
    :return:
    """
    try:
        s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
        s.connect(('8.8.8.8', 80))
        ip = s.getsockname()[0]
    finally:
        s.close()
    return ip


def send_notice(type):
    try:
        task_start_time = time.strftime("%Y%m%d%H%M%S", time.localtime())
        print(task_start_time)
        host_ip = get_host_ip()
        content = '自动化检查网络任务完成\n本机IP是:{}\n巡检类型:{}'.format(host_ip, type)
        print('send noitce to xinxi')
        webhook_api = "xxxxxx"
        data = {
            "msgtype": "markdown",
            "markdown": {
                "content": content
            }
        }
        r = requests.post(webhook_api, json=data, verify=False)
        print(r.json())
    except Exception as e:
        print(e)


def _result(result):
    return result is None


def header(header):
    try:
        if header != None:
            header['User-Agent'] = random.choice(ag)
        else:
            header = {'User-Agent': random.choice(ag)}
        return header
    except Exception as e:
        traceback.print_exc(e)


@retry(stop_max_attempt_number=5, wait_random_min=1000, wait_random_max=2000, retry_on_result=_result)
def my_request_get(url, headers=None):
    global request_count
    request_count = request_count + 1
    print('Request Count Is: {}'.format(request_count))
    response = requests.get(url, timeout=6)
    if response.status_code != 200:
        raise requests.RequestException('my_request_get error!!!!')
    return response



def open_wifi():
    global request_url
    cmd = 'networksetup -setairportpower en0 on'
    subprocess.call(cmd, shell=True)
    cmd = 'networksetup -setairportnetwork en0  {wifi_name} {wifi_password}'.format(wifi_name=wifi_name,
                                                                                    wifi_password=wifi_password)
    subprocess.call(cmd, shell=True)
    print('open Wifi after {}s'.format(wait_wifi_time))
    time.sleep(wait_wifi_time)
    try:
        r = requests.get(request_url, timeout=6)
        if r.status_code == 200:
            send_notice('重启')
    except Exception as e:
        print(e)
        open_wifi()

class PingObject():

    def job():
        try:
            print(request_url)
            if my_request_get(request_url).status_code == 200:
                send_notice('正常')
        except Exception as e:
            print(e)
            open_wifi()



if __name__ == '__main__':

    scheduler = BlockingScheduler()
    scheduler.add_job(PingObject.job, trigger='interval', minutes=30, id='my_job_id_test', next_run_time=datetime.now())

    scheduler.start()

    print('Press Ctrl+{0} to exit'.format('Break' if os.name == 'nt' else 'C'))

    try:
        while True:
            time.sleep(2)  # 其他任务是独立的线程执行
    except (KeyboardInterrupt, SystemExit):
        scheduler.shutdown()
        print('Exit The Job!')

脚本可以放到终端前台或者后台执行即可.

每 30 分钟自动监控一次,自动报警通知如下.

参考

Mac OSX 命令行下控制 Wifi 命令

https://blog.csdn.net/cn_wk/article/details/51712118