GeocompleteController.index() - Code Metrics - pyjobs/web - Measure and Improve Code Quality continuously with Scrutinizer

GeocompleteController.index() B
last analyzed 2016-09-27 21:29 UTC

↳ Parent: GeocompleteController

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Importance

Changes	7
Bugs	0	Features	0

Metric	Value
cc	6
c	7
b	0
f	0
dl	0
loc	78
rs	7.4246

How to fix Long Method

# -*- coding: utf-8 -*-
import json

from elasticsearch_dsl.aggs import A
from elasticsearch_dsl.function import SF
from elasticsearch_dsl.query import Q
from tg.decorators import expose

from pyjobsweb import model
from pyjobsweb.lib.base import BaseController


class GeocompleteController(BaseController):
    @staticmethod
    def geocomplete_town_input_parser(address_input):
        query_tokens = address_input.split(' ')

        postal_code = None
        address = None

        for token in query_tokens:
            try:
                int(token)

                if len(token) <= 5:
                    postal_code = postal_code if postal_code else token
                else:
                    return dict(results=[])
            except ValueError:
                address = u'%s %s' % (address, token) if address else token

        return address, postal_code

    @expose('json')
    def index(self, address=None, *args, **kwargs):
        if not address:
            return dict(results=[])

        search = model.Geocomplete().search()

        address_query = Q()
        postal_code_query = Q()

        (address, postal_code) = self.geocomplete_town_input_parser(address)

        if address:
            address_query = Q('match', name=address)

        if postal_code:
            postal_code_query = Q('match', postal_code=postal_code)

        weight_scoring_function = SF(
            'field_value_factor',
            factor=1,
            modifier='none',
            field='weight'
        )

        search.query = Q(
            'function_score',
            query=address_query & postal_code_query,
            functions=[weight_scoring_function]
        )

        dedup_docs = A(
            'top_hits',
            size=1,
            sort={'postal_code.raw': 'asc'}
        )

        dedup = A(
            'terms',
            field='name.raw',
            size=5,
            order={'score_sort': 'desc'}
        )

        score_sort = A(
            'max',
            script=dict(lang='expression', script='_score')
        )

        dedup.bucket('dedup_docs', dedup_docs)
        dedup.bucket('score_sort', score_sort)
        search.aggs.bucket('dedup', dedup)

        # Do not compute the results, we are only interested by the aggregations
        raw_res = search[0:0].execute()

        res = list()
        for bucket in raw_res.aggregations.dedup.buckets:
            for source_doc in bucket['dedup_docs']['hits']['hits']:
                fields = source_doc['_source']

                name = fields['name']
                complement = fields['complement']
                postal_code = fields['postal_code']
                country = 'France'

                geoloc = fields['geolocation']
                coordinates = dict(lat=geoloc['lat'], lon=geoloc['lon'])

                res.append(dict(
                    name=name,
                    complement=complement,
                    postal_code=postal_code,
                    country=country,
                    coordinates=coordinates
                ))

        return dict(results=res)


1			# -- coding: utf-8 --
2			import json
3
4			from elasticsearch_dsl.aggs import A
5			from elasticsearch_dsl.function import SF
6			from elasticsearch_dsl.query import Q
7			from tg.decorators import expose
8
9			from pyjobsweb import model
10			from pyjobsweb.lib.base import BaseController
11
12
13			class GeocompleteController(BaseController):
14			@staticmethod
15			def geocomplete_town_input_parser(address_input):
16			query_tokens = address_input.split(' ')
17
18			postal_code = None
19			address = None
20
21			for token in query_tokens:
22			try:
23			int(token)
24
25			if len(token) <= 5:
26			postal_code = postal_code if postal_code else token
27			else:
28			return dict(results=[])
29			except ValueError:
30			address = u'%s %s' % (address, token) if address else token
31
32			return address, postal_code
33
34			@expose('json')
35			def index(self, address=None, args, *kwargs):
36			if not address:
37			return dict(results=[])
38
39			search = model.Geocomplete().search()
40
41			address_query = Q()
42			postal_code_query = Q()
43
44			(address, postal_code) = self.geocomplete_town_input_parser(address)
45
46			if address:
47			address_query = Q('match', name=address)
48
49			if postal_code:
50			postal_code_query = Q('match', postal_code=postal_code)
51
52			weight_scoring_function = SF(
53			'field_value_factor',
54			factor=1,
55			modifier='none',
56			field='weight'
57			)
58
59			search.query = Q(
60			'function_score',
61			query=address_query & postal_code_query,
62			functions=[weight_scoring_function]
63			)
64
65			dedup_docs = A(
66			'top_hits',
67			size=1,
68			sort={'postal_code.raw': 'asc'}
69			)
70
71			dedup = A(
72			'terms',
73			field='name.raw',
74			size=5,
75			order={'score_sort': 'desc'}
76			)
77
78			score_sort = A(
79			'max',
80			script=dict(lang='expression', script='_score')
81			)
82
83			dedup.bucket('dedup_docs', dedup_docs)
84			dedup.bucket('score_sort', score_sort)
85			search.aggs.bucket('dedup', dedup)
86
87			# Do not compute the results, we are only interested by the aggregations
88			raw_res = search[0:0].execute()
89
90			res = list()
91			for bucket in raw_res.aggregations.dedup.buckets:
92			for source_doc in bucket['dedup_docs']['hits']['hits']:
93			fields = source_doc['_source']
94
95			name = fields['name']
96			complement = fields['complement']
97			postal_code = fields['postal_code']
98			country = 'France'
99
100			geoloc = fields['geolocation']
101			coordinates = dict(lat=geoloc['lat'], lon=geoloc['lon'])
102
103			res.append(dict(
104			name=name,
105			complement=complement,
106			postal_code=postal_code,
107			country=country,
108			coordinates=coordinates
109			))
110
111			return dict(results=res)
112

pyjobs / web

GitHub Access Token became invalid

GeocompleteController.index() B last analyzed 2016-09-27 21:29 UTC

Complexity

Size

Duplication

Importance

How to fix Long Method

Long Method

Duplication Side-by-Side

Filter issues like

GeocompleteController.index() B
last analyzed 2016-09-27 21:29 UTC