| # Licensed to the Apache Software Foundation (ASF) under one |
| # or more contributor license agreements. See the NOTICE file |
| # distributed with this work for additional information |
| # regarding copyright ownership. The ASF licenses this file |
| # to you under the Apache License, Version 2.0 (the |
| # "License"); you may not use this file except in compliance |
| # with the License. You may obtain a copy of the License at |
| # |
| # http://www.apache.org/licenses/LICENSE-2.0 |
| # |
| # Unless required by applicable law or agreed to in writing, |
| # software distributed under the License is distributed on an |
| # "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY |
| # KIND, either express or implied. See the License for the |
| # specific language governing permissions and limitations |
| # under the License. |
| |
| #-*- python -*- |
| import logging |
| from datetime import datetime |
| import urllib2 |
| import json |
| |
| # Non-stdlib imports |
| import pymongo |
| from tg import config, expose, validate, redirect, flash, jsonify |
| from tg.decorators import with_trailing_slash, without_trailing_slash |
| from pylons import tmpl_context as c, app_globals as g |
| from pylons import request, response |
| from paste.deploy.converters import asbool |
| import formencode |
| from formencode import validators |
| from webob import exc |
| from urllib import unquote |
| |
| from ming.orm import session |
| |
| # Pyforge-specific imports |
| from allura.app import Application, SitemapEntry |
| from allura.app import DefaultAdminController |
| from allura.lib import helpers as h |
| from allura.lib.search import search_app |
| from allura.lib.decorators import require_post, Property |
| from allura.lib.security import has_access, require_access |
| from allura.lib import widgets as w |
| from allura.lib.widgets.subscriptions import SubscribeForm |
| from allura.lib.widgets import form_fields as ffw |
| from allura.lib.widgets.search import SearchResults, SearchHelp |
| from allura import model as M |
| from allura.controllers import BaseController, AppDiscussionController, AppDiscussionRestController |
| from allura.controllers.feed import FeedArgs, FeedController |
| |
| # Local imports |
| from forgeblog import model as BM |
| from forgeblog import version |
| from forgeblog import widgets |
| |
| log = logging.getLogger(__name__) |
| |
| class W: |
| thread=w.Thread( |
| page=None, limit=None, page_size=None, count=None, |
| style='linear') |
| pager = widgets.BlogPager() |
| new_post_form = widgets.NewPostForm() |
| edit_post_form = widgets.EditPostForm() |
| view_post_form = widgets.ViewPostForm() |
| label_edit = ffw.LabelEdit() |
| attachment_add = ffw.AttachmentAdd() |
| attachment_list = ffw.AttachmentList() |
| preview_post_form = widgets.PreviewPostForm() |
| subscribe_form = SubscribeForm() |
| search_results = SearchResults() |
| help_modal = SearchHelp() |
| |
| class ForgeBlogApp(Application): |
| __version__ = version.__version__ |
| tool_label='Blog' |
| tool_description=""" |
| Share exciting news and progress updates with your |
| community. |
| """ |
| default_mount_label='Blog' |
| default_mount_point='blog' |
| permissions = ['configure', 'read', 'write', |
| 'unmoderated_post', 'post', 'moderate', 'admin'] |
| permissions_desc = { |
| 'read': 'View blog entries.', |
| 'write': 'Create new blog entry.', |
| 'admin': 'Set permissions. Enable/disable commenting.', |
| } |
| ordinal=14 |
| installable=True |
| exportable = True |
| config_options = Application.config_options |
| default_external_feeds = [] |
| icons={ |
| 24:'images/blog_24.png', |
| 32:'images/blog_32.png', |
| 48:'images/blog_48.png' |
| } |
| |
| def __init__(self, project, config): |
| Application.__init__(self, project, config) |
| self.root = RootController() |
| self.admin = BlogAdminController(self) |
| self.api_root = RootRestController() |
| |
| @Property |
| def external_feeds_list(): |
| def fget(self): |
| globals = BM.Globals.query.get(app_config_id=self.config._id) |
| if globals is not None: |
| external_feeds = globals.external_feeds |
| else: |
| external_feeds = self.default_external_feeds |
| return external_feeds |
| def fset(self, new_external_feeds): |
| globals = BM.Globals.query.get(app_config_id=self.config._id) |
| if globals is not None: |
| globals.external_feeds = new_external_feeds |
| elif len(new_external_feeds) > 0: |
| globals = BM.Globals(app_config_id=self.config._id, external_feeds=new_external_feeds) |
| if globals is not None: |
| session(globals).flush() |
| |
| def main_menu(self): |
| return [SitemapEntry(self.config.options.mount_label, '.')] |
| |
| @property |
| @h.exceptionless([], log) |
| def sitemap(self): |
| menu_id = self.config.options.mount_label |
| with h.push_config(c, app=self): |
| return [ |
| SitemapEntry(menu_id, '.')[self.sidebar_menu()] ] |
| |
| @property |
| def show_discussion(self): |
| if 'show_discussion' in self.config.options: |
| return self.config.options['show_discussion'] |
| else: |
| return True |
| |
| @h.exceptionless([], log) |
| def sidebar_menu(self): |
| base = c.app.url |
| links = [ |
| SitemapEntry('Home', base), |
| SitemapEntry('Search', base + 'search'), |
| ] |
| if has_access(self, 'write')(): |
| links += [ SitemapEntry('New Post', base + 'new') ] |
| return links |
| |
| def admin_menu(self): |
| import sys |
| admin_url = c.project.url() + 'admin/' + self.config.options.mount_point + '/' |
| # temporarily disabled until some bugs are fixed |
| links = super(ForgeBlogApp, self).admin_menu(force_options=True) |
| # We don't want external feeds in menu unless they're enabled |
| if asbool(config.get('forgeblog.exfeed', 'false')): |
| links.insert(0, SitemapEntry('External feeds', admin_url + 'exfeed', className='admin_modal')) |
| return links |
| #return super(ForgeBlogApp, self).admin_menu(force_options=True) |
| |
| def install(self, project): |
| 'Set up any default permissions and roles here' |
| super(ForgeBlogApp, self).install(project) |
| |
| # Setup permissions |
| role_admin = M.ProjectRole.by_name('Admin')._id |
| role_developer = M.ProjectRole.by_name('Developer')._id |
| role_auth = M.ProjectRole.by_name('*authenticated')._id |
| role_anon = M.ProjectRole.by_name('*anonymous')._id |
| self.config.acl = [ |
| M.ACE.allow(role_anon, 'read'), |
| M.ACE.allow(role_auth, 'post'), |
| M.ACE.allow(role_auth, 'unmoderated_post'), |
| M.ACE.allow(role_developer, 'write'), |
| M.ACE.allow(role_developer, 'moderate'), |
| M.ACE.allow(role_admin, 'configure'), |
| M.ACE.allow(role_admin, 'admin'), |
| ] |
| |
| def uninstall(self, project): |
| "Remove all the tool's artifacts from the database" |
| BM.Attachment.query.remove(dict(app_config_id=c.app.config._id)) |
| BM.BlogPost.query.remove(dict(app_config_id=c.app.config._id)) |
| BM.BlogPostSnapshot.query.remove(dict(app_config_id=c.app.config._id)) |
| super(ForgeBlogApp, self).uninstall(project) |
| |
| def bulk_export(self, f): |
| f.write('{"posts": [') |
| posts = BM.BlogPost.query.find(dict(app_config_id=self.config._id)) |
| for i, post in enumerate(posts): |
| if i > 0: |
| f.write(',') |
| json.dump(post, f, cls=jsonify.GenericJSON, indent=2) |
| f.write(']}') |
| |
| class RootController(BaseController, FeedController): |
| |
| def __init__(self): |
| self._discuss = AppDiscussionController() |
| |
| def _check_security(self): |
| require_access(c.app, 'read') |
| |
| @expose('jinja:forgeblog:templates/blog/index.html') |
| @with_trailing_slash |
| def index(self, page=0, limit=10, **kw): |
| query_filter = dict(app_config_id=c.app.config._id) |
| if not has_access(c.app, 'write')(): |
| query_filter['state'] = 'published' |
| q = BM.BlogPost.query.find(query_filter) |
| post_count = q.count() |
| limit, page = h.paging_sanitizer(limit, page, post_count) |
| posts = q.sort('timestamp', pymongo.DESCENDING) \ |
| .skip(page * limit).limit(limit) |
| c.form = W.preview_post_form |
| c.pager = W.pager |
| return dict(posts=posts, page=page, limit=limit, count=post_count) |
| |
| @with_trailing_slash |
| @expose('jinja:forgeblog:templates/blog/search.html') |
| @validate(dict(q=validators.UnicodeString(if_empty=None), |
| history=validators.StringBool(if_empty=False), |
| search_comments=validators.StringBool(if_empty=False), |
| project=validators.StringBool(if_empty=False))) |
| def search(self, q=None, history=None, search_comments=None, project=None, limit=None, page=0, **kw): |
| c.search_results = W.search_results |
| c.help_modal = W.help_modal |
| search_params = kw |
| search_params.update({ |
| 'q': q or '', |
| 'history': history, |
| 'search_comments': search_comments, |
| 'project': project, |
| 'limit': limit, |
| 'page': page, |
| 'allowed_types': ['Blog Post', 'Blog Post Snapshot'], |
| 'fq': ['state_s:published'] |
| }) |
| return search_app(**search_params) |
| |
| @expose('jinja:forgeblog:templates/blog/edit_post.html') |
| @without_trailing_slash |
| def new(self, **kw): |
| require_access(c.app, 'write') |
| now = datetime.utcnow() |
| post = dict( |
| state='published') |
| c.form = W.new_post_form |
| return dict(post=post) |
| |
| @expose() |
| @require_post() |
| @validate(form=W.edit_post_form, error_handler=new) |
| @without_trailing_slash |
| def save(self, **kw): |
| require_access(c.app, 'write') |
| post = BM.BlogPost.new(**kw) |
| redirect(h.really_unicode(post.url()).encode('utf-8')) |
| |
| |
| @with_trailing_slash |
| @expose('jinja:allura:templates/markdown_syntax_dialog.html') |
| def markdown_syntax_dialog(self, **kw): |
| 'Static dialog page about how to use markdown.' |
| return dict() |
| |
| @expose() |
| def _lookup(self, year=None, month=None, name=None, *rest): |
| if year is None or month is None or name is None: |
| raise exc.HTTPNotFound() |
| slug = '/'.join((year, month, urllib2.unquote(name).decode('utf-8'))) |
| post = BM.BlogPost.query.get(slug=slug, app_config_id=c.app.config._id) |
| if post is None: |
| raise exc.HTTPNotFound() |
| return PostController(post), rest |
| |
| class PostController(BaseController, FeedController): |
| |
| def __init__(self, post): |
| self.post = post |
| |
| def _check_security(self): |
| require_access(self.post, 'read') |
| |
| @expose('jinja:forgeblog:templates/blog/post.html') |
| @with_trailing_slash |
| @validate(dict(page=validators.Int(if_empty=0), |
| limit=validators.Int(if_empty=25))) |
| def index(self, page=0, limit=25, **kw): |
| if self.post.state == 'draft': |
| require_access(self.post, 'write') |
| c.form = W.view_post_form |
| c.subscribe_form = W.subscribe_form |
| c.thread = W.thread |
| post_count = self.post.discussion_thread.post_count |
| limit, page = h.paging_sanitizer(limit, page, post_count) |
| version = kw.pop('version', None) |
| post = self._get_version(version) |
| base_post = self.post |
| return dict(post=post, base_post=base_post, |
| page=page, limit=limit, count=post_count) |
| |
| @expose('jinja:forgeblog:templates/blog/edit_post.html') |
| @without_trailing_slash |
| def edit(self, **kw): |
| require_access(self.post, 'write') |
| c.form = W.edit_post_form |
| c.attachment_add = W.attachment_add |
| c.attachment_list = W.attachment_list |
| c.label_edit = W.label_edit |
| return dict(post=self.post) |
| |
| @without_trailing_slash |
| @expose('jinja:forgeblog:templates/blog/post_history.html') |
| def history(self, **kw): |
| posts = self.post.history() |
| return dict(title=self.post.title, posts=posts) |
| |
| @without_trailing_slash |
| @expose('jinja:forgeblog:templates/blog/post_diff.html') |
| def diff(self, v1, v2, **kw): |
| p1 = self._get_version(int(v1)) |
| p2 = self._get_version(int(v2)) |
| result = h.diff_text(p1.text, p2.text) |
| return dict(p1=p1, p2=p2, edits=result) |
| |
| @expose() |
| @require_post() |
| @validate(form=W.edit_post_form, error_handler=edit) |
| @without_trailing_slash |
| def save(self, delete=None, **kw): |
| require_access(self.post, 'write') |
| if delete: |
| self.post.delete() |
| flash('Post deleted', 'info') |
| redirect(h.really_unicode(c.app.url).encode('utf-8')) |
| for k,v in kw.iteritems(): |
| setattr(self.post, k, v) |
| self.post.commit() |
| redirect('.') |
| |
| @without_trailing_slash |
| @require_post() |
| @expose() |
| def revert(self, version, **kw): |
| require_access(self.post, 'write') |
| orig = self._get_version(version) |
| if orig: |
| self.post.text = orig.text |
| self.post.commit() |
| redirect('.') |
| |
| @expose() |
| @validate(W.subscribe_form) |
| def subscribe(self, subscribe=None, unsubscribe=None, **kw): |
| if subscribe: |
| self.post.subscribe(type='direct') |
| elif unsubscribe: |
| self.post.unsubscribe() |
| redirect(h.really_unicode(request.referer).encode('utf-8')) |
| |
| def get_feed(self, project, app, user): |
| """Return a :class:`allura.controllers.feed.FeedArgs` object describing |
| the xml feed for this controller. |
| |
| Overrides :meth:`allura.controllers.feed.FeedController.get_feed`. |
| |
| """ |
| return FeedArgs( |
| dict(ref_id=self.post.index_id()), |
| 'Recent changes to %s' % self.post.title, |
| self.post.url()) |
| |
| def _get_version(self, version): |
| if not version: return self.post |
| try: |
| return self.post.get_version(version) |
| except ValueError: |
| raise exc.HTTPNotFound() |
| |
| class BlogAdminController(DefaultAdminController): |
| def __init__(self, app): |
| self.app = app |
| |
| @without_trailing_slash |
| @expose('jinja:forgeblog:templates/blog/admin_options.html') |
| def options(self): |
| return dict(app=self.app, |
| allow_config=has_access(self.app, 'configure')()) |
| |
| @without_trailing_slash |
| @expose() |
| @require_post() |
| def set_options(self, show_discussion=False): |
| self.app.config.options['show_discussion'] = show_discussion and True or False |
| flash('Blog options updated') |
| redirect(h.really_unicode(c.project.url()+'admin/tools').encode('utf-8')) |
| |
| @without_trailing_slash |
| @expose('jinja:forgeblog:templates/blog/admin_exfeed.html') |
| def exfeed(self): |
| #self.app.external_feeds_list = ['feed1', 'feed2'] |
| #log.info("EXFEED: %s" % self.app.external_feeds_list) |
| feeds_list = [] |
| for feed in self.app.external_feeds_list: |
| feeds_list.append(feed) |
| return dict(app=self.app, |
| feeds_list=feeds_list, |
| allow_config=has_access(self.app, 'configure')()) |
| |
| @without_trailing_slash |
| @expose() |
| @require_post() |
| def set_exfeed(self, new_exfeed=None, **kw): |
| exfeed_val = kw.get('exfeed', []) |
| if type(exfeed_val) == unicode: |
| tmp_exfeed_list = [] |
| tmp_exfeed_list.append(exfeed_val) |
| else: |
| tmp_exfeed_list = exfeed_val |
| |
| if new_exfeed is not None and new_exfeed != '': |
| tmp_exfeed_list.append(new_exfeed) |
| |
| exfeed_list = [] |
| invalid_list = [] |
| v = validators.URL() |
| for link in tmp_exfeed_list: |
| try: |
| v.to_python(link) |
| exfeed_list.append(link) |
| except formencode.api.Invalid: |
| invalid_list.append(link) |
| |
| self.app.external_feeds_list = exfeed_list |
| flash('External feeds updated') |
| if len(invalid_list) > 0: |
| flash('Invalid link(s): %s' % ','.join(link for link in invalid_list), 'error') |
| |
| redirect(c.project.url()+'admin/tools') |
| |
| |
| class RootRestController(BaseController): |
| def __init__(self): |
| self._discuss = AppDiscussionRestController() |
| |
| def _check_security(self): |
| require_access(c.app, 'read') |
| |
| @expose('json:') |
| def index(self, title='', text='', state='draft', labels='', limit=10, page=0, **kw): |
| if request.method == 'POST': |
| require_access(c.app, 'write') |
| post = BM.BlogPost.new( |
| title=title, |
| state=state, |
| text=text, |
| labels=labels.split(','), |
| **kw) |
| return exc.HTTPCreated(headers=dict(Location=h.absurl('/rest' + post.url()))) |
| |
| else: |
| result = RootController().index(limit=limit, page=page) |
| posts = result['posts'] |
| post_titles = [] |
| for post in posts: |
| if has_access(post, 'read')(): |
| post_titles.append({'title': post.title, 'url': h.absurl('/rest' + post.url())}) |
| return dict(posts=post_titles, count=result['count'], limit=result['limit'], page=result['page']) |
| |
| @expose() |
| def _lookup(self, year=None, month=None, title=None, *rest): |
| if not (year and month and title): |
| raise exc.HTTPNotFound() |
| slug = '/'.join((year, month, urllib2.unquote(title).decode('utf-8'))) |
| post = BM.BlogPost.query.get(slug=slug, app_config_id=c.app.config._id) |
| if not post: |
| raise exc.HTTPNotFound() |
| return PostRestController(post), rest |
| |
| |
| class PostRestController(BaseController): |
| |
| def __init__(self, post): |
| self.post = post |
| |
| def _check_security(self): |
| if self.post: |
| require_access(self.post, 'read') |
| |
| @h.vardec |
| @expose('json:') |
| def index(self, **kw): |
| if request.method == 'POST': |
| return self._update_post(**kw) |
| else: |
| if self.post.state == 'draft': |
| require_access(self.post, 'write') |
| return self.post.__json__() |
| |
| def _update_post(self, **post_data): |
| require_access(self.post, 'write') |
| if 'delete' in post_data: |
| self.post.delete() |
| return {} |
| if 'title' in post_data: |
| self.post.title = post_data['title'] |
| if 'text' in post_data: |
| self.post.text = post_data['text'] |
| if 'state' in post_data: |
| self.post.state = post_data['state'] |
| if 'labels' in post_data: |
| self.post.labels = post_data['labels'].split(',') |
| self.post.commit() |
| return self.post.__json__() |
