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
|
# SPDX-License-Identifier: AGPL-3.0-or-later
# lint: pylint
"""SQLite is a small, fast and reliable SQL database engine. It does not require
any extra dependency.
Example
=======
.. _MediathekView: https://mediathekview.de/
To demonstrate the power of database engines, here is a more complex example
which reads from a MediathekView_ (DE) movie database. For this example of the
SQLite engine download the database:
- https://liste.mediathekview.de/filmliste-v2.db.bz2
and unpack into ``searx/data/filmliste-v2.db``. To search the database use e.g
Query to test: ``!mediathekview concert``
.. code:: yaml
- name: mediathekview
engine: sqlite
disabled: False
categories: general
result_template: default.html
database: searx/data/filmliste-v2.db
query_str: >-
SELECT title || ' (' || time(duration, 'unixepoch') || ')' AS title,
COALESCE( NULLIF(url_video_hd,''), NULLIF(url_video_sd,''), url_video) AS url,
description AS content
FROM film
WHERE title LIKE :wildcard OR description LIKE :wildcard
ORDER BY duration DESC
Implementations
===============
"""
import sqlite3
import contextlib
engine_type = 'offline'
database = ""
query_str = ""
limit = 10
paging = True
result_template = 'key-value.html'
def init(engine_settings):
if 'query_str' not in engine_settings:
raise ValueError('query_str cannot be empty')
if not engine_settings['query_str'].lower().startswith('select '):
raise ValueError('only SELECT query is supported')
@contextlib.contextmanager
def sqlite_cursor():
"""Implements a :py:obj:`Context Manager <contextlib.contextmanager>` for a
:py:obj:`sqlite3.Cursor`.
Open database in read only mode: if the database doesn't exist. The default
mode creates an empty file on the file system. See:
* https://docs.python.org/3/library/sqlite3.html#sqlite3.connect
* https://www.sqlite.org/uri.html
"""
uri = 'file:' + database + '?mode=ro'
with contextlib.closing(sqlite3.connect(uri, uri=True)) as connect:
connect.row_factory = sqlite3.Row
with contextlib.closing(connect.cursor()) as cursor:
yield cursor
def search(query, params):
results = []
query_params = {
'query': query,
'wildcard': r'%' + query.replace(' ', r'%') + r'%',
'limit': limit,
'offset': (params['pageno'] - 1) * limit,
}
query_to_run = query_str + ' LIMIT :limit OFFSET :offset'
with sqlite_cursor() as cur:
cur.execute(query_to_run, query_params)
col_names = [cn[0] for cn in cur.description]
for row in cur.fetchall():
item = dict(zip(col_names, map(str, row)))
item['template'] = result_template
logger.debug("append result --> %s", item)
results.append(item)
return results
|
