我处理来自不同人的数据,这些人以 csv 格式向我发送文件。当然,数据包带有错误、不完整等。在这里,我的脚本挂在我的服务器上,它会定期读取我的邮件、提取文件并按应有的方式处理它们。
很明显,为了防止程序在任何门柱的情况下崩溃,我在每个函数或类方法中都放置了try ... except并返回一个符号 - 子程序中的处理是否结束。每个方法函数具有以下结构:
def files_reading(self):
"""Read the files."""
self.response["exit_is_ok"] = False
try: # do something here
...
self.response["file"] = some_file
self.response["exit_is_ok"] = True
except Exception as error:
self.response["exit_message"] = f'{error} while reading'
return self.response
也就是在入口处,我将标志设置为False,在没有异常的情况下将其更改为True。如果功能有问题,当然应该跳过这个数据包。即在主循环中,每次调用该方法时,都会检查“Ok”或“Ok”。
那么主程序是这样的:
def main():
"""Calculate Amazon reports."""
...
for sender in senders:
...
izi_report = globals()[task_name](path_to_files_folder)
# ===data reading===
readed = izi_report.files_reading()
pd_files = readed["files"]
if not readed["exit_is_ok"]:
print(received["folder"], "--->", readed["exit_message"])
continue
# ===data_processing===
data_response = izi_report.data_processing(**pd_files)
if not data_response["exit_is_ok"]:
print(received["folder"], "--->", data_response["exit_message"])
continue
# ===write to xlsx===
files = data_response["files"]
xl_response = izi_report.excel_writer(files)
if not xl_response["exit_is_ok"]:
print(received["folder"], "--->", xl_response["exit_message"])
continue
...
可以看出,相同的行序列不断重复。稍后,将发送一封信,而不是打印。
所以这就是问题所在。在这种情况下如何遵守DRY规则。毕竟,您不能将“继续”放入函数中。或者更一般的问题。有没有办法更好地实现这个想法。
将每个函数中的异常包装在一个特定的函数中(以便根据函数出现错误消息),并在主函数中处理一次异常: