新 class 或新 .py Python
New class or new .py Python
我正在编写 python 脚本来检查文件是否已添加到带有看门狗的文件夹中，该文件是否将被添加到队列中。
我的想法是将文件名添加到 txt，然后 运行 一个新的 class 监视 txt，然后在 cmd 中执行一行并启动例如 FME。
为每个我想打开的新程序编写一个新的 .py 是不是最好的方法。例如，一个用于 FME，一个用于记事本。
我仍然希望 wachdog class 进入后台。
- 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
tl;dr 暂时将所有内容保存在一个文件中，随后在文件变大时拆分并重构。
Python 不会强制您将 类 / 函数拆分为模块。我们作为程序员做出这样的调用仅仅是为了可读性和可维护性。
在重构时，我个人会查看具有更多 ~40 - 50 行的函数和具有 ~1000 行的文件以拆分并尝试将密切相关的内容放在一起。
high cohesion and low coupling.
是好软件的特征。
此外，由于您似乎是从这个项目开始的，我建议您首先专注于制作一个可用的版本，然后重构它以提高代码质量。
premature optimization is the root of all evil.
我假设您正在寻找提高代码质量的建议，所以这里有一些您可能也感兴趣的事情：
- 遵循 pep8 标准：https://pep8.org
- 让你的函数/方法接受参数而不是硬编码它们，例如你正在观看的文件夹的路径。
- 使您的程序即使在错误/突然终止后也能够恢复操作：例如，使用文件或数据库存储状态
- 与其尝试自己实现队列，不如使用 rabbitmq 或 redis 等强大的系统。
- 编写只执行一个操作的函数/方法，并且做得很好。
这就是我所做的。现在我必须从队列中获取文件到 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')
我正在编写 python 脚本来检查文件是否已添加到带有看门狗的文件夹中，该文件是否将被添加到队列中。
我的想法是将文件名添加到 txt，然后 运行 一个新的 class 监视 txt，然后在 cmd 中执行一行并启动例如 FME。
为每个我想打开的新程序编写一个新的 .py 是不是最好的方法。例如，一个用于 FME，一个用于记事本。
我仍然希望 wachdog class 进入后台。
- 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
tl;dr 暂时将所有内容保存在一个文件中，随后在文件变大时拆分并重构。
Python 不会强制您将 类 / 函数拆分为模块。我们作为程序员做出这样的调用仅仅是为了可读性和可维护性。
在重构时，我个人会查看具有更多 ~40 - 50 行的函数和具有 ~1000 行的文件以拆分并尝试将密切相关的内容放在一起。
high cohesion and low coupling.
是好软件的特征。
此外，由于您似乎是从这个项目开始的，我建议您首先专注于制作一个可用的版本，然后重构它以提高代码质量。
premature optimization is the root of all evil.
我假设您正在寻找提高代码质量的建议，所以这里有一些您可能也感兴趣的事情：
- 遵循 pep8 标准：https://pep8.org
- 让你的函数/方法接受参数而不是硬编码它们，例如你正在观看的文件夹的路径。
- 使您的程序即使在错误/突然终止后也能够恢复操作：例如，使用文件或数据库存储状态
- 与其尝试自己实现队列，不如使用 rabbitmq 或 redis 等强大的系统。
- 编写只执行一个操作的函数/方法，并且做得很好。
这就是我所做的。现在我必须从队列中获取文件到 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')