静默上传文件到浏览器！　—— 想~ 让流星能够实现 幸福的愿望🌠

影刀更新到5.32版本之后，竟然只加了静默下载功能……没有上传？这怎么可以！😤

为了更加方便上传文件，当然要帮大家把上传的“静默魔法”也补全啦～

静默上传效果展示

因为浏览器的安全小规矩，正常情况下传文件都需要手动选择，但我们偏偏要悄悄完成！✨

其实诀窍就是：让文件提前在页面里“准备好”，上传的时候自己传给自己～

需要的影刀指令和Python模块，下面都会贴给大家～

跟着做，你也能优雅地实现“无声上传”哦！

影刀指令

# Python版本: 3.10
# websockets版本: 11.0.3

# 使用提醒:
# 1. xbot包提供软件自动化、数据表格、Excel、日志、AI等功能
# 2. package包提供访问当前应用数据的功能，如获取元素、访问全局变量、获取资源文件等功能
# 3. 当此模块作为流程独立运行时执行main函数
# 4. 可视化流程中可以通过"调用模块"的指令使用此模块

# 使用提醒:
# 1. xbot包提供软件自动化、数据表格、Excel、日志、AI等功能
# 2. package包提供访问当前应用数据的功能，如获取元素、访问全局变量、获取资源文件等功能
# 3. 当此模块作为流程独立运行时执行main函数
# 4. 可视化流程中可以通过"调用模块"的指令使用此模块

import xbot
from xbot import print, sleep
from .import package
from .package import variables as glv
import asyncio
from pathlib import Path
import websockets
import mimetypes

def main(args):
    pass

class FileTransferServer:
    def __init__(self, files, web_page):
        """初始化文件WebSocket服务器"""
        self.files = files
        self.web_page = web_page
        self.file_paths = []
        self.server_task = None
        mimetypes.init()

    def _get_mime_type(self, file_path):
        """获取文件类型"""
        path = Path(file_path)
        mime_type, _ = mimetypes.guess_type(str(path))
        return mime_type or 'application/octet-stream'

    async def send_file(self, websocket, file_path):
        """发送单个文件及其元数据"""
        path = Path(file_path)
        mime_type = self._get_mime_type(file_path)
        await websocket.send(f"{path.name}|{mime_type}")
        await websocket.send(path.read_bytes())

    async def handle_client(self, websocket, path):
        """处理文件路径输入"""
        try:
            for file_path in self.file_paths:
                await self.send_file(websocket, file_path)
            await websocket.send("[DONE]")
            await websocket.wait_closed()
        except Exception as e:
            print(f"传输错误: {e}")
        finally:
            if self.server_task:
                self.server_task.cancel()

    async def js_code(self):
        """客户端JavaScript代码"""
        code = """
        function uploadFiles() {
            const fileInput = document.querySelector('input[type="file"]');
            const socket = new WebSocket('ws://localhost:8765');

            let pendingFileMeta = null;
            const receivedFiles = [];

            socket.onopen = () => console.log("已连接服务器");

            socket.onclose = () => {
                console.log("已断开服务器");
                const dataTransfer = new DataTransfer();
                receivedFiles.forEach(f => dataTransfer.items.add(f));
                fileInput.files = dataTransfer.files;
                fileInput.dispatchEvent(new Event('change', { bubbles: true }));
            };

            socket.onmessage = (event) => {
                if (event.data === "[DONE]") {
                    socket.close();
                    return;
                }

                if (!pendingFileMeta) {
                    const [name, type] = event.data.split('|');
                    pendingFileMeta = { name, type };
                } else {
                    const file = new File([event.data], pendingFileMeta.name, { 
                        type: pendingFileMeta.type 
                    });

                    receivedFiles.push(file);
                    pendingFileMeta = null;
                }
            };

            socket.onerror = (err) => console.error("传输错误:", err);
        }
        """
        self.web_page.execute_javascript(code)

    async def start_server(self, host="localhost", port=8765):
        """WebSocket服务器"""
        self.file_paths = [Path(f) for f in (self.files if isinstance(self.files, list) else [self.files])]
        async with websockets.serve(self.handle_client, host, port) as server:
            self.server_task = asyncio.current_task()
            try:
                await self.js_code()
                await server.wait_closed()
            except asyncio.CancelledError:
                pass

    def run_server(self, host="localhost", port=8765):
        """启动文件传输服务"""
        asyncio.run(self.start_server(host, port))

def upload_files(files, web_page):
    server = FileTransferServer(files, web_page)
    server.run_server()

Python模块

总共两步，轻轻松松就能绕过对话框哟

1. 影刀开启小小服务端，把想上传的文件变成数据流，等待客户端来连接～
2. 页面里启动接收客户端，链接服务端、接过数据，然后……自己把文件送给自己！✅

教学完成！你们还可以根据自己的需要，继续优化这段代码～💖