import os import socket import random from flask import Flask, request, send_from_directory, render_template_string, jsonify, abort import threading import webbrowser from datetime import datetime import json import ipaddress import subprocess from concurrent.futures import ThreadPoolExecutor, as_completed # 创建Flask应用 app = Flask(__name__) UPLOAD_FOLDER = 'shared_files' app.config['UPLOAD_FOLDER'] = UPLOAD_FOLDER app.config['MAX_CONTENT_LENGTH'] = 16 * 1024 * 1024 * 1024 # 限制上传大小为16GB # 确保共享目录存在 if not os.path.exists(UPLOAD_FOLDER): os.makedirs(UPLOAD_FOLDER) # 确保pp目录存在（用于存放图片） if not os.path.exists('pp'): os.makedirs('pp') # 存储设备信息和扫描状态 devices = {} scanning = False max_threads = 50 total_ips = 0 scanned_ips = 0 def get_local_ip(): """获取本机在局域网中的IP地址""" try: s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) s.connect(("8.8.8.8", 80)) ip = s.getsockname()[0] s.close() return ip except Exception: return "127.0.0.1" def get_netmask(): """获取子网掩码""" try: # 使用ipconfig命令获取子网掩码 if os.name == 'nt': # Windows result = subprocess.run(["ipconfig"], capture_output=True, text=True, encoding='gbk') else: # Linux/Mac result = subprocess.run(["ifconfig"], capture_output=True, text=True) lines = result.stdout.split('\n') local_ip = get_local_ip() for i, line in enumerate(lines): if local_ip in line: # 尝试从附近行找到子网掩码 for j in range(max(0, i-2), min(len(lines), i+2)): if "子网掩码" in lines[j] or "netmask" in lines[j].lower(): parts = lines[j].split(":") if len(parts) > 1: return parts[1].strip().split()[0] except Exception: pass return "255.255.255.0" # 默认子网掩码 def calculate_network_range(): """计算局域网IP范围""" global total_ips local_ip = get_local_ip() netmask = get_netmask() try: network = ipaddress.IPv4Network(f"{local_ip}/{netmask}", strict=False) hosts = [str(host) for host in network.hosts()] total_ips = len(hosts) return hosts except: # 如果计算失败，使用常见的局域网IP范围 base_ip = ".".join(local_ip.split(".")[:3]) hosts = [f"{base_ip}.{i}" for i in range(1, 255)] total_ips = len(hosts) return hosts def ping_host(ip): """ping主机检查是否在线""" try: # 使用系统ping命令 if os.name == 'nt': # Windows command = ['ping', '-n', '1', '-w', '1000', ip] else: # Linux/Mac command = ['ping', '-c', '1', '-W', '1', ip] # 使用subprocess运行ping命令 result = subprocess.run(command, capture_output=True, text=True) return result.returncode == 0 except Exception: return False def get_hostname(ip): """尝试获取主机名""" try: socket.setdefaulttimeout(1) hostname = socket.gethostbyaddr(ip) return hostname[0] except (socket.herror, socket.gaierror, socket.timeout): return "未知设备" def get_mac_address(ip): """获取MAC地址""" try: # 使用arp命令获取MAC地址 if os.name == 'nt': # Windows result = subprocess.run(["arp", "-a", ip], capture_output=True, text=True, encoding='gbk') lines = result.stdout.split('\n') for line in lines: if ip in line: parts = line.split() if len(parts) >= 2: return parts[1] else: # Linux/Mac result = subprocess.run(["arp", "-n", ip], capture_output=True, text=True) lines = result.stdout.split('\n') for line in lines: if ip in line: parts = line.split() if len(parts) >= 3: return parts[2] except Exception: pass return "未知MAC" def scan_single_ip(ip): """扫描单个IP地址并返回设备信息（如果在线）""" global scanned_ips if not scanning: # 检查是否已停止扫描 return None try: # 检查IP是否在线 if ping_host(ip): # 获取设备信息 hostname = get_hostname(ip) mac = get_mac_address(ip) return { "ip": ip, "mac": mac, "hostname": hostname, "first_seen": datetime.now().isoformat(), "last_seen": datetime.now().isoformat() } except Exception as e: print(f"扫描 {ip} 时出错: {str(e)}") finally: # 更新已扫描计数 if scanning: scanned_ips += 1 return None def scan_network(): """扫描网络中的设备，使用多线程并行处理""" global scanning, scanned_ips scanning = True scanned_ips = 0 network_range = calculate_network_range() active_devices = [] print(f"开始扫描网络，共 {total_ips} 个IP地址") # 使用线程池管理并发扫描 with ThreadPoolExecutor(max_workers=max_threads) as executor: # 提交所有IP扫描任务 futures = {executor.submit(scan_single_ip, ip): ip for ip in network_range} # 处理完成的任务 for future in as_completed(futures): if not scanning: # 如果已停止扫描，取消所有剩余任务 executor.shutdown(wait=False, cancel_futures=True) break ip = futures[future] try: device = future.result() if device: active_devices.append(device) mac = device["mac"] if mac in devices: devices[mac]["last_seen"] = datetime.now().isoformat() else: devices[mac] = device print(f"发现设备: {device['ip']} ({device['hostname']}) - {device['mac']}") except Exception as e: print(f"处理 {ip} 结果时出错: {str(e)}") scanning = False print(f"扫描完成，发现 {len(active_devices)} 台设备") return active_devices # 添加pp目录的路由，允许访问pp文件夹中的图片 @app.route('/pp/') def serve_pp_files(filename): return send_from_directory('pp', filename) # HTML模板 - 半透明风格，自适应背景 HTML_TEMPLATE = ''' 文件共享与设备扫描
背景图片

文件共享

拖放文件到此处上传

或点击下方按钮选择文件

已共享的文件

{{ files|length }} 个文件
{% if files %} {% for file in files %}

{{ file.name }}

{{ file.size }}

下载
{% endfor %} {% else %}

暂无共享文件

上传文件以开始共享

{% endif %}

设备扫描

发现的设备

{{ devices|length }} 台设备
{% if devices %} {% for device in devices %}

{{ device.hostname }} {% if device.ip == ip_address %} 本机 {% endif %}

{{ device.ip }}

MAC: {{ device.mac }}

{% if device.ip != ip_address %} 访问该设备共享 {% endif %}
{% endfor %} {% else %}

暂无设备

点击"开始扫描"按钮发现局域网中的设备

{% endif %}
''' @app.route('/') def index(): """主页面，显示文件列表和设备列表""" files = [] for filename in os.listdir(app.config['UPLOAD_FOLDER']): filepath = os.path.join(app.config['UPLOAD_FOLDER'], filename) if os.path.isfile(filepath): size = os.path.getsize(filepath) # 格式化文件大小 if size < 1024: size_str = f"{size} B" elif size < 1024 * 1024: size_str = f"{size/1024:.2f} KB" elif size < 1024 * 1024 * 1024: size_str = f"{size/(1024*1024):.2f} MB" else: size_str = f"{size/(1024*1024*1024):.2f} GB" files.append({'name': filename, 'size': size_str}) # 准备设备列表数据 device_list = [] for mac, device_info in devices.items(): device_list.append({ 'ip': device_info['ip'], 'mac': device_info['mac'], 'hostname': device_info['hostname'] }) # 获取当前年份用于页脚 current_year = datetime.now().year local_ip = get_local_ip() port = 5000 return render_template_string( HTML_TEMPLATE, files=files, devices=device_list, ip_address=local_ip, port=port, year=current_year ) @app.route('/upload', methods=['POST']) def upload_file(): """处理文件上传""" if 'file' not in request.files: return jsonify({'error': '没有文件部分'}), 400 file = request.files['file'] if file.filename == '': return jsonify({'error': '没有选择文件'}), 400 if file: # 处理文件名重复 filename = file.filename filepath = os.path.join(app.config['UPLOAD_FOLDER'], filename) counter = 1 while os.path.exists(filepath): name, ext = os.path.splitext(filename) filename = f"{name}_{counter}{ext}" filepath = os.path.join(app.config['UPLOAD_FOLDER'], filename) counter += 1 file.save(filepath) return jsonify({'success': True, 'message': '文件上传成功'}), 200 @app.route('/download/') def download_file(filename): """处理文件下载""" try: # 安全检查，防止路径遍历 safe_path = os.path.join(app.config['UPLOAD_FOLDER'], filename) if not os.path.exists(safe_path) or not os.path.isfile(safe_path): return jsonify({'success': False, 'message': '文件不存在'}), 404 # 确保文件在上传目录内 if not os.path.abspath(safe_path).startswith(os.path.abspath(app.config['UPLOAD_FOLDER'])): return jsonify({'success': False, 'message': '非法访问'}), 403 return send_from_directory(app.config['UPLOAD_FOLDER'], filename, as_attachment=True) except Exception as e: return jsonify({'success': False, 'message': str(e)}), 500 @app.route('/delete/', methods=['DELETE']) def delete_file(filename): """处理文件删除""" try: # 安全检查 filepath = os.path.join(app.config['UPLOAD_FOLDER'], filename) if not os.path.exists(filepath): return jsonify({'success': False, 'message': '文件不存在'}), 404 if not os.path.isfile(filepath): return jsonify({'success': False, 'message': '不是文件'}), 400 # 确保文件在上传目录内 if not os.path.abspath(filepath).startswith(os.path.abspath(app.config['UPLOAD_FOLDER'])): return jsonify({'success': False, 'message': '非法操作'}), 403 os.remove(filepath) return jsonify({'success': True, 'message': '文件删除成功'}) except Exception as e: return jsonify({'success': False, 'message': str(e)}), 500 @app.route('/start_scan', methods=['POST']) def start_scan(): """开始设备扫描""" global scanning if scanning: return jsonify({'success': False, 'message': '扫描已在运行中'}) # 在后台线程中运行扫描 def run_scan(): scan_network() thread = threading.Thread(target=run_scan) thread.daemon = True thread.start() return jsonify({'success': True, 'message': '设备扫描已启动'}) @app.route('/stop_scan', methods=['POST']) def stop_scan(): """停止设备扫描""" global scanning scanning = False return jsonify({'success': True, 'message': '设备扫描已停止'}) @app.route('/scan_progress') def scan_progress(): """获取扫描进度""" device_count = len(devices) return jsonify({ 'scanning': scanning, 'scanned_ips': scanned_ips, 'total_ips': total_ips, 'device_count': device_count }) def start_server(): """启动服务器""" local_ip = get_local_ip() port = 5000 print("=" * 60) print(" 工具已启动，不要瞎几把上传奇怪的东西") print(f" 本地访问地址: http://localhost:{port}") print(f" 局域网访问地址: http://{local_ip}:{port}") print("=" * 60) print("官网工具记得star哦！github.com 支持一下吧真的没有star") print("我的邮箱naxid@caoshen.top，欢迎交流") print("在上学不一定会及时看和回复") # 自动打开浏览器 try: webbrowser.open(f"http://localhost:{port}") except: pass # 启动Flask应用 app.run(host='0.0.0.0', port=port, debug=False, threaded=True) if __name__ == '__main__': start_server()