本文整理汇总了Python中reddit_liveupdate.models.LiveUpdateStream类的典型用法代码示例。如果您正苦于以下问题:Python LiveUpdateStream类的具体用法?Python LiveUpdateStream怎么用?Python LiveUpdateStream使用的例子?那么恭喜您, 这里精选的类代码示例或许可以为您提供帮助。
在下文中一共展示了LiveUpdateStream类的14个代码示例,这些例子默认根据受欢迎程度排序。您可以为喜欢或者感觉有用的代码点赞,您的评价将有助于我们的系统推荐出更棒的Python代码示例。
示例1: POST_update
def POST_update(self, form, jquery, text):
if form.has_errors("body", errors.NO_TEXT,
errors.TOO_LONG):
return
# create and store the new update
update = LiveUpdate(data={
"author_id": c.user._id,
"body": text,
"_spam": c.user._spam,
})
hooks.get_hook("liveupdate.update").call(update=update)
LiveUpdateStream.add_update(c.liveupdate_event, update)
# tell the world about our new update
builder = LiveUpdateBuilder(None)
wrapped = builder.wrap_items([update])[0]
rendered = wrapped.render(style="api")
_broadcast(type="update", payload=rendered)
# Queue up parsing any embeds
queue_parse_embeds(c.liveupdate_event, update)
# reset the submission form
t = form.find("textarea")
t.attr('rows', 3).html("").val("")
开发者ID:hubwub,项目名称:reddit-plugin-liveupdate,代码行数:28,代码来源:controllers.py
示例2: parse_embeds
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
开发者ID:reddit,项目名称:reddit-plugin-liveupdate,代码行数:26,代码来源:media_embeds.py
示例3: POST_strike_update
def POST_strike_update(self, form, jquery, update):
if form.has_errors("id", errors.NO_THING_ID):
return
update.stricken = True
LiveUpdateStream.add_update(c.liveupdate_event, update)
send_websocket_broadcast(type="strike", payload=update._fullname)
开发者ID:Web5design,项目名称:reddit-plugin-liveupdate,代码行数:8,代码来源:controllers.py
示例4: POST_strike_update
def POST_strike_update(self, form, jquery, update):
if form.has_errors("id", errors.NO_THING_ID):
return
if not (c.liveupdate_permissions.allow("edit") or
(c.user_is_loggedin and update.author_id == c.user._id)):
abort(403)
update.stricken = True
LiveUpdateStream.add_update(c.liveupdate_event, update)
_broadcast(type="strike", payload=update._fullname)
开发者ID:hubwub,项目名称:reddit-plugin-liveupdate,代码行数:12,代码来源:controllers.py
示例5: GET_focus
def GET_focus(self, target):
try:
target = uuid.UUID(target)
except (TypeError, ValueError):
self.abort404()
try:
query = LiveUpdateStream.query_focus(c.liveupdate_event, target)
except tdb_cassandra.NotFound:
self.abort404()
builder = LiveUpdateBuilder(
query=query, skip=True, reverse=True, num=1, count=0)
listing = pages.LiveUpdateListing(builder)
wrapped_listing = listing.listing()
c.js_preload.set_wrapped(
"/live/" + c.liveupdate_event._id + ".json",
wrapped_listing,
)
content = pages.LiveUpdateFocusApp(
event=c.liveupdate_event,
listing=wrapped_listing,
)
return pages.LiveUpdateEventFocusPage(
content=content,
page_classes=["liveupdate-focus"],
).render()
开发者ID:Safturento,项目名称:reddit-plugin-liveupdate,代码行数:30,代码来源:controllers.py
示例6: GET_listing
def GET_listing(self, num, after, before, count):
reverse = False
if before:
reverse = True
after = before
query = LiveUpdateStream.query([c.liveupdate_event._id],
count=num, reverse=reverse)
if after:
query.column_start = after
builder = LiveUpdateBuilder(query=query, skip=True,
reverse=reverse, num=num,
count=count)
listing = pages.LiveUpdateListing(builder)
content = pages.LiveUpdateEvent(
event=c.liveupdate_event,
listing=listing.listing(),
)
# don't generate a url unless this is the main page of an event
websocket_url = None
if c.liveupdate_event.state == "live" and not after and not before:
websocket_url = websockets.make_url(
"/live/" + c.liveupdate_event._id, max_age=24 * 60 * 60)
return pages.LiveUpdatePage(
content=content,
websocket_url=websocket_url,
).render()
开发者ID:Web5design,项目名称:reddit-plugin-liveupdate,代码行数:30,代码来源:controllers.py
示例7: close_abandoned_threads
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)
开发者ID:madbook,项目名称:reddit-plugin-liveupdate,代码行数:28,代码来源:housekeeping.py
示例8: GET_listing
def GET_listing(self, num, after, before, count, is_embed):
reverse = False
if before:
reverse = True
after = before
query = LiveUpdateStream.query([c.liveupdate_event._id],
count=num, reverse=reverse)
if after:
query.column_start = after
builder = LiveUpdateBuilder(query=query, skip=True,
reverse=reverse, num=num,
count=count)
listing = pages.LiveUpdateListing(builder)
wrapped_listing = listing.listing()
content = pages.LiveUpdateEventPage(
event=c.liveupdate_event,
listing=wrapped_listing,
show_sidebar=not is_embed,
)
c.js_preload.set_wrapped(
"/live/" + c.liveupdate_event._id + "/about.json",
Wrapped(c.liveupdate_event),
)
c.js_preload.set_wrapped(
"/live/" + c.liveupdate_event._id + ".json",
wrapped_listing,
)
# don't generate a url unless this is the main page of an event
websocket_url = None
if c.liveupdate_event.state == "live" and not after and not before:
websocket_url = websockets.make_url(
"/live/" + c.liveupdate_event._id, max_age=24 * 60 * 60)
if not is_embed:
return pages.LiveUpdatePage(
content=content,
websocket_url=websocket_url,
page_classes=['liveupdate-event'],
).render()
else:
# ensure we're off the cookie domain before allowing embedding
if request.host != g.media_domain:
abort(404)
c.allow_framing = True
return pages.LiveUpdateEmbed(
content=content,
websocket_url=websocket_url,
page_classes=['liveupdate-event'],
).render()
开发者ID:sol2tice,项目名称:peachtree,代码行数:55,代码来源:controllers.py
示例9: POST_strike_update
def POST_strike_update(self, form, jquery, update):
"""Strike (mark incorrect and cross out) the content of an update.
Requires that specified update must have been authored by the user or
that you have the `edit` permission for this thread.
See also: [/api/live/*thread*/update](#POST_api_live_{thread}_update).
"""
if form.has_errors("id", errors.NO_THING_ID):
return
if not (c.liveupdate_permissions.allow("edit") or
(c.user_is_loggedin and update.author_id == c.user._id)):
abort(403)
update.stricken = True
LiveUpdateStream.add_update(c.liveupdate_event, update)
_broadcast(type="strike", payload=update._fullname)
开发者ID:Safturento,项目名称:reddit-plugin-liveupdate,代码行数:20,代码来源:controllers.py
示例10: POST_update
def POST_update(self, form, jquery, text):
if form.has_errors("body", errors.NO_TEXT,
errors.TOO_LONG):
return
# create and store the new update
update = LiveUpdate(data={
"author_id": c.user._id,
"body": text,
})
LiveUpdateStream.add_update(c.liveupdate_event, update)
# tell the world about our new update
builder = LiveUpdateBuilder(None)
wrapped = builder.wrap_items([update])
rendered = [w.render() for w in wrapped]
send_websocket_broadcast(type="update", payload=rendered)
# reset the submission form
t = form.find("textarea")
t.attr('rows', 3).html("").val("")
开发者ID:Web5design,项目名称:reddit-plugin-liveupdate,代码行数:21,代码来源:controllers.py
示例11: POST_delete_update
def POST_delete_update(self, form, jquery, update):
"""Delete an update from the thread.
Requires that specified update must have been authored by the user or
that you have the `edit` permission for this thread.
See also: [/api/live/*thread*/update](#POST_api_live_{thread}_update).
"""
if form.has_errors("id", errors.NO_THING_ID):
return
if not (c.liveupdate_permissions.allow("edit") or
(c.user_is_loggedin and update.author_id == c.user._id)):
abort(403)
update.deleted = True
LiveUpdateStream.add_update(c.liveupdate_event, update)
liveupdate_events.update_event(update, context=c, request=request)
_broadcast(type="delete", payload=update._fullname)
开发者ID:hoihei,项目名称:reddit-plugin-liveupdate,代码行数:21,代码来源:controllers.py
示例12: POST_update
def POST_update(self, form, jquery, text):
"""Post an update to the thread.
Requires the `update` permission for this thread.
See also: [/api/live/*thread*/strike_update]
(#POST_api_live_{thread}_strike_update), and
[/api/live/*thread*/delete_update]
(#POST_api_live_{thread}_delete_update).
"""
if form.has_errors("body", errors.NO_TEXT,
errors.TOO_LONG):
return
# create and store the new update
update = LiveUpdate(data={
"author_id": c.user._id,
"body": text,
"_spam": c.user._spam,
})
hooks.get_hook("liveupdate.update").call(update=update)
LiveUpdateStream.add_update(c.liveupdate_event, update)
# tell the world about our new update
builder = LiveUpdateBuilder(None)
wrapped = builder.wrap_items([update])[0]
rendered = wrapped.render(style="api")
_broadcast(type="update", payload=rendered)
# Queue up parsing any embeds
queue_parse_embeds(c.liveupdate_event, update)
liveupdate_events.update_event(update, context=c, request=request)
# reset the submission form
t = form.find("textarea")
t.attr('rows', 3).html("").val("")
开发者ID:hoihei,项目名称:reddit-plugin-liveupdate,代码行数:40,代码来源:controllers.py
示例13: GET_listing
def GET_listing(self, num, after, before, count, is_embed):
reverse = False
if before:
reverse = True
after = before
query = LiveUpdateStream.query([c.liveupdate_event._id],
count=num, reverse=reverse)
if after:
query.column_start = after
builder = LiveUpdateBuilder(query=query, skip=True,
reverse=reverse, num=num,
count=count)
listing = pages.LiveUpdateListing(builder)
content = pages.LiveUpdateEvent(
event=c.liveupdate_event,
listing=listing.listing(),
show_sidebar=not is_embed,
)
# don't generate a url unless this is the main page of an event
websocket_url = None
if c.liveupdate_event.state == "live" and not after and not before:
websocket_url = websockets.make_url(
"/live/" + c.liveupdate_event._id, max_age=24 * 60 * 60)
if not is_embed:
return pages.LiveUpdatePage(
content=content,
websocket_url=websocket_url,
).render()
else:
# embeds are always logged out and therefore safe for frames.
c.liveupdate_can_manage = False
c.liveupdate_can_edit = False
c.allow_framing = True
return pages.LiveUpdateEmbed(
content=content,
websocket_url=websocket_url,
).render()
开发者ID:andre-d,项目名称:reddit-plugin-liveupdate,代码行数:42,代码来源:controllers.py
示例14: GET_listing
def GET_listing(self, num, after, before, count, is_embed, style_sr):
"""Get a list of updates posted in this thread.
See also: [/api/live/*thread*/update](#POST_api_live_{thread}_update).
"""
reverse = False
if before:
reverse = True
after = before
query = LiveUpdateStream.query([c.liveupdate_event._id],
count=num, reverse=reverse)
if after:
query.column_start = after
builder = LiveUpdateBuilder(query=query, skip=True,
reverse=reverse, num=num,
count=count)
listing = pages.LiveUpdateListing(builder)
wrapped_listing = listing.listing()
if c.user_is_loggedin:
report_type = LiveUpdateReportsByAccount.get_report(
c.user, c.liveupdate_event)
else:
report_type = None
content = pages.LiveUpdateEventApp(
event=c.liveupdate_event,
listing=wrapped_listing,
show_sidebar=not is_embed,
report_type=report_type,
)
c.js_preload.set_wrapped(
"/live/" + c.liveupdate_event._id + "/about.json",
Wrapped(c.liveupdate_event),
)
c.js_preload.set_wrapped(
"/live/" + c.liveupdate_event._id + ".json",
wrapped_listing,
)
if not is_embed:
return pages.LiveUpdateEventAppPage(
content=content,
page_classes=['liveupdate-app'],
).render()
else:
# ensure we're off the cookie domain before allowing embedding
if request.host != g.media_domain:
abort(404)
c.allow_framing = True
# interstitial redirects and nsfw settings are funky on the media
# domain. just disable nsfw embeds.
if c.liveupdate_event.nsfw:
embed_page = pages.LiveUpdateEventEmbed(
content=pages.LiveUpdateNSFWEmbed(),
)
request.environ["usable_error_content"] = embed_page.render()
abort(403)
embed_page = pages.LiveUpdateEventEmbed(
content=content,
page_classes=['liveupdate-app'],
)
if style_sr and getattr(style_sr, "type", "private") != "private":
c.can_apply_styles = True
c.allow_styles = True
embed_page.subreddit_stylesheet_url = \
Reddit.get_subreddit_stylesheet_url(style_sr)
return embed_page.render()
开发者ID:Safturento,项目名称:reddit-plugin-liveupdate,代码行数:76,代码来源:controllers.py
注:本文中的reddit_liveupdate.models.LiveUpdateStream类示例由纯净天空整理自Github/MSDocs等源码及文档管理平台,相关代码片段筛选自各路编程大神贡献的开源项目,源码版权归原作者所有,传播和使用请参考对应项目的License;未经允许,请勿转载。 |
请发表评论