Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def attempt_single_item(self, item):
"""Attempts to preview or publish a single mf2 item.
Args:
item: mf2 item dict from mf2py
Returns:
CreationResult
"""
self.maybe_inject_silo_content(item)
obj = microformats2.json_to_object(item)
ignore_formatting = self.ignore_formatting(item)
if ignore_formatting:
prop = microformats2.first_props(item.get('properties', {}))
content = microformats2.get_text(prop.get('content'))
if content:
obj['content'] = content.strip()
# which original post URL to include? in order of preference:
# 1. rel-shortlink (background: https://github.com/snarfed/bridgy/issues/173)
# 2. original user-provided URL if it redirected
# 3. u-url if available
# 4. actual final fetched URL
if self.shortlink:
obj['url'] = self.shortlink
elif self.source_url() != self.fetched.url:
obj['url'] = self.source_url()
elif 'url' not in obj:
obj['url'] = self.fetched.url
logging.debug('Converted to ActivityStreams object: %s', json.dumps(obj, indent=2))
def updated_or_published(item):
props = microformats2.first_props(item.get('properties'))
return props.get('updated') or props.get('published') or ''
if self.source.is_blocked(obj):
self.abort(410, 'That user is currently blocked')
# use https for profile pictures so we don't cause SSL mixed mode errors
# when serving over https.
author = obj.get('author', {})
image = author.get('image', {})
url = image.get('url')
if url:
image['url'] = util.update_scheme(url, self)
mf2_json = microformats2.object_to_json(obj, synthesize_content=False)
# try to include the author's silo profile url
author = first_props(mf2_json.get('properties', {})).get('author', {})
author_uid = first_props(author.get('properties', {})).get('uid', '')
if author_uid:
parsed = util.parse_tag_uri(author_uid)
if parsed:
urls = author.get('properties', {}).setdefault('url', [])
try:
silo_url = self.source.gr_source.user_url(parsed[1])
if silo_url not in microformats2.get_string_urls(urls):
urls.append(silo_url)
except NotImplementedError: # from gr_source.user_url()
pass
# write the response!
self.response.headers['Access-Control-Allow-Origin'] = '*'
if format == 'html':
self.response.headers['Content-Type'] = 'text/html; charset=utf-8'
url = obj.get('url', '')
def attempt_single_item(self, item):
"""Attempts to preview or publish a single mf2 item.
Args:
item: mf2 item dict from mf2py
Returns:
CreationResult
"""
self.maybe_inject_silo_content(item)
obj = microformats2.json_to_object(item)
ignore_formatting = self.ignore_formatting(item)
if ignore_formatting:
prop = microformats2.first_props(item.get('properties', {}))
content = microformats2.get_text(prop.get('content'))
if content:
obj['content'] = content.strip()
# which original post URL to include? in order of preference:
# 1. rel-shortlink (background: https://github.com/snarfed/bridgy/issues/173)
# 2. original user-provided URL if it redirected
# 3. u-url if available
# 4. actual final fetched URL
if self.shortlink:
obj['url'] = self.shortlink
elif self.source_url() != self.fetched.url:
obj['url'] = self.source_url()
elif 'url' not in obj:
obj['url'] = self.fetched.url
logging.debug('Converted to ActivityStreams object: %s', json_dumps(obj, indent=2))
def poll(self, source):
activities = source.get_activities(group_id=as_source.SELF, fetch_likes=True)
resps = ndb.get_multi(ndb.Key('Response', util.trim_nulls(a['id']))
for a in activities)
resps = {r.key.id(): r for r in resps if r}
exception = None
for activity in activities:
obj = activity.get('object', {})
# have we already posted or started on this response?
resp = resps.get(activity['id'])
mf2 = microformats2.object_to_json(activity)
mf2_props = microformats2.first_props(mf2.get('properties', {}))
type = as_source.object_type(activity)
if mf2_props.get('in-reply-to'):
type = 'comment' # twitter reply
if type not in TYPES or (resp and resp.status == 'complete'):
continue
elif resp:
logging.info('Retrying %s', resp)
else:
resp = Response.get_or_insert(activity['id'],
activity_json=json.dumps(activity))
logging.info('Created new Response: %s', resp)
base_id = source.base_object(activity)['id']
base = source.get_activities(activity_id=base_id)[0]
# logging.info(json.dumps(base, indent=2))