def update_activity():
    events = {}
    event_counts = collections.Counter()

    query = (ev for ev in LiveUpdateEvent._all()
             if ev.state == "live" and not ev.banned)
    for chunk in utils.in_chunks(query, size=100):
        context_ids = {ev._fullname: ev._id for ev in chunk}

        view_countable = [ev._fullname for ev in chunk
                          if ev._date >= g.liveupdate_min_date_viewcounts]
        view_counts_query = ViewCountsQuery.execute_async(view_countable)

        try:
            with c.activity_service.retrying(attempts=4) as svc:
                infos = svc.count_activity_multi(context_ids.keys())
        except TTransportException:
            continue

        view_counts = view_counts_query.result()

        for context_id, info in infos.iteritems():
            event_id = context_ids[context_id]

            try:
                LiveUpdateActivityHistoryByEvent.record_activity(
                    event_id, info.count)
            except tdb_cassandra.TRANSIENT_EXCEPTIONS as e:
                g.log.warning("Failed to update activity history for %r: %s",
                              event_id, e)

            try:
                event = LiveUpdateEvent.update_activity(
                    event_id, info.count, info.is_fuzzed)
            except tdb_cassandra.TRANSIENT_EXCEPTIONS as e:
                g.log.warning("Failed to update event activity for %r: %s",
                              event_id, e)
            else:
                events[event_id] = event
                event_counts[event_id] = info.count

            websockets.send_broadcast(
                "/live/" + event_id,
                type="activity",
                payload={
                    "count": info.count,
                    "fuzzed": info.is_fuzzed,
                    "total_views": view_counts.get(context_id),
                },
            )

    top_event_ids = [event_id for event_id, count in event_counts.most_common(1000)]
    top_events = [events[event_id] for event_id in top_event_ids]
    query_ttl = datetime.timedelta(days=3)
    with CachedQueryMutator() as m:
        m.replace(get_active_events(), top_events, ttl=query_ttl)

    # ensure that all the amqp messages we've put on the worker's queue are
    # sent before we allow this script to exit.
    amqp.worker.join()

def update_activity():
    event_ids = ActiveVisitorsByLiveUpdateEvent._cf.get_range(column_count=1, filter_empty=False)

    for event_id, is_active in event_ids:
        count = 0

        if is_active:
            try:
                count = ActiveVisitorsByLiveUpdateEvent.get_count(event_id)
            except tdb_cassandra.TRANSIENT_EXCEPTIONS as e:
                g.log.warning("Failed to fetch activity count for %r: %s", event_id, e)
                return

        try:
            LiveUpdateEvent.update_activity(event_id, count)
        except tdb_cassandra.TRANSIENT_EXCEPTIONS as e:
            g.log.warning("Failed to update event activity for %r: %s", event_id, e)

        try:
            LiveUpdateActivityHistoryByEvent.record_activity(event_id, count)
        except tdb_cassandra.TRANSIENT_EXCEPTIONS as e:
            g.log.warning("Failed to update activity history for %r: %s", event_id, e)

        is_fuzzed = False
        if count < ACTIVITY_FUZZING_THRESHOLD:
            count = utils.fuzz_activity(count)
            is_fuzzed = True

        websockets.send_broadcast("/live/" + event_id, type="activity", payload={"count": count, "fuzzed": is_fuzzed})

    # ensure that all the amqp messages we've put on the worker's queue are
    # sent before we allow this script to exit.
    amqp.worker.join()

    def POST_create(self, form, jquery, title, description, resources, nsfw):
        """Create a new live thread.

        Once created, the initial settings can be modified with
        [/api/live/*thread*/edit](#POST_api_live_{thread}_edit) and new updates
        can be posted with
        [/api/live/*thread*/update](#POST_api_live_{thread}_update).

        """
        if not is_event_configuration_valid(form):
            return

        if form.has_errors("ratelimit", errors.RATELIMIT):
            return
        VRatelimit.ratelimit(
            rate_user=True, prefix="liveupdate_create_", seconds=60)

        event = LiveUpdateEvent.new(
            id=None,
            title=title,
            description=description,
            resources=resources,
            banned=c.user._spam,
            nsfw=nsfw,
        )
        event.add_contributor(c.user, ContributorPermissionSet.SUPERUSER)
        queries.create_event(event)

        form.redirect("/live/" + event._id)
        form._send_data(id=event._id)

def close_abandoned_threads():
    """Find live threads that are abandoned and close them.

    Jobs like the activity tracker iterate through all open live threads, so
    closing abandoned threads removes some effort from them and is generally
    good for cleanliness.

    """
    now = datetime.datetime.now(pytz.UTC)
    horizon = now - DERELICTION_THRESHOLD

    for event in LiveUpdateEvent._all():
        if event.state != "live" or event.banned:
            continue

        try:
            columns = LiveUpdateStream._cf.get(
                event._id, column_reversed=True, column_count=1)
        except NotFoundException:
            event_last_modified = event._date
        else:
            updates = LiveUpdateStream._column_to_obj([columns])
            most_recent_update = updates.pop()
            event_last_modified = most_recent_update._date

        if event_last_modified < horizon:
            g.log.warning("Closing %s for inactivity.", event._id)
            close_event(event)

def parse_embeds(event_id, liveupdate_id, maxwidth=_EMBED_WIDTH):
    """Find, scrape, and store any embeddable URLs in this liveupdate.

    Return the newly altered liveupdate for convenience.
    Note: This should be used in async contexts only.

    """
    if isinstance(liveupdate_id, basestring):
        liveupdate_id = uuid.UUID(liveupdate_id)

    try:
        event = LiveUpdateEvent._byID(
            event_id, read_consistency_level=tdb_cassandra.CL.QUORUM)
        liveupdate = LiveUpdateStream.get_update(
            event, liveupdate_id, read_consistency_level=tdb_cassandra.CL.QUORUM)
    except tdb_cassandra.NotFound:
        g.log.warning("Couldn't find event/liveupdate for embedding: %r / %r",
                      event_id, liveupdate_id)
        return

    urls = _extract_isolated_urls(liveupdate.body)
    liveupdate.media_objects = _scrape_media_objects(urls, maxwidth=maxwidth)
    liveupdate.mobile_objects = _scrape_mobile_media_objects(urls)
    LiveUpdateStream.add_update(event, liveupdate)

    return liveupdate

    def __before__(self, event):
        RedditController.__before__(self)

        if event:
            try:
                c.liveupdate_event = LiveUpdateEvent._byID(event)
            except tdb_cassandra.NotFound:
                pass

        if not c.liveupdate_event:
            self.abort404()

        if c.user_is_loggedin:
            c.liveupdate_permissions = \
                    c.liveupdate_event.get_permissions(c.user)

            # revoke some permissions from everyone after closing
            if c.liveupdate_event.state != "live":
                c.liveupdate_permissions = (c.liveupdate_permissions
                    .without("update")
                    .without("close")
                )

            if c.user_is_admin:
                c.liveupdate_permissions = ContributorPermissionSet.SUPERUSER
        else:
            c.liveupdate_permissions = ContributorPermissionSet.NONE

def update_activity():
    events = {}
    event_counts = collections.Counter()
    event_ids = ActiveVisitorsByLiveUpdateEvent._cf.get_range(
        column_count=1, filter_empty=False)

    for event_id, is_active in event_ids:
        count = 0

        if is_active:
            try:
                count = ActiveVisitorsByLiveUpdateEvent.get_count(event_id)
            except tdb_cassandra.TRANSIENT_EXCEPTIONS as e:
                g.log.warning("Failed to fetch activity count for %r: %s",
                              event_id, e)
                return

        try:
            LiveUpdateActivityHistoryByEvent.record_activity(event_id, count)
        except tdb_cassandra.TRANSIENT_EXCEPTIONS as e:
            g.log.warning("Failed to update activity history for %r: %s",
                          event_id, e)

        is_fuzzed = False
        if count < ACTIVITY_FUZZING_THRESHOLD:
            count = utils.fuzz_activity(count)
            is_fuzzed = True

        try:
            event = LiveUpdateEvent.update_activity(event_id, count, is_fuzzed)
        except tdb_cassandra.TRANSIENT_EXCEPTIONS as e:
            g.log.warning("Failed to update event activity for %r: %s",
                          event_id, e)
        else:
            events[event_id] = event
            event_counts[event_id] = count

        websockets.send_broadcast(
            "/live/" + event_id,
            type="activity",
            payload={
                "count": count,
                "fuzzed": is_fuzzed,
            },
        )

    top_event_ids = [event_id for event_id, count in event_counts.most_common(1000)]
    top_events = [events[event_id] for event_id in top_event_ids]
    query_ttl = datetime.timedelta(days=3)
    with CachedQueryMutator() as m:
        m.replace(get_active_events(), top_events, ttl=query_ttl)

    # ensure that all the amqp messages we've put on the worker's queue are
    # sent before we allow this script to exit.
    amqp.worker.join()

 def GET_happening_now(self):
     current_thread_id = NamedGlobals.get(HAPPENING_NOW_KEY, None)
     if current_thread_id:
         current_thread = LiveUpdateEvent._byID(current_thread_id)
     else:
         current_thread = None
     return AdminPage(
             content=pages.HappeningNowAdmin(current_thread),
             title='live: happening now',
             nav_menus=[]
         ).render()

    def __before__(self, event):
        MinimalController.__before__(self)

        if event:
            try:
                c.liveupdate_event = LiveUpdateEvent._byID(event)
            except tdb_cassandra.NotFound:
                pass

        if not c.liveupdate_event:
            self.abort404()

    def __before__(self, event):
        RedditController.__before__(self)

        if event:
            try:
                c.liveupdate_event = LiveUpdateEvent._byID(event)
            except tdb_cassandra.NotFound:
                pass

        if not c.liveupdate_event:
            self.abort404()

        c.liveupdate_can_manage = (c.liveupdate_event.state == "live" and
                                   (c.user_is_loggedin and c.user_is_admin))
        c.liveupdate_can_edit = (c.liveupdate_event.state == "live" and
                                 (c.user_is_loggedin and
                                  (c.liveupdate_event.is_editor(c.user) or
                                   c.user_is_admin)))

    def POST_create(self, form, jquery, title, description):
        if not is_event_configuration_valid(form):
            return

        if form.has_errors("ratelimit", errors.RATELIMIT):
            return
        VRatelimit.ratelimit(
            rate_user=True, prefix="liveupdate_create_", seconds=60)

        event = LiveUpdateEvent.new(
            id=None,
            title=title,
            banned=c.user._spam,
        )
        event.add_contributor(c.user, ContributorPermissionSet.SUPERUSER)
        queries.create_event(event)

        form.redirect("/live/" + event._id)

    def POST_create(self, form, jquery, title, description, resources, nsfw):
        """Create a new live thread.

        Once created, the initial settings can be modified with
        [/api/live/*thread*/edit](#POST_api_live_{thread}_edit) and new updates
        can be posted with
        [/api/live/*thread*/update](#POST_api_live_{thread}_update).

        """
        if not is_event_configuration_valid(form):
            return

        # for simplicity, set the live-thread creation threshold at the
        # subreddit creation threshold
        if not c.user_is_admin and not c.user.can_create_subreddit:
            form.set_error(errors.CANT_CREATE_SR, "")
            c.errors.add(errors.CANT_CREATE_SR, field="")
            return

        if form.has_errors("ratelimit", errors.RATELIMIT):
            return

        VRatelimit.ratelimit(
            rate_user=True, prefix="liveupdate_create_", seconds=60)

        event = LiveUpdateEvent.new(
            id=None,
            title=title,
            description=description,
            resources=resources,
            banned=c.user._spam,
            nsfw=nsfw,
        )
        event.add_contributor(c.user, ContributorPermissionSet.SUPERUSER)
        queries.create_event(event)

        form.redirect("/live/" + event._id)
        form._send_data(id=event._id)
        liveupdate_events.create_event(event, context=c, request=request)

def add_featured_live_thread(controller):
    """If we have a live thread featured, display it on the homepage."""
    if not feature.is_enabled('live_happening_now'):
        return None

    # Not on front page
    if not isinstance(c.site, DefaultSR):
        return None

    # Not on first page of front page
    if getattr(controller, 'listing_obj') and controller.listing_obj.prev:
        return None

    event_id = NamedGlobals.get(HAPPENING_NOW_KEY, None)
    if not event_id:
        return None

    try:
        event = LiveUpdateEvent._byID(event_id)
    except NotFound:
        return None
    else:
        return pages.LiveUpdateHappeningNowBar(event=event)

 def thing_lookup(self, names):
     return LiveUpdateEvent._byID(names, return_dict=False)