learning-website-django1-9  Exercise 2: A Safe Search Page
=============================================================

Turn what a visitor types into a search without ever letting their text become part
of the query language, and show results with a highlighted snippet.

Why the text cannot go in as typed: FTS5 has its own query syntax (quotes, AND, OR,
NEAR, parentheses, a leading minus, column filters such as title:x). Text that is
not valid in it is an ERROR, not a search, and text that is valid can change the
meaning. So only word characters are kept, each is quoted, the last one may be a
beginning (so "cond" finds "conditional"), and at most eight are used.

Save as apps/search/query.py:

"""Search the pages of ONE site. User text is never put into the query as it is: FTS5 has its own query
language (quotes, AND, OR, NEAR, parentheses, -), and text that is not valid in it is an error, not a search."""
import re

from django.db import connection
from django.utils.html import escape
from django.utils.safestring import mark_safe

from .text import has_cjk

MAX_TERMS = 8
START, END = "\x01", "\x02"               # snippet markers: control characters cannot appear in page text
TRIGRAM_MIN = 3


def word_query(text):
    """'conditional szer' -> '"conditional" "szer"*'  (every word must match; the last may be a beginning)."""
    words = re.findall(r"\w+", text)[:MAX_TERMS]
    if not words:
        return ""
    quoted = [f'"{w}"' for w in words]
    quoted[-1] += "*"
    return " ".join(quoted)


def cjk_terms(text):
    return [t.replace('"', "") for t in text.split() if t.replace('"', "")][:MAX_TERMS]


def _highlight(snippet):
    """Escape everything, then turn the markers into <mark> tags: page text can never inject HTML."""
    return mark_safe(escape(snippet).replace(START, "<mark>").replace(END, "</mark>"))


def search(site, text, *, limit=20, offset=0):
    """Return (total, results). A result is a dict: id, title, path, kind, url_path, snippet."""
    text = (text or "").strip()[:100]
    if not text or connection.vendor != "sqlite":
        return 0, []
    if has_cjk(text):
        return _search_cjk(site, text, limit, offset)
    match = word_query(text)
    if not match:
        return 0, []
    table = "page_fts"
    with connection.cursor() as cursor:
        cursor.execute(f"SELECT count(*) FROM {table} WHERE {table} MATCH %s AND site = %s", [match, site])
        total = cursor.fetchone()[0]
        cursor.execute(
            f"""SELECT p.id, p.title, p.path, p.kind,
                       snippet({table}, 1, %s, %s, '...', 24)
                FROM {table} JOIN content_page p ON p.id = {table}.rowid
                WHERE {table} MATCH %s AND {table}.site = %s
                ORDER BY bm25({table}, 10.0, 1.0)
                LIMIT %s OFFSET %s""",
            [START, END, match, site, limit, offset])
        rows = cursor.fetchall()
    return total, [_result(*row) for row in rows]


def _search_cjk(site, text, limit, offset):
    terms = cjk_terms(text)
    table = "page_fts_cjk"
    with connection.cursor() as cursor:
        if all(len(t) >= TRIGRAM_MIN for t in terms):
            match = " ".join(f'"{t}"' for t in terms)
            cursor.execute(f"SELECT count(*) FROM {table} WHERE {table} MATCH %s AND site = %s", [match, site])
            total = cursor.fetchone()[0]
            cursor.execute(
                f"""SELECT p.id, p.title, p.path, p.kind, snippet({table}, 1, %s, %s, '...', 24)
                    FROM {table} JOIN content_page p ON p.id = {table}.rowid
                    WHERE {table} MATCH %s AND {table}.site = %s
                    ORDER BY bm25({table}, 10.0, 1.0) LIMIT %s OFFSET %s""",
                [START, END, match, site, limit, offset])
            return total, [_result(*row) for row in cursor.fetchall()]
        # one or two characters (水, 水を): the trigram index cannot answer, so scan the Japanese pages
        likes = " AND ".join(["body LIKE %s ESCAPE '\\'"] * len(terms))
        patterns = ["%" + t.replace("\\", "\\\\").replace("%", "\\%").replace("_", "\\_") + "%" for t in terms]
        cursor.execute(f"SELECT count(*) FROM {table} WHERE site = %s AND {likes}", [site, *patterns])
        total = cursor.fetchone()[0]
        cursor.execute(
            f"""SELECT p.id, p.title, p.path, p.kind, {table}.body
                FROM {table} JOIN content_page p ON p.id = {table}.rowid
                WHERE {table}.site = %s AND {likes.replace('body', table + '.body')}
                ORDER BY p.path LIMIT %s OFFSET %s""",
            [site, *patterns, limit, offset])
        rows = cursor.fetchall()
    results = []
    for pid, title, path, kind, body in rows:
        at = body.find(terms[0])
        piece = body[max(0, at - 30): at + 60]
        marked = piece.replace(terms[0], START + terms[0] + END)
        results.append(_result(pid, title, path, kind, marked))
    return total, results


def _result(pid, title, path, kind, snippet):
    return {"id": pid, "title": title, "path": path, "kind": kind,
            "url_path": "/" + path[: -len(".html")] + "/", "snippet": _highlight(snippet)}

Save as apps/search/views.py:

from urllib.parse import urlencode

from django.shortcuts import render

from .query import search

PER_PAGE = 20


def results(request, site):
    text = request.GET.get("q", "").strip()[:100]
    try:
        page_number = max(1, int(request.GET.get("page", "1")))
    except ValueError:
        page_number = 1
    total, found = search(site, text, limit=PER_PAGE, offset=(page_number - 1) * PER_PAGE)
    last_page = max(1, -(-total // PER_PAGE))
    return render(request, "search/results.html", {
        "q": text, "results": found, "total": total, "page_number": page_number,
        "previous_url": "?" + urlencode({"q": text, "page": page_number - 1}) if page_number > 1 else None,
        "next_url": "?" + urlencode({"q": text, "page": page_number + 1}) if page_number < last_page else None,
        "too_short": bool(text) and total == 0 and len(text) < 2,
        "breadcrumbs": [], "active_folder": None,
    })

Save as apps/search/templates/search/results.html:

{% extends "theme/base.html" %}
{% block title %}Search{% if q %}: {{ q }}{% endif %}{% endblock %}
{% block content %}
  <h1 class="page-title">Search {{ site_title }}</h1>
  <form class="search-form" role="search" action="/search/" method="get">
    <label for="search-q">Search this site</label>
    <input id="search-q" name="q" type="search" value="{{ q }}" maxlength="100" autofocus>
    <button type="submit">Search</button>
  </form>

  {% if q %}
    <p class="search-count">{{ total }} result{{ total|pluralize }} for <strong>{{ q }}</strong></p>
    <ol class="search-results">
      {% for r in results %}
      <li>
        <a href="{{ r.url_path }}">{{ r.title }}</a>
        <p class="search-snippet">{{ r.snippet }}</p>
        <p class="search-path">{{ r.url_path }}</p>
      </li>
      {% empty %}
      <li class="search-none">Nothing found. Try fewer or shorter words.</li>
      {% endfor %}
    </ol>
    {% if previous_url or next_url %}
    <nav class="page-nav" aria-label="More results">
      {% if previous_url %}<a class="page-nav__prev" href="{{ previous_url }}"><span>&larr; Previous</span> page {{ page_number|add:"-1" }}</a>{% else %}<span></span>{% endif %}
      {% if next_url %}<a class="page-nav__next" href="{{ next_url }}"><span>Next &rarr;</span> page {{ page_number|add:"1" }}</a>{% endif %}
    </nav>
    {% endif %}
  {% else %}
    <p class="lead">Type a word or two. Japanese needs at least one character; for English, Hungarian, German and French the last word may be just the beginning of a word.</p>
  {% endif %}
{% endblock %}

Two details that matter:
- Snippets: the match is wrapped in control characters (they cannot occur in page
  text), the whole snippet is HTML-escaped, and only then are the markers turned into
  <mark> tags. Page text that contains <script> is shown as text.
- Ranking: bm25 with the title weighted 10 times the body, so a page called
  "Clothes" outranks a page that merely repeats the word.

Save as tests/test_search.py:

from django.core.cache import cache
from django.test import Client, SimpleTestCase, TestCase

from apps.content import importer
from apps.content.models import Page
from apps.search.indexing import index_pages, remove_pages
from apps.search.query import search, word_query
from apps.search.text import has_cjk, html_to_text
from tests.test_importer import ContentTree


def add(path, site, title, body_html, summary=""):
    page = Page.objects.create(path=path, site=site, kind="lesson", title=title, fragment=body_html, summary=summary)
    index_pages([(page, html_to_text(body_html))])
    return page


class TextTests(SimpleTestCase):
    def test_tags_are_removed_and_entities_decoded(self):
        self.assertEqual(html_to_text("<p>Fit &amp; <b>Returns</b></p>"), "Fit & Returns")

    def test_scripts_and_styles_are_not_content(self):
        self.assertEqual(html_to_text("<style>.a{color:red}</style><p>word</p><script>var secret = 1;</script>"), "word")

    def test_block_tags_separate_words(self):
        self.assertEqual(html_to_text("<td>one</td><td>two</td><li>three</li>"), "one two three")

    def test_japanese_is_detected(self):
        self.assertTrue(has_cjk("ありがとう"))
        self.assertTrue(has_cjk("水"))
        self.assertFalse(has_cjk("Szeretnék a kávét"))


class QueryBuildingTests(SimpleTestCase):
    def test_every_word_must_match_and_the_last_may_be_a_beginning(self):
        self.assertEqual(word_query("polite cond"), '"polite" "cond"*')

    def test_punctuation_and_operators_are_just_text(self):
        self.assertEqual(word_query('"; DROP TABLE x; -- OR NEAR('), '"DROP" "TABLE" "x" "OR" "NEAR"*')

    def test_nothing_searchable_gives_an_empty_query(self):
        self.assertEqual(word_query("  !!! ??? "), "")


class SearchTests(TestCase):
    def setUp(self):
        add("hungary/a.html", "languages", "Buying clothes", "<p>Szeretnék egy kabátot. The polite conditional.</p>")
        add("germany/b.html", "languages", "Beim Bäcker", "<p>Ich möchte ein Brötchen, bitte.</p>")
        add("japan/c.html", "languages", "Greetings", "<p>ありがとうございます。水をください。</p>")
        add("linux/d.html", "systems", "Installing Debian", "<p>Choose a partition scheme for the conditional install.</p>")

    def titles(self, site, text):
        return [r["title"] for r in search(site, text)[1]]

    def test_a_word_finds_its_page(self):
        self.assertEqual(self.titles("languages", "conditional"), ["Buying clothes"])

    def test_a_site_only_searches_its_own_pages(self):
        self.assertEqual(self.titles("systems", "conditional"), ["Installing Debian"])
        self.assertEqual(self.titles("systems", "Szeretnék"), [])

    def test_accents_do_not_matter(self):
        self.assertEqual(self.titles("languages", "szeretnek"), ["Buying clothes"])
        self.assertEqual(self.titles("languages", "brotchen"), ["Beim Bäcker"])
        self.assertEqual(self.titles("languages", "Backer"), ["Beim Bäcker"])

    def test_the_last_word_may_be_a_beginning(self):
        self.assertEqual(self.titles("languages", "polite cond"), ["Buying clothes"])

    def test_a_title_match_ranks_above_a_body_match(self):
        add("hungary/e.html", "languages", "Clothes", "<p>about nothing</p>")
        add("hungary/f.html", "languages", "Other", "<p>clothes clothes</p>")
        self.assertEqual(self.titles("languages", "clothes")[0], "Clothes")

    def test_japanese_is_found_by_three_or_more_characters(self):
        self.assertEqual(self.titles("languages", "ありがとう"), ["Greetings"])
        self.assertEqual(self.titles("languages", "ください"), ["Greetings"])

    def test_one_or_two_japanese_characters_are_found_by_a_scan(self):
        self.assertEqual(self.titles("languages", "水"), ["Greetings"])
        self.assertEqual(self.titles("languages", "水を"), ["Greetings"])

    def test_unsafe_text_is_a_search_not_an_error(self):
        for text in ('"', '" OR 1=1 --', "NEAR(", "a AND", "(((", "*", "title:x", "-x", "%", "_", "'"):
            search("languages", text)                           # must not raise

    def test_a_snippet_marks_the_match_and_never_lets_page_text_inject_html(self):
        add("hungary/x.html", "languages", "Trap", "<p>use &lt;script&gt;alert(1)&lt;/script&gt; and the word zebra here</p>")
        snippet = str(search("languages", "zebra")[1][0]["snippet"])
        self.assertIn("<mark>zebra</mark>", snippet)
        self.assertNotIn("<script>", snippet)
        self.assertIn("&lt;script&gt;", snippet)

    def test_reindexing_replaces_and_removing_deletes(self):
        page = Page.objects.get(path="hungary/a.html")
        page.fragment = "<p>entirely different words</p>"
        index_pages([(page, html_to_text(page.fragment))])
        self.assertEqual(self.titles("languages", "conditional"), [])
        self.assertEqual(self.titles("languages", "entirely"), ["Buying clothes"])
        remove_pages([page.pk])
        self.assertEqual(self.titles("languages", "entirely"), [])


class SearchPageTests(TestCase):
    def setUp(self):
        cache.clear()
        for n in range(25):
            add(f"hungary/p{n}.html", "languages", f"Page {n}", "<p>shared keyword here</p>")
        self.client = Client(HTTP_HOST="languages.localhost")

    def test_the_results_page_shows_the_count_and_a_link_to_each_page(self):
        html = self.client.get("/search/?q=keyword").content.decode()
        self.assertIn("25 results for", html)
        self.assertIn('href="/hungary/p0/"', html)

    def test_results_come_twenty_at_a_time_with_a_next_link(self):
        first = self.client.get("/search/?q=keyword").content.decode()
        self.assertEqual(first.count('class="search-snippet"'), 20)
        self.assertIn("?q=keyword&amp;page=2", first)
        second = self.client.get("/search/?q=keyword&page=2").content.decode()
        self.assertEqual(second.count('class="search-snippet"'), 5)
        self.assertNotIn("Next", second.split('aria-label="More results"')[-1].split("</nav>")[0] if "More results" in second else "")

    def test_a_bad_page_number_is_page_one(self):
        self.assertEqual(self.client.get("/search/?q=keyword&page=abc").status_code, 200)

    def test_an_empty_search_shows_the_help_text_and_the_query_is_escaped(self):
        self.assertIn("Type a word", self.client.get("/search/").content.decode())
        html = self.client.get('/search/?q="><script>x</script>').content.decode()
        self.assertNotIn("<script>x</script>", html)

    def test_every_page_has_a_search_box_that_goes_to_this_sites_search(self):
        html = self.client.get("/").content.decode()
        self.assertIn('action="/search/"', html)
        self.assertIn('role="search"', html)

    def test_the_search_of_another_site_does_not_see_these_pages(self):
        html = Client(HTTP_HOST="systems.localhost").get("/search/?q=keyword").content.decode()
        self.assertIn("0 results", html)


class ImporterIndexesTests(TestCase):
    def test_an_import_makes_pages_searchable_and_a_prune_removes_them(self):
        tree = ContentTree(self)
        tree.write("hungary/x/a.html", "<p>An aardvark appears.</p>")
        tree.write("hungary/x/b.html", "<p>A bonobo appears.</p>")
        importer.import_tree(tree.root)
        self.assertEqual([r["title"] for r in search("languages", "aardvark")[1]], ["A"])
        tree.remove("hungary/x/a.html")
        importer.import_tree(tree.root, prune=True)
        self.assertEqual(search("languages", "aardvark"), (0, []))
        self.assertEqual(search("languages", "bonobo")[0], 1)

    def test_an_edited_page_is_reindexed(self):
        tree = ContentTree(self)
        tree.write("hungary/x/a.html", "<p>first wording</p>")
        importer.import_tree(tree.root)
        tree.write("hungary/x/a.html", "<p>second wording</p>")
        importer.import_tree(tree.root)
        self.assertEqual(search("languages", "first")[0], 0)
        self.assertEqual(search("languages", "second")[0], 1)

Real searches, on the real database (each site only searches its own pages):

[languages] 'szeretnek': 18 results, 29 ms -> ['Buying Clothes: Sizes, Fit & Returns', 'Post Office, Bank & Paying Bills', 'Booking a Hotel']
[languages] 'Szeretnék': 18 results, 3 ms -> ['Buying Clothes: Sizes, Fit & Returns', 'Post Office, Bank & Paying Bills', 'Booking a Hotel']
[languages] 'brotchen': 1 results, 3 ms -> ['Ordering Food & Drink at a Café']
[languages] 'polite conditional': 7 results, 7 ms -> ['Wishes, Hopes & What You Would Do', 'Buying Clothes: Sizes, Fit & Returns', 'Giving Advice & Making Suggestions']
[systems] 'partition': 44 results, 39 ms -> ['Multi-Boot With More Than Two OSes & Advanced Partitioning', 'MBR vs. GPT and BIOS vs. UEFI', 'The Installation Process']
[systems] 'systemd unit': 39 results, 24 ms -> ['systemd-analyze & Troubleshooting Boot', 'Unit Files — The Basic Building Block', 'systemd in Depth — Complete Course']
[languages] 'ありがとう': 25 results, 14 ms -> ['Za Ji Zu', '🗺️ Asking for Directions — 場所をたずねる', 'Greetings & Goodbyes']
[languages] '水': 5 results, 8 ms -> ['Days, Dates & Time', 'Ordering Food & Drink at a Café', '🐾 Counting Animals — どうぶつをかぞえる']
[languages] 'ありが': 25 results, 3 ms -> ['Za Ji Zu', '🗺️ Asking for Directions — 場所をたずねる', 'Greetings & Goodbyes']
[programming] 'decorat': 63 results, 59 ms -> ['Decorators', 'Structural Patterns II: Decorator & Composite', 'Caching']
[ai] 'zzzqqq': 0 results, 1 ms -> []
[languages] '"; DROP TABLE content_page; --': 0 results, 7 ms -> []
[webdevelopment] 'grid': 63 results, 50 ms -> ['Deep Dive — Grid', 'Grid Fundamentals', 'CSS Grid']
[languages] '': 0 results, 0 ms -> []

The page itself (languages site, q=szeretnek): status 200 with 18 result snippets.
A screenshot in headless Chrome shows the highlighted word inside each snippet.

What was not verified: a one- or two-character Japanese search is a scan of the 421
Japanese pages (5 ms now; it slows down if that number grows a lot). The PostgreSQL
way (a generated tsvector column with a GIN index, and SearchQuery/SearchRank) is
described in the chapter but was not run, because this project uses SQLite.

WHY THIS WORKS AS AN ANSWER
---------------------------
The tests include hostile input (a quote, OR 1=1, NEAR(, unbalanced brackets, an
asterisk, a column filter) and require "no error", and the snippet test proves page
text cannot inject markup.