sourcing/sourcing/model.py

from flask import url_for
from .database import session
from .parse import parse_link, parse_sourcedoc_facet, parse_span
from .text import first_non_empty_line
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column, ForeignKey
from sqlalchemy.types import String, Unicode, Integer, DateTime, Boolean, UnicodeText, Enum
from sqlalchemy import func
from sqlalchemy.orm import relationship, validates, synonym, configure_mappers
from sqlalchemy.sql import exists
from flask_login import UserMixin
from werkzeug.security import generate_password_hash, check_password_hash
from sqlalchemy.ext.hybrid import hybrid_property
from sqlalchemy_continuum import make_versioned
from sqlalchemy_continuum.plugins import FlaskPlugin, ActivityPlugin
import re
from hashids import Hashids

activity_plugin = ActivityPlugin()
make_versioned(plugins=[FlaskPlugin(), activity_plugin])

doc_hashids = Hashids(min_length=8)

Base = declarative_base()
Base.query = session.query_property()

re_server_url = re.compile('^http://perma.pub/\d+/([^/]+)/([^/]+)$')

# list of disallowed usernames - maybe this should be in the database
reserved_name = ['root', 'admin', 'administrator', 'support', 'info',
    'test', 'tech', 'online', 'old', 'new', 'jobs', 'login', 'job', 'ipad'
    'iphone', 'javascript', 'script', 'host', 'mail', 'image', 'faq',
    'file', 'ftp', 'error', 'warning', 'the', 'assistance', 'maintenance',
    'controller', 'head', 'chief', 'anon']

re_username = re.compile('^\w+$', re.U)
re_full_name = re.compile('^([-.\'" ]|[^\W\d_])+$', re.U)

def item_url():
    return url_for('view.view_item',
                   username=self.user.username,
                   hashid=self.hashid)


def user_exists(field, value):
    return session.query(exists().where(field == value)).scalar()

class TimeStampedModel(Base):
    __abstract__ = True
    created = Column(DateTime, default=func.now())
    modified = Column(DateTime, default=func.now(), onupdate=func.now())

class LoginError(Exception):
    def __init__(self, msg):
        self.msg = msg

class User(TimeStampedModel, UserMixin):
    __tablename__ = 'user'
    id = Column(Integer, primary_key=True)
    username = Column(Unicode(32), unique=True, nullable=False)
    pw_hash = Column(String(160), nullable=False)
    email = Column(Unicode(64), unique=True, nullable=False)
    email_verified = Column(Boolean(), nullable=False, default=False)
    disabled = Column(Boolean(), nullable=False, default=False)
    deleted = Column(Boolean(), nullable=False, default=False)
    is_super = Column(Boolean, nullable=False, default=False)
    last_login = Column(DateTime)
    full_name = Column(Unicode(64))
    balance = Column(Integer, nullable=False, default=0)

    user_id = synonym('id')
    name = synonym('full_name')
    user_name = synonym('username')

    def __init__(self, **kwargs):
        pw_hash = generate_password_hash(kwargs.pop('password'))
        return super(User, self).__init__(pw_hash=pw_hash, **kwargs)

    def __repr__(self):
        return '<User: {!r}>'.format(self.username)

    def set_password(self, password):
        self.pw_hash = generate_password_hash(password)

    def check_password(self, password):
        return check_password_hash(self.pw_hash, password)

    def get_id(self):
        return self.id

    @validates('email')
    def validate_email(self, key, value):
        assert '@' in value
        return value

    @validates('username')
    def validate_usernane(self, key, value):
        assert re_username.match(value)
        return value

    @validates('full_name')
    def validate_full_name(self, key, value):
        if value:
            assert re_full_name.match(value)
        return value

    @hybrid_property
    def is_live(self):
        return self.email_verified & ~self.disabled & ~self.deleted

    @classmethod
    def lookup_user_or_email(cls, user_or_email):
        field = cls.email if '@' in user_or_email else cls.username
        return cls.query.filter(field == user_or_email).one_or_none()

    @property
    def mail_to_name(self):
        '''Name to use on e-mails sent to the user.'''
        return self.full_name or self.username

    @classmethod
    def attempt_login(cls, user_or_email, password):
        user = cls.lookup_user_or_email(user_or_email)
        if not user:
            raise LoginError('user not found')
        if user.disabled:
            raise LoginError('user account disabled')
        if not user.check_password(password):
            raise LoginError('incorrect password')
        return user

class Reference(Base):
    __tablename__ = 'reference'
    subject_id = Column(Integer, ForeignKey('item.id'), primary_key=True)
    object_id = Column(Integer, ForeignKey('item.id'), primary_key=True)

class Item(TimeStampedModel):
    __tablename__ = 'item'
    __versioned__ = {'base_classes': (TimeStampedModel,)}

    id = Column(Integer, primary_key=True)
    user_id = Column(Integer, ForeignKey('user.id'))
    published = Column(DateTime)
    type = Column(Enum('sourcedoc', 'xanadoc', 'xanalink', name='item_type'),
                  nullable=False)
    filename = Column(Unicode)
    text = Column(UnicodeText)

    subjects = relationship('Item',
                            lazy='dynamic',
                            secondary='reference',
                            primaryjoin=id == Reference.object_id,
                            secondaryjoin=id == Reference.subject_id)
    objects = relationship('Item',
                           lazy='dynamic',
                           secondary='reference',
                           primaryjoin=id == Reference.subject_id,
                           secondaryjoin=id == Reference.object_id)
    user = relationship('User', backref='items')

    __mapper_args__ = {
        'polymorphic_on': type,
        'with_polymorphic': '*',
    }

    @property
    def hashid(self):
        return doc_hashids.encode(self.id)

    @classmethod
    def get_by_hashid(cls, hashid):
        try:
            item_id = doc_hashids.decode(hashid)[0]
        except IndexError:
            return
        return cls.query.get(item_id)

    def view_url(self, endpoint, **kwargs):
        return url_for('view.' + endpoint,
                       username=self.user.username,
                       hashid=self.hashid,
                       **kwargs)

    @property
    def url(self):
        return self.view_url('view_item')

    def version_url(self, version):
        return self.view_url('view_item', v=version)

    @property
    def history_url(self):
        return self.view_url('history')

    @property
    def external_url(self):
        return self.view_url('view_item', _external=True)

    @property
    def edit_url(self):
        return self.view_url('edit_item')

    @property
    def set_title_url(self):
        return self.view_url('set_title')

    def title_from_link(self, titles=None):
        if not titles:
            titles = XanaLink.get_all_titles()
        return titles.get(self)

    def title(self, titles=None):
        return self.type + ': ' + (self.title_from_link(titles) or self.hashid)

    def has_title(self):
        titles = XanaLink.get_all_titles()
        return self in titles

    def set_title(self, title, user):
        title_source_doc = SourceDoc(text=title, user=user)
        session.add(title_source_doc)
        session.commit()
        link_text = '''type=title

facet=
sourcedoc: {}
facet=
span: {},start=0,length={}'''.format(self.external_url, title_source_doc.external_url, len(title))
        title_link = XanaLink(text=link_text, user=user)
        session.add(title_link)
        session.commit()

    @classmethod
    def from_external(cls, url, home=None):
        if home is None:
            home = url_for('view.home', _external=True)
        if url.startswith(home):
            username, _, hashid = url[len(home):].partition('/')
        else:
            m = re_server_url.match(url)
            if not m:
                return
            username, hashid = m.groups()
        item_id = doc_hashids.decode(hashid)[0]
        q = cls.query.filter(User.username == username, cls.id == item_id)
        return q.one_or_none()

class XanaDoc(Item):
    __tablename__ = 'xanadoc'
    __mapper_args__ = {'polymorphic_identity': 'xanadoc'}

    id = Column(Integer, ForeignKey(Item.id), primary_key=True)

    def snippet(self):
        return self.text

    @property
    def xanaedit_url(self):
        return self.view_url('xanaedit_item')

    @property
    def save_xanaedit_url(self):
        return self.view_url('save_xanaedit')

class XanaLink(Item):
    __tablename__ = 'xanalink'
    __mapper_args__ = {'polymorphic_identity': 'xanalink'}

    id = Column(Integer, ForeignKey(Item.id), primary_key=True)

    def parse(self):
        return parse_link(self.text)

    @property
    def link_type(self):
        return self.parse()['type']

    def title(self, titles=None):
        if titles is None:
            titles = XanaLink.get_all_titles()
        if self in titles:
            return self.type + ': ' + titles[self]

        parsed = self.parse()

        if parsed['type'] == 'title':
            ident = parsed['facets'][0][0].partition(': ')[2]
            item = Item.from_external(ident)
            if item in titles:
                return parsed['type'] + " link for " + item.title(titles=titles)

        if parsed['type']:
            return parsed['type'] + " link: " + self.hashid
        else:
            return "link: " + self.hashid

    def item_and_title(self, home=None):
        link = self.parse()
        if link['type'] != 'title':
            return

        try:
            facet1, facet2 = link['facets']
        except ValueError:
            return

        link_type, _, ident = facet1[0].partition(': ')
        item = Item.from_external(ident, home)

        ident2, start, length = parse_span(facet2[0])
        source_of_title = SourceDoc.from_external(ident2, home)
        if source_of_title:
            return(item, source_of_title.text[start:length + start])

    @classmethod
    def get_all_titles(cls, home=None):
        titles = {}
        for link in cls.query:
            ret = link.item_and_title(home)
            if ret is None:
                continue
            item, title = ret
            titles[item] = title
        return titles

    def snippet(self):
        return self.text

class SourceDoc(Item):
    __tablename__ = 'sourcedoc'
    __mapper_args__ = {'polymorphic_identity': 'sourcedoc'}

    id = Column(Integer, ForeignKey(Item.id), primary_key=True)
    db_price_per_character = Column(Integer)
    db_document_price = Column(Integer)

    @property
    def document_price(self):
        return self.db_document_price or self.db_price_per_character * len(self.text)

    @property
    def price_per_character(self):
        return self.db_price_per_character or self.db_document_price / len(self.text)

    def snippet(self, length=255, killwords=False, end='...', leeway=5):
        s = self.text
        assert length >= len(end), 'expected length >= %s, got %s' % (len(end), length)
        assert leeway >= 0, 'expected leeway >= 0, got %s' % leeway
        if len(s) <= length + leeway:
            return s
        if killwords:
            return s[:length - len(end)] + end
        result = s[:length - len(end)].rsplit(' ', 1)[0]
        return result + end

    def raw_title(self):
        return self.title(with_type=False)

    def title(self, titles=None, with_type=True):
        start = self.type + ': ' if with_type else ''
        titles = XanaLink.get_all_titles()
        from_link = self.title_from_link(titles=titles)
        if from_link:
            return start + from_link
        first_line = first_non_empty_line(self.text)
        if first_line:
            return start + first_line
        return start + self.hashid


configure_mappers()