描述
解决虚拟机文件互传、互访服务失灵时，文件互传、互访的问题。
操作方法
宿主机拷虚拟机文件
1.虚拟机运行。
pytho3 文件名.py2.访问虚拟机IP:8888。
3.点击要下载的文件即可。
虚拟机拷宿主机文件
1.虚拟机运行。
pytho3 文件名.py2.宿主机访问虚拟机IP:8888 。
3.宿主机点击上传即可。
源代码
import http.server
import socketserver
import os
import socket
import cgi
import subprocess
import urllib.parse
PORT = 8888
shared_message = "在此输入要同步的代码或文字..."
HTML_TEMPLATE = """
<!DOCTYPE html>
<html lang="zh-CN">
<head>
<meta charset="UTF-8">
<title>VM 极速互传中心</title>
<style>
body {{ font-family: 'Segoe UI', system-ui; background: #f0f2f5; padding: 20px; }}
.container {{ max-width: 800px; margin: 0 auto; }}
.card {{ background: white; padding: 25px; border-radius: 12px; box-shadow: 0 4px 6px rgba(0,0,0,0.05); margin-bottom: 20px; }}
h2 {{ margin-top: 0; color: #1a73e8; border-bottom: 2px solid #e8f0fe; padding-bottom: 10px; font-size: 1.1rem; }}
textarea {{ width: 100%; height: 120px; border: 1px solid #ddd; border-radius: 8px; padding: 10px; box-sizing: border-box; font-family: monospace; font-size: 14px; background: #fafafa; }}
.msg-btns {{ margin-top: 10px; display: flex; gap: 10px; }}
ul {{ list-style: none; padding: 0; }}
li {{ background: #fff; border: 1px solid #eee; margin-bottom: 8px; padding: 12px 15px; border-radius: 8px; display: flex; justify-content: space-between; align-items: center; }}
.file-link {{ text-decoration: none; color: #1a73e8; font-weight: 500; }}
button {{ background: #1a73e8; color: white; border: none; padding: 8px 18px; border-radius: 6px; cursor: pointer; font-weight: bold; }}
button:hover {{ background: #1557b0; }}
</style>
</head>
<body>
<div class="container">
<div class="card">
<h2>💬 代码/消息 互传剪贴板</h2>
<form action="/update_msg" method="post">
<textarea name="msg_content" placeholder="粘贴内容...">{shared_msg}</textarea>
<div class="msg-btns">
<button type="submit">更新保存</button>
<button type="button" onclick="document.querySelector('textarea').select();document.execCommand('copy');alert('已复制');" style="background:#5f6368">复制</button>
</div>
</form>
</div>
<div class="card">
<h2>📤 上传文件 (虚拟机接收)</h2>
<form action="/" method="post" enctype="multipart/form-data" style="display:flex; gap:10px;">
<input type="file" name="file" required style="flex-grow:1;">
<button type="submit">上传</button>
</form>
</div>
<div class="card">
<h2>📥 文件列表 (所有文件均为强制下载)</h2>
<ul>{file_list}</ul>
</div>
</div>
</body>
</html>
"""
class ProFileHandler(http.server.SimpleHTTPRequestHandler):
def do_GET(self):
# 逻辑：非主页请求均尝试作为下载处理
if self.path != '/' and not self.path.startswith('/update_msg'):
path = self.translate_path(self.path)
if os.path.isfile(path):
filename = os.path.basename(path)
self.send_response(200)
# [cite_start]强制触发下载头 [cite: 10]
self.send_header('Content-Disposition', f'attachment; filename="{filename}"')
self.send_header('Content-type', 'application/octet-stream')
self.end_headers()
with open(path, 'rb') as f:
self.wfile.write(f.read())
return
if self.path == '/':
self.send_response(200)
self.send_header("Content-type", "text/html; charset=utf-8")
self.end_headers()
files = [f for f in os.listdir('.') if os.path.isfile(f)]
file_items = "".join([f'<li><a class="file-link" href="/{f}">{f}</a> <span>{os.path.getsize(f)//1024} KB</span></li>' for f in files])
global shared_message
self.wfile.write(HTML_TEMPLATE.format(file_list=file_items or "<li>暂无文件</li>", shared_msg=shared_message).encode('utf-8'))
else:
super().do_GET()
def do_POST(self):
global shared_message
if self.path == '/update_msg':
content_length = int(self.headers['Content-Length'])
post_data = self.rfile.read(content_length).decode('utf-8')
parsed_data = urllib.parse.parse_qs(post_data)
if 'msg_content' in parsed_data:
shared_message = parsed_data['msg_content'][0]
self.send_response(303)
self.send_header('Location', '/')
self.end_headers()
return
form = cgi.FieldStorage(fp=self.rfile, headers=self.headers, environ={'REQUEST_METHOD': 'POST'})
if 'file' in form and form['file'].filename:
with open(os.path.basename(form['file'].filename), 'wb') as f:
f.write(form['file'].file.read())
self.send_response(303)
self.send_header('Location', '/')
self.end_headers()
def get_ip():
s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
try:
s.connect(('8.8.8.8', 80))
ip = s.getsockname()[0]
except:
ip = '127.0.0.1'
finally:
s.close()
return ip
def kill_port(port):
try:
# [cite_start]使用 fuser 清理旧进程 [cite: 13]
subprocess.run(["sudo", "fuser", "-k", f"{port}/tcp"], check=False, capture_output=True)
except:
pass
if __name__ == "__main__":
kill_port(PORT)
socketserver.TCPServer.allow_reuse_address = True
print(f"✅ 服务启动！宿主机访问: http://{get_ip()}:{PORT}")
with socketserver.TCPServer(("", PORT), ProFileHandler) as httpd:
httpd.serve_forever()
评论区