文章目录
- 需求
- 分析
- Tkinter
- 基本用法
- 多窗口切换
- FTP上传
- 程序打包
- 源码
需求
项目中有个小功能模块 ,需要win下实现ftp上传功能,编写一个DEMO测试
要求
- 界面简单
- 选择本地文件 上传ftp服务器
- 显示进度条
- 显示状态
- 上传完成后显示URL分享地址
分析
Tkinter
Tkinter 是 Python 的标准 GUI(图形用户界面)库,用于创建桌面应用程序。它是 Python 的官方 GUI 库,内置于Python 发行版中,因此无需额外安装即可使用。Tkinter 提供了一组用于创建窗口、对话框、按钮、文本框等 GUI
元素的类和方法。Tkinter 的外观和功能相对基础,复杂功能不使用
- Tkinter 窗口(Tk):
创建一个 Tkinter 应用程序时,首先需要创建一个顶层窗口。这通常是通过实例化 - tkinter.Tk 类来实现的。
Tkinter 小部件(Widgets):
小部件是用户界面中用于交互的元素,如按钮、文本框、标签等。Tkinter 提供了多种小部件类,如 Button、Entry、Label 等。 - 事件处理:
Tkinter 通过事件处理机制来响应用户操作,如点击按钮、输入文本等。你可以通过绑定事件处理函数到小部件上,来处理这些事件。
基本用法
import tkinter as tk
# 创建顶层窗口
root = tk.Tk()
root.title("Tkinter 示例")
root.geometry("300x200") # 设置窗口大小
# 创建一个标签
label = tk.Label(root, text="Hello, Tkinter!")
label.pack(pady=20) # 使用 pack 布局管理器,并设置上下填充
# 定义一个按钮点击事件处理函数
def on_button_click():
label.config(text="按钮已点击!")
# 创建一个按钮,并绑定事件处理函数
button = tk.Button(root, text="点击我", command=on_button_click)
button.pack(pady=10)
# 进入主事件循环
root.mainloop()
多窗口切换
实现类似Tabs 组件以及选项卡功能
import tkinter as tk
from tkinter import ttk
class Tabs(tk.Frame):
def __init__(self, master=None, **kw):
super().__init__(master, **kw)
self.master = master
self.tabs = {} # 用于存储选项卡和对应框架的字典
self.current_tab = None # 当前显示的选项卡
# 创建一个框架来放置选项卡头部
self.header_frame = tk.Frame(self)
self.header_frame.pack(fill=tk.X)
# 创建一个框架来放置选项卡内容
self.content_frame = tk.Frame(self)
self.content_frame.pack(fill=tk.BOTH, expand=True)
def add_tab(self, tab_name, content):
# 创建选项卡头部按钮
tab_button = ttk.Button(self.header_frame, text=tab_name, command=lambda: self.switch_tab(tab_name))
tab_button.pack(side=tk.LEFT, fill=tk.X, expand=True)
# 创建内容框架并添加到字典中
content_frame = tk.Frame(self.content_frame)
content_frame.grid(row=0, column=0, sticky="nsew")
self.tabs[tab_name] = content_frame
# 将内容添加到内容框架中(这里假设content是一个widget或widget的创建函数)
if callable(content):
content(content_frame)
# 如果这是第一个选项卡,则默认显示它
if self.current_tab is None:
self.current_tab = tab_name
content_frame.grid_propagate(False) # 禁止内容框架自动调整大小
content_frame.grid_rowconfigure(0, weight=1)
content_frame.grid_columnconfigure(0, weight=1)
self.content_frame.grid_rowconfigure(0, weight=1)
self.content_frame.grid_columnconfigure(0, weight=1)
content_frame.tkraise() # 将内容框架置于顶层
def switch_tab(self, tab_name):
# 隐藏当前选项卡的内容框架
if self.current_tab:
self.tabs[self.current_tab].grid_remove()
# 显示新的选项卡内容框架
self.current_tab = tab_name
self.tabs[tab_name].grid(row=0, column=0, sticky="nsew")
self.tabs[tab_name].tkraise() # 确保新的内容框架在顶层
self.tabs[tab_name].grid_propagate(False) # 禁止内容框架自动调整大小(如果需要)
# 示例使用
def create_tab_content_a(frame):
label = tk.Label(frame, text="This is content for Tab A")
label.pack(pady=20, padx=20)
def create_tab_content_b(frame):
entry = tk.Entry(frame)
entry.pack(pady=20, padx=20)
root = tk.Tk()
root.title("模拟实现TAB切换")
root.geometry("600x400") # 设置窗口大小
tabs = Tabs(root)
tabs.pack(fill=tk.BOTH, expand=True)
# 添加选项卡
tabs.add_tab("文件上传", create_tab_content_a)
tabs.add_tab("文件列表", create_tab_content_b)
root.mainloop()
FTP上传
在Python中实现FTP文件上传功能可以使用ftplib模块,这是一个内置的库,提供了对FTP协议的支持。首先,需要确保有权限访问目标FTP服务器。此外,需要准备好要上传的文件路径和FTP服务器的相关信息(如主机名、用户名、密码等)
可以先使用命令行工具进行测试确保没有问题
下面是 ftplib 中一些常用的函数和方法
- FTP 类:
构造函数: FTP(host=‘’, user=‘’, passwd=‘’, acct=‘’, timeout=None, source_address=None)
创建一个 FTP 对象,并连接到指定的主机。 - 登录相关的方法:
login(user=‘’, passwd=‘’, acct=‘’): 使用提供的用户名、密码登录到 FTP 服务器。
quit(): 发送 QUIT 命令给服务器并关闭连接。 - 目录操作:
cwd(pathname): 更改当前工作目录。
pwd(): 获取当前工作目录。
mkd(dirname): 创建新目录。
rmd(dirname): 删除目录。
nlst([arg]): 列出目录内容,如果提供了参数,则列出该目录下的内容;如果没有提供参数,则列出当前目录的内容。
dir([arg]): 打印目录列表详情。 - 文件传输:
retrbinary(cmd, callback, blocksize=8192, rest=None): 以二进制模式接收数据。
retrlines(cmd, callback=None): 以行文本模式接收数据。
storbinary(cmd, fp, blocksize=8192, callback=None, rest=None): 以二进制模式发送数据。
storlines(cmd, fp, callback=None): 以行文本模式发送数据。
- 其他常用方法:
set_pasv(val=True): 设置是否使用被动模式(默认为 True)。
getwelcome(): 返回服务器欢迎信息。
rename(fromname, toname): 重命名远程文件或目录。
delete(filename): 删除远程文件。
size(filename): 返回远程文件大小。
voidcmd(cmd): 发送任意命令但不关心响应。
示例代码
from ftplib import FTP
# 连接到FTP服务器
ftp = FTP('your.ftp.server.com')
ftp.login('username', 'password')
# 切换到某个目录
ftp.cwd('/path/to/directory')
# 列出当前目录下的所有文件
files = ftp.nlst()
print("Files in directory:", files)
# 上传文件
with open('localfile.txt', 'rb') as file:
ftp.storbinary('STOR remote_file.txt', file)
print("File has been uploaded.")
# 下载文件
with open('downloaded_file.txt', 'wb') as file:
ftp.retrbinary('RETR remote_file.txt', file.write)
print("File has been downloaded.")
# 关闭连接
ftp.quit()
程序打包
需求 Windows 环境下将python程序打包为exe
在 Windows 下将 Python 程序打包成可执行文件(.exe),最常用的方法是使用 PyInstaller
安装
pip install pyinstaller
常见使用方法
pyinstaller --onefile my_script.py
–onefile 参数告诉 PyInstaller 将所有依赖和代码打包进一个单独的 .exe 文件中。
如果不加此参数,PyInstaller 会生成一个包含多个文件的目录。
默认情况下,生成的 .exe 文件位于 dist 目录下。你可以找到 dist/my_script.exe 文件。
–windowed:如果你的应用是一个 GUI 应用(如使用 Tkinter、PyQt 或 wxPython 编写的),并且你不希望显示控制台窗口,可以添加这个选项
–icon=your_icon.ico:为你的 .exe 文件指定一个图标
–hidden-import package_name:如果你的脚本动态导入了一些模块,而这些模块没有被自动检测到,你可以手动指定它们。
源码
import tkinter as tk
from tkinter import ttk,filedialog
from ftplib import FTP
from threading import Thread
import os
import time
import datetime
class FTPUploader:
def __init__(self, file_path, ftp_server, ftp_user, ftp_password, remote_path, progress_var,status_label):
self.file_path = file_path
self.ftp_server = ftp_server
self.ftp_user = ftp_user
self.ftp_password = ftp_password
self.remote_path = remote_path
self.progress_var = progress_var
self.uploaded_bytes = 0 # Track the uploaded bytes
self.status_label = status_label
def upload(self):
try:
file_size = os.path.getsize(self.file_path)
with open(self.file_path, 'rb') as file:
ftp = FTP(self.ftp_server)
ftp.login(user=self.ftp_user, passwd=self.ftp_password)
remote_dir, remote_filename = os.path.split(self.remote_path)
if remote_dir:
ftp.cwd(remote_dir)
# Define the callback function for progress updates
def update_progress(data):
self.uploaded_bytes += len(data)
# Calculate the progress percentage
progress = self.uploaded_bytes / file_size * 100
# Update the progress variable (which is linked to the progressbar)
# Note: This may need to be done in a thread-safe way if the GUI is not thread-safe
self.progress_var.set(progress)
# Start the upload with the callback for progress updates
print(f"Uploading {self.file_path} to {self.remote_path}...")
ftp.storbinary(f'STOR {remote_filename}', file, callback=update_progress)
ftp.quit()
print(f"File {self.file_path} uploaded successfully to {self.remote_path}")
self.status_label.config(text=f"上传{self.file_path}完成,\n播放地址为 http://10.31.32.40:8087/live/record/{remote_filename}")
#self.status_label.config(text=f"上传{self.file_path}完成")
except Exception as e:
print(f"Failed to upload file: {e}")
class UploadApp:
def __init__(self, root):
self.timestamp = datetime.datetime.now().strftime("%Y%m%d%H%M%S")
self.root = root
# 创建选择文件按钮
self.choose_button = tk.Button(root, text="选择文件", command=choose_file)
self.choose_button.pack(pady=5)
# 创建进度条
self.progress_var = tk.DoubleVar()
self.progress_bar = ttk.Progressbar(root, orient="horizontal", maximum=100, mode="determinate", variable=self.progress_var)
self.progress_bar.pack(fill=tk.X, padx=20, pady=10)
#self.progress_bar.pack(pady=20)
# 创建开始上传按钮
self.start_button = tk.Button(root, text="开始上传", command=self.start_upload)
self.start_button.pack(pady=20)
# 创建状态标签
self.status_label = tk.Label(root, text="", fg="blue") # 设置文本颜色为蓝色以便更明显
self.status_label.pack(pady=10)
# FTP credentials and file paths (these should be set by the user or securely stored)
#self.local_file_path = 'ftpclient.py'
self.ftp_server = '127.0.0.1'
self.ftp_user = 'ftpuser'
self.ftp_password = '123456'
def start_upload(self):
if 'selected_file' not in globals() or not selected_file:
self.status_label.config(text="请先选择一个文件!")
return
self.status_label.config(text="上传中...")
self.progress_var.set(0)
self.local_file_path = selected_file
local_dir, local_filename = os.path.split(self.local_file_path)
print(selected_file,self.local_file_path,local_dir,local_filename)
self.remote_file_path = f'cctvlive/{local_filename}'
#self.remote_file_path = f"cctvlive/{timestamp}.mp4"
print("start_upload ",self.remote_file_path)
# Create an instance of the FTPUploader and run it in a separate thread
self.uploader = FTPUploader(self.local_file_path, self.ftp_server, self.ftp_user, self.ftp_password, self.remote_file_path, self.progress_var,self.status_label)
upload_thread = Thread(target=self.uploader.upload)
upload_thread.start()
def choose_file():
# 打开文件选择对话框
filename = filedialog.askopenfilename()
if filename:
file_label.config(text=f"选中的文件: {filename}")
global selected_file
selected_file = filename
#progress_var.set(0)
selected_file=""
if __name__ == "__main__":
root = tk.Tk()
root.title("直播录制文件上传")
# 设置窗口大小
root.geometry('400x250')
# 创建并放置文件路径标签
file_label = tk.Label(root, text="请选择要上传的文件...")
file_label.pack(pady=10)
app = UploadApp(root)
root.mainloop()