""" Elasticsearch index definition and (re)build helpers. Index design ------------ * One document per profile, denormalised: the fields a recruiter searches on (name, title, employer, skills, education, location) all live in the same document, so a query never needs a join. * Text fields are analysed with an English analyzer (lowercase, ASCII folding, stop words, stemming) so "engineering" matches "engineer". * Every field that is *filtered* or *aggregated* on has a ``keyword`` counterpart with a lowercase normaliser - analysed text cannot be used for exact term filters or facet counts. * ``full_name.autocomplete`` uses an edge n-gram analyzer for the type-ahead; the search analyzer is deliberately different (``standard``) so the query term is not itself n-grammed. * **No personal contact data is indexed.** E-mail addresses and phone numbers stay in PostgreSQL, which removes any chance of confirming someone's e-mail through a crafted search query. Rebuilds use an alias plus a timestamped physical index, so a reindex never leaves the API pointing at a half-populated index and can be rolled back. """ from __future__ import annotations import logging from collections.abc import Iterable, Iterator from datetime import UTC, datetime from typing import Any from django.conf import settings from elasticsearch import BadRequestError, Elasticsearch from elasticsearch.helpers import bulk from apps.profiles.models import Profile logger = logging.getLogger(__name__) INDEX_SETTINGS: dict[str, Any] = { "number_of_shards": 1, "number_of_replicas": 0, "refresh_interval": "1s", "max_result_window": 10_000, "analysis": { "filter": { "english_stop": {"type": "stop", "stopwords": "_english_"}, "english_stemmer": {"type": "stemmer", "language": "english"}, "english_possessive": {"type": "stemmer", "language": "possessive_english"}, "edge_ngrams": {"type": "edge_ngram", "min_gram": 2, "max_gram": 15}, }, "normalizer": { "lowercase_normalizer": { "type": "custom", "filter": ["lowercase", "asciifolding"], } }, "analyzer": { "english_text": { "type": "custom", "tokenizer": "standard", "filter": [ "lowercase", "asciifolding", "english_possessive", "english_stop", "english_stemmer", ], }, "autocomplete_index": { "type": "custom", "tokenizer": "standard", "filter": ["lowercase", "asciifolding", "edge_ngrams"], }, "autocomplete_search": { "type": "custom", "tokenizer": "standard", "filter": ["lowercase", "asciifolding"], }, }, }, } KEYWORD = {"type": "keyword", "normalizer": "lowercase_normalizer"} INDEX_MAPPINGS: dict[str, Any] = { "dynamic": "strict", # reject unexpected fields instead of guessing a type "properties": { "id": {"type": "integer"}, "full_name": { "type": "text", "analyzer": "english_text", "fields": { "keyword": KEYWORD, "autocomplete": { "type": "text", "analyzer": "autocomplete_index", "search_analyzer": "autocomplete_search", }, }, }, "headline": {"type": "text", "analyzer": "english_text"}, "job_title": { "type": "text", "analyzer": "english_text", "fields": {"keyword": KEYWORD}, }, "job_title_role": KEYWORD, "job_title_sub_role": KEYWORD, "seniority_levels": KEYWORD, "company": { "type": "text", "analyzer": "english_text", "fields": {"keyword": KEYWORD}, }, "past_companies": {"type": "text", "analyzer": "english_text"}, "industry": { "type": "text", "analyzer": "english_text", "fields": {"keyword": KEYWORD}, }, "skills": { "type": "text", "analyzer": "english_text", "fields": { "keyword": KEYWORD, "autocomplete": { "type": "text", "analyzer": "autocomplete_index", "search_analyzer": "autocomplete_search", }, }, }, "education": {"type": "text", "analyzer": "english_text"}, "degrees": KEYWORD, "majors": {"type": "text", "analyzer": "english_text", "fields": {"keyword": KEYWORD}}, "summary": {"type": "text", "analyzer": "english_text"}, "interests": {"type": "text", "analyzer": "english_text"}, "languages": KEYWORD, "location_name": {"type": "text", "analyzer": "english_text"}, "location_locality": KEYWORD, "location_region": KEYWORD, "location_country": KEYWORD, "location_continent": KEYWORD, "years_experience": {"type": "integer"}, "connections": {"type": "integer"}, "linkedin_url": {"type": "keyword", "index": False}, "indexed_at": {"type": "date"}, }, } def get_client() -> Elasticsearch: """Build an Elasticsearch client from settings (auth optional in dev).""" config = settings.SEARCH["ELASTICSEARCH"] kwargs: dict[str, Any] = { "hosts": config["HOSTS"], "request_timeout": config["TIMEOUT"], "max_retries": config["MAX_RETRIES"], "retry_on_timeout": True, } if config["USERNAME"] and config["PASSWORD"]: kwargs["basic_auth"] = (config["USERNAME"], config["PASSWORD"]) return Elasticsearch(**kwargs) def alias_name() -> str: return settings.SEARCH["ELASTICSEARCH"]["INDEX"] def physical_index_name() -> str: return f"{alias_name()}-{datetime.now(UTC):%Y%m%d%H%M%S}" def build_document(profile: Profile) -> dict[str, Any]: """Project a ``Profile`` row (plus its relations) into an index document.""" skills = [link.skill.name for link in profile.profile_skills.all()] experiences = list(profile.experiences.all()) educations = list(profile.educations.all()) degrees = sorted({degree for edu in educations for degree in edu.degrees}) majors = sorted({major for edu in educations for major in edu.majors}) schools = [edu.school_name for edu in educations if edu.school_name] return { "id": profile.id, "full_name": profile.full_name, "headline": profile.headline, "job_title": profile.job_title, "job_title_role": profile.job_title_role, "job_title_sub_role": profile.job_title_sub_role, "seniority_levels": profile.seniority_levels, "company": profile.current_company.name if profile.current_company_id else "", "past_companies": sorted({exp.company_name for exp in experiences if exp.company_name}), "industry": profile.industry, "skills": skills, "education": " ".join([*schools, *majors]), "degrees": degrees, "majors": majors, "summary": profile.summary, "interests": profile.interests, "languages": profile.languages, "location_name": profile.location_name, "location_locality": profile.location_locality, "location_region": profile.location_region, "location_country": profile.location_country, "location_continent": profile.location_continent, "years_experience": profile.years_experience, "connections": profile.connections, "linkedin_url": profile.linkedin_url, "indexed_at": datetime.now(UTC).isoformat(), } def iter_documents(queryset: Iterable[Profile] | None = None) -> Iterator[dict[str, Any]]: profiles = queryset if queryset is not None else _default_queryset() for profile in profiles: yield {"_id": profile.id, "_source": build_document(profile)} def _default_queryset(): # noqa: ANN202 return ( Profile.objects.select_related("current_company") .prefetch_related("profile_skills__skill", "experiences", "educations") .order_by("id") .iterator(chunk_size=200) ) # Creating an index waits for the master to ack and the shard to be allocated, # which can take far longer than the per-request timeout used for searches. INDEX_CREATE_TIMEOUT = 120 def create_index(client: Elasticsearch, name: str) -> None: """ Create the physical index, tolerating a retried request. ``indices.create`` is not idempotent, but the client runs with ``retry_on_timeout``: when the cluster acks the creation more slowly than the request timeout, the retry comes back as ``resource_already_exists_exception`` and the rebuild fails even though the index is there. The name carries a timestamp generated moments earlier, so an index already under that name can only be our own first attempt. """ try: client.options(request_timeout=INDEX_CREATE_TIMEOUT).indices.create( index=name, settings=INDEX_SETTINGS, mappings=INDEX_MAPPINGS ) except BadRequestError as exc: if exc.error != "resource_already_exists_exception": raise logger.warning( "index_create_retried", extra={"event": "index_create_retried", "index": name}, ) def rebuild_index(client: Elasticsearch | None = None) -> dict[str, Any]: """ Zero-downtime rebuild. Creates a fresh physical index, indexes every profile into it, then flips the alias atomically and removes the previous physical index. """ client = client or get_client() alias = alias_name() new_index = physical_index_name() create_index(client, new_index) logger.info("index_created", extra={"event": "index_created", "index": new_index}) # Anything that goes wrong from here on leaves an index nothing points at; # drop it rather than accumulating one orphan per failed attempt. try: succeeded, errors = bulk( client, ({"_index": new_index, **doc} for doc in iter_documents()), raise_on_error=False, stats_only=False, ) client.indices.refresh(index=new_index) except Exception: client.indices.delete(index=new_index, ignore_unavailable=True) raise previous = _current_indices(client, alias) actions: list[dict[str, Any]] = [{"add": {"index": new_index, "alias": alias}}] actions += [{"remove": {"index": name, "alias": alias}} for name in previous] client.indices.update_aliases(actions=actions) for name in previous: client.indices.delete(index=name, ignore_unavailable=True) result = { "alias": alias, "index": new_index, "indexed": succeeded, "errors": len(errors) if isinstance(errors, list) else errors, "replaced": previous, } logger.info("index_rebuilt", extra={"event": "index_rebuilt", **result}) return result def _current_indices(client: Elasticsearch, alias: str) -> list[str]: if not client.indices.exists_alias(name=alias): # A leftover *index* (not alias) with the same name would block the # alias creation, so remove it first. if client.indices.exists(index=alias): client.indices.delete(index=alias, ignore_unavailable=True) return [] return list(client.indices.get_alias(name=alias).keys()) def index_profile(profile: Profile, client: Elasticsearch | None = None) -> None: """Index (or re-index) a single profile - used after a targeted update.""" client = client or get_client() client.index(index=alias_name(), id=profile.id, document=build_document(profile))