python에서watchdog 파일 모니터링 및 업로드 감지 기능
전편에서 관찰자 모델의 원리를 소개했습니다. 본편은 이것에 대해 작은 응용 프로그램을 하나 더 소개하고자 합니다. 저도 오후 내내 놀았지만 그때 옵서버를 보고 마침 찾았습니다. 그리고 Django-observer도 있었지만 Django는 오랫동안 사용하지 않았기 때문에 이것을 노트로 삼았습니다.
watchdog 소개
Watchdog의 중국어'문지기', 보호의 뜻이 있습니다.최초로 Watchdog를 도입한 것은 단편기 시스템에서 단편기의 작업 환경이 외부 자장의 방해를 받기 쉽기 때문에 프로그램이'날아다니기'를 하여 전체 시스템이 정상적으로 작동하지 못하게 하기 때문에'문지기'를 도입하여 단편기의 운행 상태를 실시간으로 모니터링하고 운행 고장에 대해 보호 처리를 한다. 예를 들어 시스템을 다시 켜는 것이다.이런 Watchdog는 하드웨어 차원에 속하기 때문에 반드시 하드웨어 회로의 지원이 있어야 한다.
리눅스도 Watchdog를 도입했다. 리눅스 커널에서 Watchdog가 시작되면 타이머를 설정했다. 시간 초과 시간 내에/dev/Watchdog에 대해 쓰기 작업을 하지 않으면 시스템이 다시 시작될 수 있다.타이머를 통해 이루어진 Watchdog는 소프트웨어 차원에 속한다.
응, 그렇잖아.위의 이 말은 아무 소용이 없는 것 같아서 이야기거리가 되어도 안 된다.나도 직접 바이두 1편을 복사해서 소개로 삼았는데 습관이 그렇다.(수동 개머리)
python에서 파일 모니터링은 주로 두 개의 라이브러리가 있는데 하나는pyinotify(https://github.com/seb-m/pyinotify/wiki), 하나는 watchdog(http://pythonhosted.org/watchdog/).pyinotify는 Linux 플랫폼의 inotify에 의존하고 후자는 서로 다른 플랫폼의 이벤트를 봉인했다.
watchdog 사용
python에서 pip를 통해 직접 설치할 수 있습니다.
pip install watchdog -i https://pypi.tuna.tsinghua.edu.cn/simple이벤트 클래스(event):
watchdog.events.FileSystemEvent(event_type,
src_path,
is_directory=False) 그럼 이제 자리가 생겼어요.
def on_created(event):
print(f"hey, {event.src_path} has been created!")
def on_deleted(event):
print(f"Someone deleted {event.src_path}!")
def on_modified(event):
print(f"hey buddy, {event.src_path} has been modified")
def on_moved(event):
print(f"ok ok ok, someone moved {event.src_path} to {event.dest_path}")
path = "."
go_recursively = True
my_observer = Observer()
my_observer.schedule(my_event_handler, path, recursive=True)그러면 여기까지 알게 되면 데모 프로그램을 써서 테스트를 진행할 수 있습니다.
from watchdog.observers import Observer
from watchdog.events import *
import time
class FileEventHandler(FileSystemEventHandler):
def __init__(self):
FileSystemEventHandler.__init__(self)
def on_moved(self, event):
if event.is_directory:
print("directory moved from {0} to {1}".format(event.src_path,event.dest_path))
else:
print("file moved from {0} to {1}".format(event.src_path,event.dest_path))
def on_created(self, event):
if event.is_directory:
print("directory created:{0}".format(event.src_path))
else:
print("file created:{0}".format(event.src_path))
def on_deleted(self, event):
if event.is_directory:
print("directory deleted:{0}".format(event.src_path))
else:
print("file deleted:{0}".format(event.src_path))
def on_modified(self, event):
if event.is_directory:
print("directory modified:{0}".format(event.src_path))
else:
print("file modified:{0}".format(event.src_path))
if __name__ == "__main__":
observer = Observer()
event_handler = FileEventHandler()
observer.schedule(event_handler,r"D:\code\dingshirenwu",True)
observer.start()
try:
while True:
time.sleep(1)
except KeyboardInterrupt:
observer.stop()
observer.join()하지만 이것은 단지 하나의 모니터링일 뿐, 우리는 순환을 통해 여러 개의 폴더를 감시할 수 있다.
dirs = [r'D:\code\dingshirenwu', r'D:\code\tuiliu']
for dir in dirs:
event_handler = FileEventHandler()
observer.schedule(event_handler, dir, True)
observer.start()
import logging
import queue
import threading
import time
import watchdog.observers as observers
import watchdog.events as events
from ftplib import FTP
logger = logging.getLogger(__name__)
SENTINEL = None
def upload(f, remote_path, local_path):
fp = open(local_path, "rb")
buf_size = 1024
f.storbinary("STOR {}".format(remote_path), fp, buf_size)
fp.close()
class MyEventHandler(events.FileSystemEventHandler):
def on_any_event(self, event):
super(MyEventHandler, self).on_any_event(event)
queue.put(event)
def __init__(self, queue):
self.queue = queue
def process(queue):
while True:
event = queue.get()
logger.info(event)
print(event.key) # tuple
('modified', 'C:\\Users\\admin\\Desktop\\ \\test\\GitHub\\isadb\\.idea', True)
if (event.key)[0] == "created":
upload(ftp, remote_path, event.src_path)
if __name__ == '__main__':
logging.basicConfig(level=logging.DEBUG,
format='[%(asctime)s %(threadName)s] %(message)s',
datefmt='%H:%M:%S')
ftp = FTP()
ftp.connect("x.x.x.x", 21) # ftp ip ， ftp ， 21
ftp.login(username, password) # ftp.login()
queue = queue.Queue()
num_workers = 4
pool = [threading.Thread(target=process, args=(queue,)) for i in range(num_workers)]
for t in pool:
t.daemon = True
t.start()
event_handler = MyEventHandler(queue)
observer = observers.Observer()
observer.schedule(
event_handler,
path=r'C:\Users\admin\Desktop\ \test\GitHub\isadb',
recursive=True)
observer.start()
try:
while True:
time.sleep(1)
except KeyboardInterrupt:
observer.stop()
observer.join()그리고 Stack Overflow에서 해결 방안을 찾았습니다. 큰 파일을 업로드할 때 빈 텍스트를 업로드하고 이 파일의 크기를 기록한 다음에 이 파일을 윤문합니다. 이 파일의 크기가 더 이상 변하지 않을 때만 이 파일이 성공적으로 생성되었다고 생각합니다. 이때 업로드를 고려하지만 저도 데모를 씁니다. 너무 번거롭습니다...만약 더 좋은 방법이 있다면, 나를 평론하거나 개인적으로 믿어도 된다.
이는python에서watchdog 파일 감시와 검측에 관한 글을 소개합니다. 더 많은 관련pythonwatchdog 감시 파일 내용은 저희 이전의 글을 검색하거나 아래의 관련 글을 계속 훑어보십시오. 앞으로 많은 응원 부탁드립니다!
이 내용에 흥미가 있습니까?
현재 기사가 여러분의 문제를 해결하지 못하는 경우 AI 엔진은 머신러닝 분석(스마트 모델이 방금 만들어져 부정확한 경우가 있을 수 있음)을 통해 가장 유사한 기사를 추천합니다:
로마 숫자를 정수로 또는 그 반대로 변환그 중 하나는 로마 숫자를 정수로 변환하는 함수를 만드는 것이었고 두 번째는 그 반대를 수행하는 함수를 만드는 것이었습니다. 문자만 포함합니다'I', 'V', 'X', 'L', 'C', 'D', 'M' ; 문자열이 ...
텍스트를 자유롭게 공유하거나 복사할 수 있습니다.하지만 이 문서의 URL은 참조 URL로 남겨 두십시오.
CC BY-SA 2.5, CC BY-SA 3.0 및 CC BY-SA 4.0에 따라 라이센스가 부여됩니다.
좋은 웹페이지 즐겨찾기
