"""Minimal HTTP API for the normalized legal-document corpus.""" from __future__ import annotations import argparse import datetime import json import re import urllib.parse from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from pathlib import Path from search.minjust_opensearch import APP_VERSION, request_json from search.catalog import CATALOGS, labels from search.reviews import ReviewSnapshots, ReviewStore API_VERSION = "v1" SEARCH_ALGORITHM_VERSION = "search-1" LANGUAGES = {"ru", "ky"} CODE = re.compile(r"^[0-9]+$") MAX_PAGE_SIZE = 100 MAX_RESULT_WINDOW = 10_000 class ApiError(Exception): def __init__(self, status: int, message: str): self.status = status self.message = message def parse_positive(value: str | None, name: str, default: int, maximum: int) -> int: if value is None: return default try: parsed = int(value) except ValueError as error: raise ApiError(400, f"{name} must be an integer") from error if not 1 <= parsed <= maximum: raise ApiError(400, f"{name} must be between 1 and {maximum}") return parsed def one(query: dict[str, list[str]], name: str) -> str | None: values = query.get(name, []) if len(values) > 1: raise ApiError(400, f"{name} must be specified once") return values[0] if values else None def date(value: str | None, name: str) -> str | None: if value is None: return None try: datetime.date.fromisoformat(value) except ValueError as error: raise ApiError(400, f"{name} must be an ISO date") from error return value def openapi() -> dict: responses = {"200": {"description": "Successful response"}, "400": {"description": "Invalid request"}, "404": {"description": "Not found"}, "502": {"description": "Search backend unavailable"}} return { "openapi": "3.0.3", "info": {"title": "Akyldash Search API", "version": API_VERSION}, "paths": { "/search": {"get": {"responses": responses, "parameters": [ {"name": "q", "in": "query", "required": True, "schema": {"type": "string"}}, {"name": "language", "in": "query", "schema": {"type": "string", "enum": ["ru", "ky"]}}, {"name": "page", "in": "query", "schema": {"type": "integer", "minimum": 1}}, {"name": "page_size", "in": "query", "schema": {"type": "integer", "minimum": 1, "maximum": MAX_PAGE_SIZE}}, {"name": "document_type", "in": "query", "schema": {"type": "string"}}, {"name": "status", "in": "query", "schema": {"type": "string"}}, {"name": "authority", "in": "query", "schema": {"type": "string"}}, {"name": "legal_force", "in": "query", "schema": {"type": "string"}}, {"name": "date_from", "in": "query", "schema": {"type": "string", "format": "date"}}, {"name": "date_to", "in": "query", "schema": {"type": "string", "format": "date"}}, {"name": "sort", "in": "query", "schema": {"type": "string", "enum": ["relevance", "date"]}}, ]}}, "/search/filters": {"get": {"responses": responses}}, "/search-reviews": {"post": {"responses": {"201": {"description": "Review saved"}, "400": {"description": "Invalid review"}}}}, "/search-reviews/export": {"get": {"responses": responses}}, "/review": {"get": {"responses": {"200": {"description": "Review interface"}}}}, "/documents/{code}": {"get": {"responses": responses, "parameters": [{"name": "code", "in": "path", "required": True, "schema": {"type": "string", "pattern": "^[0-9]+$"}}]}}, "/documents/{code}/editions": {"get": {"responses": responses, "parameters": [{"name": "code", "in": "path", "required": True, "schema": {"type": "string", "pattern": "^[0-9]+$"}}]}}, "/documents/{code}/editions/{edition}": {"get": {"responses": responses, "parameters": [{"name": "code", "in": "path", "required": True, "schema": {"type": "string", "pattern": "^[0-9]+$"}}, {"name": "edition", "in": "path", "required": True, "schema": {"type": "string", "pattern": "^[0-9]+$"}}]}}, }, } class Api: def __init__(self, base_url: str, index: str, data_root: Path, reviews_db: Path | str = ":memory:", review_secret: bytes | None = None): self.base_url = base_url.rstrip("/") self.index = index self.data_root = data_root self.review_store = ReviewStore(reviews_db) self.review_snapshots = ReviewSnapshots(review_secret) def search_url(self, suffix: str) -> str: return f"{self.base_url}/{urllib.parse.quote(self.index, safe='')}/{suffix}" def query_opensearch(self, body: dict) -> dict: try: return request_json(self.search_url("_search"), "POST", json.dumps(body, ensure_ascii=False).encode(), "application/json") except RuntimeError as error: raise ApiError(502, "search backend is unavailable") from error def search(self, query: dict[str, list[str]]) -> dict: text = one(query, "q") if not text or not text.strip(): raise ApiError(400, "q is required") if len(text) > 500: raise ApiError(400, "q must not exceed 500 characters") language = one(query, "language") or "ru" if language not in LANGUAGES: raise ApiError(400, "language must be ru or ky") page = parse_positive(one(query, "page"), "page", 1, 1_000_000) page_size = parse_positive(one(query, "page_size"), "page_size", 20, MAX_PAGE_SIZE) if page * page_size >= MAX_RESULT_WINDOW: raise ApiError(400, f"page and page_size must stay within {MAX_RESULT_WINDOW} results") sort = one(query, "sort") or "relevance" if sort not in {"relevance", "date"}: raise ApiError(400, "sort must be relevance or date") filters: list[dict] = [{"term": {"language": language}}, {"term": {"is_current_edition": True}}] fields = {"document_type": "document_type_code", "status": "status_code", "authority": "authority_codes", "legal_force": "legal_force_code"} for parameter, field in fields.items(): value = one(query, parameter) if value: if value not in CATALOGS[parameter]: raise ApiError(400, f"{parameter} must be a catalog code") filters.append({"term": {field: value}}) date_from, date_to = date(one(query, "date_from"), "date_from"), date(one(query, "date_to"), "date_to") if date_from and date_to and date_from > date_to: raise ApiError(400, "date_from must not be later than date_to") if date_from or date_to: date_range = {key: value for key, value in (("gte", date_from), ("lte", date_to)) if value} filters.append({"range": {"date_adopted": date_range}}) body = { "from": (page - 1) * page_size, "size": page_size + 1, "_source": ["document_code", "edition_code", "document_name_ru", "document_name_ky", "document_type_ru", "document_type_ky", "status_ru", "status_ky", "legal_force_code", "date_adopted", "number"], "query": {"bool": {"filter": filters, "must": {"multi_match": {"query": text, "fields": [f"document_name_{language}", f"text_{language}"], "type": "cross_fields"}}}}, "collapse": {"field": "document_code"}, "highlight": {"fields": {f"text_{language}": {"number_of_fragments": 1}}}, } if sort == "date": body["sort"] = [{"date_adopted": "desc"}, {"_score": "desc"}] response = self.query_opensearch(body) try: hits = response["hits"]["hits"] except (KeyError, TypeError) as error: raise ApiError(502, "search backend returned an incomplete response") from error results = [self.search_hit(hit, language) for hit in hits[:page_size]] snapshot_results = [{"rank": rank, **result} for rank, result in enumerate(results, 1)] concrete_indexes = {hit.get("_index") for hit in hits[:page_size] if hit.get("_index")} index_name = next(iter(concrete_indexes)) if len(concrete_indexes) == 1 else self.index return {"api_version": API_VERSION, "query": text, "language": language, "page": page, "page_size": page_size, "has_next": len(hits) > page_size, "results": results, "review_token": self.review_snapshots.create(text, language, index_name, snapshot_results, SEARCH_ALGORITHM_VERSION)} @staticmethod def search_hit(hit: dict, language: str) -> dict: source = hit.get("_source") if not isinstance(source, dict) or not source.get("document_code"): raise ApiError(502, "search backend returned an incomplete result") highlight = hit.get("highlight", {}).get(f"text_{language}", []) legal_force = source.get("legal_force_code") return {"code": source["document_code"], "edition": source.get("edition_code"), "name": source.get(f"document_name_{language}"), "type": source.get(f"document_type_{language}"), "status": source.get(f"status_{language}"), "legal_force": labels("legal_force", legal_force)[language] if legal_force else None, "date_adopted": source.get("date_adopted"), "number": source.get("number"), "snippet": highlight[0] if highlight else None} def filters(self, query: dict[str, list[str]]) -> dict: language = one(query, "language") or "ru" if language not in LANGUAGES: raise ApiError(400, "language must be ru or ky") fields = {"document_types": ("document_type", "document_type_code"), "statuses": ("status", "status_code"), "authorities": ("authority", "authority_codes"), "legal_forces": ("legal_force", "legal_force_code")} body = {"size": 0, "query": {"term": {"is_current_edition": True}}, "aggs": {name: {"terms": {"field": pair[1], "size": 1000}, "aggs": {"documents": {"cardinality": {"field": "document_code", "precision_threshold": 40000}}}} for name, pair in fields.items()}} response = self.query_opensearch(body) try: aggregations = response["aggregations"] values = { name: [{"code": item["key"], "labels": labels(pair[0], item["key"]), "count": item["documents"]["value"]} for item in aggregations[name]["buckets"]] for name, pair in fields.items() } except (KeyError, TypeError) as error: raise ApiError(502, "search backend returned incomplete filters") from error return {"api_version": API_VERSION, "language": language, **values} def directory(self, code: str) -> Path: if not CODE.fullmatch(code): raise ApiError(404, "document not found") path = self.data_root / "documents" / code if not path.is_dir(): raise ApiError(404, "document not found") return path @staticmethod def read_json(path: Path, message: str) -> dict: try: value = json.loads(path.read_text(encoding="utf-8")) except (OSError, UnicodeError, json.JSONDecodeError) as error: raise ApiError(500, message) from error if not isinstance(value, dict): raise ApiError(500, message) return value def document(self, code: str) -> dict: document = self.read_json(self.directory(code) / "document.json", "document data is unavailable") editions = document.get("editions") if not isinstance(editions, list): raise ApiError(500, "document data is unavailable") return {"api_version": API_VERSION, "document": document, "current_edition": editions[-1] if editions else None} def editions(self, code: str) -> dict: document = self.read_json(self.directory(code) / "document.json", "document data is unavailable") return {"api_version": API_VERSION, "code": code, "available_languages": document.get("available_languages", []), "editions": document.get("editions", [])} def edition(self, code: str, edition: str, query: dict[str, list[str]]) -> dict: if not CODE.fullmatch(edition): raise ApiError(404, "edition not found") directory = self.directory(code) / "editions" / edition if not directory.is_dir(): raise ApiError(404, "edition not found") metadata = self.read_json(directory / "edition.json", "edition data is unavailable") language = one(query, "language") if language is not None and language not in LANGUAGES: raise ApiError(400, "language must be ru or ky") languages = [language] if language else metadata.get("available_languages", []) content = {} for item in languages: if item not in metadata.get("available_languages", []): continue try: content[item] = {"html": (directory / item / "content.html").read_text(encoding="utf-8"), "text": (directory / item / "content.txt").read_text(encoding="utf-8")} except (OSError, UnicodeError) as error: raise ApiError(500, "edition content is unavailable") from error if language and language not in content: raise ApiError(404, "edition language not found") return {"api_version": API_VERSION, "edition": metadata, "content": content} def save_review(self, body: dict) -> dict: if not isinstance(body, dict): raise ApiError(400, "request body must be an object") try: snapshot = self.review_snapshots.verify(body["review_token"]) reviewer = body["reviewer"].strip() overall_comment = body.get("overall_comment", "").strip() submitted = body["results"] except (KeyError, AttributeError, TypeError, ValueError) as error: raise ApiError(400, "review_token, reviewer and results are required") from error if not reviewer or len(reviewer) > 120: raise ApiError(400, "reviewer must be between 1 and 120 characters") if len(overall_comment) > 4000: raise ApiError(400, "overall_comment is too long") if not isinstance(submitted, list): raise ApiError(400, "results must be an array") by_rank = {item["rank"]: item for item in snapshot["results"]} if len(submitted) != len(by_rank) or {item.get("rank") for item in submitted if isinstance(item, dict)} != set(by_rank): raise ApiError(400, "all search results must be reviewed exactly once") results = [] for item in submitted: if not isinstance(item, dict) or not isinstance(item.get("rank"), int) or item["rank"] not in by_rank: raise ApiError(400, "review result rank is invalid") source = by_rank[item["rank"]] if item.get("code") != source["code"]: raise ApiError(400, "review result document does not match the search snapshot") rating = item.get("rating") if rating is not None and (isinstance(rating, bool) or not isinstance(rating, int) or not 0 <= rating <= 3): raise ApiError(400, "rating must be an integer from 0 to 3") comment = item.get("comment", "") if not isinstance(comment, str) or len(comment) > 4000: raise ApiError(400, "result comment is too long") results.append({**source, "rating": rating, "comment": comment.strip()}) if not results: raise ApiError(400, "at least one result must be reviewed") review = {"created_at": datetime.datetime.now(datetime.timezone.utc).isoformat(), "reviewer": reviewer, "query": snapshot["query"], "language": snapshot["language"], "index_name": snapshot["index_name"], "algorithm_version": snapshot["algorithm_version"], "top_result_code": snapshot["results"][0]["code"] if snapshot["results"] else None, "results": results, "overall_comment": overall_comment} review_id = self.review_store.save(review) return {"api_version": API_VERSION, "id": review_id, "created_at": review["created_at"]} @staticmethod def review_page() -> str: try: return Path(__file__).with_name("review.html").read_text(encoding="utf-8") except (OSError, UnicodeError) as error: raise ApiError(500, "review interface is unavailable") from error def handle(self, method: str, path: str, body: dict | None = None) -> tuple[int, dict]: parsed = urllib.parse.urlsplit(path) query = urllib.parse.parse_qs(parsed.query, keep_blank_values=True) parts = [urllib.parse.unquote(part) for part in parsed.path.split("/") if part] if method == "POST" and parts == ["search-reviews"]: return 201, self.save_review(body) if method == "GET" and parts == ["search-reviews", "export"]: return 200, {"api_version": API_VERSION, "reviews": self.review_store.export()} if method != "GET": raise ApiError(405, "method not allowed") if parts == ["openapi.json"]: return 200, openapi() if parts == ["search"]: return 200, self.search(query) if parts == ["search", "filters"]: return 200, self.filters(query) if len(parts) == 2 and parts[0] == "documents": return 200, self.document(parts[1]) if len(parts) == 3 and parts[:1] == ["documents"] and parts[2] == "editions": return 200, self.editions(parts[1]) if len(parts) == 4 and parts[:1] == ["documents"] and parts[2] == "editions": return 200, self.edition(parts[1], parts[3], query) raise ApiError(404, "endpoint not found") def handler(api: Api): class RequestHandler(BaseHTTPRequestHandler): def respond(self, method: str): try: if method == "GET" and urllib.parse.urlsplit(self.path).path == "/review": body = api.review_page().encode() self.send_response(200) self.send_header("Content-Type", "text/html; charset=utf-8") self.send_header("Content-Length", str(len(body))) self.end_headers() self.wfile.write(body) return body = None if method == "POST": length = int(self.headers.get("Content-Length", "0")) if length > 1_000_000: raise ApiError(413, "request body is too large") body = json.loads(self.rfile.read(length) or b"{}") status, payload = api.handle(method, self.path, body) except json.JSONDecodeError: status, payload = 400, {"api_version": API_VERSION, "error": "request body must be valid JSON"} except ApiError as error: status, payload = error.status, {"api_version": API_VERSION, "error": error.message} body = json.dumps(payload, ensure_ascii=False).encode() self.send_response(status) self.send_header("Content-Type", "application/json; charset=utf-8") self.send_header("Content-Length", str(len(body))) self.end_headers() self.wfile.write(body) def do_GET(self): self.respond("GET") def do_POST(self): self.respond("POST") def log_message(self, format: str, *args): return return RequestHandler def main() -> int: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--url", default="http://127.0.0.1:9200") parser.add_argument("--index", default="akyldash-fragments-current") parser.add_argument("--data", type=Path, default=Path("data/minjust-normalized")) parser.add_argument("--reviews-db", type=Path, default=Path("data/search-reviews.sqlite3")) parser.add_argument("--review-secret", default=None) parser.add_argument("--host", default="127.0.0.1") parser.add_argument("--port", type=int, default=8080) parser.add_argument("--version", action="version", version=APP_VERSION) arguments = parser.parse_args() secret = arguments.review_secret.encode() if arguments.review_secret else None ThreadingHTTPServer((arguments.host, arguments.port), handler(Api(arguments.url, arguments.index, arguments.data, arguments.reviews_db, secret))).serve_forever() return 0 if __name__ == "__main__": raise SystemExit(main())