How to use the granary.microformats2.first_props function in granary

To help you get started, we’ve selected a few granary examples, based on popular ways it is used in public projects.

Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.

github snarfed / bridgy / publish.py View on Github external
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))
github snarfed / bridgy / original_post_discovery.py View on Github external
def updated_or_published(item):
    props = microformats2.first_props(item.get('properties'))
    return props.get('updated') or props.get('published') or ''
github snarfed / bridgy / handlers.py View on Github external
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', '')
github snarfed / bridgy / publish.py View on Github external
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))
github snarfed / ownyourresponses / app.py View on Github external
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))