Tab.set_timezone() - Code Metrics - Inspection of "Fixed length numbers." - glogiotatidis/veresebot - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Push — master ( 8cbbd4...4cd701 )

by Giorgos

created 2015-12-01 16:11 UTC

Tab.set_timezone() A

↳ Parent: Tab

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Metric	Value
cc	1
dl	0
loc	2
rs	10

import importlib
import os
from collections import defaultdict

import ZODB
import ZODB.FileStorage
import arrow
import persistent
import transaction
import zodburi

import config
from log import logger


class User(persistent.Persistent):
    def __init__(self, user_id):
        self.id = user_id

    def update(self, first_name, last_name, username):
        self.first_name = first_name or ''
        self.last_name = last_name or ''
        self.username = username or ''
        self._p_changed__ = True


class Entry(object):
    def __init__(self, message_id, user_id, amount, date, reason=None):
        self.message_id = message_id
        self.user_id = user_id
        self.amount = amount
        self.reason = 'stuff' if not reason else reason
        self.date = date


class Tab(persistent.Persistent):
    def __init__(self, chat_id):
        self.chat_id = chat_id
        self.grandtotal = 0
        self.entries = []
        self.tz = 'UTC'
        self.users = defaultdict(int)

    def clear(self):
        self.entries = []
        self.grandtotal = 0
        self.users = defaultdict(int)
        self._p_changed__ = True

    def set_timezone(self, tz):
        self.tz = tz

    def remove(self, message_id, user_id, date, amount, reason=None):
        return self.add(message_id, user_id, date, -1 * amount, reason)

    def register_user(self, user_id):
        self.users[user_id]

    def add(self, message_id, user_id, date, amount, reason=''):
        position = 0
        for position, v in enumerate(self.entries):
            if v.message_id == message_id:
                # Already in list, ignore
                logger.debug('not adding {}, already in list'.format(amount))
                return
            elif v.message_id < message_id:
                break

        date = arrow.get(date).to(self.tz)
        entry = Entry(message_id, user_id, amount, date, reason)
        logger.debug('adding {}'.format(amount))
        self.entries.insert(position, entry)

        self.grandtotal += amount
        self.users[user_id] += amount

        self._p_changed__ = True

    def get_total(self, from_date=None, to_date=None):
        if not from_date and not to_date:
            return self.grandtotal

        if from_date and to_date and from_date > to_date:
            return -1

        total = 0
        for entry in self.entries:
            if to_date and to_date < entry.date:
                continue

            if from_date:
                if entry.date >= from_date:
                    total += entry.amount
                else:
                    break

        return total


class DB(object):

    def __init__(self):
        # Setup DB
        storage = zodburi.resolve_uri(config.database_url)[0]()
        self._db = ZODB.DB(storage)
        self._connection = self._db.open()
        self.root = self._connection.root
        self.migrate()

    def migrate(self):
        for migration_file in sorted(os.listdir('migrations')):
            if migration_file.endswith('.pyc') or not migration_file.startswith('migration_'):
                continue
            mod = importlib.import_module('migrations.{}'.format(migration_file[:-3]))
            if not mod.Migration.is_applicable(self.root):
                logger.debug('Skipping migration {}'.format(mod.Migration.DB_VERSION))
                continue
            migration = mod.Migration(self.root)
            logger.info('Applying migration {}'.format(migration.DB_VERSION))
            migration.apply()

    def get_or_create_tab(self, tab_id):

        if tab_id in self.root.tabs:
            return self.root.tabs[tab_id], False

        tab = Tab(tab_id)
        self.root.tabs[tab_id] = tab
        self.root.stats['number_of_tabs'] += 1
        logger.debug('Created tab {}'.format(tab_id))
        return tab, True

    def get_or_create_user(self, user_id):

        if user_id in self.root.users:
            return self.root.users[user_id], False

        user = User(user_id)
        self.root.users[user_id] = user
        self.root.stats['number_of_users'] += 1
        logger.debug('Created user {}'.format(user_id))
        return user, True

    def commit(self):
        transaction.commit()

    def close(self):
        self._db.close()


1			import importlib
2			import os
3			from collections import defaultdict
4
5			import ZODB
6			import ZODB.FileStorage
7			import arrow
8			import persistent
9			import transaction
10			import zodburi
11
12			import config
13			from log import logger
14
15
16			class User(persistent.Persistent):
17			def __init__(self, user_id):
18			self.id = user_id
19
20			def update(self, first_name, last_name, username):
21			self.first_name = first_name or ''
22			self.last_name = last_name or ''
23			self.username = username or ''
24			self._p_changed__ = True
25
26
27			class Entry(object):
28			def __init__(self, message_id, user_id, amount, date, reason=None):
29			self.message_id = message_id
30			self.user_id = user_id
31			self.amount = amount
32			self.reason = 'stuff' if not reason else reason
33			self.date = date
34
35
36			class Tab(persistent.Persistent):
37			def __init__(self, chat_id):
38			self.chat_id = chat_id
39			self.grandtotal = 0
40			self.entries = []
41			self.tz = 'UTC'
42			self.users = defaultdict(int)
43
44			def clear(self):
45			self.entries = []
46			self.grandtotal = 0
47			self.users = defaultdict(int)
48			self._p_changed__ = True
49
50			def set_timezone(self, tz):
51			self.tz = tz
52
53			def remove(self, message_id, user_id, date, amount, reason=None):
54			return self.add(message_id, user_id, date, -1 * amount, reason)
55
56			def register_user(self, user_id):
57			self.users[user_id]
58
59			def add(self, message_id, user_id, date, amount, reason=''):
60			position = 0
61			for position, v in enumerate(self.entries):
62			if v.message_id == message_id:
63			# Already in list, ignore
64			logger.debug('not adding {}, already in list'.format(amount))
65			return
66			elif v.message_id < message_id:
67			break
68
69			date = arrow.get(date).to(self.tz)
70			entry = Entry(message_id, user_id, amount, date, reason)
71			logger.debug('adding {}'.format(amount))
72			self.entries.insert(position, entry)
73
74			self.grandtotal += amount
75			self.users[user_id] += amount
76
77			self._p_changed__ = True
78
79			def get_total(self, from_date=None, to_date=None):
80			if not from_date and not to_date:
81			return self.grandtotal
82
83			if from_date and to_date and from_date > to_date:
84			return -1
85
86			total = 0
87			for entry in self.entries:
88			if to_date and to_date < entry.date:
89			continue
90
91			if from_date:
92			if entry.date >= from_date:
93			total += entry.amount
94			else:
95			break
96
97			return total
98
99
100			class DB(object):
101
102			def __init__(self):
103			# Setup DB
104			storage = zodburi.resolve_uri(config.database_url)[0]()
105			self._db = ZODB.DB(storage)
106			self._connection = self._db.open()
107			self.root = self._connection.root
108			self.migrate()
109
110			def migrate(self):
111			for migration_file in sorted(os.listdir('migrations')):
112			if migration_file.endswith('.pyc') or not migration_file.startswith('migration_'):
113			continue
114			mod = importlib.import_module('migrations.{}'.format(migration_file[:-3]))
115			if not mod.Migration.is_applicable(self.root):
116			logger.debug('Skipping migration {}'.format(mod.Migration.DB_VERSION))
117			continue
118			migration = mod.Migration(self.root)
119			logger.info('Applying migration {}'.format(migration.DB_VERSION))
120			migration.apply()
121
122			def get_or_create_tab(self, tab_id):
			0 ignored issues – show Duplication introduced 2015-11-27 10:24 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Duplicated code is one of the most pungent code smells. If you need to duplicate the same code in three or more different places, we strongly encourage you to look into extracting the code into a single class or operation. You can also find more detailed suggestions in the “Code” section of your repository. Loading history...
123			if tab_id in self.root.tabs:
124			return self.root.tabs[tab_id], False
125
126			tab = Tab(tab_id)
127			self.root.tabs[tab_id] = tab
128			self.root.stats['number_of_tabs'] += 1
129			logger.debug('Created tab {}'.format(tab_id))
130			return tab, True
131
132			def get_or_create_user(self, user_id):
			0 ignored issues – show Duplication introduced 2015-11-27 10:24 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Duplicated code is one of the most pungent code smells. If you need to duplicate the same code in three or more different places, we strongly encourage you to look into extracting the code into a single class or operation. You can also find more detailed suggestions in the “Code” section of your repository. Loading history...
133			if user_id in self.root.users:
134			return self.root.users[user_id], False
135
136			user = User(user_id)
137			self.root.users[user_id] = user
138			self.root.stats['number_of_users'] += 1
139			logger.debug('Created user {}'.format(user_id))
140			return user, True
141
142			def commit(self):
143			transaction.commit()
144
145			def close(self):
146			self._db.close()
147

glogiotatidis / veresebot

GitHub Access Token became invalid

Push — master ( 8cbbd4...4cd701 )

Tab.set_timezone() A

Complexity

Size

Duplication

Duplication Side-by-Side

Filter issues like