This commit is contained in:
IrvingGao 2024-05-20 23:07:29 +08:00
parent 6ce468850c
commit 155ed8b46a
2 changed files with 225 additions and 43 deletions

169
wifi_manager.py Normal file
View File

@ -0,0 +1,169 @@
# 逻辑:
# 1. 检测 Wi-Fi 连接状态,若已连接,则退出程序
# 2. 扫描 Wi-Fi 网络并获取Wi-Fi列表
# 3. 关闭Wi-Fi并启动热点和web服务器
# 4. 等待用户选择 Wi-Fi 网络并输入密码
# 5. 关闭热点并重新扫描 Wi-Fi连接到用户选择的 Wi-Fi 网络
import subprocess
import re
import os
import time
from flask import Flask, render_template, request, redirect, url_for, make_response
app = Flask(__name__)
# 检测 Wi-Fi 连接状态
def check_wifi_connection():
try:
# 使用nmcli命令检查Wi-Fi连接状态
output = subprocess.check_output(['nmcli', 'device', 'status']).decode('utf-8')
# 分析输出以查找Wi-Fi连接状态
for line in output.split('\n'):
if 'wifi' in line and 'connected' in line:
return True
except subprocess.CalledProcessError as e:
print(f"Error checking Wi-Fi status: {e}")
return False
def restart_hotspot():
try:
# sudo systemctl stop dnsmasq
# sudo systemctl restart NetworkManager
# sudo nmcli radio wifi off
# sudo nmcli radio wifi on
# nohup sudo create_ap wlan0 eth0 Takway-Toys --no-virt > create_ap.log 2>&1 &
subprocess.run(['sudo', 'systemctl', 'stop', 'dnsmasq'], check=True)
subprocess.run(['sudo', 'systemctl', 'restart', 'NetworkManager'], check=True)
subprocess.run(['sudo', 'nmcli', 'radio', 'wifi', 'off'], check=True)
subprocess.run(['sudo', 'nmcli', 'radio', 'wifi', 'on'], check=True)
subprocess.Popen('sudo nohup create_ap wlan0 eth0 Takway-Toys --no-virt > create_ap.log 2>&1 &', shell=True)
except subprocess.CalledProcessError as e:
print(f"Error setting up hotspot: {e}")
def scan_wifi():
try:
subprocess.run(['nmcli', 'dev', 'wifi', 'rescan'], check=True)
time.sleep(1)
output = subprocess.check_output(['nmcli', 'dev', 'wifi'])
wifi_list = []
lines = output.decode().splitlines()
for idx, line in enumerate(lines[1:]):
print(f"{idx}: {line}")
# 分割字符串以空格为分隔符
columns = line.split()
# 提取第一列和第三列的值
mac_address = columns[0]
ssid = " ".join(columns[1:3])
if "Infra" in ssid:
ssid = ssid.replace(" Infra", "")
if ":" in ssid:
# "72:A6:CC:8C:89:6C Takway-AI"
# 去除字符串中的MAC地址
ssid = re.sub(r'\b\w{2}(:\w{2}){5}\b', '', ssid)
# 提取第六列的强度值
last_info = columns[4:8]
# 去除所有的空格
columns = line.replace(" ", "")
# 提取"Mbit/s"后的三位以内数字
strength = int(columns[columns.index("Mbit/s")+6:columns.index("Mbit/s")+8])
# print("MAC地址:", mac_address)
# print("SSID:", ssid)
# print("强度:", strength)
wifi_list.append({'ssid': ssid, 'signal': strength})
if len(wifi_list) == 15:
break
return wifi_list
except subprocess.CalledProcessError as e:
print(f"Error scanning for Wi-Fi networks: {e}")
return []
# 连接 Wi-Fi
def connect_wifi(ssid, password):
# 关闭热点
try:
subprocess.run(['sudo', 'create_ap', '--stop', 'wlan0'], check=True)
print("Stopping create_ap service")
except subprocess.CalledProcessError as e:
print(f"Error stopping hotspot: {e}")
time.sleep(1)
# 扫描 Wi-Fi 网络
# scan_wifi()
time.sleep(1)
# 连接到用户选择的 Wi-Fi 网络
try:
subprocess.run(['nmcli', 'dev', 'wifi', 'connect', ssid, 'password', password], check=True)
print(f"Successfully connected to Wi-Fi: {ssid}")
return True
except subprocess.CalledProcessError as e:
print(f"Error connecting to Wi-Fi: {e}")
return False
# 主页
@app.route('/')
def index():
global wifi_list
print(wifi_list)
response = make_response(render_template('index.html', wifi_list=wifi_list))
response.headers.set('Content-Type', 'text/html')
response.headers.set('Apple-Web-App-Capable', 'yes')
response.headers.set('Apple-Mobile-Web-App-Status-Bar-Style', 'black-translucent')
return response
# 提交 Wi-Fi 信息
@app.route('/submit', methods=['POST'])
def submit():
ssid = request.form['ssid']
password = request.form['password']
print(f"Connecting to Wi-Fi: {ssid} with password {password}")
connected = connect_wifi(ssid, password)
time.sleep(5)
# 检查是否成功连接Wi-Fi
if check_wifi_connection() and connected:
os._exit(0) # 成功连接后退出程序
else:
print("Wi-Fi连接失败。")
return redirect(url_for('index'))
if __name__ == '__main__':
debug_mode = False # 设置为 True 以跳过 Wi-Fi 连接状态检测
if not debug_mode:
if check_wifi_connection():
print("已连接到 Wi-Fi 网络")
else:
print("未连接到 Wi-Fi 网络")
print("Starting Flask server")
print("----------------------------")
wifi_list = scan_wifi()
print("----------------------------")
# save wifi_list to file
with open('wifi_list.txt', 'w') as f:
for wifi in wifi_list:
f.write(f"{wifi['ssid']},{wifi['signal']}\n")
restart_hotspot()
# 启动 Flask 服务器
app.run(host='0.0.0.0', port=80)

View File

@ -1,4 +1,12 @@
# 逻辑:
# 1. 检测 Wi-Fi 连接状态,若已连接,则退出程序
# 2. 扫描 Wi-Fi 网络并获取Wi-Fi列表
# 3. 关闭Wi-Fi并启动热点和web服务器
# 4. 等待用户选择 Wi-Fi 网络并输入密码
# 5. 关闭热点并重新扫描 Wi-Fi连接到用户选择的 Wi-Fi 网络
import subprocess import subprocess
import re
import os import os
import time import time
from flask import Flask, render_template, request, redirect, url_for, make_response from flask import Flask, render_template, request, redirect, url_for, make_response
@ -20,13 +28,22 @@ def check_wifi_connection():
return False return False
# 设置热点 def restart_hotspot():
def set_hotspot():
try: try:
subprocess.run(['sudo', 'create_ap', 'wlan0', 'eth0', 'Takway-Toy', '--no-virt'], check=True) # sudo systemctl stop dnsmasq
# sudo systemctl restart NetworkManager
# sudo nmcli radio wifi off
# sudo nmcli radio wifi on
# nohup sudo create_ap wlan0 eth0 Takway-Toys --no-virt > create_ap.log 2>&1 &
subprocess.run(['sudo', 'systemctl', 'stop', 'dnsmasq'], check=True)
subprocess.run(['sudo', 'systemctl', 'restart', 'NetworkManager'], check=True)
subprocess.run(['sudo', 'nmcli', 'radio', 'wifi', 'off'], check=True)
subprocess.run(['sudo', 'nmcli', 'radio', 'wifi', 'on'], check=True)
subprocess.Popen('sudo nohup create_ap wlan0 eth0 Takway-Toys --no-virt > create_ap.log 2>&1 &', shell=True)
except subprocess.CalledProcessError as e: except subprocess.CalledProcessError as e:
print(f"Error setting up hotspot: {e}") print(f"Error setting up hotspot: {e}")
def scan_wifi(): def scan_wifi():
try: try:
subprocess.run(['nmcli', 'dev', 'wifi', 'rescan'], check=True) subprocess.run(['nmcli', 'dev', 'wifi', 'rescan'], check=True)
@ -46,14 +63,19 @@ def scan_wifi():
ssid = " ".join(columns[1:3]) ssid = " ".join(columns[1:3])
if "Infra" in ssid: if "Infra" in ssid:
ssid = ssid.replace(" Infra", "") ssid = ssid.replace(" Infra", "")
if ":" in ssid:
# "72:A6:CC:8C:89:6C Takway-AI"
# 去除字符串中的MAC地址
ssid = re.sub(r'\b\w{2}(:\w{2}){5}\b', '', ssid)
# 提取第六列的强度值 # 提取第六列的强度值
last_info = columns[4:8] last_info = columns[4:8]
# 去除所有的空格 # 去除所有的空格
columns = line.replace(" ", "") columns = line.replace(" ", "")
# 提取"Mbit/s"后两位数字 # 提取"Mbit/s"后的三位以内数字
strength = int(columns[columns.index("Mbit/s")+6:columns.index("Mbit/s")+8]) strength = int(columns[columns.index("Mbit/s")+6:columns.index("Mbit/s")+8])
# print("MAC地址:", mac_address) # print("MAC地址:", mac_address)
# print("SSID:", ssid) # print("SSID:", ssid)
# print("强度:", strength) # print("强度:", strength)
@ -67,31 +89,10 @@ def scan_wifi():
print(f"Error scanning for Wi-Fi networks: {e}") print(f"Error scanning for Wi-Fi networks: {e}")
return [] return []
def scan_wifi_table():
import csv
import subprocess
from io import StringIO
# 调用 nmcli dev wifi 命令并获取输出
output = subprocess.check_output(['nmcli', 'dev', 'wifi']).decode()
# 创建一个内存中的文件对象
file_object = StringIO(output)
# 使用csv模块读取文件对象并指定分隔符为whitespace
reader = csv.reader(file_object, delimiter=' ')
# 将数据存储到二维列表中
data = [row for row in reader]
# 打印表格数据
for row in data:
print(row)
# 连接 Wi-Fi # 连接 Wi-Fi
def connect_wifi(ssid, password): def connect_wifi(ssid, password):
scan_wifi() # 关闭热点
time.sleep(1)
try: try:
subprocess.run(['sudo', 'create_ap', '--stop', 'wlan0'], check=True) subprocess.run(['sudo', 'create_ap', '--stop', 'wlan0'], check=True)
print("Stopping create_ap service") print("Stopping create_ap service")
@ -100,23 +101,26 @@ def connect_wifi(ssid, password):
time.sleep(1) time.sleep(1)
# 扫描 Wi-Fi 网络
# scan_wifi()
time.sleep(1)
# 连接到用户选择的 Wi-Fi 网络
try: try:
subprocess.run(['nmcli', 'dev', 'wifi', 'connect', ssid, 'password', password], check=True) subprocess.run(['nmcli', 'dev', 'wifi', 'connect', ssid, 'password', password], check=True)
print(f"Successfully connected to Wi-Fi: {ssid}")
return True
except subprocess.CalledProcessError as e: except subprocess.CalledProcessError as e:
print(f"Error connecting to Wi-Fi: {e}") print(f"Error connecting to Wi-Fi: {e}")
return False
time.sleep(5)
# 检查是否成功连接Wi-Fi
if check_wifi_connection():
print("成功连接到Wi-Fi程序即将退出。")
os._exit(0) # 成功连接后退出程序
else:
print("Wi-Fi连接失败。")
# 主页 # 主页
@app.route('/') @app.route('/')
def index(): def index():
wifi_list = scan_wifi() global wifi_list
print(wifi_list)
response = make_response(render_template('index.html', wifi_list=wifi_list)) response = make_response(render_template('index.html', wifi_list=wifi_list))
response.headers.set('Content-Type', 'text/html') response.headers.set('Content-Type', 'text/html')
response.headers.set('Apple-Web-App-Capable', 'yes') response.headers.set('Apple-Web-App-Capable', 'yes')
@ -129,8 +133,15 @@ def submit():
ssid = request.form['ssid'] ssid = request.form['ssid']
password = request.form['password'] password = request.form['password']
print(f"Connecting to Wi-Fi: {ssid} with password {password}") print(f"Connecting to Wi-Fi: {ssid} with password {password}")
connect_wifi(ssid, password) connected = connect_wifi(ssid, password)
stop_hotspot() # 关闭热点
time.sleep(5)
# 检查是否成功连接Wi-Fi
if check_wifi_connection() and connected:
os._exit(0) # 成功连接后退出程序
else:
print("Wi-Fi连接失败。")
return redirect(url_for('index')) return redirect(url_for('index'))
@ -142,15 +153,17 @@ if __name__ == '__main__':
print("已连接到 Wi-Fi 网络") print("已连接到 Wi-Fi 网络")
else: else:
print("未连接到 Wi-Fi 网络") print("未连接到 Wi-Fi 网络")
set_hotspot()
print("Starting Flask server") print("Starting Flask server")
print("----------------------------") print("----------------------------")
wf_list = scan_wifi() wifi_list = scan_wifi()
print("----------------------------") print("----------------------------")
print(wf_list) # save wifi_list to file
with open('wifi_list.txt', 'w') as f:
for wifi in wifi_list:
f.write(f"{wifi['ssid']},{wifi['signal']}\n")
restart_hotspot()
# 启动 Flask 服务器 # 启动 Flask 服务器
app.run(host='0.0.0.0', port=80, debug=True) app.run(host='0.0.0.0', port=80)