Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
except Exception as e:
code, _ = util.interpret_http_exception(e)
if code in Instagram.RATE_LIMIT_HTTP_CODES:
self.messages.add(
'<a href="https://github.com/snarfed/bridgy/issues/665#issuecomment-524977427">Apologies, Instagram is temporarily blocking us.</a> Please try again later!')
return self.redirect('/')
else:
raise
if not actor:
self.messages.add("Couldn't find Instagram user '%s'. Please check your site's rel-me link and your Instagram account." % username)
return self.redirect('/')
canonicalize = util.UrlCanonicalizer(redirects=False)
website = canonicalize(auth_entity.key.id())
urls = [canonicalize(u) for u in microformats2.object_urls(actor)]
logging.info('Looking for %s in %s', website, urls)
if website not in urls:
self.messages.add("Please add %s to your Instagram profile's website or bio field and try again." % website)
return self.redirect('/')
# check that the instagram account is public
if not gr_source.Source.is_public(actor):
self.messages.add('Your Instagram account is private. Bridgy only supports public accounts.')
return self.redirect('/')
self.maybe_add_or_delete_source(Instagram, auth_entity, state, actor=actor)
type = att.get('objectType')
if type == 'image':
att['image'] = util.get_first(att, 'image')
image_atts.append(att['image'])
continue
image_urls_seen |= set(util.get_urls(att, 'image'))
if type in ('note', 'article'):
html = microformats2.render_content(
att, include_location=reader, render_attachments=True,
white_space_pre=False)
author = att.get('author')
if author:
name = microformats2.maybe_linked_name(
microformats2.object_to_json(author).get('properties') or {})
html = '%s: %s' % (name.strip(), html)
children.append(html)
# render image(s) that we haven't already seen
for image in image_atts + util.get_list(obj, 'image'):
if not image:
continue
url = image.get('url')
parsed = urllib.parse.urlparse(url)
rest = urllib.parse.urlunparse(('', '') + parsed[2:])
img_src_re = re.compile(r"""src *= *['"] *((https?:)?//%s)?%s *['"]""" %
(re.escape(parsed.netloc),
_encode_ampersands(re.escape(rest))))
if (url and url not in image_urls_seen and
not img_src_re.search(obj['rendered_content'])):
children.append(microformats2.img(url))
if not obj:
self.abort(404, 'Not found: %s:%s %s %s' %
(source_short_name, string_id, type, ids))
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!
Args:
html: unicode string
url: string URL html came from, optional
fetch_author: boolean, whether to make HTTP request to fetch rel-author link
reader: boolean, whether the output will be rendered in a feed reader.
Currently just includes location if True, not otherwise.
Returns:
unicode string with Atom XML
"""
if fetch_author:
assert url, 'fetch_author=True requires url!'
parsed = util.parse_mf2(html, url=url)
actor = microformats2.find_author(parsed, fetch_mf2_func=util.fetch_mf2)
return activities_to_atom(
microformats2.html_to_activities(html, url, actor),
actor,
title=microformats2.get_title(parsed),
xml_base=util.base_url(url),
host_url=url,
reader=reader)
logging.info('Logged in as %s (%s)',
actor.get('username'), actor.get('displayName'))
else:
logging.warning("Couldn't determine Instagram user!")
activities = resp.get('items', [])
format = self.request.get('format', 'atom')
if format == 'atom':
title = 'instagram-atom feed for %s' % ig.actor_name(actor)
self.response.headers['Content-Type'] = 'application/atom+xml'
self.response.out.write(atom.activities_to_atom(
activities, actor, title=title, host_url=host_url,
request_url=self.request.path_url, xml_base='https://www.instagram.com/'))
elif format == 'html':
self.response.headers['Content-Type'] = 'text/html'
self.response.out.write(microformats2.activities_to_html(activities))
else:
self.abort(400, 'format must be either atom or html; got %s' % format)
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]