Каждый раз, когда я пытаюсь запустить свою программу, я получаю всплывающее окно с надписью «Python неожиданно завершил работу», но без обратной трассировки.
cli_manager.py
import importlib.util
import sys
import PySimpleGUI as sg
import multiprocessing
next_task_id = 0
def generate_data(tasks):
data = []
for i,task in enumerate(tasks):
name = task.name
description = task.description
timetoarchive = task.time_to_archive()
task_id = task.task_id
row = [name, description, timetoarchive, task_id]
data.append(row)
return data
def update_tasks(tasks):
global next_task_id
for i, task in enumerate(tasks):
if task.stopped:
tasks.pop(i)
if not task.running:
task.task_id = next_task_id
next_task_id+=1
task.start()
def import_task(path):
spec = importlib.util.spec_from_file_location("task", path)
foo = importlib.util.module_from_spec(spec)
spec.loader.exec_module(foo)
return foo.Task
def prompt_action(tasks):
print("What would you like to do?")
print('''
(C)reate new task
(K)ill running task
(L)ist tasks
(Q)uit
''')
acted = False
while not acted:
choice = str(input()).lower()
if choice == 'c':
acted = True
f = input('Enter path to file: ')
task = import_task(f)()
tasks.append(task)
elif choice == 'k':
acted = True
elif choice == 'l':
acted = True
print('Name | TTA | ID')
for task in tasks:
print(f'{task.name} | {task.time_to_archive()} | {task.task_id}')
elif choice == 'Q':
acted = True
sys.exit()
print('exit')
else:
continue
if __name__=="__main__":
tasks = []
while True:
prompt_action(tasks)
update_tasks(tasks)
Task.py
import multiprocessing
class BaseTask:
def __init__(self):
self.name = 'Task'
self.description = ''
self.constant = False
self.running = False
self.stopped = False
self.task_id = None
self.tta = multiprocessing.Value('i', -3)
self.STOP = multiprocessing.Value('i', 0)
def time_to_archive(self):
if self.constant:
return "Constant"
elif self.tta.value == -1:
return "Updating"
elif self.tta.value == -2:
self.stopped = True
return "Stopping"
elif self.tta.value == -3:
return "Starting"
else:
return self.tta.value
def start(self):
p = multiprocessing.Process(target=self.run, args=(self.tta, self.STOP))
self.p = p
self.p.daemon = True
self.p.start()
self.running = True
def stop(self):
self.STOP.value = 1
MCBackup.py
import requests
import datetime
import os
import time
from pytz import timezone
from Task import BaseTask
class Task(BaseTask):
def __init__(self):
super(Task, self).__init__()
def run(self, tta, STOP):
try:
url = 'http://18.225.13.246/backup'
tz = timezone('EST')
def backup(url):
r = requests.get(url)
ctime = datetime.datetime.now(tz).strftime("%Y%m%d-%H%M%S")
path = os.path.join('backups', f'{ctime}-world.zip')
open(path, 'wb').write(r.content)
while STOP.value == 0:
# print(f'NOT STOPPED, {self.task_id}')
backuptime = datetime.datetime.now()
print('backing up')
tta.value = -1
backup(url)
print('SUCCESSFULLY BACKED UP WORLD')
waited = False
towait = 30
prevwait = 30
while (not waited):
if STOP.value != 0:
break
newtime = datetime.datetime.now()
minsincebackup = (newtime - backuptime).seconds//60
te = minsincebackup == prevwait
if not te:
prevwait = minsincebackup
tta.value = (towait - minsincebackup)
# print(STOP.value)
if minsincebackup >= towait:
waited = True
# print(f'STOPPED {self.task_id}')
tta.value = -2
return 0
except Exception as e:
print(e)
if __name__ == '__main__':
t = Task()
t.start()
prevtta = -1
while True:
tim = t.time_to_archive()
if tim != prevtta:
print(f'Next backup in {tim} minutes')
prevtta = tim
time.sleep(1)
Я просто запускаю cli_manager.py, ввод "c", затем ввод "MCBackup.py". Нет обратной связи, просто диалог, который говорит: «Python неожиданно завершил работу». Я понятия не имею, как идти об отладке этого. Любая помощь будет оценена.
Я использую последнюю версию macOS Mojave и Python 3.7
Мне кажется, проблема в модуле запросов, учитывая, что когда я его комментирую, программа работает безупречно. Может ли быть так, что я использую request.get в multiprocessing.process?