gistfile1.txt

# -*- coding: utf-8 -*-

import re, random

from sqlalchemy.exc import IntegrityError

from smapo.common.models.event import EventOperation
from smapo.common.models.base import (
    DBSession, RedirectPage, RedirectUniqueCode, UserActionHistory,
)


CHARS = 'abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789'
assert len(CHARS) == 26 * 2 + 10

def randstr(n):
    """指定された長さのランダムな文字列を生成して返す"""
    return ''.join( random.choice(CHARS) for _ in xrange(n) )


class RedirectOperation(object):

    def redirect_url_and_code(self, page_id, user_agent):
        """リダイレクト用のURLとランダムなコードを生成して返す"""
        page = RedirectPage.filter_by(code=page_id).first()
        if not page:
            redirect_url = code = None
            return redirect_url, code
        redirect_url = self._redirect_url_for(page, user_agent)
        if redirect_url:
            code = self._new_unique_code(page)
            redirect_url = redirect_url.replace('{UNIQUE_ID}', code)
        else:
            code = None
        return redirect_url, code

    def _redirect_url_for(self, page, user_agent):
        """ユーザエージェントを調べて、それに合ったリダイレクト用URLを返す"""
        if re.search('Android', user_agent):
            return page.redirect_to_android
        elif re.search(r'(iOS|iPhone|iPod|iPad)', user_agent):
            return page.redirect_to_ios
        else:
            return page.redirect_to_pc

    def _new_uniqe_code(self, page):
        """一意なランダム文字列を作って返す。またそれをDBに保存する"""
        def _new_code(page):
            try:
                code = randstr(36)
                uniq = RedirectUniqueCode(code=code, page=page)
                uniq.add()        # may raise IntegrityError when code is not unique
                return code
            except IntegrityError:
                return None
        #
        for i in xrange(1,5):
            code = _new_code(page)
            if code:
                return
        raise RuntimeError('Failed to generate unique code')

    def mark_as_redirected(self, code, user, datetime):
        """リダイレクトされたコードを、使用済みにし、イベントを発火する"""
        succeeded = self._consume_unique_code(code, user, datetime)
        if not succeeded:
            return
        uniq = RedirectUniqueCode.filter_by(code=code).first()
        if not uniq:
            raise Exception('Uniq code not found %s' % code)
        self._fire_event(uniq)

    def _consume_unique_code(self, code, user, datetime):
        """コードを使用済みにする。成功すればTrueを返し、失敗するとFalseを返す。"""
        uniq_code = RedirectUniqueCode.__table__
        stmt = (uniq_code.update()
                .where(uniq_code.c.code==code)
                .where(uniq_code.c.consumed_at == None)
                .where(uniq_code.c.user_id == None)
                ).values(user_id=user.id, consumed_at=datetime)
        result = DBSession.execute(stmt)
        succeeded = result.rowcount == 1
        return succeeded

    def _fire_event(self, uniq):
        """イベントを発火する"""
        action_hist = self._new_action_history(uniq)
        EventOperation().execute(action_hist)

    def _new_action_history(self, uniq):
        return UserActionHistory.create(
            UserActionHistory.REDIRECT_CONSUMED,
            latitude   = 0,
            longitude  = 0,
            user       = user,
            created_at = uniq.consumed_at,
            trails     = {"code": uniq.code, "page_id": str(uniq.page.id)},
        )

CHARS = 'abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789'
may can be used as
string.ascii_letters + string.digits