本文整理汇总了Python中pyspider.libs.utils.text函数的典型用法代码示例。如果您正苦于以下问题：Python text函数的具体用法？Python text怎么用？Python text使用的例子？那么恭喜您, 这里精选的函数代码示例或许可以为您提供帮助。
在下文中一共展示了text函数的20个代码示例，这些例子默认根据受欢迎程度排序。您可以为喜欢或者感觉有用的代码点赞，您的评价将有助于我们的系统推荐出更棒的Python代码示例。
示例1: test_a10_counter
def test_a10_counter(self):
for i in range(30):
time.sleep(1)
if self.rpc.counter('5m', 'sum')\
.get('test_project', {}).get('success', 0) > 5:
break
rv = self.app.get('/counter?time=5m&type=sum')
self.assertEqual(rv.status_code, 200)
data = json.loads(utils.text(rv.data))
self.assertGreater(len(data), 0)
self.assertGreater(data['test_project']['success'], 3)
rv = self.app.get('/counter?time=1h&type=sum')
self.assertEqual(rv.status_code, 200)
data = json.loads(utils.text(rv.data))
self.assertGreater(len(data), 0)
self.assertGreater(data['test_project']['success'], 3)
rv = self.app.get('/counter?time=1d&type=sum')
self.assertEqual(rv.status_code, 200)
data = json.loads(utils.text(rv.data))
self.assertGreater(len(data), 0)
self.assertGreater(data['test_project']['success'], 3)
rv = self.app.get('/counter?time=all&type=sum')
self.assertEqual(rv.status_code, 200)
data = json.loads(utils.text(rv.data))
self.assertGreater(len(data), 0)
self.assertGreater(data['test_project']['success'], 3)
开发者ID:SMWARREN，项目名称:pyspider，代码行数:30，代码来源:test_webui.py
示例2: test_a10_counter
def test_a10_counter(self):
for i in range(30):
time.sleep(1)
if self.rpc.counter("5m", "sum").get("test_project", {}).get("success", 0) > 5:
break
rv = self.app.get("/counter?time=5m&type=sum")
self.assertEqual(rv.status_code, 200)
data = json.loads(utils.text(rv.data))
self.assertGreater(len(data), 0)
self.assertGreater(data["test_project"]["success"], 3)
rv = self.app.get("/counter?time=1h&type=sum")
self.assertEqual(rv.status_code, 200)
data = json.loads(utils.text(rv.data))
self.assertGreater(len(data), 0)
self.assertGreater(data["test_project"]["success"], 3)
rv = self.app.get("/counter?time=1d&type=sum")
self.assertEqual(rv.status_code, 200)
data = json.loads(utils.text(rv.data))
self.assertGreater(len(data), 0)
self.assertGreater(data["test_project"]["success"], 3)
rv = self.app.get("/counter?time=all&type=sum")
self.assertEqual(rv.status_code, 200)
data = json.loads(utils.text(rv.data))
self.assertGreater(len(data), 0)
self.assertGreater(data["test_project"]["success"], 3)
开发者ID:syhrte，项目名称:pyspider，代码行数:29，代码来源:test_webui.py
示例3: _parse
def _parse(data):
if six.PY3:
for key, value in list(six.iteritems(data)):
if isinstance(value, six.binary_type):
data[utils.text(key)] = utils.text(value)
else:
data[utils.text(key)] = value
return data
开发者ID:CrazyTeaFs，项目名称:pyspider，代码行数:8，代码来源:projectdb.py
示例4: test_50_get
def test_50_get(self):
io = BytesIO()
self.webdav.download('handler.py', io)
self.assertEqual(utils.text(inspect.getsource(data_handler)), utils.text(io.getvalue()))
io.close()
io = BytesIO()
self.webdav.download('sample_handler.py', io)
self.assertEqual(utils.text(inspect.getsource(data_sample_handler)), utils.text(io.getvalue()))
io.close()
开发者ID:Dmitry-Kucher，项目名称:pyspider，代码行数:10，代码来源:test_webdav.py
示例5: test_20_debug
def test_20_debug(self):
rv = self.app.get("/debug/test_project")
self.assertEqual(rv.status_code, 200)
self.assertIn(b"debugger", rv.data)
self.assertIn(b"var task_content = ", rv.data)
self.assertIn(b"var script_content = ", rv.data)
m = re.search(r"var task_content = (.*);\n", utils.text(rv.data))
self.assertIsNotNone(m)
self.__class__.task_content = json.loads(m.group(1))
m = re.search(r"var script_content = (.*);\n", utils.text(rv.data))
self.assertIsNotNone(m)
self.__class__.script_content = json.loads(m.group(1))
开发者ID:0xa-cc，项目名称:pyspider，代码行数:13，代码来源:test_webui.py
示例6: test_20_debug
def test_20_debug(self):
rv = self.app.get('/debug/test_project')
self.assertEqual(rv.status_code, 200)
self.assertIn(b'debugger', rv.data)
self.assertIn(b'var task_content = ', rv.data)
self.assertIn(b'var script_content = ', rv.data)
m = re.search(r'var task_content = (.*);\n', utils.text(rv.data))
self.assertIsNotNone(m)
self.assertIn('test_project', json.loads(m.group(1)))
m = re.search(r'var script_content = (.*);\n', utils.text(rv.data))
self.assertIsNotNone(m)
self.assertIn('__START_URL__', json.loads(m.group(1)))
开发者ID:widy28，项目名称:pyspider，代码行数:14，代码来源:test_webui.py
示例7: test_20_debug
def test_20_debug(self):
rv = self.app.get('/debug/test_project')
self.assertEqual(rv.status_code, 200)
self.assertIn(b'debugger', rv.data)
self.assertIn(b'var task_content = ', rv.data)
self.assertIn(b'var script_content = ', rv.data)
m = re.search(r'var task_content = (.*);\n', utils.text(rv.data))
self.assertIsNotNone(m)
self.__class__.task_content = json.loads(m.group(1))
m = re.search(r'var script_content = (.*);\n', utils.text(rv.data))
self.assertIsNotNone(m)
self.__class__.script_content = (json.loads(m.group(1))
.replace('http://scrapy.org/',
'http://127.0.0.1:14887/pyspider/test.html'))
开发者ID:SMWARREN，项目名称:pyspider，代码行数:15，代码来源:test_webui.py
示例8: test_20_debug
def test_20_debug(self):
rv = self.app.get("/debug/test_project")
self.assertEqual(rv.status_code, 200)
self.assertIn(b"debugger", rv.data)
self.assertIn(b"var task_content = ", rv.data)
self.assertIn(b"var script_content = ", rv.data)
m = re.search(r"var task_content = (.*);\n", utils.text(rv.data))
self.assertIsNotNone(m)
self.__class__.task_content = json.loads(m.group(1))
m = re.search(r"var script_content = (.*);\n", utils.text(rv.data))
self.assertIsNotNone(m)
self.__class__.script_content = json.loads(m.group(1)).replace(
"http://scrapy.org/", "http://127.0.0.1:14887/pyspider/test.html"
)
开发者ID:syhrte，项目名称:pyspider，代码行数:15，代码来源:test_webui.py
示例9: load_tasks
def load_tasks(self, status, project=None, fields=None):
if project is None:
project = self.projects
elif not isinstance(project, list):
project = [project, ]
if self.scan_available:
scan_method = self.redis.sscan_iter
else:
scan_method = self.redis.smembers
if fields:
def get_method(key):
obj = self.redis.hmget(key, fields)
if all(x is None for x in obj):
return None
return dict(zip(fields, obj))
else:
get_method = self.redis.hgetall
for p in project:
status_key = self._gen_status_key(p, status)
for taskid in scan_method(status_key):
obj = get_method(self._gen_key(p, utils.text(taskid)))
if not obj:
#self.redis.srem(status_key, taskid)
continue
else:
yield self._parse(obj)
开发者ID:runt18，项目名称:pyspider，代码行数:29，代码来源:taskdb.py
示例10: readonly
def readonly(self):
projectdb = self.app.config["projectdb"]
if not projectdb:
return True
if (
"lock" in projectdb.split_group(self.project.get("group"))
and self.app.config.get("webui_username")
and self.app.config.get("webui_password")
):
authheader = self.environ.get("HTTP_AUTHORIZATION")
if not authheader:
return True
authheader = authheader[len("Basic ") :]
try:
username, password = text(base64.b64decode(authheader)).split(":", 1)
except Exception as e:
self.app.logger.error("wrong api key: %r, %r", authheader, e)
return True
if username == self.app.config["webui_username"] and password == self.app.config["webui_password"]:
return False
else:
return True
return False
开发者ID:RacoonBattle，项目名称:pyspider，代码行数:25，代码来源:webdav.py
示例11: _parse
def _parse(self, data):
for key, value in list(six.iteritems(data)):
if isinstance(value, (bytearray, six.binary_type)):
data[key] = utils.text(value)
if 'result' in data:
data['result'] = json.loads(data['result'])
return data
开发者ID:0xa-cc，项目名称:pyspider，代码行数:7，代码来源:resultdb.py
示例12: test_90_run
def test_90_run(self):
time.sleep(0.5)
rv = self.app.post('/run', data={
'project': 'test_project',
})
self.assertEqual(rv.status_code, 200)
self.assertEqual(json.loads(utils.text(rv.data))['result'], True)
开发者ID:SMWARREN，项目名称:pyspider，代码行数:7，代码来源:test_webui.py
示例13: _select2dic
def _select2dic(self, tablename=None, what="*", where="", where_values=[],
order=None, offset=0, limit=None):
tablename = self.escape(tablename or self.__tablename__)
if isinstance(what, list) or isinstance(what, tuple) or what is None:
what = ','.join(self.escape(f) for f in what) if what else '*'
sql_query = "SELECT %s FROM %s" % (what, tablename)
if where:
sql_query += " WHERE %s" % where
if order:
sql_query += ' ORDER BY %s' % order
if limit:
sql_query += " LIMIT %d, %d" % (offset, limit)
elif offset:
sql_query += " LIMIT %d, %d" % (offset, self.maxlimit)
logger.debug("<sql: %s>", sql_query)
dbcur = self._execute(sql_query, where_values)
# f[0] may return bytes type
# https://github.com/mysql/mysql-connector-python/pull/37
fields = [utils.text(f[0]) for f in dbcur.description]
for row in dbcur:
yield dict(zip(fields, row))
开发者ID:QxxDmaku，项目名称:pyspider，代码行数:25，代码来源:basedb.py
示例14: crawl
def crawl(self, url=None, track=None, **kwargs):
if url is None and kwargs.get('callback'):
url = dataurl.encode(utils.text(kwargs.get('callback')))
project_data = self.processor.project_manager.get(self.project_name)
assert project_data, "can't find project: %s" % self.project_name
instance = project_data['instance']
instance._reset()
task = instance.crawl(url, **kwargs)
if isinstance(task, list):
task = task[0]
task['track'] = track
result = self.fetcher.fetch(task)
self.processor.on_task(task, result)
status = None
while not self.status_queue.empty():
status = self.status_queue.get()
newtasks = []
while not self.newtask_queue.empty():
newtasks = self.newtask_queue.get()
result = None
while not self.result_queue.empty():
_, result = self.result_queue.get()
return status, newtasks, result
开发者ID:1271281914，项目名称:pyspider，代码行数:25，代码来源:test_fetcher_processor.py
示例15: doc
def doc(self):
"""Returns a PyQuery object of the response's content"""
if hasattr(self, '_doc'):
return self._doc
elements = self.etree
doc = self._doc = PyQuery(elements)
doc.make_links_absolute(utils.text(self.url))
return doc
开发者ID:01jiagnwei01，项目名称:pyspider，代码行数:8，代码来源:response.py
示例16: test_35_run_http_task
def test_35_run_http_task(self):
rv = self.app.post('/debug/test_project/run', data={
'script': self.script_content,
'task': json.dumps(self.task_content2)
})
self.assertEqual(rv.status_code, 200)
data = json.loads(utils.text(rv.data))
self.assertIn(b'follows', rv.data)
开发者ID:SMWARREN，项目名称:pyspider，代码行数:8，代码来源:test_webui.py
示例17: test_a20_tasks
def test_a20_tasks(self):
rv = self.app.get('/tasks')
self.assertEqual(rv.status_code, 200, rv.data)
self.assertIn(b'SUCCESS</span>', rv.data)
self.assertNotIn(b'>ERROR</span>', rv.data)
m = re.search(r'/task/test_project:[^"]+', utils.text(rv.data))
self.assertIsNotNone(m)
self.__class__.task_url = m.group(0)
self.assertIsNotNone(self.task_url)
m = re.search(r'/debug/test_project[^"]+', utils.text(rv.data))
self.assertIsNotNone(m)
self.__class__.debug_task_url = m.group(0)
self.assertIsNotNone(self.debug_task_url)
rv = self.app.get('/tasks?project=test_project')
self.assertEqual(rv.status_code, 200)
self.assertIn(b'SUCCESS</span>', rv.data)
self.assertNotIn(b'>ERROR</span>', rv.data)
开发者ID:SMWARREN，项目名称:pyspider，代码行数:18，代码来源:test_webui.py
示例18: _parse
def _parse(data):
for key, value in list(six.iteritems(data)):
if isinstance(value, six.binary_type):
data[key] = utils.text(value)
if "result" in data:
if isinstance(data["result"], bytearray):
data["result"] = str(data["result"])
data["result"] = json.loads(data["result"])
return data
开发者ID:syhrte，项目名称:pyspider，代码行数:9，代码来源:resultdb.py
示例19: test_33_run_bad_script
def test_33_run_bad_script(self):
rv = self.app.post('/debug/test_project/run', data={
'script': self.script_content+'adfasfasdf',
'task': self.task_content
})
self.assertEqual(rv.status_code, 200)
data = json.loads(utils.text(rv.data))
self.assertGreater(len(data['logs']), 0)
self.assertEqual(len(data['follows']), 0)
开发者ID:widy28，项目名称:pyspider，代码行数:9，代码来源:test_webui.py
示例20: __init__
def __init__(self, path, environ, app, project=None):
super(ScriptResource, self).__init__(path, environ)
self.app = app
self.new_project = False
self._project = project
self.project_name = text(self.name)
self.writebuffer = None
if self.project_name.endswith('.py'):
self.project_name = self.project_name[:-len('.py')]
开发者ID:ZhiweiWang，项目名称:pyspider，代码行数:10，代码来源:webdav.py
注：本文中的pyspider.libs.utils.text函数示例由纯净天空整理自Github/MSDocs等源码及文档管理平台，相关代码片段筛选自各路编程大神贡献的开源项目，源码版权归原作者所有，传播和使用请参考对应项目的License；未经允许，请勿转载。 |
客服电话
APP下载
官方微信
请发表评论