dffd1c3b-7d7a-48d0-ac24-921e1ddd6643

Committed 19 Feb 2025 02:41PM UTC coverage: 85.836%. Remained the same

Build # dffd1c3b-7d7a-48d0-ac24-921e1ddd6643

Build Type

push

circleci

Committed by

web-flow

Commit Message

Update README.md (#257)

* Update README.md

* Update README.md

Run Details

3539 of 4123 relevant lines covered (85.84%)

10.16 hits per line

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

80.0

/scrapydweb/views/api.py

# coding: utf-8
import re
import time

from .baseview import BaseView


API_MAP = dict(start='schedule', stop='cancel', forcestop='cancel', liststats='logs/stats')


class ApiView(BaseView):

    def __init__(self):
        super(ApiView, self).__init__()

        self.opt = self.view_args['opt']
        self.project = self.view_args['project']
        self.version_spider_job = self.view_args['version_spider_job']

        self.url = 'http://{}/{}.json'.format(self.SCRAPYD_SERVER, API_MAP.get(self.opt, self.opt))
        self.data = None
        self.status_code = 0
        self.js = {}

    def dispatch_request(self, **kwargs):
        self.update_url()
        self.update_data()
        self.get_result()
        self.handle_result()
        return self.json_dumps(self.js, sort_keys=False, as_response=True)

    def update_url(self):
        if self.opt in ['listversions', 'listjobs']:
            self.url += '?project=%s' % self.project
        elif self.opt == 'listspiders':
            if self.version_spider_job == self.DEFAULT_LATEST_VERSION:
                self.url += '?project=%s' % self.project
            else:
                # Should be _version
                self.url += '?project=%s&_version=%s' % (self.project, self.version_spider_job)

    def update_data(self):
        self.data = dict(project=self.project)
        if self.opt == 'start':
            self.data['spider'] = self.version_spider_job
            self.data['jobid'] = self.get_now_string()
        elif self.opt in ['stop', 'forcestop']:
            self.data['job'] = self.version_spider_job
        elif self.opt == 'delversion':
            self.data['version'] = self.version_spider_job
        elif self.opt == 'delproject':
            pass
        else:
            self.data = None

    def get_result(self):
        timeout = 3 if self.opt == 'daemonstatus' else 60
        dumps_json = self.opt not in ['daemonstatus', 'liststats']
        times = 2 if self.opt == 'forcestop' else 1
        for __ in range(times):
            self.status_code, self.js = self.make_request(self.url, data=self.data, auth=self.AUTH,
                                                          as_json=True, dumps_json=dumps_json, timeout=timeout)
            if times != 1:
                self.js['times'] = times
                time.sleep(2)

    def handle_result(self):
        if self.status_code != 200:
            if self.opt == 'liststats':
                if self.project and self.version_spider_job:  # 'List Stats' in the Servers page
                    if self.status_code == 404:
                        self.js = dict(status=self.OK, tip="'pip install logparser' and run command 'logparser'")
                else:  # XMLHttpRequest in the Jobs page; POST in jobs.py
                    self.js['tip'] = ("'pip install logparser' on host '%s' and run command 'logparser' "
                                      "to show crawled_pages and scraped_items. ") % self.SCRAPYD_SERVER
            else:
                self.js['tip'] = "Make sure that your Scrapyd server is accessable. "
        elif self.js['status'] != self.OK:
            if re.search('No such file|no active project', self.js.get('message', '')):
                self.js['tip'] = "Maybe the project had been deleted, check out the Projects page. "
            elif self.opt == 'listversions':
                self.js['tip'] = (
                    "Maybe it's caused by failing to compare versions, "
                    "you can check out the HELP section in the Deploy Project page for more info, "
                    "and solve the problem in the Projects page. "
                )
            elif self.opt == 'listspiders' and re.search("TypeError: 'tuple'", self.js.get('message', '')):
                self.js['tip'] = "Maybe it's a broken project, check out the Projects page to delete it. "
        elif self.opt == 'liststats':
            if self.js.get('logparser_version') != self.LOGPARSER_VERSION:
                if self.project and self.version_spider_job:  # 'List Stats' in the Servers page
                    tip = "'pip install --upgrade logparser' to update LogParser to v%s" % self.LOGPARSER_VERSION
                    self.js = dict(status=self.OK, tip=tip)
                else:  # XMLHttpRequest in the Jobs page; POST in jobs.py
                    self.js['tip'] = ("'pip install --upgrade logparser' on host '%s' and run command 'logparser' "
                                      "to update LogParser to v%s") % (self.SCRAPYD_SERVER, self.LOGPARSER_VERSION)
                    self.js['status'] = self.ERROR
            elif self.project and self.version_spider_job:  # 'List Stats' in the Servers page
                self.extract_pages_items()

    def extract_pages_items(self):
        details = None
        if self.project in self.js['datas']:
            for spider in self.js['datas'][self.project]:
                for jobid in self.js['datas'][self.project][spider]:
                    if jobid == self.version_spider_job:
                        details = self.js['datas'][self.project][spider][self.version_spider_job]
                        self.js['project'] = self.project
                        self.js['spider'] = spider
                        self.js['jobid'] = jobid
                        break
        if not details:
            details = dict(pages=self.NA, items=self.NA)
        details.setdefault('project', self.project)
        details.setdefault('spider', self.NA)
        details.setdefault('jobid', self.version_spider_job)
        details['logparser_version'] = self.js.get('logparser_version', None)
        self.js = dict(status=self.OK, details=details)

1	# coding: utf-8
2	import re	12✔
3	import time	12✔
4
5	from .baseview import BaseView	12✔
6
7
8	API_MAP = dict(start='schedule', stop='cancel', forcestop='cancel', liststats='logs/stats')	12✔
9
10
11	class ApiView(BaseView):	12✔
12
13	def __init__(self):	12✔
14	super(ApiView, self).__init__()	12✔
15
16	self.opt = self.view_args['opt']	12✔
17	self.project = self.view_args['project']	12✔
18	self.version_spider_job = self.view_args['version_spider_job']	12✔
19
20	self.url = 'http://{}/{}.json'.format(self.SCRAPYD_SERVER, API_MAP.get(self.opt, self.opt))	12✔
21	self.data = None	12✔
22	self.status_code = 0	12✔
23	self.js = {}	12✔
24
25	def dispatch_request(self, **kwargs):	12✔
26	self.update_url()	12✔
27	self.update_data()	12✔
28	self.get_result()	12✔
29	self.handle_result()	12✔
30	return self.json_dumps(self.js, sort_keys=False, as_response=True)	12✔
31
32	def update_url(self):	12✔
33	if self.opt in ['listversions', 'listjobs']:	12✔
34	self.url += '?project=%s' % self.project	12✔
35	elif self.opt == 'listspiders':	12✔
36	if self.version_spider_job == self.DEFAULT_LATEST_VERSION:	12✔
37	self.url += '?project=%s' % self.project	12✔
38	else:
39	# Should be _version
40	self.url += '?project=%s&_version=%s' % (self.project, self.version_spider_job)	12✔
41
42	def update_data(self):	12✔
43	self.data = dict(project=self.project)	12✔
44	if self.opt == 'start':	12✔
45	self.data['spider'] = self.version_spider_job	12✔
46	self.data['jobid'] = self.get_now_string()	12✔
47	elif self.opt in ['stop', 'forcestop']:	12✔
48	self.data['job'] = self.version_spider_job	12✔
49	elif self.opt == 'delversion':	12✔
50	self.data['version'] = self.version_spider_job	12✔
51	elif self.opt == 'delproject':	12✔
52	pass	12✔
53	else:
54	self.data = None	12✔
55
56	def get_result(self):	12✔
57	timeout = 3 if self.opt == 'daemonstatus' else 60	12✔
58	dumps_json = self.opt not in ['daemonstatus', 'liststats']	12✔
59	times = 2 if self.opt == 'forcestop' else 1	12✔
60	for __ in range(times):	12✔
61	self.status_code, self.js = self.make_request(self.url, data=self.data, auth=self.AUTH,	12✔
62	as_json=True, dumps_json=dumps_json, timeout=timeout)
63	if times != 1:	12✔
64	self.js['times'] = times	12✔
65	time.sleep(2)	12✔
66
67	def handle_result(self):	12✔
68	if self.status_code != 200:	12✔
69	if self.opt == 'liststats':	×
70	if self.project and self.version_spider_job: # 'List Stats' in the Servers page	×
71	if self.status_code == 404:	×
72	self.js = dict(status=self.OK, tip="'pip install logparser' and run command 'logparser'")	×
73	else: # XMLHttpRequest in the Jobs page; POST in jobs.py
74	self.js['tip'] = ("'pip install logparser' on host '%s' and run command 'logparser' "	×
75	"to show crawled_pages and scraped_items. ") % self.SCRAPYD_SERVER
76	else:
77	self.js['tip'] = "Make sure that your Scrapyd server is accessable. "	×
78	elif self.js['status'] != self.OK:	12✔
79	if re.search('No such file\|no active project', self.js.get('message', '')):	12✔
80	self.js['tip'] = "Maybe the project had been deleted, check out the Projects page. "	2✔
81	elif self.opt == 'listversions':	12✔
82	self.js['tip'] = (	×
83	"Maybe it's caused by failing to compare versions, "
84	"you can check out the HELP section in the Deploy Project page for more info, "
85	"and solve the problem in the Projects page. "
86	)
87	elif self.opt == 'listspiders' and re.search("TypeError: 'tuple'", self.js.get('message', '')):	12✔
88	self.js['tip'] = "Maybe it's a broken project, check out the Projects page to delete it. "	×
89	elif self.opt == 'liststats':	12✔
90	if self.js.get('logparser_version') != self.LOGPARSER_VERSION:	12✔
91	if self.project and self.version_spider_job: # 'List Stats' in the Servers page	×
92	tip = "'pip install --upgrade logparser' to update LogParser to v%s" % self.LOGPARSER_VERSION	×
93	self.js = dict(status=self.OK, tip=tip)	×
94	else: # XMLHttpRequest in the Jobs page; POST in jobs.py
95	self.js['tip'] = ("'pip install --upgrade logparser' on host '%s' and run command 'logparser' "	×
96	"to update LogParser to v%s") % (self.SCRAPYD_SERVER, self.LOGPARSER_VERSION)
97	self.js['status'] = self.ERROR	×
98	elif self.project and self.version_spider_job: # 'List Stats' in the Servers page	12✔
99	self.extract_pages_items()	12✔
100
101	def extract_pages_items(self):	12✔
102	details = None	12✔
103	if self.project in self.js['datas']:	12✔
104	for spider in self.js['datas'][self.project]:	12✔
105	for jobid in self.js['datas'][self.project][spider]:	12✔
106	if jobid == self.version_spider_job:	12✔
107	details = self.js['datas'][self.project][spider][self.version_spider_job]	×
108	self.js['project'] = self.project	×
109	self.js['spider'] = spider	×
110	self.js['jobid'] = jobid	×
111	break	×
112	if not details:	12✔
113	details = dict(pages=self.NA, items=self.NA)	12✔
114	details.setdefault('project', self.project)	12✔
115	details.setdefault('spider', self.NA)	12✔
116	details.setdefault('jobid', self.version_spider_job)	12✔
117	details['logparser_version'] = self.js.get('logparser_version', None)	12✔
118	self.js = dict(status=self.OK, details=details)	12✔

my8100 / scrapydweb / dffd1c3b-7d7a-48d0-ac24-921e1ddd6643

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