CDR._get_total_sell() - Code Metrics - mwolff44/pyfreebilling - Measure and Improve Code Quality continuously with Scrutinizer

CDR._get_total_sell() A
last analyzed 2018-05-31 07:49 UTC

↳ Parent: CDR

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
dl	0
loc	8
c	0
b	0
f	0
rs	9.2
cc	4

# Copyright 2013 Mathias WOLFF
# This file is part of pyfreebilling.
#
# pyfreebilling is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# pyfreebilling is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with pyfreebilling.  If not, see <http://www.gnu.org/licenses/>

from django.db import models
from django.db.models import Sum, Count
from django.utils.translation import ugettext_lazy as _

import decimal
import math
import qsstats

from pyfreebilling.pyfreebill.models import Company, SofiaGateway, RateCard, LCRGroup

from pyfreebilling.customerdirectory.models import CustomerDirectory


class CDR(models.Model):
    """ CDR Model    """
    customer = models.ForeignKey(
        Company,
        verbose_name=_(u"customer"),
        null=True,
        related_name="customer_related")
    customer_ip = models.CharField(
        _(u"customer IP address"),
        max_length=100,
        null=True,
        help_text=_(u"Customer IP address."))
    uuid = models.CharField(
        _(u"UUID"),
        max_length=100,
        null=True)
    bleg_uuid = models.CharField(
        _(u"b leg UUID"),
        null=True,
        default="",
        max_length=100)
    kam_uuid = models.CharField(
        _(u"SIP Server unique call-ID"),
        max_length=100,
        null=True)
    caller_id_number = models.CharField(
        _(u"caller ID num"),
        max_length=100,
        null=True)
    destination_number = models.CharField(
        _(u"Dest. number"),
        max_length=100,
        null=True)
    chan_name = models.CharField(
        _(u"channel name"),
        max_length=100,
        null=True)
    start_stamp = models.DateTimeField(
        _(u"start time"),
        null=True,
        db_index=True)
    answered_stamp = models.DateTimeField(
        _(u"answered time"),
        null=True)
    end_stamp = models.DateTimeField(
        _(u"hangup time"),
        null=True)
    duration = models.IntegerField(
        _(u"global duration"),
        null=True)
    effectiv_duration = models.IntegerField(
        _(u"total duration"),
        null=True,
        help_text=_(
            u"Global call duration since call has been received by the switch in ms."))
    effective_duration = models.IntegerField(
        _(u"effective duration"),
        null=True,
        help_text=_(u"real call duration in s."))
    billsec = models.IntegerField(
        _(u"billed duration"),
        null=True,
        help_text=_(u"billed call duration in s."))
    read_codec = models.CharField(
        _(u"read codec"),
        max_length=20,
        null=True)
    write_codec = models.CharField(
        _(u"write codec"),
        max_length=20,
        null=True)
    hangup_cause = models.CharField(
        _(u"hangup cause"),
        max_length=50,
        null=True,
        db_index=True)
    hangup_cause_q850 = models.IntegerField(
        _(u"q.850"),
        null=True)
    gateway = models.ForeignKey(
        SofiaGateway,
        verbose_name=_(u"gateway"),
        null=True)
    cost_rate = models.DecimalField(
        _(u'buy rate'),
        max_digits=11,
        decimal_places=5,
        default="0",
        null=True)
    total_sell = models.DecimalField(
        _(u'total sell'),
        max_digits=11,
        decimal_places=5,
        default="0",
        null=True)
    total_cost = models.DecimalField(
        _(u'total cost'),
        max_digits=11,
        decimal_places=5,
        default="0",
        null=True)
    prefix = models.CharField(
        _(u'Prefix'),
        max_length=30,
        null=True)
    country = models.CharField(
        _(u'Country'),
        max_length=100,
        null=True)
    rate = models.DecimalField(
        _(u'sell rate'),
        max_digits=11,
        decimal_places=5,
        null=True)
    init_block = models.DecimalField(
        _(u'Connection fee'),
        max_digits=11,
        decimal_places=5,
        null=True)
    block_min_duration = models.IntegerField(
        _(u'increment'),
        null=True)
    lcr_carrier_id = models.ForeignKey(
        Company,
        verbose_name=_(u"provider"),
        null=True,
        related_name="carrier_related")
    ratecard_id = models.ForeignKey(
        RateCard,
        null=True,
        verbose_name=_(u"ratecard"))
    lcr_group_id = models.ForeignKey(
        LCRGroup,
        null=True,
        verbose_name=_(u"lcr group"))
    sip_user_agent = models.CharField(
        _(u'sip user agent'),
        null=True,
        max_length=100)
    sip_rtp_rxstat = models.CharField(
        _(u'sip rtp rx stat'),
        null=True,
        max_length=30)
    sip_rtp_txstat = models.CharField(
        _(u'sip rtp tx stat'),
        null=True,
        max_length=30)
    switchname = models.CharField(
        _(u"Media server name"),
        null=True,
        default="",
        max_length=100)
    sipserver_name = models.CharField(
        _(u"FSIP server name"),
        null=True,
        default="",
        max_length=100)
    switch_ipv4 = models.CharField(
        _(u"switch ipv4"),
        null=True,
        default="",
        max_length=100)
    hangup_disposition = models.CharField(
        _(u"hangup disposition"),
        null=True,
        default="",
        max_length=100,
        help_text=_(
            u"""Interpretation of these values differs on incoming and
            outgoing calls since FreeSWITCH is at different ends of the
            session.

            <b>Incoming :</b>
            send_bye    FS sent BYE to the caller (we hung up)
            recv_bye    FS received BYE from the caller (they hung up)
            send_refuse FS rejected the call (e.g. 4xx or 5xx)
            send_cancel n/a

            <b>Outgoing :</b>
            send_bye    FS sent BYE to the endpoint (we hung up)
            recv_bye    FS received BYE from the endpoint (they hung up)
            send_refuse Endpoint rejected the call (e.g. 4xx or 5xx)
            send_cancel FS aborted the call (we sent CANCEL)"""))
    sip_hangup_cause = models.CharField(
        _(u"SIP hangup cause"),
        null=True,
        default="",
        max_length=100)
    sell_destination = models.CharField(
        _(u'sell destination'),
        blank=True,
        default='',
        null=True,
        max_length=128,
        db_index=True)
    cost_destination = models.CharField(
        _(u'cost destination'),
        blank=True,
        default='',
        null=True,
        max_length=128,
        db_index=True)
    insee_code = models.CharField(
        _(u'Special code for routing urgency numbers'),
        null=True,
        blank=True,
        max_length=10,
        help_text=_(u"""Postal code, INSEE code ... for routing
          urgency number to the right urgency call center."""))
    customerdirectory_id = models.CharField(
        max_length=50,
        null=True,
        verbose_name=_(u"sip account"))
    rctype = models.CharField(
        _(u"Type of call"),
        max_length=10,
        null=True,
        blank=True,
        help_text=_(u"""Type of calls."""))

    class Meta:
        db_table = 'cdr'
        app_label = 'cdr'
        ordering = ('start_stamp', 'customer')
        indexes = [
            models.Index(fields=['customerdirectory_id']),
        ]
        verbose_name = _(u"CDR")
        verbose_name_plural = _(u"CDRs")

    def __unicode__(self):
        if self.start_stamp:
            return unicode(self.start_stamp)
        else:
            return self.custom_alias_name

    def hangup_cause_colored(self):
        if self.billsec == 0:
            color = "red"
        else:
            color = "green"
        return " <span style=color:%s>%s</span>" % (color, self.hangup_cause)
    hangup_cause_colored.allow_tags = True

    @property
    def daily_total_answered_calls(self):
        return qsstats.QuerySetStats(
            self.objects.all()
                .exclude(effective_duration="0")
                .filter(hangup_cause="NORMAL_CLEARING"),
            'start_stamp',
            aggregate=Count('id')
        ).this_day()

    @property
    def daily_total_calls(self):
        return qsstats.QuerySetStats(
            self.objects.all(),
            'start_stamp',
            aggregate=Count('id')
        ).this_day()

    @property
    def daily_total_effective_duration_calls(self):
        return qsstats.QuerySetStats(
            self.objects.all()
                .exclude(effective_duration="0")
                .filter(hangup_cause="NORMAL_CLEARING"),
            'start_stamp',
            aggregate=Sum('effective_duration')
        ).this_day()

    @property
    def daily_total_sell_calls(self):
        return qsstats.QuerySetStats(
            self.objects.all()
                .exclude(effective_duration="0")
                .filter(hangup_cause="NORMAL_CLEARING"),
            'start_stamp',
            aggregate=Sum('total_sell')
        ).this_day()

    @property
    def daily_total_cost_calls(self):
        return qsstats.QuerySetStats(
            self.objects.all()
                .exclude(effective_duration="0")
                .filter(hangup_cause="NORMAL_CLEARING"),
            'start_stamp',
            aggregate=Sum('total_cost')
        ).this_day()

    def _get_min_effective_duration(self):
        if self.effective_duration:
            min = int(self.effective_duration / 60)
            sec = int(self.effective_duration % 60)
        else:
            min = 0
            sec = 0

        return "%02d:%02d" % (min, sec)
    min_effective_duration = property(_get_min_effective_duration)

    def _get_total_sell(self):
        if self.rate and self.rate != 0:
            totalsell = decimal.Decimal(self.billsec) * decimal.Decimal(self.rate) / 60
        else:
            totalsell = 0.000000
        if self.init_block:
            totalsell = decimal.Decimal(totalsell) + decimal.Decimal(self.init_block)
        return round(totalsell, 6)
    total_sell_py = property(_get_total_sell)

    def _get_total_cost(self):
        if self.cost_rate:
            totalcost = decimal.Decimal(self.effective_duration) * decimal.Decimal(self.cost_rate) / 60
        else:
            totalcost = 0.000000
        return round(totalcost, 6)
    total_cost_py = property(_get_total_cost)

    def _get_effective_duration(self):
        if self.effectiv_duration:
            effdur = math.ceil(self.effectiv_duration / 1000.0)
        else:
            effdur = 0
        return int(effdur)
    effective_duration_py = property(_get_effective_duration)

    def _get_billsec(self):
        if self.block_min_duration and self.effective_duration:
            if self.effective_duration < self.block_min_duration:
                billsec = self.block_min_duration
            else:
                billsec = math.ceil(self.effective_duration / self.block_min_duration) * self.block_min_duration
        else:
            billsec = self.effective_duration
        return int(billsec)
    billsec_py = property(_get_billsec)

    def success_cdr(self):
        return self.CDR.objects.exclude(effective_duration="0")


1			# Copyright 2013 Mathias WOLFF
2			# This file is part of pyfreebilling.
3			#
4			# pyfreebilling is free software: you can redistribute it and/or modify
5			# it under the terms of the GNU 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			# pyfreebilling 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 General Public License for more details.
13			#
14			# You should have received a copy of the GNU General Public License
15			# along with pyfreebilling. If not, see <http://www.gnu.org/licenses/>
16
17			from django.db import models
18			from django.db.models import Sum, Count
19			from django.utils.translation import ugettext_lazy as _
20
21			import decimal
22			import math
23			import qsstats
24
25			from pyfreebilling.pyfreebill.models import Company, SofiaGateway, RateCard, LCRGroup
26
27			from pyfreebilling.customerdirectory.models import CustomerDirectory
28
29
30			class CDR(models.Model):
31			""" CDR Model """
32			customer = models.ForeignKey(
33			Company,
34			verbose_name=_(u"customer"),
35			null=True,
36			related_name="customer_related")
37			customer_ip = models.CharField(
38			_(u"customer IP address"),
39			max_length=100,
40			null=True,
41			help_text=_(u"Customer IP address."))
42			uuid = models.CharField(
43			_(u"UUID"),
44			max_length=100,
45			null=True)
46			bleg_uuid = models.CharField(
47			_(u"b leg UUID"),
48			null=True,
49			default="",
50			max_length=100)
51			kam_uuid = models.CharField(
52			_(u"SIP Server unique call-ID"),
53			max_length=100,
54			null=True)
55			caller_id_number = models.CharField(
56			_(u"caller ID num"),
57			max_length=100,
58			null=True)
59			destination_number = models.CharField(
60			_(u"Dest. number"),
61			max_length=100,
62			null=True)
63			chan_name = models.CharField(
64			_(u"channel name"),
65			max_length=100,
66			null=True)
67			start_stamp = models.DateTimeField(
68			_(u"start time"),
69			null=True,
70			db_index=True)
71			answered_stamp = models.DateTimeField(
72			_(u"answered time"),
73			null=True)
74			end_stamp = models.DateTimeField(
75			_(u"hangup time"),
76			null=True)
77			duration = models.IntegerField(
78			_(u"global duration"),
79			null=True)
80			effectiv_duration = models.IntegerField(
81			_(u"total duration"),
82			null=True,
83			help_text=_(
84			u"Global call duration since call has been received by the switch in ms."))
85			effective_duration = models.IntegerField(
86			_(u"effective duration"),
87			null=True,
88			help_text=_(u"real call duration in s."))
89			billsec = models.IntegerField(
90			_(u"billed duration"),
91			null=True,
92			help_text=_(u"billed call duration in s."))
93			read_codec = models.CharField(
94			_(u"read codec"),
95			max_length=20,
96			null=True)
97			write_codec = models.CharField(
98			_(u"write codec"),
99			max_length=20,
100			null=True)
101			hangup_cause = models.CharField(
102			_(u"hangup cause"),
103			max_length=50,
104			null=True,
105			db_index=True)
106			hangup_cause_q850 = models.IntegerField(
107			_(u"q.850"),
108			null=True)
109			gateway = models.ForeignKey(
110			SofiaGateway,
111			verbose_name=_(u"gateway"),
112			null=True)
113			cost_rate = models.DecimalField(
114			_(u'buy rate'),
115			max_digits=11,
116			decimal_places=5,
117			default="0",
118			null=True)
119			total_sell = models.DecimalField(
120			_(u'total sell'),
121			max_digits=11,
122			decimal_places=5,
123			default="0",
124			null=True)
125			total_cost = models.DecimalField(
126			_(u'total cost'),
127			max_digits=11,
128			decimal_places=5,
129			default="0",
130			null=True)
131			prefix = models.CharField(
132			_(u'Prefix'),
133			max_length=30,
134			null=True)
135			country = models.CharField(
136			_(u'Country'),
137			max_length=100,
138			null=True)
139			rate = models.DecimalField(
140			_(u'sell rate'),
141			max_digits=11,
142			decimal_places=5,
143			null=True)
144			init_block = models.DecimalField(
145			_(u'Connection fee'),
146			max_digits=11,
147			decimal_places=5,
148			null=True)
149			block_min_duration = models.IntegerField(
150			_(u'increment'),
151			null=True)
152			lcr_carrier_id = models.ForeignKey(
153			Company,
154			verbose_name=_(u"provider"),
155			null=True,
156			related_name="carrier_related")
157			ratecard_id = models.ForeignKey(
158			RateCard,
159			null=True,
160			verbose_name=_(u"ratecard"))
161			lcr_group_id = models.ForeignKey(
162			LCRGroup,
163			null=True,
164			verbose_name=_(u"lcr group"))
165			sip_user_agent = models.CharField(
166			_(u'sip user agent'),
167			null=True,
168			max_length=100)
169			sip_rtp_rxstat = models.CharField(
170			_(u'sip rtp rx stat'),
171			null=True,
172			max_length=30)
173			sip_rtp_txstat = models.CharField(
174			_(u'sip rtp tx stat'),
175			null=True,
176			max_length=30)
177			switchname = models.CharField(
178			_(u"Media server name"),
179			null=True,
180			default="",
181			max_length=100)
182			sipserver_name = models.CharField(
183			_(u"FSIP server name"),
184			null=True,
185			default="",
186			max_length=100)
187			switch_ipv4 = models.CharField(
188			_(u"switch ipv4"),
189			null=True,
190			default="",
191			max_length=100)
192			hangup_disposition = models.CharField(
193			_(u"hangup disposition"),
194			null=True,
195			default="",
196			max_length=100,
197			help_text=_(
198			u"""Interpretation of these values differs on incoming and
199			outgoing calls since FreeSWITCH is at different ends of the
200			session.
201
202			<b>Incoming :</b>
203			send_bye FS sent BYE to the caller (we hung up)
204			recv_bye FS received BYE from the caller (they hung up)
205			send_refuse FS rejected the call (e.g. 4xx or 5xx)
206			send_cancel n/a
207
208			<b>Outgoing :</b>
209			send_bye FS sent BYE to the endpoint (we hung up)
210			recv_bye FS received BYE from the endpoint (they hung up)
211			send_refuse Endpoint rejected the call (e.g. 4xx or 5xx)
212			send_cancel FS aborted the call (we sent CANCEL)"""))
213			sip_hangup_cause = models.CharField(
214			_(u"SIP hangup cause"),
215			null=True,
216			default="",
217			max_length=100)
218			sell_destination = models.CharField(
219			_(u'sell destination'),
220			blank=True,
221			default='',
222			null=True,
223			max_length=128,
224			db_index=True)
225			cost_destination = models.CharField(
226			_(u'cost destination'),
227			blank=True,
228			default='',
229			null=True,
230			max_length=128,
231			db_index=True)
232			insee_code = models.CharField(
233			_(u'Special code for routing urgency numbers'),
234			null=True,
235			blank=True,
236			max_length=10,
237			help_text=_(u"""Postal code, INSEE code ... for routing
238			urgency number to the right urgency call center."""))
239			customerdirectory_id = models.CharField(
240			max_length=50,
241			null=True,
242			verbose_name=_(u"sip account"))
243			rctype = models.CharField(
244			_(u"Type of call"),
245			max_length=10,
246			null=True,
247			blank=True,
248			help_text=_(u"""Type of calls."""))
249
250			class Meta:
251			db_table = 'cdr'
252			app_label = 'cdr'
253			ordering = ('start_stamp', 'customer')
254			indexes = [
255			models.Index(fields=['customerdirectory_id']),
256			]
257			verbose_name = _(u"CDR")
258			verbose_name_plural = _(u"CDRs")
259
260			def __unicode__(self):
261			if self.start_stamp:
262			return unicode(self.start_stamp)
263			else:
264			return self.custom_alias_name
265
266			def hangup_cause_colored(self):
267			if self.billsec == 0:
268			color = "red"
269			else:
270			color = "green"
271			return " <span style=color:%s>%s</span>" % (color, self.hangup_cause)
272			hangup_cause_colored.allow_tags = True
273
274			@property
275			def daily_total_answered_calls(self):
276			return qsstats.QuerySetStats(
277			self.objects.all()
278			.exclude(effective_duration="0")
279			.filter(hangup_cause="NORMAL_CLEARING"),
280			'start_stamp',
281			aggregate=Count('id')
282			).this_day()
283
284			@property
285			def daily_total_calls(self):
286			return qsstats.QuerySetStats(
287			self.objects.all(),
288			'start_stamp',
289			aggregate=Count('id')
290			).this_day()
291
292			@property
293			def daily_total_effective_duration_calls(self):
294			return qsstats.QuerySetStats(
295			self.objects.all()
296			.exclude(effective_duration="0")
297			.filter(hangup_cause="NORMAL_CLEARING"),
298			'start_stamp',
299			aggregate=Sum('effective_duration')
300			).this_day()
301
302			@property
303			def daily_total_sell_calls(self):
304			return qsstats.QuerySetStats(
305			self.objects.all()
306			.exclude(effective_duration="0")
307			.filter(hangup_cause="NORMAL_CLEARING"),
308			'start_stamp',
309			aggregate=Sum('total_sell')
310			).this_day()
311
312			@property
313			def daily_total_cost_calls(self):
314			return qsstats.QuerySetStats(
315			self.objects.all()
316			.exclude(effective_duration="0")
317			.filter(hangup_cause="NORMAL_CLEARING"),
318			'start_stamp',
319			aggregate=Sum('total_cost')
320			).this_day()
321
322			def _get_min_effective_duration(self):
323			if self.effective_duration:
324			min = int(self.effective_duration / 60)
325			sec = int(self.effective_duration % 60)
326			else:
327			min = 0
328			sec = 0
329
330			return "%02d:%02d" % (min, sec)
331			min_effective_duration = property(_get_min_effective_duration)
332
333			def _get_total_sell(self):
334			if self.rate and self.rate != 0:
335			totalsell = decimal.Decimal(self.billsec) * decimal.Decimal(self.rate) / 60
336			else:
337			totalsell = 0.000000
338			if self.init_block:
339			totalsell = decimal.Decimal(totalsell) + decimal.Decimal(self.init_block)
340			return round(totalsell, 6)
341			total_sell_py = property(_get_total_sell)
342
343			def _get_total_cost(self):
344			if self.cost_rate:
345			totalcost = decimal.Decimal(self.effective_duration) * decimal.Decimal(self.cost_rate) / 60
346			else:
347			totalcost = 0.000000
348			return round(totalcost, 6)
349			total_cost_py = property(_get_total_cost)
350
351			def _get_effective_duration(self):
352			if self.effectiv_duration:
353			effdur = math.ceil(self.effectiv_duration / 1000.0)
354			else:
355			effdur = 0
356			return int(effdur)
357			effective_duration_py = property(_get_effective_duration)
358
359			def _get_billsec(self):
360			if self.block_min_duration and self.effective_duration:
361			if self.effective_duration < self.block_min_duration:
362			billsec = self.block_min_duration
363			else:
364			billsec = math.ceil(self.effective_duration / self.block_min_duration) * self.block_min_duration
365			else:
366			billsec = self.effective_duration
367			return int(billsec)
368			billsec_py = property(_get_billsec)
369
370			def success_cdr(self):
371			return self.CDR.objects.exclude(effective_duration="0")
372

mwolff44 / pyfreebilling

CDR._get_total_sell() A last analyzed 2018-05-31 07:49 UTC

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like

CDR._get_total_sell() A
last analyzed 2018-05-31 07:49 UTC