Python에서 데이터에 대한 TCP 스트리밍 피드를 지속적으로 모니터링

Question

파이썬 3.5에서 파이썬 스크립트를 작성하고 있는데 호스트와 포트가 있으며 스크립트를 작성하여 제공된 호스트를 지속적으로 모니터링합니다. 데이터. 데이터는 xml 형식의 TCP 스트리밍 피드를 통해 배포되며 태그는 이벤트의 시작과 끝을 표시합니다.

그래서 기본적으로 XML 시작과 끝 태그 사이에 표시된 새 이벤트에 대한 TCP 피드를 모니터링 한 다음 이벤트를 검색하고 스크립트에서 적절히 처리합니다. 또한 이상적으로는 밀리 초 내에 피드의 새 데이터에 액세스해야합니다.

피드는 경고를 배포하는 정부 피드이며 피드는 streaming1.naad-adna.pelmorex.com이고 포트는 8080입니다. 새 피드에 대한 피드를 모니터링하고 액세스 할 수 있습니다. 경고를 처리하고 그에 따라 Python에서 처리합니다. 피드는 연결이 살아 있음을 나타 내기 위해 매분 하트 비트를 보냅니다.

이 특정 사용 사례에서 구현하는 방법을 잘 모르지만 최선의 방법은 소켓을 사용하는 것이라고 생각합니다. 필자는 TCP 피드에 대한 많은 경험이 없으며, 특정 유즈 케이스에서 파이썬으로 TCP 피드를 처리하는 방법과 관련하여 온라인을 많이 찾을 수 없었습니다. xml을 처리 할 수는 있었지만 TCP 피드에서 가져옵니다.

도움을 주시면 감사하겠습니다.

Answer 1

몇 가지 기술적 인 문제가 귀하의 질문에 있습니다.

먼저 서버에 연결하고 데이터를 검색하는 간단한 문제입니다. 아래의 connect()에서 볼 수 있듯이 매우 간단합니다. 소켓 (s = socket.socket())을 만들고 연결하십시오 (s.connect(('hostname', port_number))).

다음 문제는 유용한 양식으로 데이터를 검색하는 것입니다. 소켓은 기본적으로 .recv()을 제공하지만 파일과 비슷한 인터페이스로 무언가를 원했습니다. 소켓 모듈은 파이썬 고유의 메소드 인 .makefile()을 제공합니다. (return s.makefile('rb'))

이제 우리는 어려운 부분에 도달합니다. XML 문서는 일반적으로 파일 당 하나의 문서 또는 TCP 전송 당 하나의 문서로 저장됩니다. 따라서 문서의 끝은 파일 끝 표시 또는 Content-Length: 헤더로 쉽게 발견 할 수 있습니다. 결과적으로 파이썬 XML API는 하나의 파일이나 하나의 문자열에서 여러 XML 문서를 다루는 메커니즘을 가지고 있지 않다. 나는이 문제를 해결하기 위해 xml_partition()라고 썼다. xml_partition()은 파일과 유사한 객체에서 데이터를 사용하고 스트림에서 각 XML 문서를 생성합니다. (참고 : XML 문서는 함께 눌러야하며, 마지막 공백 뒤에는 공백이 허용되지 않습니다. >).

마지막으로 간단한 테스트 프로그램 (alerts())이 스트림에 연결되고 몇 개의 XML 문서를 읽고 각 파일을 자체 파일에 저장합니다.

여기 전체적으로 Pelmorex의 National Alert Aggregation & 보급 시스템에서 긴급 알리미를 다운로드하는 프로그램입니다.

import socket 
import xml.etree.ElementTree as ET 

def connect(): 
    'Connect to pelmorex data stream and return a file-like object' 
    # Set up the socket 
    s = socket.socket() 
    s.connect(('streaming1.naad-adna.pelmorex.com', 8080)) 
    return s.makefile('rb') 

# We have to consume the XML data in bits and pieces 
# so that we can stop precisely at the boundary between 
# streamed XML documents. This function ensures that 
# nothing follows a '>' in any XML fragment. 
def partition(s, pattern): 
    'Consume a file-like object, and yield parts defined by pattern' 
    data = s.read(2048) 
    while data: 
     left, middle, data = data.partition(pattern) 
     while left or middle: 
      yield left 
      yield middle 
      left, middle, data = data.partition(pattern) 
     data = s.read(2048) 

# Split the incoming XML stream into fragments (much smaller 
# than an XML document.) The end of each XML document 
# is guaranteed to align with the end of a fragment. 
# Use an XML parser to determine the actual end of 
# a document. Whenever the parser signals the end 
# of an XML document, yield what we have so far and 
# start a new parser. 
def xml_partition(s): 
    'Read multiple XML documents from one data stream' 
    parser = None 
    for part in partition(s, b'>'): 
     if parser is None: 
      parser = ET.XMLPullParser(['start', 'end']) 
      starts = ends = 0 
      xml = [] 
     xml.append(part) 
     parser.feed(part) 
     for event, elem in parser.read_events(): 
      starts += event == "start" 
      ends += event == "end" 
      if starts == ends > 0: 
       # We have reached the end of the XML doc 
       parser.close() 
       parser = None 
       yield b''.join(xml) 

# Typical usage: 
def alerts(): 
    for i, xml in enumerate(xml_partition(connect())): 
     # The XML is a bytes object that contains the undecoded 
     # XML stream. You'll probably want to parse it and 
     # somehow display the alert. 

     # I'm just saving it to a file. 
     with open('alert%d.xml' % i, 'wb') as fp: 
      fp.write(xml) 
     if i == 3: 
      break 

def test(): 
    # A test function that uses multiple XML documents in one 
    # file. This avoids the wait for a natural-disaster alert. 
    with open('multi.xml', 'rb') as fp: 
     print(list(xml_partition(fp))) 

alerts()