mediagoblin/util.py

   1 # GNU MediaGoblin -- federated, autonomous media hosting
   2 # Copyright (C) 2011 Free Software Foundation, Inc
   3 #
   4 # This program is free software: you can redistribute it and/or modify
   5 # it under the terms of the GNU Affero General Public License as published by
   6 # the Free Software Foundation, either version 3 of the License, or
   7 # (at your option) any later version.
   8 #
   9 # This program is distributed in the hope that it will be useful,
  10 # but WITHOUT ANY WARRANTY; without even the implied warranty of
  11 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  12 # GNU Affero General Public License for more details.
  13 #
  14 # You should have received a copy of the GNU Affero General Public License
  15 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
  16
  17 from email.MIMEText import MIMEText
  18 import gettext
  19 import pkg_resources
  20 import smtplib
  21 import sys
  22 import re
  23 import jinja2
  24 from mediagoblin.db.util import ObjectId
  25 import translitcodec
  26
  27 from mediagoblin import globals as mgoblin_globals
  28
  29 import urllib
  30 from math import ceil
  31 import copy
  32 import decorators
  33 from webob import exc
  34
  35 TESTS_ENABLED = False
  36 def _activate_testing():
  37     """
  38     Call this to activate testing in util.py
  39     """
  40     global TESTS_ENABLED
  41     TESTS_ENABLED = True
  42
  43
  44 def get_jinja_loader(user_template_path=None):
  45     """
  46     Set up the Jinja template loaders, possibly allowing for user
  47     overridden templates.
  48
  49     (In the future we may have another system for providing theming;
  50     for now this is good enough.)
  51     """
  52     if user_template_path:
  53         return jinja2.ChoiceLoader(
  54             [jinja2.FileSystemLoader(user_template_path),
  55              jinja2.PackageLoader('mediagoblin', 'templates')])
  56     else:
  57         return jinja2.PackageLoader('mediagoblin', 'templates')
  58
  59
  60 def get_jinja_env(template_loader, locale):
  61     """
  62     Set up the Jinja environment,
  63
  64     (In the future we may have another system for providing theming;
  65     for now this is good enough.)
  66     """
  67     setup_gettext(locale)
  68
  69     template_env = jinja2.Environment(
  70         loader=template_loader, autoescape=True,
  71         extensions=['jinja2.ext.i18n'])
  72
  73     template_env.install_gettext_callables(
  74         mgoblin_globals.translations.gettext,
  75         mgoblin_globals.translations.ngettext)
  76
  77     return template_env
  78
  79
  80 def setup_user_in_request(request):
  81     """
  82     Examine a request and tack on a request.user parameter if that's
  83     appropriate.
  84     """
  85     if not request.session.has_key('user_id'):
  86         request.user = None
  87         return
  88
  89     user = None
  90     user = request.app.db.User.one(
  91         {'_id': ObjectId(request.session['user_id'])})
  92
  93     if not user:
  94         # Something's wrong... this user doesn't exist?  Invalidate
  95         # this session.
  96         request.session.invalidate()
  97
  98     request.user = user
  99
 100
 101 def import_component(import_string):
 102     """
 103     Import a module component defined by STRING.  Probably a method,
 104     class, or global variable.
 105
 106     Args:
 107      - import_string: a string that defines what to import.  Written
 108        in the format of "module1.module2:component"
 109     """
 110     module_name, func_name = import_string.split(':', 1)
 111     __import__(module_name)
 112     module = sys.modules[module_name]
 113     func = getattr(module, func_name)
 114     return func
 115
 116 _punct_re = re.compile(r'[\t !"#$%&\'()*\-/<=>?@\[\\\]^_`{|},.]+')
 117
 118 def slugify(text, delim=u'-'):
 119     """
 120     Generates an ASCII-only slug. Taken from http://flask.pocoo.org/snippets/5/
 121     """
 122     result = []
 123     for word in _punct_re.split(text.lower()):
 124         word = word.encode('translit/long')
 125         if word:
 126             result.append(word)
 127     return unicode(delim.join(result))
 128
 129 ### ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
 130 ### Special email test stuff begins HERE
 131 ### ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
 132
 133 # We have two "test inboxes" here:
 134 #
 135 # EMAIL_TEST_INBOX:
 136 # ----------------
 137 #   If you're writing test views, you'll probably want to check this.
 138 #   It contains a list of MIMEText messages.
 139 #
 140 # EMAIL_TEST_MBOX_INBOX:
 141 # ----------------------
 142 #   This collects the messages from the FakeMhost inbox.  It's reslly
 143 #   just here for testing the send_email method itself.
 144 #
 145 #   Anyway this contains:
 146 #    - from
 147 #    - to: a list of email recipient addresses
 148 #    - message: not just the body, but the whole message, including
 149 #      headers, etc.
 150 #
 151 # ***IMPORTANT!***
 152 # ----------------
 153 # Before running tests that call functions which send email, you should
 154 # always call _clear_test_inboxes() to "wipe" the inboxes clean.
 155
 156 EMAIL_TEST_INBOX = []
 157 EMAIL_TEST_MBOX_INBOX = []
 158
 159
 160 class FakeMhost(object):
 161     """
 162     Just a fake mail host so we can capture and test messages
 163     from send_email
 164     """
 165     def connect(self):
 166         pass
 167
 168     def sendmail(self, from_addr, to_addrs, message):
 169         EMAIL_TEST_MBOX_INBOX.append(
 170             {'from': from_addr,
 171              'to': to_addrs,
 172              'message': message})
 173
 174 def _clear_test_inboxes():
 175     global EMAIL_TEST_INBOX
 176     global EMAIL_TEST_MBOX_INBOX
 177     EMAIL_TEST_INBOX = []
 178     EMAIL_TEST_MBOX_INBOX = []
 179
 180 ### ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
 181 ### </Special email test stuff>
 182 ### ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
 183
 184 def send_email(from_addr, to_addrs, subject, message_body):
 185     """
 186     Simple email sending wrapper, use this so we can capture messages
 187     for unit testing purposes.
 188
 189     Args:
 190      - from_addr: address you're sending the email from
 191      - to_addrs: list of recipient email addresses
 192      - subject: subject of the email
 193      - message_body: email body text
 194     """
 195     # TODO: make a mock mhost if testing is enabled
 196     if TESTS_ENABLED or mgoblin_globals.email_debug_mode:
 197         mhost = FakeMhost()
 198     elif not mgoblin_globals.email_debug_mode:
 199         mhost = smtplib.SMTP()
 200
 201     mhost.connect()
 202
 203     message = MIMEText(message_body.encode('utf-8'), 'plain', 'utf-8')
 204     message['Subject'] = subject
 205     message['From'] = from_addr
 206     message['To'] = ', '.join(to_addrs)
 207
 208     if TESTS_ENABLED:
 209         EMAIL_TEST_INBOX.append(message)
 210
 211     if getattr(mgoblin_globals, 'email_debug_mode', False):
 212         print u"===== Email ====="
 213         print u"From address: %s" % message['From']
 214         print u"To addresses: %s" % message['To']
 215         print u"Subject: %s" % message['Subject']
 216         print u"-- Body: --"
 217         print message.get_payload(decode=True)
 218
 219     return mhost.sendmail(from_addr, to_addrs, message.as_string())
 220
 221
 222 ###################
 223 # Translation tools
 224 ###################
 225
 226
 227 TRANSLATIONS_PATH = pkg_resources.resource_filename(
 228     'mediagoblin', 'translations')
 229
 230
 231 def locale_to_lower_upper(locale):
 232     """
 233     Take a locale, regardless of style, and format it like "en-us"
 234     """
 235     if '-' in locale:
 236         lang, country = locale.split('-', 1)
 237         return '%s_%s' % (lang.lower(), country.upper())
 238     elif '_' in locale:
 239         lang, country = locale.split('_', 1)
 240         return '%s_%s' % (lang.lower(), country.upper())
 241     else:
 242         return locale.lower()
 243
 244
 245 def locale_to_lower_lower(locale):
 246     """
 247     Take a locale, regardless of style, and format it like "en_US"
 248     """
 249     if '_' in locale:
 250         lang, country = locale.split('_', 1)
 251         return '%s-%s' % (lang.lower(), country.lower())
 252     else:
 253         return locale.lower()
 254
 255
 256 def get_locale_from_request(request):
 257     """
 258     Figure out what target language is most appropriate based on the
 259     request
 260     """
 261     request_form = request.GET or request.POST
 262
 263     if request_form.has_key('lang'):
 264         return locale_to_lower_upper(request_form['lang'])
 265
 266     accept_lang_matches = request.accept_language.best_matches()
 267
 268     # Your routing can explicitly specify a target language
 269     if request.matchdict.has_key('locale'):
 270         target_lang = request.matchdict['locale']
 271     elif request.session.has_key('target_lang'):
 272         target_lang = request.session['target_lang']
 273     # Pull the first acceptable language
 274     elif accept_lang_matches:
 275         target_lang = accept_lang_matches[0]
 276     # Fall back to English
 277     else:
 278         target_lang = 'en'
 279
 280     return locale_to_lower_upper(target_lang)
 281
 282
 283 def setup_gettext(locale):
 284     """
 285     Setup the gettext instance based on this locale
 286     """
 287     # Later on when we have plugins we may want to enable the
 288     # multi-translations system they have so we can handle plugin
 289     # translations too
 290
 291     # TODO: fallback nicely on translations from pt_PT to pt if not
 292     # available, etc.
 293     this_gettext = gettext.translation(
 294         'mediagoblin', TRANSLATIONS_PATH, [locale], fallback=True)
 295
 296     mgoblin_globals.setup_globals(
 297         translations=this_gettext)
 298
 299
 300 class Pagination(object):
 301     """
 302     Pagination class,
 303     initialization through __init__(self, cursor, page=1, per_page=2):
 304     get actual data slice through __call__()
 305     """
 306
 307     def __init__(self, page, cursor, per_page=2):
 308         """
 309         Initializes Pagination
 310
 311         Args:
 312          - page: requested page
 313          - per_page: number of objects per page
 314          - cursor: db cursor
 315         """
 316         self.page = page
 317         self.per_page = per_page
 318         self.cursor = cursor
 319         self.total_count = self.cursor.count()
 320
 321     def __call__(self):
 322         """
 323         Returns slice of objects for the requested page
 324         """
 325         return self.cursor.skip(
 326             (self.page - 1) * self.per_page).limit(self.per_page)
 327
 328     @property
 329     def pages(self):
 330         return int(ceil(self.total_count / float(self.per_page)))
 331
 332     @property
 333     def has_prev(self):
 334         return self.page > 1
 335
 336     @property
 337     def has_next(self):
 338         return self.page < self.pages
 339
 340     def iter_pages(self, left_edge=2, left_current=2,
 341                    right_current=5, right_edge=2):
 342         last = 0
 343         for num in xrange(1, self.pages + 1):
 344             if num <= left_edge or \
 345                (num > self.page - left_current - 1 and \
 346                 num < self.page + right_current) or \
 347                num > self.pages - right_edge:
 348                 if last + 1 != num:
 349                     yield None
 350                 yield num
 351                 last = num
 352
 353     def get_page_url(self, path_info, page_no, get_params=None):
 354         """
 355         Get a new page based of the path_info, the new page number,
 356         and existing get parameters.
 357         """
 358         new_get_params = copy.copy(get_params or {})
 359         new_get_params['page'] = page_no
 360         return "%s?%s" % (
 361             path_info, urllib.urlencode(new_get_params))
 362