本文整理汇总了Python中tornado.escape._unicode函数的典型用法代码示例。如果您正苦于以下问题：Python _unicode函数的具体用法？Python _unicode怎么用？Python _unicode使用的例子？那么恭喜您, 这里精选的函数代码示例或许可以为您提供帮助。
在下文中一共展示了_unicode函数的20个代码示例，这些例子默认根据受欢迎程度排序。您可以为喜欢或者感觉有用的代码点赞，您的评价将有助于我们的系统推荐出更棒的Python代码示例。
示例1: __after__
def __after__(self):
try:
v = self._values
# strip / to avoid treating http://example.com and http://example.com/
# as different sites
v["website"] = v["website"].strip("/")
user_id = self._handler.current_user.id
usecase_md = markdown.markdown(escape._unicode(v["usecase"]))
if v.get("site", ""):
stmt = "UPDATE site SET sitename = %s,website = %s,description = %s, "\
"usecase = %s,usecase_md = %s,source_url = %s, logo = %s, "\
"updated = UTC_TIMESTAMP() where uuid_ = %s"
args = (v["sitename"], v["website"], v["desc"], \
v["usecase"], usecase_md, v["source_url"], v["logo"], \
v["site"])
self._handler.db.execute(stmt, *args)
else:
if self._handler.db.get("select * from site where website = %s", v["website"]):
self.add_error("website", "This web site already be registered.")
else:
if (not self._handler.is_staff):
user_projects = self._handler.db.get("select count(*) as c from site where user_id = %s", \
self._handler.current_user.id)
if user_projects is not None and user_projects.c >= 3:
# It will need approve if someone create more than three sites
status = const.Status.PENDING
else:
status = const.Status.UNVERIFIED
else:
status = const.Status.ACTIVE
slug = v["website"].lower().strip()
slug = _domain_prefix_re.sub("", slug)
slug = unicodedata.normalize("NFKD", escape._unicode(slug)).encode("ascii", "ignore")
slug = re.sub(r"[^\w]+", " ", slug)
slug = "-".join(slug.split())
if not slug:
slug = "site"
while True:
e = self._handler.db.get("SELECT * FROM site WHERE slug = %s", slug)
if not e:
break
slug += "-" + uuid.uuid4().hex[0:2]
stmt = "INSERT INTO site (sitename,website,description,usecase,usecase_md,source_url,"\
"user_id,logo,uuid_,created,updated,updated_ss,status_,slug) "\
"VALUES (%s,%s,%s,%s,%s,%s,"\
"%s,%s,%s,UTC_TIMESTAMP(),UTC_TIMESTAMP(),UTC_TIMESTAMP(),%s,%s)"
args = (v["sitename"], v["website"], v["desc"], v["usecase"], usecase_md, v["source_url"], \
user_id, v["logo"], uuid.uuid4().hex, status, slug)
self._handler.db.execute(stmt, *args)
except Exception, e:
logging.error(str(e))
self.add_error("sitename", "Submit project error, please try it later.")
开发者ID:HarrisonHDU，项目名称:poweredsites，代码行数:60，代码来源:submit.py
示例2: match_list
def match_list(self, match_list):
arguments = self.request.arguments
if '__NODELAY__' in arguments:
del arguments['__NODELAY__']
if '__BLACKLIST__' in arguments:
del arguments['__BLACKLIST__']
if '__ASYNCLIST__' in arguments:
del arguments['__ASYNCLIST__']
if '__ASYNC_RESULT__' in arguments:
del arguments['__ASYNC_RESULT__']
if '__ENABLE_DEBUG__' in arguments:
del arguments['__ENABLE_DEBUG__']
if '__CONTENT_TYPE__' in arguments:
del arguments['__CONTENT_TYPE__']
if '__JSONP_CALLBACK_VARNAME__' in arguments:
del arguments['__JSONP_CALLBACK_VARNAME__']
match = "|".join(match_list)
self.logging.info(match)
p = re.compile(r'%s'%(match,),re.I|re.M)
for k in arguments.keys():
arguments_string = _unicode(" ".join(arguments[k]))
arguments_string = re.sub("\r|\n","",arguments_string)
if p.search(arguments_string):
return True
if hasattr(self.request,'body'):
self.logging.info(_unicode(self.request.body))
body = re.sub("\r|\n","",_unicode(self.request.body))
if p.search(body):
return True
return False
开发者ID:liruikaiyao，项目名称:workshop，代码行数:33，代码来源:trouter.py
示例3: prepare_request
def prepare_request(cls, request, default_host):
parsed = urlparse.urlsplit(_unicode(request.url))
if request.method not in cls._SUPPORTED_METHODS and not request.allow_nonstandard_methods:
raise KeyError("unknown method %s" % request.method)
request.follow_redirects = False
for key in (
"network_interface",
"proxy_host",
"proxy_port",
"proxy_username",
"proxy_password",
"expect_100_continue",
"body_producer",
):
if getattr(request, key, None):
raise NotImplementedError("%s not supported" % key)
request.headers.pop("Connection", None)
if "Host" not in request.headers:
if not parsed.netloc:
request.headers["Host"] = default_host
elif "@" in parsed.netloc:
request.headers["Host"] = parsed.netloc.rpartition("@")[-1]
else:
request.headers["Host"] = parsed.netloc
username, password = None, None
if parsed.username is not None:
username, password = parsed.username, parsed.password
elif request.auth_username is not None:
username = request.auth_username
password = request.auth_password or ""
if username is not None:
if request.auth_mode not in (None, "basic"):
raise ValueError("unsupported auth_mode %s", request.auth_mode)
auth = utf8(username) + b":" + utf8(password)
request.headers["Authorization"] = b"Basic " + base64.b64encode(auth)
if request.user_agent:
request.headers["User-Agent"] = request.user_agent
if not request.allow_nonstandard_methods:
# Some HTTP methods nearly always have bodies while others
# almost never do. Fail in this case unless the user has
# opted out of sanity checks with allow_nonstandard_methods.
body_expected = request.method in ("POST", "PATCH", "PUT")
body_present = request.body is not None or request.body_producer is not None
if (body_expected and not body_present) or (body_present and not body_expected):
raise ValueError(
"Body must %sbe None for method %s (unless "
"allow_nonstandard_methods is true)" % ("not " if body_expected else "", request.method)
)
if request.body is not None:
# When body_producer is used the caller is responsible for
# setting Content-Length (or else chunked encoding will be used).
request.headers["Content-Length"] = str(len(request.body))
if request.method == "POST" and "Content-Type" not in request.headers:
request.headers["Content-Type"] = "application/x-www-form-urlencoded"
if request.decompress_response:
request.headers["Accept-Encoding"] = "gzip"
request.url = (parsed.path or "/") + (("?" + parsed.query) if parsed.query else "")
return request
开发者ID:mSOHU，项目名称:http2，代码行数:60，代码来源:tornado4.py
示例4: __init__
def __init__(self, io_loop, client, request, release_callback,
final_callback, max_buffer_size, tcp_client,
max_header_size):
self.start_time = io_loop.time() # 连接开始时间
self.io_loop = io_loop
self.client = client # 请求的client，一般都是同一个
self.request = request # 代理的request
self.release_callback = release_callback # 如果取消了fetch，执行的回调
self.final_callback = final_callback # 如果结束了，执行的回调
self.max_buffer_size = max_buffer_size
self.tcp_client = tcp_client # 一个tcp的连接
self.max_header_size = max_header_size
self.code = None
self.headers = None
self.chunks = []
self._decompressor = None
# Timeout handle returned by IOLoop.add_timeout
self._timeout = None
self._sockaddr = None
with stack_context.ExceptionStackContext(self._handle_exception):
self.parsed = urlparse.urlsplit(_unicode(self.request.url))
if self.parsed.scheme not in ("http", "https"):
raise ValueError("Unsupported url scheme: %s" %
self.request.url)
# urlsplit results have hostname and port results, but they
# didn't support ipv6 literals until python 2.7.
netloc = self.parsed.netloc
if "@" in netloc:
userpass, _, netloc = netloc.rpartition("@")
match = re.match(r'^(.+):(\d+)$', netloc)
if match:
host = match.group(1)
port = int(match.group(2))
else:
host = netloc
port = 443 if self.parsed.scheme == "https" else 80
if re.match(r'^\[.*\]$', host):
# raw ipv6 addresses in urls are enclosed in brackets
host = host[1:-1]
self.parsed_hostname = host # save final host for _on_connect
if request.allow_ipv6 is False:
af = socket.AF_INET
else:
af = socket.AF_UNSPEC
ssl_options = self._get_ssl_options(self.parsed.scheme)
timeout = min(self.request.connect_timeout, self.request.request_timeout)
# 这里执行添加一个timeout到io_loop，也就是如果请求过时什么的
if timeout:
self._timeout = self.io_loop.add_timeout(
self.start_time + timeout,
stack_context.wrap(self._on_timeout))
# 这个函数connect好像是没有回调的一个函数，最后一个callback是怎么回事儿
# 这个connect是一个被@gen.coroutine的方法
self.tcp_client.connect(host, port, af=af,
ssl_options=ssl_options,
callback=self._on_connect)
开发者ID:ColorFuzzy，项目名称:tornado_code，代码行数:60，代码来源:simple_httpclient.py
示例5: __init__
def __init__(self, io_loop, client, request, release_callback,
final_callback, max_buffer_size, tcp_client,
max_header_size, max_body_size):
self.start_time = io_loop.time()
self.io_loop = io_loop
self.client = client
self.request = request
self.release_callback = release_callback
self.final_callback = final_callback
self.max_buffer_size = max_buffer_size
self.tcp_client = tcp_client
self.max_header_size = max_header_size
self.max_body_size = max_body_size
self.code = None
self.headers = None
self.chunks = []
self._decompressor = None
# Timeout handle returned by IOLoop.add_timeout
self._timeout = None
self._sockaddr = None
with stack_context.ExceptionStackContext(self._handle_exception):
self.parsed = urlparse.urlsplit(_unicode(self.request.url))
af = socket.AF_UNSPEC
netloc = self.parsed.netloc
host, port = httputil.split_host_and_port(netloc)
timeout = min(self.request.connect_timeout, self.request.request_timeout)
if timeout:
self._timeout = self.io_loop.add_timeout(
self.start_time + timeout,
stack_context.wrap(self._on_timeout))
self.tcp_client.connect(host,port, af=af,
max_buffer_size=self.max_buffer_size,
callback=self._on_connect)
开发者ID:bluerover，项目名称:attack_on_titan，代码行数:34，代码来源:dmtp.py
示例6: __init__
def __init__(self, io_loop, client, request, callback):
self.start_time = time.time()
self.io_loop = io_loop
self.client = client
self.request = request
self.callback = callback
self.code = None
self.headers = None
self.chunks = None
self._decompressor = None
# Timeout handle returned by IOLoop.add_timeout
self._timeout = None
with stack_context.StackContext(self.cleanup):
parsed = urlparse.urlsplit(_unicode(self.request.url))
# urlsplit results have hostname and port results, but they
# didn't support ipv6 literals until python 2.7.
netloc = parsed.netloc
if "@" in netloc:
userpass, _, netloc = netloc.rpartition("@")
match = re.match(r"^(.+):(\d+)$", netloc)
if match:
host = match.group(1)
port = int(match.group(2))
else:
host = netloc
port = 443 if parsed.scheme == "https" else 80
if re.match(r"^\[.*\]$", host):
# raw ipv6 addresses in urls are enclosed in brackets
host = host[1:-1]
if self.client.hostname_mapping is not None:
host = self.client.hostname_mapping.get(host, host)
if request.allow_ipv6:
af = socket.AF_UNSPEC
else:
# We only try the first IP we get from getaddrinfo,
# so restrict to ipv4 by default.
af = socket.AF_INET
addrinfo = socket.getaddrinfo(host, port, af, socket.SOCK_STREAM, 0, 0)
af, socktype, proto, canonname, sockaddr = addrinfo[0]
if parsed.scheme == "https":
ssl_options = {}
if request.validate_cert:
ssl_options["cert_reqs"] = ssl.CERT_REQUIRED
if request.ca_certs is not None:
ssl_options["ca_certs"] = request.ca_certs
else:
ssl_options["ca_certs"] = _DEFAULT_CA_CERTS
self.stream = SSLIOStream(
socket.socket(af, socktype, proto), io_loop=self.io_loop, ssl_options=ssl_options
)
else:
self.stream = IOStream(socket.socket(af, socktype, proto), io_loop=self.io_loop)
timeout = min(request.connect_timeout, request.request_timeout)
if timeout:
self._connect_timeout = self.io_loop.add_timeout(self.start_time + timeout, self._on_timeout)
self.stream.set_close_callback(self._on_close)
self.stream.connect(sockaddr, functools.partial(self._on_connect, parsed))
开发者ID:enki，项目名称:tornado，代码行数:60，代码来源:simple_httpclient.py
示例7: __call__
def __call__(self, *fields, **kwargs):
_params = []
_cols = []
if len(fields)<1 and not kwargs:
raise database.OperationalError("Update need some date")
# raise database.OperationalError, 'Update Need some data'
elif kwargs:
for k in kwargs.keys():
if type(kwargs[k]).__name__ == 'unicode':
cond = eval(''.join(['self._queryer.', k, "==u'''", kwargs[k].replace("'''", "\\'\\'\\'"), "'''"]))
elif type(kwargs[k]).__name__ == 'int' or type(kwargs[k]).__name__ == 'long':
cond = eval(''.join(['self._queryer.', k, "==u'''", str(kwargs[k]), "'''"]))
elif type(kwargs[k]).__name__ == 'datetime':
cond = eval(''.join(['self._queryer.', k, "==u'", unicode(kwargs[k]), "'"]))
else:
cond = eval(''.join(['self._queryer.', k, "==u'", _unicode(kwargs[k]), "'"]))
_cols.append(cond.get_sql())
_params.append(cond.get_params()[0])
else:
for f in fields:
_cols.append(f.get_sql())
_params.append(f.get_params()[0])
_sql_slice = ['UPDATE `', self._table, '` SET ', ','.join(_cols)]
if self._where:
_sql_slice.append(' WHERE ')
_sql_slice.append(self._where.get_sql())
for p in self._where.get_params():
_params.append(p)
_sql = ''.join(_sql_slice)
return self._db.execute(_sql,*_params)
raise database.OperationalError("Update need where")
开发者ID:zh68319220，项目名称:tornado-framework，代码行数:31，代码来源:db.py
示例8: respond_post
def respond_post(self, cmd_name):
stime = time.time()
user = yield objs.User.find_one(
{'login_key': self.get_argument('login', '')})
self.set_header('Content-Type', 'application/json; charset=utf-8')
if not cmd_name:
defer.returnValue(json.dumps(dict(
ok=True,
commands=api_handlers.handlers.keys())))
if cmd_name not in api_handlers.handlers:
defer.returnValue(json.dumps(dict(
ok=False,
desc='unknown command')))
callogtuple = (id(self), cmd_name, user['name'] if user else '')
print "API call %d '%s' by '%s' started." % callogtuple
handler = api_handlers.handlers[cmd_name]
args = dict(
(utf8(k), _unicode(v[0]))
for k, v in self.request.arguments.iteritems())
if 'login' in args:
del args['login']
self.set_header('Cache-Control', 'no-cache')
cors_origin = self.request.headers.get('Origin', None)
if cors_origin:
self.set_header('Access-Control-Allow-Origin', '*')
print "API call %d '%s' by '%s' is CORS from '%s'" % (callogtuple + (cors_origin,))
try:
result = yield handler(ApiRequest(user), **args)
except BnwResponse as br:
print "API call %d '%s' by '%s' failed with exception. %f.s" % (callogtuple + ((time.time() - stime),))
defer.returnValue(json.dumps(dict(
ok=False,
desc=str(br)), ensure_ascii=False))
print "API call %d '%s' by '%s' completed. %f.s" % (callogtuple + ((time.time() - stime),))
defer.returnValue(json.dumps(result, ensure_ascii=False))
开发者ID:anarchypride，项目名称:bnw，代码行数:35，代码来源:api.py
示例9: decode_argument
def decode_argument(value, name=None):
"""Decodes an argument from the request.
"""
try:
return _unicode(value)
except UnicodeDecodeError:
raise HTTPError(400, "Invalid unicode in %s: %r" %
(name or "url", value[:40]))
开发者ID:NSConfArg，项目名称:NSOperations，代码行数:8，代码来源:tornadoparser.py
示例10: make_content
def make_content(text, extra_params='rel="nofollow"'):
"""https://github.com/facebook/tornado/blob/master/tornado/escape.py#L238
"""
if extra_params:
extra_params = " " + extra_params.strip()
def make_link(m):
url = m.group(1)
proto = m.group(2)
href = m.group(1)
if not proto:
href = "http://" + href
params = extra_params
if '.' in href:
name_extension = href.split('.')[-1].lower()
if name_extension in ('jpg', 'png', 'gif', 'jpeg'):
return u'<img src="%s" />' % href
return u'<a href="%s"%s>%s</a>' % (href, params, url)
def cover_email(m):
data = {'mail': m.group(1),
'end': m.group(2)}
return u'<a href="mailto:%(mail)s">%(mail)s</a>%(end)s' % data
def convert_mention(m):
data = {}
data['begin'], data['user'] = m.group(1).split('@')
t = u'%(begin)s<a href="/member/%(user)s" class="mention">' \
u'@%(user)s</a>'
return t % data
def convert_floor(m):
data = {}
data['begin'], data['floor'] = m.group(1).split('#')
t = u'%(begin)s<a href="#reply%(floor)s"' \
' class="mention mention_floor">#%(floor)s</a>'
return t % data
def convert_topic(m):
data = {}
data['begin'], data['topic_link'] = m.group(1).split('t')
data['topic_link_short'] = data['topic_link'][:6]
t = u"""%(begin)s<a href="%(topic_link)s"
class="mention mention_topic"
_id=%(topic_link)s>t%(topic_link_short)s</a>"""
return t % data
text = _unicode(text).replace('\n', '<br />')
text = text.replace(''', '\'')
text = _EMAIL_RE.sub(cover_email, text)
text = _MENTION_RE.sub(convert_mention, text)
text = _FLOOR_RE.sub(convert_floor, text)
text = _TOPIC_RE.sub(convert_topic, text)
return _URL_RE.sub(make_link, text)
开发者ID:JmPotato，项目名称:College，代码行数:58，代码来源:utils.py
示例11: post
def post(self):
self.finish(
{
"header": self.request.headers["X-Header-Encoding-Test"],
"argument": self.get_argument("argument"),
"filename": self.request.files["files"][0].filename,
"filebody": _unicode(self.request.files["files"][0]["body"]),
}
)
开发者ID:nickwong，项目名称:tornado，代码行数:9，代码来源:httpserver_test.py
示例12: __init__
def __init__(self, **kwargs):
if 'form' == self.__datatype__:
self._raw_data = {}
# Processing on handler.request.arguments, utf-8 values
for k in kwargs:
if isinstance(kwargs[k], list):
if len(kwargs[k]) > 1:
self._raw_data[k] = map(_unicode, kwargs[k])
else:
self._raw_data[k] = _unicode(kwargs[k][0])
else:
self._raw_data[k] = _unicode(kwargs[k])
else:
self._raw_data = kwargs
self.data = {}
self.errors = []
self.validate()
开发者ID:xoxoj，项目名称:torext，代码行数:19，代码来源:params.py
示例13: __init__
def __init__(self, io_loop, client, request, release_callback,
final_callback, max_buffer_size, resolver):
self.start_time = io_loop.time()
self.io_loop = io_loop
self.client = client
self.request = request
self.release_callback = release_callback
self.final_callback = final_callback
self.max_buffer_size = max_buffer_size
self.resolver = resolver
self.code = None
self.headers = None
self.chunks = None
self._decompressor = None
# Timeout handle returned by IOLoop.add_timeout
self._timeout = None
with stack_context.ExceptionStackContext(self._handle_exception):
self.parsed = urlparse.urlsplit(_unicode(self.request.url))
if self.parsed.scheme not in ("http", "https"):
raise ValueError("Unsupported url scheme: %s" %
self.request.url)
# urlsplit results have hostname and port results, but they
# didn't support ipv6 literals until python 2.7.
netloc = self.parsed.netloc
if "@" in netloc:
userpass, _, netloc = netloc.rpartition("@")
match = re.match(r'^(.+):(\d+)$', netloc)
if match:
host = match.group(1)
port = int(match.group(2))
else:
host = netloc
port = 443 if self.parsed.scheme == "https" else 80
if re.match(r'^\[.*\]$', host):
# raw ipv6 addresses in urls are enclosed in brackets
host = host[1:-1]
self.parsed_hostname = host # save final host for _on_connect
if request.allow_ipv6:
af = socket.AF_UNSPEC
else:
# We only try the first IP we get from getaddrinfo,
# so restrict to ipv4 by default.
af = socket.AF_INET
# 支持request的timeout
timeout = min(self.request.connect_timeout, self.request.request_timeout)
if timeout:
self._timeout = self.io_loop.add_timeout(
self.start_time + timeout,
stack_context.wrap(self._on_timeout))
# dns解析模块，解析ip地址
# Tornado 本身提供两个dns解析库，一个是基于多线程的，一个就是直接同步的blocking版本
self.resolver.resolve(host, port, af, callback=self._on_resolve)
开发者ID:zhkzyth，项目名称:tornado-reading-notes，代码行数:55，代码来源:simple_httpclient.py
示例14: __init__
def __init__(self, io_loop, client, request, release_callback,
final_callback, max_buffer_size, tcp_client,
max_header_size, max_body_size):
self.start_time = io_loop.time()
self.io_loop = io_loop
self.client = client
self.request = request
self.release_callback = release_callback
self.final_callback = final_callback
self.max_buffer_size = max_buffer_size
self.tcp_client = tcp_client
self.max_header_size = max_header_size
self.max_body_size = max_body_size
self.code = None
self.headers = None
self.chunks = []
self._decompressor = None
# Timeout handle returned by IOLoop.add_timeout
self._timeout = None
self._sockaddr = None
with stack_context.ExceptionStackContext(self._handle_exception):
self.parsed = urlparse.urlsplit(_unicode(self.request.url))
if self.parsed.scheme not in ("http", "https"):
raise ValueError("Unsupported url scheme: %s" %
self.request.url)
# urlsplit results have hostname and port results, but they
# didn't support ipv6 literals until python 2.7.
netloc = self.parsed.netloc
if "@" in netloc:
userpass, _, netloc = netloc.rpartition("@")
host, port = httputil.split_host_and_port(netloc)
if port is None:
port = 443 if self.parsed.scheme == "https" else 80
if re.match(r'^\[.*\]$', host):
# raw ipv6 addresses in urls are enclosed in brackets
host = host[1:-1]
self.parsed_hostname = host # save final host for _on_connect
if request.allow_ipv6 is False:
af = socket.AF_INET
else:
af = socket.AF_UNSPEC
ssl_options = self._get_ssl_options(self.parsed.scheme)
timeout = min(self.request.connect_timeout, self.request.request_timeout)
if timeout:
self._timeout = self.io_loop.add_timeout(
self.start_time + timeout,
stack_context.wrap(self._on_timeout))
self.tcp_client.connect(host, port, af=af,
ssl_options=ssl_options,
max_buffer_size=self.max_buffer_size,
callback=self._on_connect)
开发者ID:G-Moriarty，项目名称:tornado，代码行数:54，代码来源:simple_httpclient.py
示例15: getlist
def getlist(self, name, strip=True):
values = []
for v in self.get(name, []):
v = _unicode(v)
if isinstance(v, unicode):
# Get rid of any weird control chars (unless decoding gave
# us bytes, in which case leave it alone)
v = re.sub(r"[\x00-\x08\x0e-\x1f]", " ", v)
if strip:
v = v.strip()
values.append(v)
return values
开发者ID:Mondego，项目名称:pyreco，代码行数:12，代码来源:allPythonContent.py
示例16: __init__
def __init__(self, io_loop, client, request, release_callback,
final_callback, max_buffer_size):
self.start_time = io_loop.time()
self.io_loop = io_loop
self.client = client
self.request = request
self.release_callback = release_callback
self.final_callback = final_callback
self.max_buffer_size = max_buffer_size
self.code = None
self.headers = None
self.chunks = None
self._decompressor = None
# Timeout handle returned by IOLoop.add_timeout
self._timeout = None
with stack_context.ExceptionStackContext(self._handle_exception):
self.parsed = urlparse.urlsplit(_unicode(self.request.url))
if ssl is None and self.parsed.scheme == "https":
raise ValueError("HTTPS requires either python2.6+ or "
"curl_httpclient")
if self.parsed.scheme not in ("http", "https"):
raise ValueError("Unsupported url scheme: %s" %
self.request.url)
# urlsplit results have hostname and port results, but they
# didn't support ipv6 literals until python 2.7.
netloc = self.parsed.netloc
if "@" in netloc:
userpass, _, netloc = netloc.rpartition("@")
match = re.match(r'^(.+):(\d+)$', netloc)
if match:
host = match.group(1)
port = int(match.group(2))
else:
host = netloc
port = 443 if self.parsed.scheme == "https" else 80
if re.match(r'^\[.*\]$', host):
# raw ipv6 addresses in urls are enclosed in brackets
host = host[1:-1]
self.parsed_hostname = host # save final host for _on_connect
if self.client.hostname_mapping is not None:
host = self.client.hostname_mapping.get(host, host)
if request.allow_ipv6:
af = socket.AF_UNSPEC
else:
# We only try the first IP we get from getaddrinfo,
# so restrict to ipv4 by default.
af = socket.AF_INET
self.client.resolver.getaddrinfo(
host, port, af, socket.SOCK_STREAM, 0, 0,
callback=self._on_resolve)
开发者ID:CNCBASHER，项目名称:tornado，代码行数:52，代码来源:simple_httpclient.py
示例17: __init__
def __init__(self, **kwargs):
self._raw_data = {}
# handler.request.arguments, utf-8 values
for k in kwargs:
if isinstance(kwargs[k], list):
self._raw_data[k] = map(_unicode, kwargs[k])
else:
self._raw_data[k] = _unicode(kwargs[k])
self.data = {}
self.errors = []
self.validate()
开发者ID:linyaquan，项目名称:torext，代码行数:13，代码来源:params.py
示例18: make_content
def make_content(text, extra_params='rel="nofollow"'):
"""https://github.com/facebook/tornado/blob/master/tornado/escape.py#L238
"""
if extra_params:
extra_params = " " + extra_params.strip()
def make_link(m):
url = m.group(1)
proto = m.group(2)
href = m.group(1)
if not proto:
href = "http://" + href # no proto specified, use http
params = extra_params
if '.' in href:
name_extension = href.split('.')[-1].lower()
if name_extension in ('jpg', 'png', 'git', 'jpeg'):
return u'<img src="%s" />' % href
return u'<a href="%s"%s>%s</a>' % (href, params, url)
def cover_email(m):
data = {'mail': m.group(1),
'end': m.group(2)}
return u'<a href="mailto:%(mail)s">%(mail)s</a>%(end)s' % data
def convert_mention(m):
data = {}
data['begin'], data['user'] = m.group(1).split('@')
t = u'%(begin)s<a href="/member/%(user)s" class="mention">' \
u'@%(user)s</a>'
return t % data
def highligt(m):
try:
name = m.group(1)
lexer = get_lexer_by_name(name)
except ValueError:
lexer = TextLexer()
text = m.group(2).replace('"', '"').replace('&', '&')
text = text.replace('<', '<').replace('>', '>')
text = text.replace(' ', ' ')
return highlight(text, lexer, formatter)
text = _unicode(xhtml_escape(text)).replace(' ', ' ')
text = _CODE_RE.sub(highligt, text).replace('\n', '<br />')
text = _EMAIL_RE.sub(cover_email, text)
text = _MENTION_RE.sub(convert_mention, text)
return _URL_RE.sub(make_link, text)
开发者ID:VTScn，项目名称:PBB，代码行数:51，代码来源:utils.py
示例19: match_list
def match_list(self, match_list):
arguments = self.request.arguments
if '__NODELAY__' in arguments:
del arguments['__NODELAY__']
if '__BLACKLIST__' in arguments:
del arguments['__BLACKLIST__']
if '__ASYNCLIST__' in arguments:
del arguments['__ASYNCLIST__']
if '__ASYNC_RESULT__' in arguments:
del arguments['__ASYNC_RESULT__']
if '__ENABLE_DEBUG__' in arguments:
del arguments['__ENABLE_DEBUG__']
if '__CONTENT_TYPE__' in arguments:
del arguments['__CONTENT_TYPE__']
if '__JSONP_CALLBACK_VARNAME__' in arguments:
del arguments['__JSONP_CALLBACK_VARNAME__']
match = "|".join(match_list)
for k in arguments.keys():
if re.match(match,_unicode(" ".join(arguments[k]))):
return True
if server_request.body != '' and re.match(match,_unicode(server_request.body)):
return True
return False
开发者ID:icatholic，项目名称:icc，代码行数:24，代码来源:trouter.py
示例20: make_content
def make_content(text, extra_params='rel="nofollow"'):
"""https://github.com/facebook/tornado/blob/master/tornado/escape.py#L238
"""
if extra_params:
extra_params = " " + extra_params.strip()
def make_link(m):
url = m.group(1)
proto = m.group(2)
href = m.group(1)
if not proto:
href = "http://" + href # no proto specified, use http
params = extra_params
if "." in href:
name_extension = href.split(".")[-1]
if name_extension in ("jpg", "png", "git", "jpeg"):
return u'<img src="%s" />' % href
return u'<a href="%s"%s>%s</a>' % (href, params, url)
def cover_email(m):
data = {"mail": m.group(1), "end": m.group(2)}
return u'<a href="mailto:%(mail)s">%(mail)s</a>%(end)s' % data
def convert_mention(m):
data = {}
data["begin"], data["user"] = m.group(1).split("@")
t = u'%(begin)s<a href="/member/%(user)s" class="mention">' u"@%(user)s</a>"
return t % data
def highligt(m):
try:
name = m.group(1)
lexer = get_lexer_by_name(name)
except ValueError:
lexer = TextLexer()
text = m.group(2).replace(""", '"').replace("&", "&")
text = text.replace("<", "<").replace(">", ">")
return highlight(text, lexer, formatter)
text = _unicode(xhtml_escape(text))
text = _CODE_RE.sub(highligt, text).replace("\n", "<br />")
text = _EMAIL_RE.sub(cover_email, text)
text = _MENTION_RE.sub(convert_mention, text)
return _URL_RE.sub(make_link, text)
开发者ID:alphachoi，项目名称:PBB，代码行数:48，代码来源:utils.py
注：本文中的tornado.escape._unicode函数示例由纯净天空整理自Github/MSDocs等源码及文档管理平台，相关代码片段筛选自各路编程大神贡献的开源项目，源码版权归原作者所有，传播和使用请参考对应项目的License；未经允许，请勿转载。 |
客服电话
APP下载
官方微信
请发表评论