web/utils.py at master · kun-fang/web · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
"""web util functions"""
import difflib
from sqlalchemy.orm.exc import NoResultFound

from web import models


class InvalidEntity(Exception):
    """Exception for invalid input"""
    pass


def new_document(user, sess=None):
    """create a new document"""
    with models.managed_session(sess=sess) as sess:
        doc = models.Document(user)
        doc = sess.merge(doc, load=True)
        sess.commit()
        return doc


def get_document(name, sess=None):
    """get document by random name"""
    with models.managed_session(sess=sess) as sess:
        try:
            doc = sess.query(models.Document).filter_by(name=name).one()
            return doc
        except NoResultFound:
            return None


def new_user(name, email, sess=None):
    """create a new user"""
    with models.managed_session(sess=sess) as sess:
        user = models.User(name, email)
        user = sess.merge(user, load=True)
        sess.commit()
        return user


def get_user(name, email, sess=None):
    """get all the documents for a user"""
    with models.managed_session(sess=sess) as sess:
        try:
            user = sess.query(models.User).filter_by(name=name, email=email).one()
        except NoResultFound:
            user = new_user(name, email, sess=sess)
        return user


def get_history(user_id, document_id, sess=None):
    """get edit history of a document"""
    with models.managed_session(sess=sess) as sess:
        document = sess.query(models.Document).filter_by(document_id=document_id).one()
        if document.user_id != user_id:
            raise InvalidEntity("document and user do not match")
        return document.history


def diff_text(old_text, new_text):
    """compare two texts return diff in dictionary"""
    new_text = new_text.split("\n")
    old_text = old_text.split("\n")
    matcher = difflib.SequenceMatcher(a=old_text, b=new_text)
    diff = []
    for tag, i1, i2, j1, j2 in matcher.get_opcodes():  # pylint: disable = invalid-name
        if tag == 'equal':
            continue
        diff.append({
            'old': [i1, i2, old_text[i1:i2]],
            'new': [j1, j2, new_text[j1:j2]]
        })
    return diff


def revert_diff(item):
    """revert diff direction"""
    return {
        'old': item['new'],
        'new': item['old']
    }


def convert(text, diff, back=False):
    """convert the text according to diff"""
    text = text.split("\n")
    new_text = []
    prev = 0
    for item in diff:
        if back:
            item = revert_diff(item)
        start, end, content = item['old']
        if text[start:end] != content:
            print text[start:end]
            print content
            raise ValueError("diff doesn't match")
        new_text.extend(text[prev:start])
        new_text.extend(item['new'][2])
        prev = end
    new_text.extend(text[prev:])
    return "\n".join(new_text)


def get_room(room, sess=None, create_room=True):
    with models.managed_session(sess=sess) as sess:
        try:
            room = sess.query(models.Room).filter_by(room=room).one()
        except NoResultFound:
            if create_room:
                room = models.Room(room)
                room = sess.merge(room, load=True)
                sess.commit()
            else:
                raise
        return room


def add_to_room(room, user, sess=None):
    with models.managed_session(sess=sess) as sess:
        room = get_room(room, sess=sess)
        room.add_user(user)
        sess.commit()
        return room


def remove_from_room(room, user, sess=None):
    with models.managed_session(sess=sess) as sess:
        room = get_room(room, sess=sess, create_room=False)
        if room:
            room.remove_user(user)
            sess.commit()
        return room