Новый класс или новый .py Python

Question

Я пишу скрипт на python, чтобы проверить, будет ли файл добавлен в папку с помощью watchdog, этот файл будет добавлен в очередь.

Моя идея - добавить имя файла в текстовый файлзатем либо запустите новый класс, который просматривает txt, а затем выполняет строку в cmd и запускает, например, FME.

Это лучший способ написать новый .py для каждой новой программы, которую я хочу открыть.Например, один для FME и один для блокнота.

Я все еще хочу, чтобы класс wachdog перешел в фоновый режим.

• looking_for_files_and_adding_to_queue py
• looking_in_queue_for_the_next_in_line_and_direct_to_3_party py
• *1013* 1013* Блокнот py

и так далее ...

Или на all.py

class looking_for_files_and_adding_to_queue
class looking_in_queue_for_the_next_in_line_and_direct_to_3_party
class FME
class Notepad

Сегодня мой скрипт выглядит так:

import time
import sys
import os
import datetime
from watchdog.observers import Observer  
from watchdog.events import PatternMatchingEventHandler

class MyHandler(PatternMatchingEventHandler):
    patterns = ["*.tif"]
    count_move = 0

    def process(self, event):
        if self.count_move == 1:
            # the file will be processed there
            folder = "P:\\03_auto\\Indata"
            indata = event.src_path

            #Makes a new folder in Utdata based on filename
            newfolder = os.path.join(folder[:11], str("Utdata\\orto"), event.src_path[18:29])
            if not os.path.exists(newfolder):
               os.makedirs(newfolder)

            #Logg and print start of FME
            print(time.strftime('%a %H:%M:%S') + ": FME " + event.src_path[18:] + " startats i FME.")
            log_file = open("P:\\03_auto\\log.txt", "a")
            log_file.write(time.strftime('%a %H:%M:%S') + ": FME " + event.src_path[18:] + " startats i FME.\n")
            log_file.close()

            #Starting and excequting FME
            var_fme = str('fme.exe "P:\\03_auto\\Script\\tiff_to_milti_jpg_tiff\\tif_to_multi-jpg-tiff.fmw" --SourceDataset_TIFF "') + indata + str('" --FEATURE_TYPES "" --DestDataset_JPEG "') + newfolder + str('" --DestDataset_JPEG_5 "') + newfolder + str('" --DestDataset_JPEG_4 "') + newfolder + str('" --DestDataset_GEOTIFF "') + newfolder + str('" --DestDataset_GEOTIFF_3 "') + newfolder + str('"')
            os.system(var_fme)

            #Logg and pring move file
            print(time.strftime('%a %H:%M:%S') + ": Flytt " + event.src_path[18:] + " har flyttats till" + newfolder + "\nTransformering klar\n")
            log_file = open("P:\\03_auto\\log.txt", "a")
            log_file.write(time.strftime('%a %H:%M:%S') + ": Flytt " + event.src_path[18:] + " har flyttats till" + newfolder + "\nTransformering klar\n\n")
            log_file.close()

            #Move org file to Utdata\orto
            file_move = newfolder + indata[17:]
            os.rename(indata, file_move)

            #Restets script
            self.count_move = 0
        else:
            #Logg and pring loadning file while transfering
            print(time.strftime('%a %H:%M:%S') + ": Laddar " + event.src_path[18:] + " startar inladdning.")
            log_file = open("P:\\03_auto\\log.txt", "a")
            log_file.write(time.strftime('%a %H:%M:%S') + ": Laddar " + event.src_path[18:] + " startar inladdning.\n")
            log_file.close()

            #Sets counter to 1 wich enables the FME part
            self.count_move += 1


    def on_modified(self, event):
        self.process(event)

if __name__ == '__main__':
    path = "P:\\03_auto\\Indata"
    observer = Observer()
    observer.schedule(MyHandler(), path, recursive=True)
    observer.start()

    try:
        while True:
            time.sleep(1)
    except KeyboardInterrupt:
        observer.stop()

    observer.join()
enter code here

Answer 1

Вот как далеко я проделалТеперь я должен получить файлы из очереди в FME

import time
import sys
import os
import datetime
import arrow

from watchdog.observers import Observer  
from watchdog.events import PatternMatchingEventHandler
from shutil import copy

class Queue:
    def __init__(self):
        self.items =[]

    def isEmpty(self):
        return self.items == []

    def enqueue(self, item):
        self.items.insert(0, item)

    def dequeue(self):
        self.items.pop()

    def size(self):
        return len(self.items)

    def printqueue(self):
        i = 0
        for items in self.items:
            i += 1
            print(str(i) + ": " + items)

class MyHandler(PatternMatchingEventHandler):
    patterns = ["*.tif","*.pdf"]
    q = Queue()

    def on_created(self, event):
        file_name = os.path.basename(event.src_path)
        file_type = file_name.split(".")[-1]
        file_path = "path"
        file_name_path = event.src_path
        endwith = file_name.endswith("_mosaic_group1.tif")
        new_folder = "C:\\FME_workdir\\"
        new_path = new_folder + file_name

        #create new temp folder for FME 
        if not os.path.exists(new_folder):
            os.makedirs(new_folder)

        #get tif file from project
        if file_name.endswith("_mosaic_group1.tif") and not             os.path.exists(new_path):
            print("Queue:")
            self.q.enqueue("[" + file_name + ", " + file_name_path + ", " + new_path + ", " + file_type + "]")
            self.q.printqueue()
            print("\n")
            #fme = Fme()
            #return fme.runfme(file_name, file_path, file_name_path)

        #copy file to FME folder    
            if not os.path.exists(new_path):
                copy(file_name_path, new_path)

        #get the PDF report        
        elif file_name.endswith("_report.pdf") and "1_initial" in file_name_path:
            pdf_path = os.path.dirname(file_name_path)
            pdf_path_new_path = "\\".join(pdf_path.split("\\")[:3])
            pdf_path_new_dir = "\\".join(pdf_path.split("\\")[5:6])
            date_now = str(time.strftime("%y%m%d"))
            pdf_new_path = pdf_path_new_path + "\\03_leverans\\" + pdf_path_new_dir + "_" + date_now
            pdf_new_path_filename = pdf_new_path + "\\" + file_name

            if not os.path.exists(pdf_new_path):
                os.makedirs(pdf_new_path)

            copy(file_name_path, pdf_new_path_filename)

        #put inte que system
            self.q.enqueue("[" + file_name + ", " + file_name_path + ", " + pdf_new_path + ", " + file_type + "]")
            self.q.printqueue()

class Fme:
    def runfme(self, file_name, file_path, file_name_path):
        print("FME: " + self.file_name)

if __name__ == '__main__':
    path = "P:\\"
    observer = Observer()
    observer.schedule(MyHandler(), path, recursive=True)
    observer.start()

    try:
        while True:
            time.sleep(1)
    except KeyboardInterrupt:
        observer.stop()
    print("stop")

    observer.join()

    input('Press ENTER to exit')

Answer 2

tl; dr пока держите все в одном файле, затем делите его во время рефакторинга, когда размер файла увеличивается.

---

Python не заставляет вас разделять классы / функции на модули.Мы, программисты, делаем этот вызов исключительно с целью удобочитаемости и удобства обслуживания.

При рефакторинге я лично смотрю на функции с ~ 40 - 50 строками и файлами с ~ 1000 строками для разделения и стараюсь сохранить тесно связанные вещивместе.

высокая когезия и низкая связь.

является характерной чертой хорошего программного обеспечения.

Кроме того, поскольку вы, похоже, начинаете сВ этом проекте я бы порекомендовал вам сначала сконцентрироваться на создании работающей версии, а затем провести ее рефакторинг для улучшения качества кода.

Преждевременная оптимизация - корень всех зол.

---

Я предполагаю, что вы ищете предложения по улучшению качества кода, поэтому вот несколько вещей, которые могут вас заинтересовать:

• следуйте стандартам pep8: https://pep8.org
• заставьте ваши функции / методы принимать параметры вместо их жесткого кодирования, например, пути к папке, которую вы просматриваете.
• Сделайте вашу программу способной возобновлять операции даже после ошибочного / внезапного завершения: например, сохранить состояние с файлом или базой данных
• , вместо того, чтобы пытаться создать очередь самостоятельно, используйте надежные системы, такие как rabbitmq или redis.
• написать функции / методы, которые выполняют только одну операцию и делают это хорошо.