How to use the granary.source.object_type function in granary

trim_nulls: boolean, whether to remove elements with null or empty values
    entry_class: string or sequence, the mf2 class(es) that entries should be
      given (e.g. 'h-cite' when parsing a reference to a foreign entry).
      defaults to 'h-entry'
    default_object_type: string, the ActivityStreams objectType to use if one
      is not present. defaults to None
    synthesize_content: whether to generate synthetic content if the object
      doesn't have its own, e.g. 'likes this.' or 'shared this.'

  Returns:
    dict, decoded microformats2 JSON
  """
  if not obj or not isinstance(obj, dict):
    return {}

  obj_type = source.object_type(obj) or default_object_type
  # if the activity type is a post, then it's really just a conduit
  # for the object. for other verbs, the activity itself is the
  # interesting thing
  if obj_type == 'post':
    primary = obj.get('object', {})
    obj_type = source.object_type(primary) or default_object_type
  else:
    primary = obj

  # TODO: extract snippet
  name = primary.get('displayName', primary.get('title'))
  summary = primary.get('summary')
  author = obj.get('author', obj.get('actor', {}))

  in_reply_tos = obj.get('inReplyTo', obj.get('context', {}).get('inReplyTo', []))
  is_rsvp = obj_type in ('rsvp-yes', 'rsvp-no', 'rsvp-maybe')

def _prepare_activity(a, reader=True):
  """Preprocesses an activity to prepare it to be rendered as Atom.

  Modifies a in place.

  Args:
    a: ActivityStreams 1 activity dict
    reader: boolean, whether the output will be rendered in a feed reader.
      Currently just includes location if True, not otherwise.
  """
  act_type = source.object_type(a)
  obj = util.get_first(a, 'object', default={})
  primary = obj if (not act_type or act_type == 'post') else a

  # Render content as HTML; escape &s
  obj['rendered_content'] = _encode_ampersands(microformats2.render_content(
    primary, include_location=reader, render_attachments=True,
    # Readers often obey CSS white-space: pre strictly and don't even line wrap,
    # so don't use it.
    # https://forum.newsblur.com/t/android-cant-read-line-pre-formatted-lines/6116
    white_space_pre=False))

  # Make sure every activity has the title field, since Atom  requires
  # the title element.
  if not a.get('title'):
    a['title'] = util.ellipsize(_encode_ampersands(
      a.get('displayName') or a.get('content') or obj.get('title') or

def render(source, activity, base):
    obj = activity.get('object') or activity
    content = microformats2.render_content(obj)
    embed = source.embed_post(base)

    type = as_source.object_type(activity)
    content = activity.get('content', '')
    if type == 'share' and not content:
      content = 'retweeted this.'

    rendered = embed + content if type == 'comment' else content + embed

    mf2_class = {'like': 'u-like-of',
                 'share': 'u-repost-of',
                 }.get(type, 'in-reply-to')
    url = (obj.get('inReplyTo') or [{}])[0].get('url') or base.get('url')
    rendered += """
<a href="%s" class="%s"></a>
<a href="%s" class="u-syndication"></a>
""" % (mf2_class, url, activity.get('url'))

    return rendered

trim_nulls: boolean, whether to remove elements with null or empty values
    entry_class: string or sequence, the mf2 class(es) that entries should be
      given (e.g. 'h-cite' when parsing a reference to a foreign entry).
      defaults to 'h-entry'
    default_object_type: string, the ActivityStreams objectType to use if one
      is not present. defaults to None
    synthesize_content: whether to generate synthetic content if the object
      doesn't have its own, e.g. 'likes this.' or 'shared this.'

  Returns:
    dict, decoded microformats2 JSON
  """
  if not obj or not isinstance(obj, dict):
    return {}

  obj_type = source.object_type(obj) or default_object_type
  # if the activity type is a post, then it's really just a conduit
  # for the object. for other verbs, the activity itself is the
  # interesting thing
  if obj_type == 'post':
    primary = obj.get('object', {})
    obj_type = source.object_type(primary) or default_object_type
  else:
    primary = obj

  # TODO: extract snippet
  name = primary.get('displayName', primary.get('title'))
  summary = primary.get('summary')
  author = obj.get('author', obj.get('actor', {}))

  in_reply_tos = obj.get('inReplyTo', obj.get('context', {}).get('inReplyTo', []))
  is_rsvp = obj_type in ('rsvp-yes', 'rsvp-no', 'rsvp-maybe')

def is_activity_public(self, activity):
    """Returns True if the given activity is public, False otherwise.

    Uses the :attr:`post_publics_json` cache if we can't tell otherwise.
    """
    obj = activity.get('object', {})
    fb_id = activity.get('fb_id') or obj.get('fb_id')
    if fb_id and gr_source.object_type(activity) not in ('comment', 'like', 'share'):
      fb_id = self.cached_resolve_object_id(fb_id, activity=activity)

    post_publics = self._load_cache('post_publics')
    public = gr_source.Source.is_public(activity)

    if not fb_id:
      return public
    elif public is not None:
      post_publics[fb_id] = public    # write cache
      return public
    else:
      return post_publics.get(fb_id)  # read cache

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))

# linkify embedded links. ignore the "mention" tags that we added ourselves.
  # TODO: fix the bug in test_linkify_broken() in webutil/tests/test_util.py, then
  # uncomment this.
  # if content:
  #   content = util.linkify(content)

  # attachments, e.g. links (aka articles)
  # TODO: use oEmbed? http://oembed.com/ , http://code.google.com/p/python-oembed/
  if render_attachments:
    atts = [a for a in obj.get('attachments', [])
            if a.get('objectType') not in ('note', 'article')]
    content += _render_attachments(atts + tags.pop('article', []), obj)

  # generate share/like contexts if the activity does not have content
  # of its own
  obj_type = source.object_type(obj)
  for as_type, verb in (
      ('favorite', 'Favorites'), ('like', 'Likes'), ('share', 'Shared')):
    if (not synthesize_content or obj_type != as_type or 'object' not in obj or
        'content' in obj):
      continue

    targets = get_list(obj, 'object')
    if not targets:
      continue

    for target in targets:
      # sometimes likes don't have enough content to render anything
      # interesting
      if 'url' in target and set(target) <= set(['url', 'objectType']):
        content += '<a href="%s">%s this.</a>' % (
          target.get('url'), verb.lower())