-4

我正在编写一个 python 脚本来检查是否将文件添加到带有看门狗的文件夹中，该文件将被添加到队列中。

我的想法是将文件名添加到 txt，然后运行一个监视 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
  • FME py
  • 记事本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
4

2 回答 2

0

tl;博士暂时将所有内容保存在一个文件中，随后在文件变大时进行拆分，同时进行重构。

Python 不会强制您将类/函数拆分为模块。作为程序员，我们提出这个要求仅仅是为了可读性和可维护性。

在重构时，我个人会查看大约 40 - 50 行的函数和大约 1000 行的文件来拆分并尝试将密切相关的东西放在一起。

高内聚和低耦合。

是好软件的一个特征。

此外，由于您似乎是从这个项目开始的，我建议您首先专注于制作一个有效的版本，然后对其进行重构以提高代码质量。

过早的优化是万恶之源。

我假设您正在这里寻找提高代码质量的建议，所以这里有一些您可能也感兴趣的事情：

  • 遵循 pep8 标准：https ://pep8.org
  • 让你的函数/方法接受参数而不是硬编码它们，例如你正在观看的文件夹的路径。
  • 使您的程序即使在错误/突然终止后也能够恢复操作：例如使用文件或数据库存储状态
  • 与其尝试自己实现队列，不如使用强大的系统，如 rabbitmq 或 redis。
  • 编写只执行一项操作并做好的函数/方法。
于 2018-09-26T17:36:22.413 回答
0

这是我做了多远。现在我必须将文件从 que 到 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')
于 2018-10-24T13:11:38.883 回答