SearchJobsController._compute_keyword_queries() - Code Metrics - Inspection of "Research and companies" - pyjobs/web - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Pull Request — master (#10)

unknown

created 2016-09-15 14:06 UTC

SearchJobsController._compute_keyword_queries() A

↳ Parent: SearchJobsController

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Importance

Changes	5
Bugs	0	Features	0

Metric	Value
cc	1
c	5
b	0
f	0
dl	0
loc	67
rs	9.2815

How to fix Long Method

# -*- coding: utf-8 -*-
import json
import logging
from elasticsearch_dsl import Q, SF

from sqlalchemy.orm.exc import NoResultFound
from tg.decorators import expose, redirect, paginate
from tg.exceptions import HTTPNotFound

from pyjobsweb.model import JobAlchemy
from pyjobsweb.model import JobElastic
from pyjobsweb.model.data import SOURCES
from pyjobsweb.lib.base import BaseController
from pyjobsweb.lib.elasticsearch_ import PaginatedSearch
from pyjobsweb.forms.research_forms import JobsResearchForm


class SearchJobsController(BaseController):
    items_per_page = 10

    def __init__(self, items_per_page=10):
        self.items_per_page = items_per_page

    @staticmethod
    def _compute_keyword_queries(terms):
        search_on = dict(
            description=[
                'description',
                'description.technologies'
            ],
            title=[
                'title',
                'title.technologies'
             ],
            company=['company']
        )

        description_query = Q(
            'multi_match',
            type='most_fields',
            query=terms,
            fields=search_on['description'],
            fuzziness='AUTO',
            operator='or',
            minimum_should_match='1<2 2<2 3<3 4<3 5<4 6<5 7<5 8<6 9<6',
            boost=len(terms.split(','))
        )

        title_query = Q(
            'multi_match',
            type='most_fields',
            query=terms,
            fields=search_on['title'],
            fuzziness='AUTO',
            operator='or',
            minimum_should_match='1<1',
            boost=20 - len(terms.split(',')) + 1
        )

        company_name_query = Q(
            'multi_match',
            type='best_fields',
            query=terms,
            fields=search_on['company'],
            fuzziness='AUTO',
            operator='or',
            minimum_should_match='1<1',
            boost=50
        )

        keyword_queries = Q(
            'bool',
            must=[
                company_name_query
            ],
            should=[
                title_query,
                description_query
            ]
        ) | Q(
            'bool',
            must=[
                description_query
            ],
            should=[
                title_query,
                company_name_query
            ]
        )
        return keyword_queries

    @staticmethod
    def _compute_decay_functions():
        decay_function = SF(
            'gauss',
            publication_datetime=dict(
                origin='now',
                scale='30d',
                offset='7d',
                decay='0.1'
            )
        )

        return [decay_function]

    @staticmethod
    def _apply_geolocation_filters(query, (lat, lon), radius):
        query = query.filter(
            'geo_distance',
            geolocation=[lon, lat],
            distance='%skm' % float(radius)
        )

        query = query.filter(
            'term',
            geolocation_is_valid=True
        )

        return query

    @staticmethod
    def _apply_date_sort(query):
        query = query.sort(
            '-publication_datetime',
            '-_score'
        )

        return query

    @expose('pyjobsweb.templates.jobs.list')
    @paginate('jobs', items_per_page=items_per_page)
    def index(self, query=None, radius=None, center=None, sort_by=None,
              *args, **kwargs):
        if not query and not radius and not center:
            redirect('/jobs')

        search_query = JobElastic().search()

        if query:
            keyword_queries = self._compute_keyword_queries(query)
            decay_functions = self._compute_decay_functions()

            search_query.query = Q(
                'function_score',
                query=keyword_queries,
                functions=decay_functions
            )

        try:
            geoloc_query = json.loads(center)
            coordinates = geoloc_query['coordinates']
            lat, lon = (coordinates['lat'], coordinates['lon'])
        except (ValueError, TypeError):
            # One of the following case has occurred:
            #     - Center wasn't a valid json string
            #     - Radius couldn't be converted to float
            # Since both these information are required to set a geolocation
            # filter are required, we ignore it.
            pass
        else:
            search_query = self._apply_geolocation_filters(
                search_query, (lat, lon), radius if radius else 5.0)

        if sort_by == 'dates':
            search_query = self._apply_date_sort(search_query)

        return dict(sources=SOURCES, jobs=PaginatedSearch(search_query),
                    job_offer_search_form=JobsResearchForm)


class JobsController(BaseController):
    items_per_page = 10
    search = SearchJobsController(items_per_page)

    @expose('pyjobsweb.templates.jobs.list')
    @paginate('jobs', items_per_page=items_per_page)
    def index(self, *args, **kwargs):
        try:
            job_offers = JobAlchemy.get_all_job_offers()
        except NoResultFound:
            job_offers = None

        return dict(sources=SOURCES, jobs=job_offers,
                    job_offer_search_form=JobsResearchForm)

    @expose('pyjobsweb.templates.jobs.details')
    def details(self, offer_id, *args, **kwargs):
        try:
            job = JobAlchemy.get_job_offer(offer_id)
        except NoResultFound:
            raise HTTPNotFound()
        except Exception as exc:
            logging.getLogger(__name__).log(logging.ERROR, exc)
            raise HTTPNotFound()
        else:
            return dict(
                job=job,
                sources=SOURCES
            )



1			# -- coding: utf-8 --
2			import json
3			import logging
4			from elasticsearch_dsl import Q, SF
5
6			from sqlalchemy.orm.exc import NoResultFound
7			from tg.decorators import expose, redirect, paginate
8			from tg.exceptions import HTTPNotFound
9
10			from pyjobsweb.model import JobAlchemy
11			from pyjobsweb.model import JobElastic
12			from pyjobsweb.model.data import SOURCES
13			from pyjobsweb.lib.base import BaseController
14			from pyjobsweb.lib.elasticsearch_ import PaginatedSearch
15			from pyjobsweb.forms.research_forms import JobsResearchForm
16
17
18			class SearchJobsController(BaseController):
19			items_per_page = 10
20
21			def __init__(self, items_per_page=10):
22			self.items_per_page = items_per_page
23
24			@staticmethod
25			def _compute_keyword_queries(terms):
26			search_on = dict(
27			description=[
28			'description',
29			'description.technologies'
30			],
31			title=[
32			'title',
33			'title.technologies'
34			],
35			company=['company']
36			)
37
38			description_query = Q(
39			'multi_match',
40			type='most_fields',
41			query=terms,
42			fields=search_on['description'],
43			fuzziness='AUTO',
44			operator='or',
45			minimum_should_match='1<2 2<2 3<3 4<3 5<4 6<5 7<5 8<6 9<6',
46			boost=len(terms.split(','))
47			)
48
49			title_query = Q(
50			'multi_match',
51			type='most_fields',
52			query=terms,
53			fields=search_on['title'],
54			fuzziness='AUTO',
55			operator='or',
56			minimum_should_match='1<1',
57			boost=20 - len(terms.split(',')) + 1
58			)
59
60			company_name_query = Q(
61			'multi_match',
62			type='best_fields',
63			query=terms,
64			fields=search_on['company'],
65			fuzziness='AUTO',
66			operator='or',
67			minimum_should_match='1<1',
68			boost=50
69			)
70
71			keyword_queries = Q(
72			'bool',
73			must=[
74			company_name_query
75			],
76			should=[
77			title_query,
78			description_query
79			]
80			) \| Q(
81			'bool',
82			must=[
83			description_query
84			],
85			should=[
86			title_query,
87			company_name_query
88			]
89			)
90			return keyword_queries
91
92			@staticmethod
93			def _compute_decay_functions():
94			decay_function = SF(
95			'gauss',
96			publication_datetime=dict(
97			origin='now',
98			scale='30d',
99			offset='7d',
100			decay='0.1'
101			)
102			)
103
104			return [decay_function]
105
106			@staticmethod
107			def _apply_geolocation_filters(query, (lat, lon), radius):
108			query = query.filter(
109			'geo_distance',
110			geolocation=[lon, lat],
111			distance='%skm' % float(radius)
112			)
113
114			query = query.filter(
115			'term',
116			geolocation_is_valid=True
117			)
118
119			return query
120
121			@staticmethod
122			def _apply_date_sort(query):
123			query = query.sort(
124			'-publication_datetime',
125			'-_score'
126			)
127
128			return query
129
130			@expose('pyjobsweb.templates.jobs.list')
131			@paginate('jobs', items_per_page=items_per_page)
132			def index(self, query=None, radius=None, center=None, sort_by=None,
133			args, *kwargs):
134			if not query and not radius and not center:
135			redirect('/jobs')
136
137			search_query = JobElastic().search()
138
139			if query:
140			keyword_queries = self._compute_keyword_queries(query)
141			decay_functions = self._compute_decay_functions()
142
143			search_query.query = Q(
144			'function_score',
145			query=keyword_queries,
146			functions=decay_functions
147			)
148
149			try:
150			geoloc_query = json.loads(center)
151			coordinates = geoloc_query['coordinates']
152			lat, lon = (coordinates['lat'], coordinates['lon'])
153			except (ValueError, TypeError):
154			# One of the following case has occurred:
155			# - Center wasn't a valid json string
156			# - Radius couldn't be converted to float
157			# Since both these information are required to set a geolocation
158			# filter are required, we ignore it.
159			pass
160			else:
161			search_query = self._apply_geolocation_filters(
162			search_query, (lat, lon), radius if radius else 5.0)
163
164			if sort_by == 'dates':
165			search_query = self._apply_date_sort(search_query)
166
167			return dict(sources=SOURCES, jobs=PaginatedSearch(search_query),
168			job_offer_search_form=JobsResearchForm)
169
170
171			class JobsController(BaseController):
172			items_per_page = 10
173			search = SearchJobsController(items_per_page)
174
175			@expose('pyjobsweb.templates.jobs.list')
176			@paginate('jobs', items_per_page=items_per_page)
177			def index(self, args, *kwargs):
178			try:
179			job_offers = JobAlchemy.get_all_job_offers()
180			except NoResultFound:
181			job_offers = None
182
183			return dict(sources=SOURCES, jobs=job_offers,
184			job_offer_search_form=JobsResearchForm)
185
186			@expose('pyjobsweb.templates.jobs.details')
187			def details(self, offer_id, args, *kwargs):
188			try:
189			job = JobAlchemy.get_job_offer(offer_id)
190			except NoResultFound:
191			raise HTTPNotFound()
192			except Exception as exc:
193			logging.getLogger(__name__).log(logging.ERROR, exc)
194			raise HTTPNotFound()
195			else:
196			return dict(
197			job=job,
198			sources=SOURCES
199			)
200
201

pyjobs / web

GitHub Access Token became invalid

Pull Request — master (#10)

SearchJobsController._compute_keyword_queries() A

Complexity

Size

Duplication

Importance

How to fix Long Method

Long Method

Duplication Side-by-Side

Filter issues like