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 os
  22 import sys
  23 import re
  24 import urllib
  25 from math import ceil
  26 import copy
  27
  28 from babel.localedata import exists
  29 import jinja2
  30 import translitcodec
  31 from paste.deploy.loadwsgi import NicerConfigParser
  32
  33 from mediagoblin import globals as mgoblin_globals
  34 from mediagoblin.db.util import ObjectId
  35
  36
  37 TESTS_ENABLED = False
  38 def _activate_testing():
  39     """
  40     Call this to activate testing in util.py
  41     """
  42     global TESTS_ENABLED
  43     TESTS_ENABLED = True
  44
  45
  46 def get_jinja_loader(user_template_path=None):
  47     """
  48     Set up the Jinja template loaders, possibly allowing for user
  49     overridden templates.
  50
  51     (In the future we may have another system for providing theming;
  52     for now this is good enough.)
  53     """
  54     if user_template_path:
  55         return jinja2.ChoiceLoader(
  56             [jinja2.FileSystemLoader(user_template_path),
  57              jinja2.PackageLoader('mediagoblin', 'templates')])
  58     else:
  59         return jinja2.PackageLoader('mediagoblin', 'templates')
  60
  61
  62 SETUP_JINJA_ENVS = {}
  63
  64
  65 def get_jinja_env(template_loader, locale):
  66     """
  67     Set up the Jinja environment,
  68
  69     (In the future we may have another system for providing theming;
  70     for now this is good enough.)
  71     """
  72     setup_gettext(locale)
  73
  74     # If we have a jinja environment set up with this locale, just
  75     # return that one.
  76     if SETUP_JINJA_ENVS.has_key(locale):
  77         return SETUP_JINJA_ENVS[locale]
  78
  79     template_env = jinja2.Environment(
  80         loader=template_loader, autoescape=True,
  81         extensions=['jinja2.ext.i18n'])
  82
  83     template_env.install_gettext_callables(
  84         mgoblin_globals.translations.gettext,
  85         mgoblin_globals.translations.ngettext)
  86
  87     if exists(locale):
  88         SETUP_JINJA_ENVS[locale] = template_env
  89
  90     return template_env
  91
  92
  93 # We'll store context information here when doing unit tests
  94 TEMPLATE_TEST_CONTEXT = {}
  95
  96
  97 def render_template(request, template, context):
  98     """
  99     Render a template with context.
 100
 101     Always inserts the request into the context, so you don't have to.
 102     Also stores the context if we're doing unit tests.  Helpful!
 103     """
 104     template = request.template_env.get_template(
 105         template)
 106     context['request'] = request
 107     rendered = template.render(context)
 108
 109     if TESTS_ENABLED:
 110         TEMPLATE_TEST_CONTEXT[template] = context
 111
 112     return rendered
 113
 114
 115 def clear_test_template_context():
 116     global TEMPLATE_TEST_CONTEXT
 117     TEMPLATE_TEST_CONTEXT = {}
 118
 119
 120 def setup_user_in_request(request):
 121     """
 122     Examine a request and tack on a request.user parameter if that's
 123     appropriate.
 124     """
 125     if not request.session.has_key('user_id'):
 126         request.user = None
 127         return
 128
 129     user = None
 130     user = request.app.db.User.one(
 131         {'_id': ObjectId(request.session['user_id'])})
 132
 133     if not user:
 134         # Something's wrong... this user doesn't exist?  Invalidate
 135         # this session.
 136         request.session.invalidate()
 137
 138     request.user = user
 139
 140
 141 def import_component(import_string):
 142     """
 143     Import a module component defined by STRING.  Probably a method,
 144     class, or global variable.
 145
 146     Args:
 147      - import_string: a string that defines what to import.  Written
 148        in the format of "module1.module2:component"
 149     """
 150     module_name, func_name = import_string.split(':', 1)
 151     __import__(module_name)
 152     module = sys.modules[module_name]
 153     func = getattr(module, func_name)
 154     return func
 155
 156 _punct_re = re.compile(r'[\t !"#$%&\'()*\-/<=>?@\[\\\]^_`{|},.]+')
 157
 158 def slugify(text, delim=u'-'):
 159     """
 160     Generates an ASCII-only slug. Taken from http://flask.pocoo.org/snippets/5/
 161     """
 162     result = []
 163     for word in _punct_re.split(text.lower()):
 164         word = word.encode('translit/long')
 165         if word:
 166             result.append(word)
 167     return unicode(delim.join(result))
 168
 169 ### ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
 170 ### Special email test stuff begins HERE
 171 ### ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
 172
 173 # We have two "test inboxes" here:
 174 #
 175 # EMAIL_TEST_INBOX:
 176 # ----------------
 177 #   If you're writing test views, you'll probably want to check this.
 178 #   It contains a list of MIMEText messages.
 179 #
 180 # EMAIL_TEST_MBOX_INBOX:
 181 # ----------------------
 182 #   This collects the messages from the FakeMhost inbox.  It's reslly
 183 #   just here for testing the send_email method itself.
 184 #
 185 #   Anyway this contains:
 186 #    - from
 187 #    - to: a list of email recipient addresses
 188 #    - message: not just the body, but the whole message, including
 189 #      headers, etc.
 190 #
 191 # ***IMPORTANT!***
 192 # ----------------
 193 # Before running tests that call functions which send email, you should
 194 # always call _clear_test_inboxes() to "wipe" the inboxes clean.
 195
 196 EMAIL_TEST_INBOX = []
 197 EMAIL_TEST_MBOX_INBOX = []
 198
 199
 200 class FakeMhost(object):
 201     """
 202     Just a fake mail host so we can capture and test messages
 203     from send_email
 204     """
 205     def connect(self):
 206         pass
 207
 208     def sendmail(self, from_addr, to_addrs, message):
 209         EMAIL_TEST_MBOX_INBOX.append(
 210             {'from': from_addr,
 211              'to': to_addrs,
 212              'message': message})
 213
 214 def _clear_test_inboxes():
 215     global EMAIL_TEST_INBOX
 216     global EMAIL_TEST_MBOX_INBOX
 217     EMAIL_TEST_INBOX = []
 218     EMAIL_TEST_MBOX_INBOX = []
 219
 220 ### ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
 221 ### </Special email test stuff>
 222 ### ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
 223
 224 def send_email(from_addr, to_addrs, subject, message_body):
 225     """
 226     Simple email sending wrapper, use this so we can capture messages
 227     for unit testing purposes.
 228
 229     Args:
 230      - from_addr: address you're sending the email from
 231      - to_addrs: list of recipient email addresses
 232      - subject: subject of the email
 233      - message_body: email body text
 234     """
 235     # TODO: make a mock mhost if testing is enabled
 236     if TESTS_ENABLED or mgoblin_globals.email_debug_mode:
 237         mhost = FakeMhost()
 238     elif not mgoblin_globals.email_debug_mode:
 239         mhost = smtplib.SMTP()
 240
 241     mhost.connect()
 242
 243     message = MIMEText(message_body.encode('utf-8'), 'plain', 'utf-8')
 244     message['Subject'] = subject
 245     message['From'] = from_addr
 246     message['To'] = ', '.join(to_addrs)
 247
 248     if TESTS_ENABLED:
 249         EMAIL_TEST_INBOX.append(message)
 250
 251     if getattr(mgoblin_globals, 'email_debug_mode', False):
 252         print u"===== Email ====="
 253         print u"From address: %s" % message['From']
 254         print u"To addresses: %s" % message['To']
 255         print u"Subject: %s" % message['Subject']
 256         print u"-- Body: --"
 257         print message.get_payload(decode=True)
 258
 259     return mhost.sendmail(from_addr, to_addrs, message.as_string())
 260
 261
 262 ###################
 263 # Translation tools
 264 ###################
 265
 266
 267 TRANSLATIONS_PATH = pkg_resources.resource_filename(
 268     'mediagoblin', 'translations')
 269
 270
 271 def locale_to_lower_upper(locale):
 272     """
 273     Take a locale, regardless of style, and format it like "en-us"
 274     """
 275     if '-' in locale:
 276         lang, country = locale.split('-', 1)
 277         return '%s_%s' % (lang.lower(), country.upper())
 278     elif '_' in locale:
 279         lang, country = locale.split('_', 1)
 280         return '%s_%s' % (lang.lower(), country.upper())
 281     else:
 282         return locale.lower()
 283
 284
 285 def locale_to_lower_lower(locale):
 286     """
 287     Take a locale, regardless of style, and format it like "en_US"
 288     """
 289     if '_' in locale:
 290         lang, country = locale.split('_', 1)
 291         return '%s-%s' % (lang.lower(), country.lower())
 292     else:
 293         return locale.lower()
 294
 295
 296 def get_locale_from_request(request):
 297     """
 298     Figure out what target language is most appropriate based on the
 299     request
 300     """
 301     request_form = request.GET or request.POST
 302
 303     if request_form.has_key('lang'):
 304         return locale_to_lower_upper(request_form['lang'])
 305
 306     accept_lang_matches = request.accept_language.best_matches()
 307
 308     # Your routing can explicitly specify a target language
 309     if request.matchdict.has_key('locale'):
 310         target_lang = request.matchdict['locale']
 311     elif request.session.has_key('target_lang'):
 312         target_lang = request.session['target_lang']
 313     # Pull the first acceptable language
 314     elif accept_lang_matches:
 315         target_lang = accept_lang_matches[0]
 316     # Fall back to English
 317     else:
 318         target_lang = 'en'
 319
 320     return locale_to_lower_upper(target_lang)
 321
 322
 323 def read_config_file(conf_file):
 324     """
 325     Read a paste deploy style config file and process it.
 326     """
 327     if not os.path.exists(conf_file):
 328         raise IOError(
 329             "MEDIAGOBLIN_CONFIG not set or file does not exist")
 330
 331     parser = NicerConfigParser(conf_file)
 332     parser.read(conf_file)
 333     parser._defaults.setdefault(
 334         'here', os.path.dirname(os.path.abspath(conf_file)))
 335     parser._defaults.setdefault(
 336         '__file__', os.path.abspath(conf_file))
 337
 338     mgoblin_conf = dict(
 339         [(section_name, dict(parser.items(section_name)))
 340          for section_name in parser.sections()])
 341
 342     return mgoblin_conf
 343
 344
 345 SETUP_GETTEXTS = {}
 346
 347 def setup_gettext(locale):
 348     """
 349     Setup the gettext instance based on this locale
 350     """
 351     # Later on when we have plugins we may want to enable the
 352     # multi-translations system they have so we can handle plugin
 353     # translations too
 354
 355     # TODO: fallback nicely on translations from pt_PT to pt if not
 356     # available, etc.
 357     if SETUP_GETTEXTS.has_key(locale):
 358         this_gettext = SETUP_GETTEXTS[locale]
 359     else:
 360         this_gettext = gettext.translation(
 361             'mediagoblin', TRANSLATIONS_PATH, [locale], fallback=True)
 362         if exists(locale):
 363             SETUP_GETTEXTS[locale] = this_gettext
 364
 365     mgoblin_globals.setup_globals(
 366         translations=this_gettext)
 367
 368
 369 PAGINATION_DEFAULT_PER_PAGE = 30
 370
 371 class Pagination(object):
 372     """
 373     Pagination class for mongodb queries.
 374
 375     Initialization through __init__(self, cursor, page=1, per_page=2),
 376     get actual data slice through __call__().
 377     """
 378
 379     def __init__(self, page, cursor, per_page=PAGINATION_DEFAULT_PER_PAGE):
 380         """
 381         Initializes Pagination
 382
 383         Args:
 384          - page: requested page
 385          - per_page: number of objects per page
 386          - cursor: db cursor
 387         """
 388         self.page = page
 389         self.per_page = per_page
 390         self.cursor = cursor
 391         self.total_count = self.cursor.count()
 392
 393     def __call__(self):
 394         """
 395         Returns slice of objects for the requested page
 396         """
 397         return self.cursor.skip(
 398             (self.page - 1) * self.per_page).limit(self.per_page)
 399
 400     @property
 401     def pages(self):
 402         return int(ceil(self.total_count / float(self.per_page)))
 403
 404     @property
 405     def has_prev(self):
 406         return self.page > 1
 407
 408     @property
 409     def has_next(self):
 410         return self.page < self.pages
 411
 412     def iter_pages(self, left_edge=2, left_current=2,
 413                    right_current=5, right_edge=2):
 414         last = 0
 415         for num in xrange(1, self.pages + 1):
 416             if num <= left_edge or \
 417                (num > self.page - left_current - 1 and \
 418                 num < self.page + right_current) or \
 419                num > self.pages - right_edge:
 420                 if last + 1 != num:
 421                     yield None
 422                 yield num
 423                 last = num
 424
 425     def get_page_url_explicit(self, base_url, get_params, page_no):
 426         """
 427         Get a page url by adding a page= parameter to the base url
 428         """
 429         new_get_params = copy.copy(get_params or {})
 430         new_get_params['page'] = page_no
 431         return "%s?%s" % (
 432             base_url, urllib.urlencode(new_get_params))
 433
 434     def get_page_url(self, request, page_no):
 435         """
 436         Get a new page url based of the request, and the new page number.
 437
 438         This is a nice wrapper around get_page_url_explicit()
 439         """
 440         return self.get_page_url_explicit(
 441             request.path_info, request.GET, page_no)