Page MenuHomePhorge
Files F85650140

No OneTemporary
Actions

Size
17 KB
Referenced Files
None
Subscribers
None

View Options

diff --git a/changelog.d/search-indexing.skip b/changelog.d/search-indexing.skip
new file mode 100644
index 000000000..e69de29bb
diff --git a/lib/mix/tasks/pleroma/search/meilisearch.ex b/lib/mix/tasks/pleroma/search/meilisearch.ex
index edce9e871..facc38815 100644
--- a/lib/mix/tasks/pleroma/search/meilisearch.ex
+++ b/lib/mix/tasks/pleroma/search/meilisearch.ex
@@ -1,145 +1,145 @@
# Pleroma: A lightweight social networking server
# Copyright © 2017-2021 Pleroma Authors <https://pleroma.social/>
# SPDX-License-Identifier: AGPL-3.0-only
defmodule Mix.Tasks.Pleroma.Search.Meilisearch do
require Pleroma.Constants
import Mix.Pleroma
import Ecto.Query
import Pleroma.Search.Meilisearch,
only: [meili_put: 2, meili_get: 1, meili_delete: 1]
def run(["index"]) do
start_pleroma()
Pleroma.HTML.compile_scrubbers()
meili_version =
(
{:ok, result} = meili_get("/version")
result["pkgVersion"]
)
# The ranking rule syntax was changed but nothing about that is mentioned in the changelog
if not Version.match?(meili_version, ">= 0.25.0") do
raise "Meilisearch <0.24.0 not supported"
end
{:ok, _} =
meili_put(
"/indexes/objects/settings/ranking-rules",
[
"published:desc",
"words",
"exactness",
"proximity",
"typo",
"attribute",
"sort"
]
)
{:ok, _} =
meili_put(
"/indexes/objects/settings/searchable-attributes",
[
"content"
]
)
IO.puts("Created indices. Starting to insert posts.")
chunk_size = Pleroma.Config.get([Pleroma.Search.Meilisearch, :initial_indexing_chunk_size])
Pleroma.Repo.transaction(
fn ->
query =
from(Pleroma.Object,
# Only index public and unlisted posts which are notes and have some text
where:
fragment("data->>'type' = 'Note'") and
(fragment("data->'to' \\? ?", ^Pleroma.Constants.as_public()) or
fragment("data->'cc' \\? ?", ^Pleroma.Constants.as_public())),
order_by: [desc: fragment("data->'published'")]
)
count = query |> Pleroma.Repo.aggregate(:count, :data)
IO.puts("Entries to index: #{count}")
Pleroma.Repo.stream(
query,
timeout: :infinity
)
- |> Stream.map(&Pleroma.Search.Meilisearch.object_to_search_data/1)
+ |> Stream.map(&Pleroma.Search.object_to_search_data/1)
|> Stream.filter(fn o -> not is_nil(o) end)
|> Stream.chunk_every(chunk_size)
|> Stream.transform(0, fn objects, acc ->
new_acc = acc + Enum.count(objects)
# Reset to the beginning of the line and rewrite it
IO.write("\r")
IO.write("Indexed #{new_acc} entries")
{[objects], new_acc}
end)
|> Stream.each(fn objects ->
result =
meili_put(
"/indexes/objects/documents",
objects
)
with {:ok, res} <- result do
if not Map.has_key?(res, "uid") do
IO.puts("\nFailed to index: #{inspect(result)}")
end
else
e -> IO.puts("\nFailed to index due to network error: #{inspect(e)}")
end
end)
|> Stream.run()
end,
timeout: :infinity
)
IO.write("\n")
end
def run(["clear"]) do
start_pleroma()
meili_delete("/indexes/objects/documents")
end
def run(["show-keys", master_key]) do
start_pleroma()
endpoint = Pleroma.Config.get([Pleroma.Search.Meilisearch, :url])
{:ok, result} =
Pleroma.HTTP.get(
Path.join(endpoint, "/keys"),
[{"Authorization", "Bearer #{master_key}"}]
)
decoded = Jason.decode!(result.body)
if decoded["results"] do
Enum.each(decoded["results"], fn %{"description" => desc, "key" => key} ->
IO.puts("#{desc}: #{key}")
end)
else
IO.puts("Error fetching the keys, check the master key is correct: #{inspect(decoded)}")
end
end
def run(["stats"]) do
start_pleroma()
{:ok, result} = meili_get("/indexes/objects/stats")
IO.puts("Number of entries: #{result["numberOfDocuments"]}")
IO.puts("Indexing? #{result["isIndexing"]}")
end
end
diff --git a/lib/pleroma/search.ex b/lib/pleroma/search.ex
index 8477873f9..9cd2768c4 100644
--- a/lib/pleroma/search.ex
+++ b/lib/pleroma/search.ex
@@ -1,43 +1,80 @@
defmodule Pleroma.Search do
alias Pleroma.Activity
alias Pleroma.Object
alias Pleroma.Web.ActivityPub.Visibility
alias Pleroma.Workers.SearchIndexingWorker
@spec add_to_index(Activity.t()) :: {:ok, Oban.Job.t() | :noop} | {:error, Oban.Job.changeset()}
def add_to_index(%Activity{id: activity_id, object: %Object{} = object} = activity) do
with {_, true} <- {:indexable, indexable?(activity)},
{_, "public"} <- {:visibility, Visibility.get_visibility(object)} do
SearchIndexingWorker.new(%{"op" => "add_to_index", "activity" => activity_id})
|> Oban.insert()
else
_ -> {:ok, :noop}
end
end
def add_to_index(%Activity{id: activity_id}) do
case Activity.get_by_id_with_object(activity_id) do
%Activity{} = preloaded -> add_to_index(preloaded)
- _ -> :ok
+ _ -> {:ok, :noop}
end
end
@spec remove_from_index(Object.t()) :: {:ok, Oban.Job.t()} | {:error, Oban.Job.changeset()}
def remove_from_index(%Pleroma.Object{id: object_id}) do
SearchIndexingWorker.new(%{"op" => "remove_from_index", "object" => object_id})
|> Oban.insert()
end
def search(query, options) do
search_module = Pleroma.Config.get([Pleroma.Search, :module])
search_module.search(options[:for_user], query, options)
end
def healthcheck_endpoints do
search_module = Pleroma.Config.get([Pleroma.Search, :module])
search_module.healthcheck_endpoints()
end
- defp indexable?(%Activity{data: %{"type" => "Create"}}), do: true
+ def object_to_search_data(%Object{} = object) do
+ data = object.data
+
+ content_str =
+ case data["content"] do
+ [nil | rest] -> to_string(rest)
+ str -> str
+ end
+
+ content =
+ with {:ok, scrubbed} <-
+ FastSanitize.Sanitizer.scrub(content_str, Pleroma.HTML.Scrubber.SearchIndexing),
+ trimmed <- String.trim(scrubbed) do
+ trimmed
+ end
+
+ # Make sure we have a non-empty string
+ if content != "" do
+ {:ok, published, _} = DateTime.from_iso8601(data["published"])
+
+ %{
+ id: object.id,
+ content: content,
+ ap: data["id"],
+ published: published |> DateTime.to_unix()
+ }
+ end
+ end
+
+ defp indexable?(%Activity{
+ data: %{"type" => "Create"},
+ object: %Object{
+ data: %{"content" => content, "published" => published, "type" => "Note"}
+ }
+ })
+ when not is_nil(content) and content not in ["", "."] and not is_nil(published),
+ do: true
+
defp indexable?(_), do: false
end
diff --git a/lib/pleroma/search/meilisearch.ex b/lib/pleroma/search/meilisearch.ex
index 4541ef14a..dc10076e1 100644
--- a/lib/pleroma/search/meilisearch.ex
+++ b/lib/pleroma/search/meilisearch.ex
@@ -1,195 +1,159 @@
defmodule Pleroma.Search.Meilisearch do
require Logger
require Pleroma.Constants
alias Pleroma.Activity
alias Pleroma.Config.Getting, as: Config
alias Pleroma.Object
+ alias Pleroma.Search
import Pleroma.Search.DatabaseSearch
import Ecto.Query
@behaviour Pleroma.Search.SearchBackend
@impl true
def create_index, do: :ok
@impl true
def drop_index, do: :ok
defp meili_headers do
private_key = Config.get([Pleroma.Search.Meilisearch, :private_key])
[{"Content-Type", "application/json"}] ++
if is_nil(private_key), do: [], else: [{"Authorization", "Bearer #{private_key}"}]
end
def meili_get(path) do
endpoint = Config.get([Pleroma.Search.Meilisearch, :url])
result =
Pleroma.HTTP.get(
Path.join(endpoint, path),
meili_headers()
)
with {:ok, res} <- result do
{:ok, Jason.decode!(res.body)}
end
end
def meili_post(path, params) do
endpoint = Config.get([Pleroma.Search.Meilisearch, :url])
result =
Pleroma.HTTP.post(
Path.join(endpoint, path),
Jason.encode!(params),
meili_headers()
)
with {:ok, res} <- result do
{:ok, Jason.decode!(res.body)}
end
end
def meili_put(path, params) do
endpoint = Config.get([Pleroma.Search.Meilisearch, :url])
result =
Pleroma.HTTP.request(
:put,
Path.join(endpoint, path),
Jason.encode!(params),
meili_headers(),
[]
)
with {:ok, res} <- result do
{:ok, Jason.decode!(res.body)}
end
end
def meili_delete(path) do
endpoint = Config.get([Pleroma.Search.Meilisearch, :url])
with {:ok, _} <-
Pleroma.HTTP.request(
:delete,
Path.join(endpoint, path),
"",
meili_headers(),
[]
) do
:ok
else
_ -> {:error, "Could not remove from index"}
end
end
@impl true
def search(user, query, options \\ []) do
limit = Enum.min([Keyword.get(options, :limit), 40])
offset = Keyword.get(options, :offset, 0)
author = Keyword.get(options, :author)
res =
meili_post(
"/indexes/objects/search",
%{q: query, offset: offset, limit: limit}
)
with {:ok, result} <- res do
hits = result["hits"] |> Enum.map(& &1["ap"])
try do
hits
|> Activity.create_by_object_ap_id()
|> Activity.with_preloaded_object()
|> Activity.restrict_deactivated_users()
|> maybe_restrict_local(user)
|> maybe_restrict_author(author)
|> maybe_restrict_blocked(user)
|> maybe_fetch(user, query)
|> order_by([object: obj], desc: obj.data["published"])
|> Pleroma.Repo.all()
rescue
_ -> maybe_fetch([], user, query)
end
end
end
- def object_to_search_data(object) do
- # Only index public or unlisted Notes
- if not is_nil(object) and object.data["type"] == "Note" and
- not is_nil(object.data["content"]) and
- not is_nil(object.data["published"]) and
- (Pleroma.Constants.as_public() in object.data["to"] or
- Pleroma.Constants.as_public() in object.data["cc"]) and
- object.data["content"] not in ["", "."] do
- data = object.data
-
- content_str =
- case data["content"] do
- [nil | rest] -> to_string(rest)
- str -> str
- end
-
- content =
- with {:ok, scrubbed} <-
- FastSanitize.Sanitizer.scrub(content_str, Pleroma.HTML.Scrubber.SearchIndexing),
- trimmed <- String.trim(scrubbed) do
- trimmed
- end
-
- # Make sure we have a non-empty string
- if content != "" do
- {:ok, published, _} = DateTime.from_iso8601(data["published"])
-
- %{
- id: object.id,
- content: content,
- ap: data["id"],
- published: published |> DateTime.to_unix()
- }
- end
- end
- end
-
@impl true
def add_to_index(%Activity{object: %Object{} = object} = activity) do
- search_data = object_to_search_data(object)
+ search_data = Search.object_to_search_data(object)
result =
meili_put(
"/indexes/objects/documents",
[search_data]
)
with {:ok, %{"status" => "enqueued"}} <- result do
# Added successfully
:ok
else
_ ->
# There was an error, report it
Logger.error("Failed to add activity #{activity.id} to index: #{inspect(result)}")
{:error, result}
end
end
@impl true
def remove_from_index(object) do
meili_delete("/indexes/objects/documents/#{object.id}")
end
@impl true
def healthcheck_endpoints do
endpoint =
Config.get([Pleroma.Search.Meilisearch, :url])
|> URI.parse()
|> Map.put(:path, "/health")
|> URI.to_string()
[endpoint]
end
end
diff --git a/lib/pleroma/search/qdrant_search.ex b/lib/pleroma/search/qdrant_search.ex
index 06f5b1983..4d57cfa88 100644
--- a/lib/pleroma/search/qdrant_search.ex
+++ b/lib/pleroma/search/qdrant_search.ex
@@ -1,207 +1,207 @@
defmodule Pleroma.Search.QdrantSearch do
@behaviour Pleroma.Search.SearchBackend
import Ecto.Query
alias Pleroma.Activity
alias Pleroma.Config.Getting, as: Config
alias Pleroma.Object
+ alias Pleroma.Search
alias __MODULE__.OpenAIClient
alias __MODULE__.QdrantClient
- import Pleroma.Search.Meilisearch, only: [object_to_search_data: 1]
import Pleroma.Search.DatabaseSearch, only: [maybe_fetch: 3]
@impl true
def create_index do
payload = Config.get([Pleroma.Search.QdrantSearch, :qdrant_index_configuration])
with {:ok, %{status: 200}} <- QdrantClient.put("/collections/posts", payload) do
:ok
else
e -> {:error, e}
end
end
@impl true
def drop_index do
with {:ok, %{status: 200}} <- QdrantClient.delete("/collections/posts") do
:ok
else
e -> {:error, e}
end
end
def get_embedding(text) do
with {:ok, %{body: %{"data" => [%{"embedding" => embedding}]}}} <-
OpenAIClient.post("/v1/embeddings", %{
input: text,
model: Config.get([Pleroma.Search.QdrantSearch, :openai_model])
}) do
{:ok, embedding}
else
_ ->
{:error, "Failed to get embedding"}
end
end
defp actor_from_activity(%{data: %{"actor" => actor}}) do
actor
end
defp actor_from_activity(_), do: nil
defp build_index_payload(activity, embedding) do
actor = actor_from_activity(activity)
published_at = activity.data["published"]
%{
points: [
%{
id: activity.id |> FlakeId.from_string() |> Ecto.UUID.cast!(),
vector: embedding,
payload: %{actor: actor, published_at: published_at}
}
]
}
end
defp build_search_payload(embedding, options) do
base = %{
vector: embedding,
limit: options[:limit] || 20,
offset: options[:offset] || 0
}
if author = options[:author] do
Map.put(base, :filter, %{
must: [%{key: "actor", match: %{value: author.ap_id}}]
})
else
base
end
end
@impl true
def add_to_index(%Activity{object: %Object{} = object} = activity) do
- search_data = object_to_search_data(object)
+ search_data = Search.object_to_search_data(object)
with {:ok, embedding} <- get_embedding(search_data.content),
{:ok, %{status: 200}} <-
QdrantClient.put(
"/collections/posts/points",
build_index_payload(activity, embedding)
) do
:ok
else
e -> {:error, e}
end
end
@impl true
def remove_from_index(object) do
activity = Activity.get_by_object_ap_id_with_object(object.data["id"])
id = activity.id |> FlakeId.from_string() |> Ecto.UUID.cast!()
with {:ok, %{status: 200}} <-
QdrantClient.post("/collections/posts/points/delete", %{"points" => [id]}) do
:ok
else
e -> {:error, e}
end
end
@impl true
def search(user, original_query, options) do
query = "Represent this sentence for searching relevant passages: #{original_query}"
with {:ok, embedding} <- get_embedding(query),
{:ok, %{body: %{"result" => result}}} <-
QdrantClient.post(
"/collections/posts/points/search",
build_search_payload(embedding, options)
) do
ids =
Enum.map(result, fn %{"id" => id} ->
Ecto.UUID.dump!(id)
end)
from(a in Activity, where: a.id in ^ids)
|> Activity.with_preloaded_object()
|> Activity.restrict_deactivated_users()
|> Ecto.Query.order_by([a], fragment("array_position(?, ?)", ^ids, a.id))
|> Pleroma.Repo.all()
|> maybe_fetch(user, original_query)
else
_ ->
[]
end
end
@impl true
def healthcheck_endpoints do
qdrant_health =
Config.get([Pleroma.Search.QdrantSearch, :qdrant_url])
|> URI.parse()
|> Map.put(:path, "/healthz")
|> URI.to_string()
openai_health = Config.get([Pleroma.Search.QdrantSearch, :openai_healthcheck_url])
[qdrant_health, openai_health] |> Enum.filter(& &1)
end
end
defmodule Pleroma.Search.QdrantSearch.OpenAIClient do
alias Pleroma.Config.Getting, as: Config
def post(path, body) do
Tesla.post(client(), path, body)
end
defp client do
Tesla.client(middleware())
end
defp middleware do
[
{Tesla.Middleware.BaseUrl, Config.get([Pleroma.Search.QdrantSearch, :openai_url])},
Tesla.Middleware.JSON,
{Tesla.Middleware.Headers,
[
{"Authorization", "Bearer #{Config.get([Pleroma.Search.QdrantSearch, :openai_api_key])}"}
]}
]
end
end
defmodule Pleroma.Search.QdrantSearch.QdrantClient do
alias Pleroma.Config.Getting, as: Config
def delete(path) do
Tesla.delete(client(), path)
end
def post(path, body) do
Tesla.post(client(), path, body)
end
def put(path, body) do
Tesla.put(client(), path, body)
end
defp client do
Tesla.client(middleware())
end
defp middleware do
[
{Tesla.Middleware.BaseUrl, Config.get([Pleroma.Search.QdrantSearch, :qdrant_url])},
Tesla.Middleware.JSON,
{Tesla.Middleware.Headers,
[
{"api-key", Pleroma.Config.get([Pleroma.Search.QdrantSearch, :qdrant_api_key])}
]}
]
end
end

File Metadata

Mime Type
text/x-diff
Expires
Sat, Aug 29, 9:58 PM (1 d, 12 h)
Storage Engine
blob
Storage Format
Raw Data
Storage Handle
1737508
Default Alt Text
(17 KB)

Event Timeline