1、tk界面 2、获取到py列表 3、runpy使用多线程运行多个py 4、使用sys.stdout将输出流依赖tempfile临时文件中转向tk界面
import multiprocessing
import os
import tempfile
import time
from multiprocessing import Process
from tkinter import *
from os import path, listdir
import tkinter.messagebox
from runpy import run_path
import sys
import psutil
# 遇到的问题,当页面中调用执行py时,如果py运行程序长久则会导致页面卡主
# \App\ --> display() -->if do() --> display() -->if do() 单进程问题
# 解决 给if do()重新启动一个进程
# 界面展示
def view_show():
app = Tk()
app.geometry('750x650')
app.config(bg='white')
app.title('运行py脚本')
text=StringVar(value='请输入一个包含py程序路径')
Entry(app,name='url_entry',bg='#EAEAEA',state='normal',textvariable=text).grid(row = 0,column=0,sticky = W)
Button(app,name='url_btn',text='获取py程序',command=get_py_program).grid(row = 0,column=1,sticky = W)
Text(app,name='text_py',width=80,height=20).grid(row=1,column=0,sticky=W)
Button(app,name='start',text="开启",command=run).grid(row=3,column=0,sticky=W)
Button(app,name='stop',text="停止",command=stop).grid(row=3,column=1,sticky=W)
Text(app, name='result_py', width=80, height=20).grid(row=2, column=0, sticky=W)
return app
# 解析路径并获取py程序
def get_py_program():
entry = app.children['url_entry'] # /Volumes/mac资料/Python-env/env-python/lib/python3.7
url=entry.get()
if path.exists(url):
list=listdir(url)
for li in list:
suffix=path.splitext(li)[1]
if suffix=='.py' :
ss.add(path.join(url,li))
else:
tkinter.messagebox.showerror("❌","路径并不存在!!")
text=app.children['text_py']
if ss:
for s in ss:
text.insert('end',s)
text.insert(tkinter.INSERT,'\n')
# 运行
def run():
if ss:
savedStdout = sys.stdout # 保存标准输出流
# with open('/Volumes/mac资料/Python-env/env-python/Projects/hello.txt', 'w+') as file:
# file=tempfile.TemporaryFile(mode='w+', suffix='temp', prefix='标准输出_临时文件',
# dir='/Volumes/mac资料/Python-env/env-python/Projects/')
file = tempfile.mkstemp(prefix='标准输出_临时文件', text=True,
dir='/Volumes/mac资料/Python-env/env-python/Projects/')
with open(file[1],'r+') as f:
sys.stdout = f # 标准输出重定向至文件
watcher()
i=1
for s in ss:
p = Process(name='print'+str(i), target=lambda: run_path(s,
run_name='__main__'))
i=i+1
p.start()
sys.stdout = savedStdout # 恢复标准输出流
global tempurl
tempurl=file[1]
# print(multiprocessing.active_children())
# print(os.getpid())
# print('子进程',psutil.Process(os.getpid()).children)
# 停止
def stop():
for p in multiprocessing.active_children():
if p.name=="print":
p.terminate()
def watcher():
global tempurl
childrens=multiprocessing.active_children()
if childrens:
print(childrens)
if not len(childrens): # 所有进程都已执行完
if tempurl:
print(tempurl)
with open(tempurl, 'r+') as file:
file.seek(os.SEEK_SET)
process_result = file.read()
result_text = app.children['result_py']
if process_result:
result_text.insert('end', process_result)
result_text.insert(tkinter.INSERT, '\n')
os.remove(tempurl)
tempurl=None
app.after(1000,watcher)
def temp():
global tempurl
tempurl=None
if __name__=='__main__':
ss = set()
tempurl=None
app=view_show()
# app.after(0,watcher)
app.mainloop()