Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
# -*- coding: utf-8 -*-
from __future__ import unicode_literals
from __future__ import print_function
import re
import datetime
from boltons.iterutils import unique
from hyperlink import parse as parse_url
import requests
MW_API_URL = parse_url('https://en.wikipedia.org/w/api.php')
REST_API_BASE_URL = parse_url('https://en.wikipedia.org/api/rest_v1/')
REF_API_BASE_URL = REST_API_BASE_URL.child('page', 'references')
from log import tlog
def format_datetime(dt):
if isinstance(dt, datetime.date):
dt = datetime.datetime(dt.year, dt.month, dt.day, 0, 0, 0)
return dt.isoformat().split('.')[0] + 'Z'
## PTArticle-based Metrics
def get_revid(pta):
# -*- coding: utf-8 -*-
from __future__ import unicode_literals
from __future__ import print_function
import re
import datetime
from boltons.iterutils import unique
from hyperlink import parse as parse_url
import requests
MW_API_URL = parse_url('https://en.wikipedia.org/w/api.php')
REST_API_BASE_URL = parse_url('https://en.wikipedia.org/api/rest_v1/')
REF_API_BASE_URL = REST_API_BASE_URL.child('page', 'references')
from log import tlog
def format_datetime(dt):
if isinstance(dt, datetime.date):
dt = datetime.datetime(dt.year, dt.month, dt.day, 0, 0, 0)
return dt.isoformat().split('.')[0] + 'Z'
## PTArticle-based Metrics
def get_revid(pta):
return _get_revid_at_timestamp(pta.title, format_datetime(pta.timestamp))
def topic_links(realm_filters_key: int, topic_name: str) -> List[str]:
matches = [] # type: List[str]
realm_filters = realm_filters_for_realm(realm_filters_key)
for realm_filter in realm_filters:
pattern = prepare_realm_pattern(realm_filter[0])
for m in re.finditer(pattern, topic_name):
matches += [realm_filter[1] % m.groupdict()]
# Also make raw urls navigable.
for sub_string in basic_link_splitter.split(topic_name):
link_match = re.match(get_web_link_regex(), sub_string)
if link_match:
url = link_match.group('url')
url_object = parse(url)
if not url_object.scheme:
url = url_object.replace(scheme='https').to_text()
matches.append(url)
return matches
def _create_client_service(self, nodeurl, api_token):
url = parse(nodeurl)
wsurl = url.replace(scheme="ws").child("private", "logs", "v1")
factory = WebSocketClientFactory(
url=wsurl.to_uri().to_text(),
headers={
"Authorization": "{} {}".format("tahoe-lafs", api_token),
},
)
factory.protocol = TahoeLogReader
factory.streamedlogs = self
endpoint = TCP4ClientEndpoint(self._reactor, url.host, url.port)
client_service = ClientService(endpoint, factory, clock=self._reactor)
return client_service
def parse_valid_url(url, schemes=None):
url_obj = url_parse(url)
url_obj = url_obj.normalize()
assert url_obj.scheme
if schemes is not None:
assert url_obj.scheme in schemes
assert url_obj.host
return url_obj
import attr
import treq
from txacme.interfaces import IResponder
from txacme.challenges._libcloud import _validation
from zope.interface import implementer
from twisted.internet.defer import inlineCallbacks
from twisted.internet.task import deferLater
from hyperlink import parse
from ._common import ConsistencyChecker
base = parse("https://api.cloudflare.com/client/v4/")
def global_reactor():
from twisted.internet import reactor
return reactor
@attr.s(hash=False)
@implementer(IResponder)
class CloudflareV4Responder(object):
"""
Cloudflare API V4 responder.
"""
_email = attr.ib()
_api_key = attr.ib()
_zone_name = attr.ib()
_reactor = attr.ib(default=attr.Factory(global_reactor))
def parse_valid_url(url, schemes=None):
url_obj = url_parse(url)
url_obj = url_obj.normalize()
assert url_obj.scheme
if schemes is not None:
assert url_obj.scheme in schemes
assert url_obj.host
return url_obj