lektor_ng.sourcesearch

src/lektor_ng/sourcesearch.py
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
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
import sqlite3
from collections.abc import Sized

from lektor_ng.constants import PRIMARY_ALT


def _iter_parents(path):
    path = path.strip("/")
    if path:
        pieces = path.split("/")
        for x in range(len(pieces)):
            yield "/" + "/".join(pieces[:x])


def _find_info(infos, alt, lang):
    for info in infos:
        if info["alt"] == alt and info["lang"] == lang:
            return info
    return None


def _id_from_path(path):
    try:
        return path.strip("/").split("/")[-1]
    except IndexError:
        return ""


def _mapping_from_cursor(cur):
    rv = {}
    for path, alt, lang, type, title in cur.fetchall():
        rv.setdefault(path, []).append(
            {
                "id": _id_from_path(path),
                "path": path,
                "alt": alt,
                "type": type,
                "lang": lang,
                "title": title,
            }
        )
    return rv


def _find_best_info(infos, alt, lang):
    for _alt, _lang in [
        (alt, lang),
        (PRIMARY_ALT, lang),
        (alt, "en"),
        (PRIMARY_ALT, "en"),
    ]:
        rv = _find_info(infos, _alt, _lang)
        if rv is not None:
            return rv
    return None


def _build_parent_path(path, mapping, alt, lang):
    rv = []
    for parent in _iter_parents(path):
        info = _find_best_info(mapping.get(parent) or [], alt, lang)
        id = _id_from_path(parent)
        if info is None:
            title = id or "(Index)"
        else:
            title = info.get("title")
        rv.append({"id": id, "path": parent, "title": title})
    return rv


def _placeholders(values: Sized) -> str:
    """Return SQL '?' placeholders for an array or set of values."""
    return ",".join(["?"] * len(values))


def _process_search_results(builder, cur, alt, lang, limit):
    mapping = _mapping_from_cursor(cur)
    rv = []

    files_needed = set()

    for path, infos in mapping.items():
        info = _find_best_info(infos, alt, lang)
        if info is None:
            continue

        for parent in _iter_parents(path):
            if parent not in mapping:
                files_needed.add(parent)

        rv.append(info)
        if len(rv) == limit:
            break

    if files_needed:
        cur.execute(
            f"""
            select path, alt, lang, type, title
              from source_info
             where path in ({_placeholders(files_needed)})
            """,
            list(files_needed),
        )
        mapping.update(_mapping_from_cursor(cur))

    for info in rv:
        info["parents"] = _build_parent_path(info["path"], mapping, alt, lang)

    return rv


def find_files(builder, query, alt=PRIMARY_ALT, lang=None, limit=50, types=None):
    if types is None:
        types = ["page"]
    else:
        types = list(types)
    languages = ["en"]
    if lang not in ("en", None):
        languages.append(lang)
    else:
        lang = "en"
    alts = [PRIMARY_ALT]
    if alt != PRIMARY_ALT:
        alts.append(alt)

    query = query.strip()
    title_like = "%" + query + "%"
    path_like = "/%" + query.rstrip("/") + "%"

    con = sqlite3.connect(builder.buildstate_database_filename, timeout=10)
    try:
        cur = con.cursor()
        cur.execute(
            f"""
            select path, alt, lang, type, title
              from source_info
             where (title like ? or path like ?)
               and lang in ({_placeholders(languages)})
               and alt in ({_placeholders(alts)})
               and type in ({_placeholders(types)})
          order by title
           collate nocase
             limit ?
            """,
            [title_like, path_like] + languages + alts + types + [limit * 2],
        )
        return _process_search_results(builder, cur, alt, lang, limit)
    finally:
        con.close()