你是否曾经想过,如何能方便地将在电脑上存储的照片,通过手机或平板在局域网内快速浏览?今天介绍的这个 python 脚本就能帮你轻松实现!它巧妙地结合了 wxpython 构建的图形用户界面(gui)和 python 内建的 web 服务器功能,让你在本地网络中搭建一个私人的、即开即用的网页相册。
让我们一起深入代码,看看它是如何一步步实现的。
项目目标
这个脚本的核心目标是:
- 提供一个简单的桌面应用程序,让用户可以选择包含图片的本地文件夹。
- 启动一个本地 http 服务器,该服务器能生成一个展示所选文件夹内图片缩略图的 html 页面。
- 允许同一局域网内的其他设备(如手机、平板)通过浏览器访问这个 html 页面。
- 提供一个具备现代功能的网页浏览界面,例如图片懒加载、点击缩略图弹出大图预览、以及在大图模式下切换图片等。
核心技术栈
python 3: 作为主要的编程语言。
wxpython: 一个跨平台的 python gui 工具库,用于创建桌面应用程序窗口。
http.server & socketserver: python 标准库,用于创建基础的 http web 服务器。
threading: 用于在后台线程中运行 http 服务器,避免阻塞 gui 界面。
socket: 用于获取本机的局域网 ip 地址。
os & pathlib: 用于文件系统操作(列出目录、检查文件、获取路径、大小等)。pathlib 被导入但实际未使用,代码主要使用了 os.path。
webbrowser: 用于在脚本启动服务后自动打开默认浏览器访问页面。
mimetypes: 用于猜测图片文件的 mime 类型(如 image/jpeg),以便浏览器正确显示。
html, css, javascript: 用于构建用户在浏览器中看到的图片浏览前端界面。
代码深度解析
让我们逐一拆解脚本的关键组成部分:
1. 导入模块与全局变量
import wx import os import http.server import socketserver import threading import socket import webbrowser from pathlib import path # 导入但未使用 import mimetypes # 全局变量,用于存储选择的图片文件夹路径 selected_folder = "" server_thread = none server_instance = none
脚本首先导入了所有必需的库。
定义了三个全局变量:
- selected_folder: 存储用户通过 gui 选择的图片文件夹路径。
- server_thread: 用于保存运行 http 服务器的线程对象。
- server_instance: 用于保存实际的 tcpserver 服务器实例。
在这个场景下使用全局变量简化了状态管理,但在更复杂的应用中可能需要更精细的状态管理机制。
2. 自定义 http 请求处理器 (imagehandler)
class imagehandler(http.server.simplehttprequesthandler):
def __init__(self, *args, **kwargs):
# 确保每次请求都使用最新的文件夹路径
global selected_folder
# 将directory参数传递给父类的__init__方法
super().__init__(directory=selected_folder, *args, **kwargs)
def do_get(self):
# ... (处理 "/" 根路径请求,生成 html 页面) ...
# ... (处理 "/images/..." 图片文件请求) ...
# ... (处理其他路径,返回 404) ...
这个类继承自 http.server.simplehttprequesthandler,它提供了处理静态文件请求的基础功能。
__init__ (构造函数): 这是个关键点。每次处理新的 http 请求时,这个构造函数都会被调用。它会读取当前的 selected_folder 全局变量的值,并将其作为 directory 参数传递给父类的构造函数。这意味着服务器始终从 gui 中最新选择的文件夹提供服务。(注意:当前代码逻辑下,用户选择新文件夹后需要重新点击“启动服务器”按钮才会生效)。
do_get 方法: 这个方法负责处理所有传入的 http get 请求。
请求根路径 (/):
1.当用户访问服务器的根地址时(例如 http://<ip>:8000/),此部分代码被执行。
2.它会扫描 selected_folder 文件夹,找出所有具有常见图片扩展名(如 .jpg, .png, .gif 等)的文件。
3.计算每个图片文件的大小(转换为 mb 或 kb)。
4.按文件名对图片列表进行字母排序。
5.动态生成一个完整的 html 页面,页面内容包括:
css 样式: 定义了页面的外观,包括响应式的网格布局 (.gallery)、图片容器样式、用于大图预览的模态弹出框 (.modal)、导航按钮、加载指示器以及图片懒加载的淡入效果。
html 结构: 包含一个标题 (<h1>)、一个加载进度条 (<div>)、一个图片画廊区域 (<div class="gallery">),其中填充了每个图片项(包含一个 img 标签用于懒加载、图片文件名和文件大小),以及模态框的 html 结构。
javascript 脚本: 实现了前端的交互逻辑:
- 图片懒加载 (lazy loading): 利用现代浏览器的 intersectionobserver api(并为旧浏览器提供后备方案),仅当图片滚动到可视区域时才加载其 src,极大地提高了包含大量图片时的初始页面加载速度。同时,还实现了一个简单的加载进度条。
- 模态框预览: 当用户点击任意缩略图时,会弹出一个覆盖全屏的模态框,显示对应的大图。
- 图片导航: 在模态框中,用户可以通过点击“上一张”/“下一张”按钮,或使用键盘的左右箭头键来切换浏览图片。按 escape 键可以关闭模态框。
- 图片预加载: 在打开模态框显示某张图片时,脚本会尝试预加载其相邻(上一张和下一张)的图片,以提升导航切换时的流畅度。
6.最后,服务器将生成的 html 页面内容连同正确的 http 头部(content-type: text/html, cache-control 设置为缓存 1 小时)发送给浏览器。
请求图片路径 (/images/...):
- 当浏览器请求的路径以 /images/ 开头时(这是 html 中 <img> 标签的 src 指向的路径),服务器认为它是在请求一个具体的图片文件。
- 代码从路径中提取出图片文件名,并结合 selected_folder 构建出完整的文件系统路径。
- 检查该文件是否存在且确实是一个文件。
- 使用 mimetypes.guess_type 来推断文件的 mime 类型(例如 image/jpeg),并为 png 和未知类型提供回退。
- 将图片文件的二进制内容读取出来,并连同相应的 http 头部(content-type, content-length, cache-control 设置为缓存 1 天以提高性能, accept-ranges 表示支持范围请求)发送给浏览器。
请求其他路径:
对于所有其他无法识别的请求路径,服务器返回 404 “file not found” 错误。
3. 启动服务器函数 (start_server)
def start_server(port=8000):
global server_instance
# 设置允许地址重用,解决端口被占用的问题
socketserver.tcpserver.allow_reuse_address = true
# 创建服务器实例
server_instance = socketserver.tcpserver(("", port), imagehandler)
server_instance.serve_forever()
这个函数被设计用来在一个单独的线程中运行。
socketserver.tcpserver.allow_reuse_address = true 是一个重要的设置,它允许服务器在关闭后立即重新启动时可以快速重用相同的端口,避免常见的“地址已被使用”错误。
它创建了一个 tcpserver 实例,监听本机的所有网络接口 ("") 上的指定端口(默认为 8000),并指定使用我们自定义的 imagehandler 类来处理所有接收到的请求。
server_instance.serve_forever() 启动了服务器的主循环,持续监听和处理连接请求,直到 shutdown() 方法被调用。
4. 获取本机 ip 函数 (get_local_ip)
def get_local_ip():
try:
# 创建一个临时套接字连接到外部地址,以获取本机ip
s = socket.socket(socket.af_inet, socket.sock_dgram)
s.connect(("8.8.8.8", 80)) # 连接到一个公共地址(如谷歌dns),无需实际发送数据
ip = s.getsockname()[0] # 获取用于此连接的本地套接字地址
s.close()
return ip
except:
return "127.0.0.1" # 如果获取失败,返回本地回环地址
这是一个实用工具函数,用于查找运行脚本的计算机在局域网中的 ip 地址。这对于告诉用户(以及其他设备)应该访问哪个 url 非常重要。
它使用了一个常用技巧:创建一个 udp 套接字,并尝试“连接”(这并不会实际发送数据)到一个已知的外部 ip 地址(例如 google 的公共 dns 服务器 8.8.8.8)。操作系统为了完成这个(虚拟的)连接,会确定应该使用哪个本地 ip 地址,然后我们就可以通过 getsockname() 获取这个地址。
如果尝试获取 ip 失败(例如,没有网络连接),它会回退到返回 127.0.0.1 (localhost)。
5. wxpython 图形用户界面 (photoserverapp, photoserverframe)
photoserverapp (应用程序类):
class photoserverapp(wx.app):
def oninit(self):
self.frame = photoserverframe("图片服务器", (600, 400))
self.frame.show()
return true
标准的 wx.app 子类。它的 oninit 方法负责创建并显示主应用程序窗口 (photoserverframe)。
photoserverframe (主窗口类):
class photoserverframe(wx.frame):
def __init__(self, title, size):
# ... 创建界面控件 (文本框, 按钮, 静态文本) ...
# ... 使用 wx.boxsizer 进行布局管理 ...
# ... 绑定事件处理函数 (on_browse, on_start_server, on_stop_server, on_close) ...
# ... 初始化设置 (禁用停止按钮, 显示欢迎信息) ...
def on_browse(self, event):
# ... 弹出文件夹选择对话框 (wx.dirdialog) ...
# ... 更新全局变量 selected_folder 和界面上的文本框 ...
def on_start_server(self, event):
# ... 检查是否已选择文件夹 ...
# ... 如果服务器已运行,先停止旧的再启动新的(实现简单的重启逻辑)...
# ... 在新的后台守护线程 (daemon thread) 中启动服务器 ...
# ... 获取本机 ip, 更新界面按钮状态, 在状态区记录日志, 自动打开浏览器 ...
# ... 包含基本的错误处理 ...
def on_stop_server(self, event):
# ... 调用 server_instance.shutdown() 关闭服务器 ...
# ... 等待服务器线程结束 (join) ...
# ... 更新界面按钮状态, 在状态区记录日志 ...
def log_status(self, message):
# ... 将消息追加到状态显示文本框 (self.status_txt) ...
def on_close(self, event):
# ... 绑定窗口关闭事件,确保退出程序前尝试关闭服务器 ...
# ... event.skip() 允许默认的窗口关闭行为继续执行 ...
这个类定义了应用程序的主窗口。
__init__: 创建所有的可视化元素:一个只读文本框显示选定的文件夹路径,“选择文件夹” 按钮,“启动服务器” 和 “停止服务器” 按钮,以及一个多行只读文本框用于显示服务器状态和日志信息。它还使用 wx.boxsizer 来组织这些控件的布局,并绑定了按钮点击事件和窗口关闭事件到相应的方法。初始时,“停止服务器”按钮是禁用的。
on_browse: 处理 “选择文件夹” 按钮的点击事件。它会弹出一个标准的文件夹选择对话框。如果用户选择了文件夹并确认,它会更新 selected_folder 全局变量,并将路径显示在界面文本框中,同时记录一条日志。
on_start_server: 处理 “启动服务器” 按钮的点击事件。
- 首先检查用户是否已经选择了文件夹。
- 检查服务器是否已在运行。如果是,它会先尝试 shutdown() 当前服务器实例并等待线程结束,然后才启动新的服务器线程(提供了一种重启服务的方式)。
- 创建一个新的 threading.thread 来运行 start_server 函数。将线程设置为 daemon=true,这样主程序退出时,这个后台线程也会自动结束。
- 调用 get_local_ip() 获取本机 ip。
- 更新 gui 按钮的状态(禁用“启动”和“选择文件夹”,启用“停止”)。
- 在状态文本框中打印服务器已启动、ip 地址、端口号以及供手机访问的 url。
- 使用 webbrowser.open() 自动在用户的默认浏览器中打开服务器地址。
- 包含了一个 try...except 块来捕获并显示启动过程中可能出现的错误。
on_stop_server: 处理 “停止服务器” 按钮的点击事件。
- 如果服务器实例存在 (server_instance 不为 none),调用 server_instance.shutdown() 来请求服务器停止。shutdown() 会使 serve_forever() 循环退出。
- 等待服务器线程 (server_thread) 结束(使用 join() 并设置了短暂的超时)。
- 重置全局变量 server_instance 和 server_thread 为 none。
- 更新 gui 按钮状态(启用“启动”和“选择文件夹”,禁用“停止”)。
- 记录服务器已停止的日志。
log_status: 一个简单的辅助方法,将传入的消息追加到状态文本框 self.status_txt 中,并在末尾添加换行符。
on_close: 当用户点击窗口的关闭按钮时触发。它会检查服务器是否仍在运行,如果是,则尝试调用 shutdown() 来关闭服务器,以确保资源被正确释放。
event.skip() 允许 wxpython 继续执行默认的窗口关闭流程。
6. 程序入口 (if __name__ == "__main__":)
if __name__ == "__main__":
app = photoserverapp(false)
app.mainloop()
这是标准的 python 脚本入口点。
它创建了 photoserverapp 的实例。
调用 app.mainloop() 启动了 wxpython 的事件循环。这个循环会监听用户的交互(如按钮点击、窗口关闭等)并分派事件给相应的处理函数,直到应用程序退出。
完整代码
# -*- coding: utf-8 -*-
# (在此处粘贴完整的 python 代码)
import wx
import os
import http.server
import socketserver
import threading
import socket
import webbrowser
from pathlib import path # 实际未使用 os.path
import mimetypes
# 全局变量,用于存储选择的图片文件夹路径
selected_folder = ""
server_thread = none
server_instance = none
# 自定义http请求处理器
class imagehandler(http.server.simplehttprequesthandler):
def __init__(self, *args, **kwargs):
# 确保每次请求都使用最新的文件夹路径
global selected_folder
# 将directory参数传递给父类的__init__方法
# 注意:simplehttprequesthandler 在 python 3.7+ 才接受 directory 参数
# 如果在更早版本运行,需要修改此处的实现方式(例如,在 do_get 中处理路径)
super().__init__(directory=selected_folder, *args, **kwargs)
def do_get(self):
# 使用 os.path.join 来确保路径分隔符正确
requested_path = os.path.normpath(self.translate_path(self.path))
if self.path == "/":
# 显示图片列表的主页
self.send_response(200)
self.send_header("content-type", "text/html; charset=utf-8") # 指定utf-8编码
self.send_header("cache-control", "max-age=3600") # 缓存1小时,提高加载速度
self.end_headers()
# 获取图片文件列表
image_extensions = ['.jpg', '.jpeg', '.png', '.gif', '.bmp', '.webp']
image_files = []
current_directory = self.directory # 使用 __init__ 中设置的目录
try:
# 确保目录存在
if not os.path.isdir(current_directory):
self.wfile.write(f"错误:目录 '{current_directory}' 不存在或不是一个目录。".encode('utf-8'))
return
for file in os.listdir(current_directory):
file_path = os.path.join(current_directory, file)
if os.path.isfile(file_path) and os.path.splitext(file)[1].lower() in image_extensions:
# 获取文件大小用于显示预加载信息
try:
file_size = os.path.getsize(file_path) / (1024 * 1024) # 转换为mb
image_files.append((file, file_size))
except oserror as e:
self.log_error(f"获取文件大小出错: {file} - {str(e)}")
except exception as e:
self.log_error(f"读取目录出错: {current_directory} - {str(e)}")
# 可以向浏览器发送一个错误信息
self.wfile.write(f"读取目录时发生错误: {str(e)}".encode('utf-8'))
return
# 按文件名排序 (考虑自然排序可能更好,如 '1.jpg', '2.jpg', '10.jpg')
image_files.sort(key=lambda x: x[0].lower())
# 生成html页面
# 使用 f-string 或模板引擎生成 html 会更清晰
html_parts = []
html_parts.append("""
<!doctype html>
<html>
<head>
<meta charset="utf-8">
<meta name="viewport" content="width=device-width, initial-scale=1.0">
<title>图片浏览</title>
<style>
body { font-family: arial, sans-serif; margin: 0; padding: 20px; background-color: #f0f0f0; }
h1 { color: #333; text-align: center; }
.gallery { display: grid; grid-template-columns: repeat(auto-fill, minmax(200px, 1fr)); grid-gap: 15px; margin-top: 20px; }
.image-item { background-color: #fff; border-radius: 5px; box-shadow: 0 2px 5px rgba(0,0,0,0.1); overflow: hidden; } /* 添加 overflow hidden */
.image-container { width: 100%; padding-bottom: 75%; /* 4:3 aspect ratio */ position: relative; overflow: hidden; cursor: pointer; background-color: #eee; /* placeholder color */ }
.image-container img { position: absolute; top: 0; left: 0; width: 100%; height: 100%; object-fit: cover; /* use cover for better thumbnail */ transition: transform 0.3s, opacity 0.3s; opacity: 0; /* start hidden for lazy load */ }
.image-container img.lazy-loaded { opacity: 1; } /* fade in when loaded */
.image-container:hover img { transform: scale(1.05); }
.image-info { padding: 8px 10px; } /* group name and size */
.image-name { text-align: center; font-size: 12px; overflow: hidden; white-space: nowrap; text-overflow: ellipsis; margin-bottom: 3px; }
.image-size { text-align: center; font-size: 11px; color: #666; }
.modal { display: none; position: fixed; z-index: 1000; left: 0; top: 0; width: 100%; height: 100%; background-color: rgba(0,0,0,0.9); }
.modal-content { display: block; max-width: 90%; max-height: 90%; margin: auto; position: absolute; top: 50%; left: 50%; transform: translate(-50%, -50%); object-fit: contain; }
.modal-close { position: absolute; top: 15px; right: 35px; color: #f1f1f1; font-size: 40px; font-weight: bold; cursor: pointer; }
.modal-caption { color: white; position: absolute; bottom: 20px; width: 100%; text-align: center; font-size: 14px; }
.nav-button { position: absolute; top: 50%; transform: translatey(-50%); color: white; font-size: 30px; font-weight: bold; cursor: pointer; background: rgba(0,0,0,0.4); border-radius: 50%; width: 45px; height: 45px; text-align: center; line-height: 45px; user-select: none; transition: background 0.2s; }
.nav-button:hover { background: rgba(0,0,0,0.7); }
.prev { left: 15px; }
.next { right: 15px; }
.loading-indicator { position: fixed; top: 0; left: 0; width: 100%; height: 3px; background-color: #4caf50; z-index: 2000; transform: scalex(0); transform-origin: left; transition: transform 0.3s ease-out, opacity 0.5s 0.5s; /* fade out after completion */ opacity: 1; }
.loading-indicator.hidden { opacity: 0; }
@media (max-width: 600px) {
.gallery { grid-template-columns: repeat(auto-fill, minmax(120px, 1fr)); /* smaller thumbnails on mobile */ }
.nav-button { width: 40px; height: 40px; line-height: 40px; font-size: 25px; }
.modal-close { font-size: 30px; top: 10px; right: 20px; }
}
</style>
</head>
<body>
<h1>图片浏览</h1>
<div class="loading-indicator" id="loadingbar"></div>
<div class="gallery" id="imagegallery">
""")
if not image_files:
html_parts.append("<p style='text-align:center; color: #555;'>未在此文件夹中找到图片。</p>")
# 使用 urllib.parse.quote 来编码文件名,防止特殊字符问题
from urllib.parse import quote
for idx, (image, size) in enumerate(image_files):
# 显示文件名和大小信息
size_display = f"{size:.2f} mb" if size >= 1 else f"{size*1024:.1f} kb"
# encode the image filename for use in url
image_url_encoded = quote(image)
html_parts.append(f"""
<div class="image-item" data-index="{idx}" data-src="/images/{image_url_encoded}" data-filename="{image.replace('"', '"')}">
<div class="image-container">
<img class="lazy-image" data-src="/images/{image_url_encoded}" alt="{image.replace('"', '"')}" loading="lazy">
</div>
<div class="image-info">
<div class="image-name" title="{image.replace('"', '"')}">{image}</div>
<div class="image-size">{size_display}</div>
</div>
</div>
""")
html_parts.append("""
</div>
<div id="imagemodal" class="modal">
<span class="modal-close" title="关闭 (esc)">×</span>
<img class="modal-content" id="modalimage" alt="预览图片">
<div class="modal-caption" id="modalcaption"></div>
<div class="nav-button prev" id="prevbutton" title="上一张 (←)">❮</div>
<div class="nav-button next" id="nextbutton" title="下一张 (→)">❯</div>
</div>
<script>
document.addeventlistener('domcontentloaded', function() {
const lazyimages = document.queryselectorall('.lazy-image');
const loadingbar = document.getelementbyid('loadingbar');
const imagegallery = document.getelementbyid('imagegallery');
const modal = document.getelementbyid('imagemodal');
const modalimg = document.getelementbyid('modalimage');
const captiontext = document.getelementbyid('modalcaption');
const prevbutton = document.getelementbyid('prevbutton');
const nextbutton = document.getelementbyid('nextbutton');
const closebutton = document.queryselector('.modal-close');
let loadedcount = 0;
let currentindex = 0;
let allimageitems = []; // will be populated after dom ready
function updateloadingbar() {
if (lazyimages.length === 0) {
loadingbar.style.transform = 'scalex(1)';
settimeout(() => { loadingbar.classlist.add('hidden'); }, 500);
return;
}
const progress = math.min(loadedcount / lazyimages.length, 1);
loadingbar.style.transform = `scalex(${progress})`;
if (loadedcount >= lazyimages.length) {
settimeout(() => { loadingbar.classlist.add('hidden'); }, 500); // hide after a short delay
}
}
// --- lazy loading ---
if ('intersectionobserver' in window) {
const observeroptions = { rootmargin: '0px 0px 200px 0px' }; // load images 200px before they enter viewport
const imageobserver = new intersectionobserver((entries, observer) => {
entries.foreach(entry => {
if (entry.isintersecting) {
const img = entry.target;
img.src = img.dataset.src;
img.onload = () => {
img.classlist.add('lazy-loaded');
loadedcount++;
updateloadingbar();
};
img.onerror = () => {
// optionally handle image load errors
img.alt = "图片加载失败";
loadedcount++; // still count it to finish loading bar
updateloadingbar();
}
observer.unobserve(img);
}
});
}, observeroptions);
lazyimages.foreach(img => imageobserver.observe(img));
} else {
// fallback for older browsers
lazyimages.foreach(img => {
img.src = img.dataset.src;
img.onload = () => {
img.classlist.add('lazy-loaded');
loadedcount++;
updateloadingbar();
};
img.onerror = () => { loadedcount++; updateloadingbar(); }
});
}
updateloadingbar(); // initial call for case of 0 images
// --- modal logic ---
// get all image items once dom is ready
allimageitems = array.from(document.queryselectorall('.image-item'));
function preloadimage(index) {
if (index >= 0 && index < allimageitems.length) {
const img = new image();
img.src = allimageitems[index].dataset.src;
}
}
function openmodal(index) {
if (index < 0 || index >= allimageitems.length) return;
currentindex = index;
const item = allimageitems[index];
const imgsrc = item.dataset.src;
const filename = item.dataset.filename;
modalimg.src = imgsrc; // set src immediately
modalimg.alt = filename;
captiontext.textcontent = `${filename} (${index + 1}/${allimageitems.length})`; // use textcontent for security
modal.style.display = 'block';
document.body.style.overflow = 'hidden'; // prevent background scrolling
// preload adjacent images
preloadimage(index - 1);
preloadimage(index + 1);
}
function closemodal() {
modal.style.display = 'none';
modalimg.src = ""; // clear src to stop loading/free memory
document.body.style.overflow = ''; // restore background scrolling
}
function showprevimage() {
const newindex = (currentindex - 1 + allimageitems.length) % allimageitems.length;
openmodal(newindex);
}
function shownextimage() {
const newindex = (currentindex + 1) % allimageitems.length;
openmodal(newindex);
}
// event listeners
imagegallery.addeventlistener('click', function(e) {
const item = e.target.closest('.image-item');
if (item) {
const index = parseint(item.dataset.index, 10);
openmodal(index);
}
});
closebutton.addeventlistener('click', closemodal);
prevbutton.addeventlistener('click', showprevimage);
nextbutton.addeventlistener('click', shownextimage);
// close modal if background is clicked
modal.addeventlistener('click', function(e) {
if (e.target === modal) {
closemodal();
}
});
// keyboard navigation
document.addeventlistener('keydown', function(e) {
if (modal.style.display === 'block') {
if (e.key === 'arrowleft') {
showprevimage();
} else if (e.key === 'arrowright') {
shownextimage();
} else if (e.key === 'escape') {
closemodal();
}
}
});
});
</script>
</body>
</html>
""")
# combine and send html
full_html = "".join(html_parts)
self.wfile.write(full_html.encode('utf-8')) # ensure utf-8 encoding
# --- serve image files ---
# check if the requested path seems like an image file request within our structure
elif self.path.startswith("/images/"):
# decode the url path component
from urllib.parse import unquote
try:
image_name = unquote(self.path[len("/images/"):])
except exception as e:
self.send_error(400, f"bad image path encoding: {e}")
return
# construct the full path using the selected directory
# important: sanitize image_name to prevent directory traversal attacks
# os.path.join on its own is not enough if image_name contains '..' or starts with '/'
image_path_unsafe = os.path.join(self.directory, image_name)
# basic sanitization: ensure the resolved path is still within the base directory
base_dir_real = os.path.realpath(self.directory)
image_path_real = os.path.realpath(image_path_unsafe)
if not image_path_real.startswith(base_dir_real):
self.send_error(403, "forbidden: path traversal attempt?")
return
if os.path.exists(image_path_real) and os.path.isfile(image_path_real):
try:
# get mime type
content_type, _ = mimetypes.guess_type(image_path_real)
if content_type is none:
# guess common types again or default
ext = os.path.splitext(image_name)[1].lower()
if ext == '.png': content_type = 'image/png'
elif ext in ['.jpg', '.jpeg']: content_type = 'image/jpeg'
elif ext == '.gif': content_type = 'image/gif'
elif ext == '.webp': content_type = 'image/webp'
else: content_type = 'application/octet-stream'
# get file size
file_size = os.path.getsize(image_path_real)
# send headers
self.send_response(200)
self.send_header('content-type', content_type)
self.send_header('content-length', str(file_size))
self.send_header('cache-control', 'max-age=86400') # cache for 1 day
self.send_header('accept-ranges', 'bytes') # indicate support for range requests
self.end_headers()
# send file content
with open(image_path_real, 'rb') as file:
# simple send - for large files consider shutil.copyfileobj
self.wfile.write(file.read())
except ioerror as e:
self.log_error(f"ioerror serving file: {image_path_real} - {str(e)}")
self.send_error(500, f"error reading file: {str(e)}")
except exception as e:
self.log_error(f"error serving file: {image_path_real} - {str(e)}")
self.send_error(500, f"server error serving file: {str(e)}")
else:
self.send_error(404, "image not found")
else:
# for any other path, let the base class handle it (or send 404)
# super().do_get() # if you want base class behavior for other files
self.send_error(404, "file not found") # or just send 404 directly
# 启动http服务器
def start_server(port=8000):
global server_instance
# 设置允许地址重用,解决端口被占用的问题
socketserver.tcpserver.allow_reuse_address = true
try:
# 创建服务器实例
server_instance = socketserver.tcpserver(("", port), imagehandler)
print(f"服务器启动于端口 {port}...")
server_instance.serve_forever()
print("服务器已停止。") # this line will be reached after shutdown()
except oserror as e:
print(f"!!! 启动服务器失败(端口 {port}): {e}")
# optionally notify the gui thread here if needed
# wx.callafter(frame.notify_server_start_failed, str(e))
server_instance = none # ensure instance is none if failed
except exception as e:
print(f"!!! 启动服务器时发生意外错误: {e}")
server_instance = none
# 获取本机ip地址
def get_local_ip():
ip = "127.0.0.1" # default fallback
try:
# create a socket object
s = socket.socket(socket.af_inet, socket.sock_dgram)
# doesn't need to be reachable
s.connect(("8.8.8.8", 80))
ip = s.getsockname()[0]
s.close()
except exception as e:
print(f"无法自动获取本机ip: {e},将使用 {ip}")
return ip
# 主应用程序类
class photoserverapp(wx.app):
def oninit(self):
# setappname helps with some platform integrations
self.setappname("photoserver")
self.frame = photoserverframe(none, title="本地图片服务器", size=(650, 450)) # slightly larger window
self.frame.show()
return true
# 主窗口类
class photoserverframe(wx.frame):
def __init__(self, parent, title, size):
super().__init__(parent, title=title, size=size)
# 创建面板
self.panel = wx.panel(self)
# 创建控件
folder_label = wx.statictext(self.panel, label="图片文件夹:")
self.folder_txt = wx.textctrl(self.panel, style=wx.te_readonly | wx.border_static) # use static border
self.browse_btn = wx.button(self.panel, label="选择文件夹(&b)...", id=wx.id_open) # use standard id and mnemonic
self.start_btn = wx.button(self.panel, label="启动服务(&s)")
self.stop_btn = wx.button(self.panel, label="停止服务(&t)")
status_label = wx.statictext(self.panel, label="服务器状态:")
self.status_txt = wx.textctrl(self.panel, style=wx.te_multiline | wx.te_readonly | wx.hscroll | wx.border_theme) # add scroll and theme border
# 设置停止按钮初始状态为禁用
self.stop_btn.disable()
# 绑定事件
self.bind(wx.evt_button, self.on_browse, self.browse_btn)
self.bind(wx.evt_button, self.on_start_server, self.start_btn)
self.bind(wx.evt_button, self.on_stop_server, self.stop_btn)
self.bind(wx.evt_close, self.on_close)
# --- 使用 sizers 进行布局 ---
# 主垂直 sizer
main_sizer = wx.boxsizer(wx.vertical)
# 文件夹选择行 (水平 sizer)
folder_sizer = wx.boxsizer(wx.horizontal)
folder_sizer.add(folder_label, 0, wx.align_center_vertical | wx.right, 5)
folder_sizer.add(self.folder_txt, 1, wx.expand | wx.right, 5) # 让文本框扩展
folder_sizer.add(self.browse_btn, 0, wx.align_center_vertical)
main_sizer.add(folder_sizer, 0, wx.expand | wx.all, 10) # add padding around this row
# 控制按钮行 (水平 sizer) - 居中
buttons_sizer = wx.boxsizer(wx.horizontal)
buttons_sizer.add(self.start_btn, 0, wx.right, 5)
buttons_sizer.add(self.stop_btn, 0)
main_sizer.add(buttons_sizer, 0, wx.align_center | wx.bottom, 10) # center align and add bottom margin
# 状态标签和文本框
main_sizer.add(status_label, 0, wx.left | wx.right | wx.top, 10)
main_sizer.add(self.status_txt, 1, wx.expand | wx.left | wx.right | wx.bottom, 10) # let status text expand
# 设置面板 sizer 并适应窗口
self.panel.setsizer(main_sizer)
self.panel.layout()
# self.fit() # optional: adjust window size to fit content initially
# 居中显示窗口
self.centre(wx.both) # center on screen
# 显示初始信息
self.log_status("欢迎使用图片服务器!请选择一个包含图片的文件夹,然后启动服务器。")
def on_browse(self, event):
# 弹出文件夹选择对话框
# use the current value as the default path if available
default_path = self.folder_txt.getvalue() if self.folder_txt.getvalue() else os.getcwd()
dialog = wx.dirdialog(self, "选择图片文件夹", defaultpath=default_path,
style=wx.dd_default_style | wx.dd_dir_must_exist | wx.dd_change_dir)
if dialog.showmodal() == wx.id_ok:
global selected_folder
new_folder = dialog.getpath()
# only update if the folder actually changed
if new_folder != selected_folder:
selected_folder = new_folder
self.folder_txt.setvalue(selected_folder)
self.log_status(f"已选择文件夹: {selected_folder}")
# if server is running, maybe prompt user to restart?
# or automatically enable start button if it was disabled due to no folder?
if not self.start_btn.isenabled() and not (server_thread and server_thread.is_alive()):
self.start_btn.enable()
dialog.destroy()
def on_start_server(self, event):
global server_thread, selected_folder, server_instance
# 检查是否已选择文件夹
if not selected_folder or not os.path.isdir(selected_folder):
wx.messagebox("请先选择一个有效的图片文件夹!", "错误", wx.ok | wx.icon_error, self)
return
# 检查服务器是否已经在运行 (更可靠的方式是检查 server_instance)
if server_instance is not none and server_thread is not none and server_thread.is_alive():
self.log_status("服务器已经在运行中。请先停止。")
# wx.messagebox("服务器已经在运行中。如果需要使用新文件夹,请先停止。", "提示", wx.ok | wx.icon_information, self)
return # don't restart automatically here, let user stop first
port = 8000 # you might want to make this configurable
self.log_status(f"正在尝试启动服务器在端口 {port}...")
try:
# 清理旧线程引用 (以防万一)
if server_thread and not server_thread.is_alive():
server_thread = none
# 创建并启动服务器线程
# pass the frame or a callback mechanism if start_server needs to report failure back to gui
server_thread = threading.thread(target=start_server, args=(port,), daemon=true)
server_thread.start()
# --- 短暂等待,看服务器是否启动成功 ---
# 这是一种简单的方法,更健壮的是使用事件或队列从线程通信
threading.timer(0.5, self.check_server_status_after_start, args=(port,)).start()
except exception as e:
self.log_status(f"!!! 启动服务器线程时出错: {str(e)}")
wx.messagebox(f"启动服务器线程时出错: {str(e)}", "严重错误", wx.ok | wx.icon_error, self)
def check_server_status_after_start(self, port):
# this runs in a separate thread (from timer), use wx.callafter to update gui
global server_instance
if server_instance is not none:
ip_address = get_local_ip()
url = f"http://{ip_address}:{port}"
def update_gui_success():
self.log_status("服务器已成功启动!")
self.log_status(f"本机 ip 地址: {ip_address}")
self.log_status(f"端口: {port}")
self.log_status(f"请在浏览器中访问: {url}")
self.start_btn.disable()
self.stop_btn.enable()
self.browse_btn.disable() # disable browse while running
try:
webbrowser.open(url)
except exception as wb_e:
self.log_status(f"自动打开浏览器失败: {wb_e}")
wx.callafter(update_gui_success)
else:
def update_gui_failure():
self.log_status("!!! 服务器未能成功启动,请检查端口是否被占用或查看控制台输出。")
# ensure buttons are in correct state if start failed
self.start_btn.enable()
self.stop_btn.disable()
self.browse_btn.enable()
wx.callafter(update_gui_failure)
def on_stop_server(self, event):
global server_thread, server_instance
if server_instance:
self.log_status("正在停止服务器...")
try:
# shutdown must be called from a different thread than serve_forever
# so, start a small thread just to call shutdown
def shutdown_server():
try:
server_instance.shutdown() # request shutdown
# server_instance.server_close() # close listening socket immediately
except exception as e:
# use callafter to log from this thread
wx.callafter(self.log_status, f"关闭服务器时出错: {e}")
shutdown_thread = threading.thread(target=shutdown_server)
shutdown_thread.start()
shutdown_thread.join(timeout=2.0) # wait briefly for shutdown command
# now wait for the main server thread to exit
if server_thread:
server_thread.join(timeout=2.0) # wait up to 2 seconds
if server_thread.is_alive():
self.log_status("警告:服务器线程未能及时停止。")
server_thread = none
server_instance = none # mark as stopped
# update ui
self.start_btn.enable()
self.stop_btn.disable()
self.browse_btn.enable()
self.log_status("服务器已停止!")
except exception as e:
self.log_status(f"!!! 停止服务器时发生错误: {str(e)}")
# attempt to force button state reset even if error occurred
self.start_btn.enable()
self.stop_btn.disable()
self.browse_btn.enable()
else:
self.log_status("服务器当前未运行。")
# ensure button states are correct if already stopped
self.start_btn.enable()
self.stop_btn.disable()
self.browse_btn.enable()
def log_status(self, message):
# ensure ui updates happen on the main thread
def append_text():
# optional: add timestamp
# import datetime
# timestamp = datetime.datetime.now().strftime("%h:%m:%s")
# self.status_txt.appendtext(f"[{timestamp}] {message}\n")
self.status_txt.appendtext(f"{message}\n")
self.status_txt.setinsertionpointend() # scroll to end
# if called from background thread, use callafter
if wx.ismainthread():
append_text()
else:
wx.callafter(append_text)
def on_close(self, event):
# 关闭窗口时确保服务器也被关闭
if server_instance and server_thread and server_thread.is_alive():
msg_box = wx.messagedialog(self, "服务器仍在运行。是否停止服务器并退出?", "确认退出",
wx.yes_no | wx.cancel | wx.icon_question)
result = msg_box.showmodal()
msg_box.destroy()
if result == wx.id_yes:
self.on_stop_server(none) # call stop logic
# check again if stop succeeded before destroying
if server_instance is none:
self.destroy() # proceed with close
else:
wx.messagebox("无法完全停止服务器,请手动检查。", "警告", wx.ok | wx.icon_warning, self)
# don't destroy if stop failed, let user retry maybe
elif result == wx.id_no:
self.destroy() # exit without stopping server (daemon thread will die)
else: # wx.id_cancel
# don't close the window
if event.canveto():
event.veto() # stop the close event
else:
# server not running, just exit cleanly
self.destroy() # explicitly destroy frame
if __name__ == "__main__":
# ensure we handle high dpi displays better if possible
try:
# this might need adjustment based on wxpython version and os
if hasattr(wx, 'enableasserts'): wx.enableasserts(false) # optional: disable asserts for release
# some systems might need this for high dpi scaling:
# if hasattr(wx, 'app'): wx.app.setthreadsafety(wx.app_thread_safety_none)
# if 'wxmsw' in wx.platforminfo:
# import ctypes
# try:
# ctypes.windll.shcore.setprocessdpiawareness(1) # try for win 8.1+
# except exception:
# try:
# ctypes.windll.user32.setprocessdpiaware() # try for older windows
# except exception: pass
pass # keep it simple for now
except exception as e:
print(f"无法设置 dpi 感知: {e}")
app = photoserverapp(redirect=false) # redirect=false for easier debugging output
app.mainloop()工作流程
用户使用这个工具的典型流程如下:
- 运行 python 脚本。
- 出现一个带有 “图片服务器” 标题的窗口。
- 点击 “选择文件夹” 按钮,在弹出的对话框中找到并选择一个包含图片的文件夹。
- 选中的文件夹路径会显示在文本框中。
- 点击 “启动服务器” 按钮。
- 脚本获取本机 ip 地址,在后台启动 http 服务器。
- 状态日志区域会显示服务器已启动、本机 ip 地址和端口号(通常是 8000),并提示用户可以通过 http://<本机ip>:8000 访问。
- 脚本会自动打开系统的默认浏览器,并访问上述地址。
- 浏览器中会显示一个包含所选文件夹中所有图片缩略图的网页。图片会随着滚动懒加载。
- 用户可以在浏览器中滚动浏览缩略图。
- 点击任意缩略图,会弹出一个大图预览模态框。
- 在模态框中,可以使用左右箭头或点击两侧按钮切换图片。
- 在桌面应用程序窗口中,点击 “停止服务器” 可以关闭后台服务。
- 关闭桌面应用程序窗口时,后台服务也会自动尝试停止。
主要功能与优势
简单易用: 提供图形界面,操作直观。
本地网络共享: 轻松将电脑上的图片共享给局域网内的手机、平板等设备浏览。
无需安装额外服务器软件: 利用 python 内建库,绿色便携。
跨平台潜力: python 和 wxpython 都是跨平台的,理论上可以在 windows, macos, linux 上运行(需安装相应依赖)。
现代化的 web 界面: 提供了懒加载、模态预览、键盘导航等功能,提升了浏览体验。
性能考虑: 通过懒加载和 http 缓存(针对图片文件设置了 1 天缓存,html 页面 1 小时缓存)来优化性能。
潜在改进与思考
虽然这个脚本已经相当实用,但仍有一些可以改进的地方:
更健壮的错误处理: 对文件读取、网络错误等进行更细致的处理和用户反馈。
安全性: 目前服务器对局域网内的所有设备开放,没有任何访问控制。对于敏感图片,可能需要添加密码验证等安全措施。
处理超大目录: 如果文件夹包含成千上万张图片,一次性读取所有文件名和大小可能仍然会造成短暂卡顿,可以考虑分批加载或更优化的目录扫描方式。
可配置端口: 将端口号 8000 硬编码在了代码中,可以将其改为用户可在界面上配置或通过命令行参数指定。
支持更多文件类型: 目前只处理了常见的图片格式,可以扩展支持视频预览或其他媒体类型。
异步服务器: 对于高并发场景(虽然在本应用中不太可能),可以考虑使用基于 asyncio 的 web 框架(如 aiohttp, fastapi 等)代替 socketserver,以获得更好的性能。
界面美化: wxpython 界面和 html 界面都可以进一步美化。
运行结果

总结
这个 python 脚本是一个非常实用的小工具,它完美地结合了桌面 gui 的易用性和 web 技术的灵活性,为在本地网络中快速浏览电脑上的图片提供了一个优雅的解决方案。代码结构清晰,功能完善,并且展示了 python 在快速开发网络应用方面的强大能力。无论你是想学习 gui 编程、网络服务,还是仅仅需要这样一个方便的工具,这个项目都值得一看。
以上就是使用python开发一个简单的本地图片服务器的详细内容,更多关于python本地图片服务器的资料请关注代码网其它相关文章!
发表评论