如何调试正在运行的python程序_如何调试作为服务运行的Python程序?
我有一個python腳本,在控制臺中運行時效果很好。但是,在使用pywin32它以使其作為服務運行時,服務將安裝并開始正常,但不會生成所需的輸出。所以必定會出現問題 - 但我無法看到發現原因的原因。
該腳本執行以下操作:在給定的輸入目錄中搜索* .csv文件
如果沒有找到這樣的文件,則等待1分鐘。如果找到一個,則將其用作步驟3的輸入。如果找到多個csv文件,則使用第一個。
對列的順序進行一些轉換
將轉換后的內容寫入輸出csv
將輸入csv移動到子目錄中并重命名。
我先向您展示作為服務實現的版本的代碼:#!/usr/bin/env python3
import sys
import os
import csv
from pathlib import Path
import time
import win32service
import win32serviceutil
import win32event
def reorder_AT_csv(ifile, ofile):
"?ffnet die CSV Datei, überführt sie in das neue Format und exportiert sie."
print('[i] Lese aus ' + ifile.name + ' und schreibe in ' +
ofile.name)
with open(
ifile, mode='r') as infile, open(
ofile, mode='w') as outfile:
reader = csv.DictReader(
infile,
fieldnames=[
'Post Nummer', 'Sendungsnummer', 'Referenz1', 'Referenz2',
'Gewicht', 'Empf.Name1', 'Empf.Adresse', 'Empf.PLZ',
'Empf.Ort', 'Kostenstelle', 'Produkt'
],
delimiter=';')
fn = [
'Post Nummer', 'Referenz1', 'Referenz2', 'Sendungsnummer',
'Gewicht', 'Empf.Name1', 'Empf.Adresse', 'Empf.PLZ', 'Empf.Ort',
'Kostenstelle', 'Produkt'
]
writer = csv.DictWriter(
outfile, extrasaction='ignore', fieldnames=fn, delimiter=';')
# reorder the header first
try:
for row in reader:
# writes the reordered rows to the new file
writer.writerow(row)
except Exception as e:
print('[!] Fehler bei der Bearbeitung der CSV Datei:')
print('[!] ' + str(e))
print(
'[!] Bitte ueberpruefen, ob es sich um eine korrekte CSV Datei handelt!'
)
sys.exit(1)
def checkInputFromUser(path):
"überprüfe ob das Verzeichnis existiert."
if not path.exists():
print(
'[!] Die Eingabe ist kein Verzeichnis. Bitte ein gueltiges Verzeichnis eingeben.'
)
sys.exit(1)
return True
def findCSVFile(path):
"Finde alle CSV Dateien im Verzeichnis path."
all_files = []
all_files.extend(Path(path).glob('*.csv'))
if len(all_files) == 0:
# print('[!] Keine CSV Dateien gefunden. Bitte Pfad überprüfen.')
# sys.exit(1)
return None
elif len(all_files) > 1:
print('[i] Mehrere CSV Dateien gefunden. Nehme ersten Fund:')
return all_files[0]
def moveInputFile(input):
"Verschiebe Input Datei in Unterordner und füge Suffix hinzu."
movepath = Path(input.parent / 'processed')
targetname = input.with_suffix(input.suffix + '.success')
fulltarget = movepath / targetname.name
input.replace(fulltarget)
class CSVConvertSvc(win32serviceutil.ServiceFramework):
# you can NET START/STOP the service by the following name
_svc_name_ = "blub"
# this text shows up as the service name in the Service
# Control Manager (SCM)
_svc_display_name_ = "bar der AT CSV Dateien."
# this text shows up as the description in the SCM
_svc_description_ = "Dieser Dienst ?ffnet die AT CSV Datei und überführt sie in das DPD Format."
def __init__(self, args):
win32serviceutil.ServiceFramework.__init__(self, args)
# create an event to listen for stop requests on
self.hWaitStop = win32event.CreateEvent(None, 0, 0, None)
# core logic of the service
def SvcDoRun(self):
import servicemanager
rc = None
inputpath = Path(r'C:\Users\Dennis\Documents')
outputpath = Path(r'C:\Users\Dennis\Desktop')
file = None
# if the stop event hasn't been fired keep looping
while rc != win32event.WAIT_OBJECT_0:
checkInputFromUser(inputpath)
while file is None:
file = findCSVFile(inputpath)
if file is None:
time.sleep(60)
inputfile = file
outputfile = outputpath / 'out.csv'
reorder_AT_csv(inputfile, outputfile)
moveInputFile(inputfile)
# block for 5 seconds and listen for a stop event
rc = win32event.WaitForSingleObject(self.hWaitStop, 5000)
# called when we're being shut down
def SvcStop(self):
# tell the SCM we're shutting down
self.ReportServiceStatus(win32service.SERVICE_STOP_PENDING)
# fire the stop event
win32event.SetEvent(self.hWaitStop)
if __name__ == '__main__':
win32serviceutil.HandleCommandLine(CSVConvertSvc)
這個版本沒有做它應該做的事情。但是,我開始使用它可以作為非服務版本,而是一個簡單的python腳本,它可以執行它應該執行的操作:#!/usr/bin/env python3
import sys
import os
import csv
import time
from pathlib import Path
def reorder_AT_csv(ifile, ofile):
"?ffnet die CSV Datei, überführt sie in das neue Format und exportiert sie."
print('[i] Lese aus ' + ifile.name + ' und schreibe in ' +
ofile.name)
with open(
ifile, mode='r') as infile, open(
ofile, mode='w') as outfile:
reader = csv.DictReader(
infile,
fieldnames=[
'Post Nummer', 'Sendungsnummer', 'Referenz1', 'Referenz2',
'Gewicht', 'Empf.Name1', 'Empf.Adresse', 'Empf.PLZ',
'Empf.Ort', 'Kostenstelle', 'Produkt'
],
delimiter=';')
fn = [
'Post Nummer', 'Referenz1', 'Referenz2', 'Sendungsnummer',
'Gewicht', 'Empf.Name1', 'Empf.Adresse', 'Empf.PLZ', 'Empf.Ort',
'Kostenstelle', 'Produkt'
]
writer = csv.DictWriter(
outfile, extrasaction='ignore', fieldnames=fn, delimiter=';')
# reorder the header first
try:
for row in reader:
# writes the reordered rows to the new file
writer.writerow(row)
except Exception as e:
print('[!] Fehler bei der Bearbeitung der CSV Datei:')
print('[!] ' + str(e))
print(
'[!] Bitte ueberpruefen, ob es sich um eine korrekte CSV Datei handelt!'
)
sys.exit(1)
def checkInputFromUser(path):
"überprüfe ob das Verzeichnis existiert."
if not path.exists():
print(
'[!] Die Eingabe ist kein Verzeichnis. Bitte ein gueltiges Verzeichnis eingeben.'
)
sys.exit(1)
return True
def findCSVFile(path):
"Finde alle CSV Dateien im Verzeichnis path."
all_files = []
all_files.extend(Path(path).glob('*.csv'))
if len(all_files) == 0:
# print('[!] Keine CSV Dateien gefunden. Bitte Pfad überprüfen.')
# sys.exit(1)
return None
elif len(all_files) > 1:
print('[i] Mehrere CSV Dateien gefunden. Nehme ersten Fund:')
return all_files[0]
def moveInputFile(input):
movepath = Path(input.parent / 'processed')
targetname = input.with_suffix(input.suffix + '.success')
fulltarget = movepath / targetname.name
input.replace(fulltarget)
def main():
inputpath = Path(r'C:\Users\Dennis\Documents')
outputpath = Path(r'C:\Users\Dennis\Desktop')
file = None
checkInputFromUser(inputpath)
while file is None:
file = findCSVFile(inputpath)
if file is None:
time.sleep(60)
inputfile = file
outputfile = outputpath / 'out.csv'
reorder_AT_csv(inputfile, outputfile)
moveInputFile(inputfile)
if __name__ == '__main__':
main()
作為Python的新手,我無法弄清楚我缺少什么。該服務已正確安裝,也可以順利啟動。我使用的是隨pywin32 libs一起提供的ActiveState Python 3.6。
總結
以上是生活随笔為你收集整理的如何调试正在运行的python程序_如何调试作为服务运行的Python程序?的全部內容,希望文章能夠幫你解決所遇到的問題。
- 上一篇: AI for Science的上半场来了
- 下一篇: 敏感词屏蔽——AC自动机