90f68d38-3b4e-4f37-a4d8-85d436e6e297

Committed 01 Jan 2025 10:39AM UTC coverage: 87.405% (+6.2%) from 81.215%

Build # 90f68d38-3b4e-4f37-a4d8-85d436e6e297

Build Type

push

circleci

Committed by

web-flow

Commit Message

Support telnet for Python 3.13 (#29)

Run Details

15 of 22 new or added lines in 1 file covered. (68.18%)

1 existing line in 1 file now uncovered.

805 of 921 relevant lines covered (87.4%)

5.2 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

84.91

/logparser/telnet.py

# coding: utf-8
import io
import logging
import os
import platform
import re
import sys

# DeprecationWarning: 'telnetlib' is deprecated and slated for removal in Python 3.13
try:
    import telnetlib
except ImportError:
    telnetlib = None
    
import traceback

import pexpect

from .common import Common
from .utils import get_logger


logger = get_logger(__name__)

SUPPORTED_SCRAPY_VERSION = '1.5.1'
TELNET_TIMEOUT = 10
TELNET_LOG_FILE = 'telnet_log'
TELNETCONSOLE_DEFAULT_USERNAME = 'scrapy'
TELNETCONSOLE_COMMAND_MAP = dict(
    log_file=b'settings.attributes["LOG_FILE"].value',
    crawler_stats=b'stats.get_stats()',
    crawler_engine=b'est()'
)


# noinspection PyBroadException
class MyTelnet(Common):
    logger = logger
    # Linux-5.0.9-301.fc30.x86_64-x86_64-with-fedora-30-Thirty'
    on_fedora = 'fedora' in platform.platform()

    def __init__(self, data, override_telnet_console_host, verbose):
        self.data = data
        self.OVERRIDE_TELNET_CONSOLE_HOST = override_telnet_console_host
        self.verbose = verbose
        if self.verbose:
            self.logger.setLevel(logging.DEBUG)
        else:
            self.logger.setLevel(logging.INFO)

        self.scrapy_version = self.data['latest_matches']['scrapy_version'] or '0.0.0'
        self.telnet_console = self.data['latest_matches']['telnet_console']
        self.telnet_username = self.data['latest_matches']['telnet_username'] or TELNETCONSOLE_DEFAULT_USERNAME
        self.telnet_password = self.data['latest_matches']['telnet_password']
        self.host = None
        self.port = None
        self.tn = None
        self.crawler_stats = {}
        self.crawler_engine = {}

    def main(self):
        try:
            self.run()
        # Cannot catch error directly in setup_pexpect()
        # pexpect.exceptions.EOF: End Of File (EOF). Exception style platform.
        # pexpect.exceptions.TIMEOUT: Timeout exceeded  # Old logfile but connected to 1.5.1
        # except (pexpect.exceptions.EOF, pexpect.exceptions.TIMEOUT) as err:
        # In setup_telnet(): # except ConnectionRefusedError as err:  # Python 2: <class 'socket.error'>
        except Exception as err:
            self.logger.error("Fail to telnet to %s:%s for %s (%s). Maybe the job was stopped: %s",
                              self.host, self.port, self.data['log_path'], self.scrapy_version, err)
            if self.verbose:
                self.logger.error(traceback.format_exc())
        finally:
            if self.tn is not None:
                try:
                    self.tn.close()
                except:
                    pass
            self.tn = None

        return self.crawler_stats, self.crawler_engine

    # https://stackoverflow.com/questions/18547412/python-telnetlib-to-connect-to-scrapy-telnet-to-read-stats
    def run(self):
        self.logger.debug("scrapy_version: %s", self.scrapy_version)
        # Telnet via pexpect would cause '[twisted] CRITICAL: Unhandled Error' in Scrapy log on Fedora:
        # twisted/conch/telnet.py line 585, in dataReceived
        # raise ValueError("Stumped", b)
        # builtins.ValueError: ('Stumped', b'\\xec')
        if (self.ON_WINDOWS or self.on_fedora) and self.scrapy_version > SUPPORTED_SCRAPY_VERSION:
            self.logger.error("Telnet only supports scrapy<=%s if you are running Scrapyd on Windows and Fedora, "
                              "current scrapy_version: %s", SUPPORTED_SCRAPY_VERSION, self.scrapy_version)
            return
        # Telnet console listening on 127.0.0.1:6023
        m = re.search(r'^(.+):(\d+)$', self.telnet_console)
        if not m:
            self.logger.warning("Fail to extract host and port from %s", self.telnet_console)
            return
        self.host, self.port = m.groups()
        self.host = self.OVERRIDE_TELNET_CONSOLE_HOST or self.host

        self.logger.debug("Try to telnet to %s:%s for %s", self.host, self.port, self.data['log_path'])
        if self.telnet_password or telnetlib is None:
            self.setup_pexpect()
            if self.tn is not None:
                self.pexpect_io()
        else:
            self.setup_telnet()
            if self.tn is not None:
                self.telnet_io()

    def setup_pexpect(self):
        # Cannot catch error directly here, see main()
        self.tn = pexpect.spawn('telnet %s %s' % (self.host, self.port), encoding='utf-8', timeout=TELNET_TIMEOUT)
        # logfile: <open file '<stdout>', mode 'w' at 0x7fe160149150>
        # logfile_read: None
        # logfile_send: None
        if self.verbose:
            self.tn.logfile = sys.stdout
        else:
            self.tn.logfile = io.open(os.path.join(self.CWD, TELNET_LOG_FILE), 'w')

    @staticmethod
    def telnet_callback(tn, command, option):
        if command == telnetlib.DO and option == telnetlib.TTYPE:
            tn.sendall(telnetlib.IAC + telnetlib.WILL + telnetlib.TTYPE)
            tn.sendall(telnetlib.IAC + telnetlib.SB + telnetlib.TTYPE + '\0' + 'LogParser' + telnetlib.IAC + telnetlib.SE)
        elif command in (telnetlib.DO, telnetlib.DONT):
            tn.sendall(telnetlib.IAC + telnetlib.WILL + option)
        elif command in (telnetlib.WILL, telnetlib.WONT):
            tn.sendall(telnetlib.IAC + telnetlib.DO + option)

    def setup_telnet(self):
        self.tn = telnetlib.Telnet(self.host, int(self.port), timeout=TELNET_TIMEOUT)
        # [twisted] CRITICAL: Unhandled Error
        # Failure: twisted.conch.telnet.OptionRefused: twisted.conch.telnet.OptionRefused
        # https://github.com/jookies/jasmin-web/issues/2
        self.tn.set_option_negotiation_callback(self.telnet_callback)
        if self.verbose:
            self.tn.set_debuglevel(logging.DEBUG)

    def parse_output(self, text):
        m = re.search(r'{.+}', text)
        if m:
            result = self.parse_crawler_stats(m.group())
        else:
            lines = [line for line in re.split(r'\r\n|\n|\r', text) if ':' in line]
            result = dict([re.split(r'\s*:\s*', line, maxsplit=1) for line in lines])
            for k, v in result.items():
                if k == 'engine.spider.name':
                    continue
                elif v == 'True':
                    result[k] = True
                elif v == 'False':
                    result[k] = False
                else:
                    try:
                        result[k] = int(float(v))
                    except (TypeError, ValueError):
                        pass
        if result:
            return self.get_ordered_dict(result, source='telnet')
        else:
            return {}

    def pexpect_io(self):
        def bytes_to_str(src):
            if self.PY2:
                return src
            return src.decode('utf-8')
        # TypeError: got <type 'str'> ('Username: ') as pattern,
        # must be one of: <type 'unicode'>, pexpect.EOF, pexpect.TIMEOUT
        if self.telnet_password:
            self.tn.expect(u'Username: ', timeout=TELNET_TIMEOUT)
            self.tn.sendline(self.telnet_username)
            self.tn.expect(u'Password: ', timeout=TELNET_TIMEOUT)
            self.tn.sendline(self.telnet_password)
            self.tn.expect(u'>>>', timeout=TELNET_TIMEOUT)
            self.logger.debug("Login successfully")
        else:
            self.tn.expect(u'>>>', timeout=TELNET_TIMEOUT)
            self.logger.debug("Connect successfully")

        self.tn.sendline(bytes_to_str(TELNETCONSOLE_COMMAND_MAP['log_file']))
        self.tn.expect(re.compile(r'[\'"].+>>>', re.S), timeout=TELNET_TIMEOUT)
        log_file = self.tn.after
        self.logger.debug("settings['LOG_FILE'] found via telnet: %s", log_file)
        if not self.verify_log_file_path(self.parse_log_path(self.data['log_path']), log_file):
            self.logger.warning("Skip telnet due to mismatching: %s AND %s", self.data['log_path'], log_file)
            return

        self.tn.sendline(bytes_to_str(TELNETCONSOLE_COMMAND_MAP['crawler_stats']))
        self.tn.expect(re.compile(r'{.+>>>', re.S), timeout=TELNET_TIMEOUT)
        self.crawler_stats = self.parse_output(self.tn.after)

        self.tn.sendline(bytes_to_str(TELNETCONSOLE_COMMAND_MAP['crawler_engine']))
        self.tn.expect(re.compile(r'Execution engine status.+>>>', re.S), timeout=TELNET_TIMEOUT)
        self.crawler_engine = self.parse_output(self.tn.after)

    def _telnet_io(self, command):
        # Microsoft Telnet> o
        # ( to )127.0.0.1 6023
        # >>>stats.get_stats()
        # >>>est()
        self.tn.write(b'%s\n' % command)
        content = self.tn.read_until(b'\n>>>', timeout=TELNET_TIMEOUT)
        # print(repr(content))
        # b"\x1bc>>> \x1b[4hstats.get_stats()\r\r\r\n{'log_count/INFO': 61,
        # 'start_time': datetime.datetime(2019, 1, 22, 9, 7, 14, 998126),
        # 'httperror/response_ignored_status_count/404': 1}\r\r\r\n>>>"
        # b' est()\r\r\r\nExecution engine status\r\r\r\n\r\r\r\n
        # time()-engine.start_time                        : 3249.7548048496246
        # engine.scraper.slot.needs_backout()             : False\r\r\r\n\r\r\r\n\r\r\r\n>>>'
        return content.decode('utf-8')

    def telnet_io(self):
        # spider._job, spider._version, settings.attributes["BOT_NAME"].value, JOB, SPIDER, PROJECT
        # '\'logs\\\\demo_persistent\\\\test\\\\2019-01-23T18_25_34.log\'\r\r\r\n>>>'
        log_file = self._telnet_io(TELNETCONSOLE_COMMAND_MAP['log_file'])
        self.logger.debug("settings['LOG_FILE'] found via telnet: %s", log_file)
        # Username: Password:
        if 'Username:' in log_file:
            self.logger.error("Telnet with auth is not supported on Windows. You can use scrapy<=%s instead: %s",
                              SUPPORTED_SCRAPY_VERSION, log_file)
            return
        if not self.verify_log_file_path(self.parse_log_path(self.data['log_path']), log_file):
            self.logger.warning("Skip telnet due to mismatching: %s vs %s", self.data['log_path'], log_file)
            return
        self.crawler_stats = self.parse_output(self._telnet_io(TELNETCONSOLE_COMMAND_MAP['crawler_stats']))
        self.crawler_engine = self.parse_output(self._telnet_io(TELNETCONSOLE_COMMAND_MAP['crawler_engine']))

    def verify_log_file_path(self, parts, log_file):
        for part in parts:
            if part not in log_file:
                self.logger.warning("%s not found in settings['LOG_FILE']: %s", part, log_file)
                return False
        return True

1	# coding: utf-8
2	import io	6✔
3	import logging	6✔
4	import os	6✔
5	import platform	6✔
6	import re	6✔
7	import sys	6✔
8
9	# DeprecationWarning: 'telnetlib' is deprecated and slated for removal in Python 3.13
10	try:	6✔
11	import telnetlib	6✔
NEW 12	except ImportError:	1✔
NEW 13	telnetlib = None	1✔
14
15	import traceback	6✔
16
17	import pexpect	6✔
18
19	from .common import Common	6✔
20	from .utils import get_logger	6✔
21
22
23	logger = get_logger(__name__)	6✔
24
25	SUPPORTED_SCRAPY_VERSION = '1.5.1'	6✔
26	TELNET_TIMEOUT = 10	6✔
27	TELNET_LOG_FILE = 'telnet_log'	6✔
28	TELNETCONSOLE_DEFAULT_USERNAME = 'scrapy'	6✔
29	TELNETCONSOLE_COMMAND_MAP = dict(	6✔
30	log_file=b'settings.attributes["LOG_FILE"].value',
31	crawler_stats=b'stats.get_stats()',
32	crawler_engine=b'est()'
33	)
34
35
36	# noinspection PyBroadException
37	class MyTelnet(Common):	6✔
38	logger = logger	6✔
39	# Linux-5.0.9-301.fc30.x86_64-x86_64-with-fedora-30-Thirty'
40	on_fedora = 'fedora' in platform.platform()	6✔
41
42	def __init__(self, data, override_telnet_console_host, verbose):	6✔
43	self.data = data	6✔
44	self.OVERRIDE_TELNET_CONSOLE_HOST = override_telnet_console_host	6✔
45	self.verbose = verbose	6✔
46	if self.verbose:	6✔
47	self.logger.setLevel(logging.DEBUG)	×
48	else:
49	self.logger.setLevel(logging.INFO)	6✔
50
51	self.scrapy_version = self.data['latest_matches']['scrapy_version'] or '0.0.0'	6✔
52	self.telnet_console = self.data['latest_matches']['telnet_console']	6✔
53	self.telnet_username = self.data['latest_matches']['telnet_username'] or TELNETCONSOLE_DEFAULT_USERNAME	6✔
54	self.telnet_password = self.data['latest_matches']['telnet_password']	6✔
55	self.host = None	6✔
56	self.port = None	6✔
57	self.tn = None	6✔
58	self.crawler_stats = {}	6✔
59	self.crawler_engine = {}	6✔
60
61	def main(self):	6✔
62	try:	6✔
63	self.run()	6✔
64	# Cannot catch error directly in setup_pexpect()
65	# pexpect.exceptions.EOF: End Of File (EOF). Exception style platform.
66	# pexpect.exceptions.TIMEOUT: Timeout exceeded # Old logfile but connected to 1.5.1
67	# except (pexpect.exceptions.EOF, pexpect.exceptions.TIMEOUT) as err:
68	# In setup_telnet(): # except ConnectionRefusedError as err: # Python 2: <class 'socket.error'>
69	except Exception as err:	6✔
70	self.logger.error("Fail to telnet to %s:%s for %s (%s). Maybe the job was stopped: %s",	6✔
71	self.host, self.port, self.data['log_path'], self.scrapy_version, err)
72	if self.verbose:	6✔
73	self.logger.error(traceback.format_exc())	×
74	finally:
75	if self.tn is not None:	6✔
76	try:	6✔
77	self.tn.close()	6✔
78	except:	×
79	pass	×
80	self.tn = None	6✔
81
82	return self.crawler_stats, self.crawler_engine	6✔
83
84	# https://stackoverflow.com/questions/18547412/python-telnetlib-to-connect-to-scrapy-telnet-to-read-stats
85	def run(self):	6✔
86	self.logger.debug("scrapy_version: %s", self.scrapy_version)	6✔
87	# Telnet via pexpect would cause '[twisted] CRITICAL: Unhandled Error' in Scrapy log on Fedora:
88	# twisted/conch/telnet.py line 585, in dataReceived
89	# raise ValueError("Stumped", b)
90	# builtins.ValueError: ('Stumped', b'\\xec')
91	if (self.ON_WINDOWS or self.on_fedora) and self.scrapy_version > SUPPORTED_SCRAPY_VERSION:	6✔
92	self.logger.error("Telnet only supports scrapy<=%s if you are running Scrapyd on Windows and Fedora, "	×
93	"current scrapy_version: %s", SUPPORTED_SCRAPY_VERSION, self.scrapy_version)
94	return	×
95	# Telnet console listening on 127.0.0.1:6023
96	m = re.search(r'^(.+):(\d+)$', self.telnet_console)	6✔
97	if not m:	6✔
98	self.logger.warning("Fail to extract host and port from %s", self.telnet_console)	6✔
99	return	6✔
100	self.host, self.port = m.groups()	6✔
101	self.host = self.OVERRIDE_TELNET_CONSOLE_HOST or self.host	6✔
102
103	self.logger.debug("Try to telnet to %s:%s for %s", self.host, self.port, self.data['log_path'])	6✔
104	if self.telnet_password or telnetlib is None:	6✔
105	self.setup_pexpect()	6✔
106	if self.tn is not None:	6✔
107	self.pexpect_io()	6✔
108	else:
109	self.setup_telnet()	5✔
110	if self.tn is not None:	5✔
111	self.telnet_io()	5✔
112
113	def setup_pexpect(self):	6✔
114	# Cannot catch error directly here, see main()
115	self.tn = pexpect.spawn('telnet %s %s' % (self.host, self.port), encoding='utf-8', timeout=TELNET_TIMEOUT)	6✔
116	# logfile: <open file '<stdout>', mode 'w' at 0x7fe160149150>
117	# logfile_read: None
118	# logfile_send: None
119	if self.verbose:	6✔
120	self.tn.logfile = sys.stdout	×
121	else:
122	self.tn.logfile = io.open(os.path.join(self.CWD, TELNET_LOG_FILE), 'w')	6✔
123
124	@staticmethod	6✔
125	def telnet_callback(tn, command, option):	6✔
126	if command == telnetlib.DO and option == telnetlib.TTYPE:	5✔
NEW 127	tn.sendall(telnetlib.IAC + telnetlib.WILL + telnetlib.TTYPE)	×
NEW 128	tn.sendall(telnetlib.IAC + telnetlib.SB + telnetlib.TTYPE + '\0' + 'LogParser' + telnetlib.IAC + telnetlib.SE)	×
129	elif command in (telnetlib.DO, telnetlib.DONT):	5✔
NEW 130	tn.sendall(telnetlib.IAC + telnetlib.WILL + option)	×
131	elif command in (telnetlib.WILL, telnetlib.WONT):	5✔
132	tn.sendall(telnetlib.IAC + telnetlib.DO + option)	5✔
133
134	def setup_telnet(self):	6✔
135	self.tn = telnetlib.Telnet(self.host, int(self.port), timeout=TELNET_TIMEOUT)	5✔
136	# [twisted] CRITICAL: Unhandled Error
137	# Failure: twisted.conch.telnet.OptionRefused: twisted.conch.telnet.OptionRefused
138	# https://github.com/jookies/jasmin-web/issues/2
139	self.tn.set_option_negotiation_callback(self.telnet_callback)	5✔
140	if self.verbose:	5✔
141	self.tn.set_debuglevel(logging.DEBUG)	×
142
143	def parse_output(self, text):	6✔
144	m = re.search(r'{.+}', text)	6✔
145	if m:	6✔
146	result = self.parse_crawler_stats(m.group())	6✔
147	else:
148	lines = [line for line in re.split(r'\r\n\|\n\|\r', text) if ':' in line]	6✔
149	result = dict([re.split(r'\s:\s', line, maxsplit=1) for line in lines])	6✔
150	for k, v in result.items():	6✔
151	if k == 'engine.spider.name':	6✔
152	continue	6✔
153	elif v == 'True':	6✔
154	result[k] = True	×
155	elif v == 'False':	6✔
156	result[k] = False	6✔
157	else:
158	try:	6✔
159	result[k] = int(float(v))	6✔
160	except (TypeError, ValueError):	6✔
161	pass	6✔
162	if result:	6✔
163	return self.get_ordered_dict(result, source='telnet')	6✔
164	else:
165	return {}	×
166
167	def pexpect_io(self):	6✔
168	def bytes_to_str(src):	6✔
169	if self.PY2:	6✔
170	return src	×
171	return src.decode('utf-8')	6✔
172	# TypeError: got <type 'str'> ('Username: ') as pattern,
173	# must be one of: <type 'unicode'>, pexpect.EOF, pexpect.TIMEOUT
174	if self.telnet_password:	6✔
175	self.tn.expect(u'Username: ', timeout=TELNET_TIMEOUT)	6✔
176	self.tn.sendline(self.telnet_username)	6✔
177	self.tn.expect(u'Password: ', timeout=TELNET_TIMEOUT)	6✔
178	self.tn.sendline(self.telnet_password)	6✔
179	self.tn.expect(u'>>>', timeout=TELNET_TIMEOUT)	6✔
180	self.logger.debug("Login successfully")	6✔
181	else:
NEW 182	self.tn.expect(u'>>>', timeout=TELNET_TIMEOUT)	1✔
NEW 183	self.logger.debug("Connect successfully")	×
184
185	self.tn.sendline(bytes_to_str(TELNETCONSOLE_COMMAND_MAP['log_file']))	6✔
186	self.tn.expect(re.compile(r'[\'"].+>>>', re.S), timeout=TELNET_TIMEOUT)	6✔
187	log_file = self.tn.after	6✔
188	self.logger.debug("settings['LOG_FILE'] found via telnet: %s", log_file)	6✔
189	if not self.verify_log_file_path(self.parse_log_path(self.data['log_path']), log_file):	6✔
190	self.logger.warning("Skip telnet due to mismatching: %s AND %s", self.data['log_path'], log_file)	×
191	return	×
192
193	self.tn.sendline(bytes_to_str(TELNETCONSOLE_COMMAND_MAP['crawler_stats']))	6✔
194	self.tn.expect(re.compile(r'{.+>>>', re.S), timeout=TELNET_TIMEOUT)	6✔
195	self.crawler_stats = self.parse_output(self.tn.after)	6✔
196
197	self.tn.sendline(bytes_to_str(TELNETCONSOLE_COMMAND_MAP['crawler_engine']))	6✔
198	self.tn.expect(re.compile(r'Execution engine status.+>>>', re.S), timeout=TELNET_TIMEOUT)	6✔
199	self.crawler_engine = self.parse_output(self.tn.after)	6✔
200
201	def _telnet_io(self, command):	6✔
202	# Microsoft Telnet> o
203	# ( to )127.0.0.1 6023
204	# >>>stats.get_stats()
205	# >>>est()
206	self.tn.write(b'%s\n' % command)	5✔
207	content = self.tn.read_until(b'\n>>>', timeout=TELNET_TIMEOUT)	5✔
208	# print(repr(content))
209	# b"\x1bc>>> \x1b[4hstats.get_stats()\r\r\r\n{'log_count/INFO': 61,
210	# 'start_time': datetime.datetime(2019, 1, 22, 9, 7, 14, 998126),
211	# 'httperror/response_ignored_status_count/404': 1}\r\r\r\n>>>"
212	# b' est()\r\r\r\nExecution engine status\r\r\r\n\r\r\r\n
213	# time()-engine.start_time : 3249.7548048496246
214	# engine.scraper.slot.needs_backout() : False\r\r\r\n\r\r\r\n\r\r\r\n>>>'
215	return content.decode('utf-8')	5✔
216
217	def telnet_io(self):	6✔
218	# spider._job, spider._version, settings.attributes["BOT_NAME"].value, JOB, SPIDER, PROJECT
219	# '\'logs\\\\demo_persistent\\\\test\\\\2019-01-23T18_25_34.log\'\r\r\r\n>>>'
220	log_file = self._telnet_io(TELNETCONSOLE_COMMAND_MAP['log_file'])	5✔
221	self.logger.debug("settings['LOG_FILE'] found via telnet: %s", log_file)	5✔
222	# Username: Password:
223	if 'Username:' in log_file:	5✔
224	self.logger.error("Telnet with auth is not supported on Windows. You can use scrapy<=%s instead: %s",	5✔
225	SUPPORTED_SCRAPY_VERSION, log_file)
226	return	5✔
227	if not self.verify_log_file_path(self.parse_log_path(self.data['log_path']), log_file):	×
228	self.logger.warning("Skip telnet due to mismatching: %s vs %s", self.data['log_path'], log_file)	×
229	return	×
230	self.crawler_stats = self.parse_output(self._telnet_io(TELNETCONSOLE_COMMAND_MAP['crawler_stats']))	×
231	self.crawler_engine = self.parse_output(self._telnet_io(TELNETCONSOLE_COMMAND_MAP['crawler_engine']))	×
232
233	def verify_log_file_path(self, parts, log_file):	6✔
234	for part in parts:	6✔
235	if part not in log_file:	6✔
236	self.logger.warning("%s not found in settings['LOG_FILE']: %s", part, log_file)	×
237	return False	×
238	return True	6✔

my8100 / logparser / 90f68d38-3b4e-4f37-a4d8-85d436e6e297

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous