fix: expose curated search query

2026-08-20 15:08:42 +03:00
parent dc5399de0a
commit e3f08426d5
4 changed files with 117 additions and 72 deletions
--- a/backend/README.md
+++ b/backend/README.md
@@ -189,6 +189,13 @@ PYTHONPATH=backend python3 -m search.evaluate_relevance \
 Менять веса или анализаторы следует только после фиксации этого baseline и
 разбора ошибок выдачи.

+Для проверки текущей выдачи без будущего HTTP API используйте CLI:
+
+```bash
+PYTHONPATH=backend python3 -m search.query "как открыть ОсОО" --language ru
+PYTHONPATH=backend python3 -m search.query "ЖЧК ачуу тартиби" --language ky
+```
+
 ---

 Акылдаш · Backend v0.5.2 · Frontend — не создан
--- a/backend/search/evaluate_relevance.py
+++ b/backend/search/evaluate_relevance.py
@@ -5,64 +5,10 @@ from __future__ import annotations
 import argparse
 import json
 import sys
-import urllib.parse
 from pathlib import Path

-from search.minjust_opensearch import APP_VERSION, request_json
-
-
-def company_registration_clauses(language: str, query: str) -> list[dict]:
-    normalized = query.casefold()
-    intent_terms = {
-        "ru": (("осоо",), ("откр", "созд", "зарегистр")),
-        "ky": (("жчк",), ("ач", "түз", "катто")),
-    }
-    entity_terms, action_terms = intent_terms[language]
-    if not any(term in normalized for term in entity_terms) or not any(term in normalized for term in action_terms):
-        return []
-
-    status = {"ru": "Действует", "ky": "Күчүндө"}[language]
-
-    # ponytail: curated legal mapping; replace with a reviewed intent catalog when coverage expands.
-    def clause(document_code: str, boost: int) -> dict:
-        return {
-            "constant_score": {
-                "filter": {
-                    "bool": {
-                        "filter": [
-                            {"term": {"document_code": document_code}},
-                            {"term": {f"status_{language}": status}},
-                        ]
-                    }
-                },
-                "boost": boost,
-            }
-        }
-
-    return [clause("230044970", 2000), clause("667", 1000)]
-
-
-def search_body(language: str, query: str, top_k: int) -> bytes:
-    full_text = {
-        "multi_match": {
-            "query": query,
-            "fields": [f"document_name_{language}", f"text_{language}"],
-            "type": "cross_fields",
-        }
-    }
-    clauses = company_registration_clauses(language, query)
-    bool_query = {"filter": {"term": {"language": language}}}
-    if clauses:
-        bool_query.update({"should": [full_text, *clauses], "minimum_should_match": 1})
-    else:
-        bool_query["must"] = full_text
-    return json.dumps({
-        "size": top_k,
-        "track_total_hits": False,
-        "_source": ["document_code"],
-        "query": {"bool": bool_query},
-        "collapse": {"field": "document_code"},
-    }, ensure_ascii=False).encode()
+from search.minjust_opensearch import APP_VERSION
+from search.query import search_documents


 def load_queries(path: Path) -> list[dict]:
@@ -97,14 +43,7 @@ def load_queries(path: Path) -> list[dict]:


 def search(base_url: str, index: str, item: dict, top_k: int) -> list[str]:
-    language = item["language"]
-    body = search_body(language, item["query"], top_k)
-    url = f"{base_url.rstrip('/')}/{urllib.parse.quote(index, safe='')}/_search"
-    response = request_json(url, "POST", body, "application/json")
-    try:
-        return [hit["_source"]["document_code"] for hit in response["hits"]["hits"]]
-    except (KeyError, TypeError) as error:
-        raise RuntimeError("OpenSearch search response is incomplete") from error
+    return search_documents(base_url, index, item["language"], item["query"], top_k)


 def evaluate(queries: list[dict], base_url: str, index: str, top_k: int) -> dict:
--- a/backend/search/query.py
+++ b/backend/search/query.py
@@ -0,0 +1,93 @@
+"""Run document searches against the local OpenSearch index."""
+
+from __future__ import annotations
+
+import argparse
+import json
+import re
+import urllib.parse
+
+from search.minjust_opensearch import APP_VERSION, request_json
+
+
+def company_registration_clauses(language: str, query: str) -> list[dict]:
+    tokens = set(re.findall(r"\w+", query.casefold()))
+    intent = {
+        "ru": ("осоо", ("как", "порядок", "процедура"), ("откр", "созд", "зарегистр"), "Действует"),
+        "ky": ("жчк", ("кантип", "тартиби"), ("ач", "түз", "кат"), "Күчүндө"),
+    }[language]
+    entity, procedure_terms, action_prefixes, status = intent
+    if entity not in tokens or not tokens.intersection(procedure_terms) or not any(
+        token.startswith(prefix) for token in tokens for prefix in action_prefixes
+    ):
+        return []
+
+    # ponytail: curated legal mapping; replace with a reviewed intent catalog when coverage expands.
+    def clause(document_code: str, boost: int) -> dict:
+        return {
+            "constant_score": {
+                "filter": {
+                    "bool": {
+                        "filter": [
+                            {"term": {"document_code": document_code}},
+                            {"term": {f"status_{language}": status}},
+                        ]
+                    }
+                },
+                "boost": boost,
+            }
+        }
+
+    return [clause("230044970", 2000), clause("667", 1000)]
+
+
+def build_search_body(language: str, query: str, top_k: int) -> bytes:
+    full_text = {
+        "multi_match": {
+            "query": query,
+            "fields": [f"document_name_{language}", f"text_{language}"],
+            "type": "cross_fields",
+        }
+    }
+    clauses = company_registration_clauses(language, query)
+    bool_query = {"filter": {"term": {"language": language}}}
+    if clauses:
+        bool_query.update({"should": [full_text, *clauses], "minimum_should_match": 1})
+    else:
+        bool_query["must"] = full_text
+    return json.dumps({
+        "size": top_k,
+        "track_total_hits": False,
+        "_source": ["document_code"],
+        "query": {"bool": bool_query},
+        "collapse": {"field": "document_code"},
+    }, ensure_ascii=False).encode()
+
+
+def search_documents(base_url: str, index: str, language: str, query: str, top_k: int) -> list[str]:
+    url = f"{base_url.rstrip('/')}/{urllib.parse.quote(index, safe='')}/_search"
+    response = request_json(url, "POST", build_search_body(language, query, top_k), "application/json")
+    try:
+        return [hit["_source"]["document_code"] for hit in response["hits"]["hits"]]
+    except (KeyError, TypeError) as error:
+        raise RuntimeError("OpenSearch search response is incomplete") from error
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("query")
+    parser.add_argument("--language", choices=("ru", "ky"), required=True)
+    parser.add_argument("--url", default="http://127.0.0.1:9200")
+    parser.add_argument("--index", default="akyldash-fragments-v1")
+    parser.add_argument("--top-k", type=int, default=10)
+    parser.add_argument("--version", action="version", version=APP_VERSION)
+    arguments = parser.parse_args()
+    if arguments.top_k <= 0:
+        raise SystemExit("--top-k must be greater than zero")
+    print(json.dumps(search_documents(arguments.url, arguments.index, arguments.language, arguments.query, arguments.top_k), ensure_ascii=False))
+    print(f"Akyldash Backend v{APP_VERSION} · Frontend — not created")
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())
--- a/backend/test_search_relevance.py
+++ b/backend/test_search_relevance.py
@@ -4,7 +4,8 @@ import unittest
 from pathlib import Path
 from unittest.mock import patch

-from search.evaluate_relevance import evaluate, load_queries, search_body
+from search.evaluate_relevance import evaluate, load_queries
+from search.query import build_search_body


 class SearchRelevanceTest(unittest.TestCase):
@@ -28,16 +29,12 @@ class SearchRelevanceTest(unittest.TestCase):
            path.write_text(json.dumps(queries, ensure_ascii=False), encoding="utf-8")
            loaded = load_queries(path)

-        responses = [
-            {"hits": {"hits": [{"_source": {"document_code": code}} for code in ["7", "10", "8"]]}},
-            {"hits": {"hits": [{"_source": {"document_code": code}} for code in ["10", "9"]]}},
-        ]
-        with patch("search.evaluate_relevance.request_json", side_effect=responses) as request:
+        with patch("search.evaluate_relevance.search_documents", side_effect=[["7", "10", "8"], ["10", "9"]]):
            result = evaluate(loaded, "http://127.0.0.1:9200", "test", 10)

        self.assertEqual(result["summary"], {"query_count": 2, "recall_at_10": 1.0, "mrr_at_10": 0.75})
        self.assertEqual(result["queries"][0]["reciprocal_rank_at_10"], 1.0)
-        body = json.loads(request.call_args_list[0].args[2])
+        body = json.loads(build_search_body("ru", "трудовой договор", 10))
        self.assertFalse(body["track_total_hits"])
        self.assertEqual(body["collapse"], {"field": "document_code"})
        self.assertEqual(body["query"]["bool"]["must"]["multi_match"]["type"], "cross_fields")
@@ -47,7 +44,7 @@ class SearchRelevanceTest(unittest.TestCase):
            ("ru", "как открыть ОсОО", "Действует"),
            ("ky", "ЖЧК ачуу тартиби", "Күчүндө"),
        ):
-            body = json.loads(search_body(language, query, 10))
+            body = json.loads(build_search_body(language, query, 10))
            search_query = body["query"]["bool"]
            self.assertEqual(search_query["minimum_should_match"], 1)
            boosts = [clause["constant_score"] for clause in search_query["should"][1:]]
@@ -58,6 +55,15 @@ class SearchRelevanceTest(unittest.TestCase):
            )
            self.assertTrue(all(item["filter"]["bool"]["filter"][1] == {"term": {f"status_{language}": status}} for item in boosts))

+    def test_company_registration_intent_ignores_non_procedural_queries(self):
+        for language, query in (
+            ("ru", "ОсОО зарегистрирован?"),
+            ("ru", "кто зарегистрировал ОсОО"),
+            ("ky", "ЖЧК ачык маалымат"),
+        ):
+            body = json.loads(build_search_body(language, query, 10))
+            self.assertNotIn("should", body["query"]["bool"])
+

 if __name__ == "__main__":
    unittest.main()