fix(wiki): convert reassembles pages from existing chunks (crawler/remote)

convert_source_to_wiki now builds pages from the source's already-ingested vector-store chunks instead of re-parsing storage files, so it works for crawler/remote/connector sources (which have no stored files; file_path empty). Groups by metadata file_path/file_name (URL- and connector-aware; URLs normalized to virtual paths) rather than the raw source. Passes embeddings_key to create_vectorstore in BOTH convert and reembed_wiki_page (fixes a TypeError crash on faiss/elasticsearch). Conservative overlap-trim (min length). Deletes the original chunks after reassembling so retrieval has no duplicate/stale content.
This commit is contained in:
Alex committed 2026-06-22 23:03:52 +01:00
1 parent 80533f9383
commit 8a0f7024e3
2 files changed
+420 -230

No files matched your search

+152 -74
View File
@@ -12,7 +12,7 @@ import zipfile
import uuid
from collections import Counter
from urllib.parse import urljoin
from urllib.parse import urljoin, urlsplit
import requests
@@ -2012,7 +2012,9 @@ def reembed_wiki_page_worker(self, source_id, path, content_hash, user):
source_id = str(source["id"])
cfg = SourceConfig.parse(source.get("config"))
store = VectorCreator.create_vectorstore(settings.VECTOR_STORE, source_id)
store = VectorCreator.create_vectorstore(
settings.VECTOR_STORE, source_id, settings.EMBEDDINGS_KEY
)
deleted = store.delete_chunks_by_source_path(path)
with db_readonly() as conn:
@@ -2073,16 +2075,98 @@ def _wiki_page_path_from_rel(rel_path):
return validate_tool_path(raw)
def convert_source_to_wiki_worker(self, source_id, user):
"""Convert an ingested source into a wiki by materializing files as pages.
def _chunk_text(chunk) -> str:
if isinstance(chunk, dict):
return chunk.get("text") or chunk.get("page_content") or ""
return getattr(chunk, "text", None) or getattr(chunk, "page_content", "") or ""
Re-parses the source's stored originals (the same storage file-load +
``SimpleDirectoryReader`` machinery reingest uses), creating one wiki page
per file with extractable text. Non-text/binary/unsupported files are
skipped and reported. Embedding is enqueued per page (async); the source's
``config.kind`` flips to ``wiki`` only after at least one page is
materialized — a source with no extractable pages keeps its original kind
and directory structure.
def _chunk_metadata(chunk) -> dict:
if isinstance(chunk, dict):
meta = chunk.get("metadata")
else:
meta = getattr(chunk, "metadata", None)
return meta if isinstance(meta, dict) else {}
def _chunk_doc_id(chunk):
if isinstance(chunk, dict):
return chunk.get("doc_id")
return getattr(chunk, "doc_id", None)
def _url_to_virtual_path(value: str) -> str:
parts = urlsplit(value)
path = parts.path.strip("/")
if path:
return path
if parts.netloc:
return parts.netloc
return value.replace("://", "/")
def _chunk_page_path(metadata) -> str:
for key in ("file_path", "file_name", "filename", "title", "source"):
value = metadata.get(key)
if value:
value = str(value)
if "://" in value:
value = _url_to_virtual_path(value)
return value
return ""
def _chunk_order_hint(metadata):
for key in ("chunk", "chunk_index", "index", "start", "start_index", "offset"):
value = metadata.get(key)
if isinstance(value, (int, float)):
return value
if isinstance(value, str):
try:
return float(value)
except ValueError:
continue
return None
MIN_CHUNK_OVERLAP_TRIM = 32
def _join_chunk_texts(texts: list[str]) -> str:
"""Concatenate chunk texts, trimming an exact overlapping boundary.
Only an exact suffix/prefix overlap of at least ``MIN_CHUNK_OVERLAP_TRIM``
characters between consecutive chunks is removed (the common
``chunk_overlap`` case); shorter, possibly legitimate repeats are kept.
"""
parts: list[str] = []
for text in texts:
if parts:
prev = parts[-1]
limit = min(len(prev), len(text))
overlap = 0
for size in range(limit, MIN_CHUNK_OVERLAP_TRIM - 1, -1):
if prev.endswith(text[:size]):
overlap = size
break
if overlap:
text = text[overlap:]
if text:
parts.append(text)
return "\n\n".join(parts)
def convert_source_to_wiki_worker(self, source_id, user):
"""Convert an ingested source into a wiki by reassembling pages from chunks.
Groups the source's existing vector-store chunks by their page path
(``metadata.source``), concatenating each group into one wiki page. This
works uniformly for every source type, including crawler/remote/connector
sources that hold no stored originals. Chunks with a missing or invalid
path are skipped and reported. Embedding is enqueued per page (async); the
source's ``config.kind`` flips to ``wiki`` only after at least one page is
materialized — a source with no usable chunks keeps its original kind and
directory structure.
Args:
self: Celery task instance.
@@ -2091,10 +2175,11 @@ def convert_source_to_wiki_worker(self, source_id, user):
Returns:
A summary dict ``{"status": "converted", "pages_created", "skipped"}``,
``{"status": "no_pages", ...}`` when nothing was extractable, or
``{"status": "no_pages", ...}`` when nothing was reassembled, or
``{"status": "already_wiki"}`` when the source is already a wiki.
"""
from application.api.user.tasks import reembed_wiki_page
from application.vectorstore.vector_creator import VectorCreator
source_id = str(source_id)
@@ -2108,77 +2193,61 @@ def convert_source_to_wiki_worker(self, source_id, user):
if cfg.kind == "wiki":
return {"status": "already_wiki", "pages_created": 0, "skipped": []}
storage = StorageCreator.get_storage()
source_file_path = source.get("file_path", "")
file_name_map = _normalize_file_name_map(source.get("file_name_map"))
created_pages: list[tuple[str, str]] = []
skipped: list[dict] = []
stored_files: set[str] = set()
if storage.is_directory(source_file_path):
for stored in storage.list_files(source_file_path):
if storage.is_directory(stored):
continue
stored_files.add(os.path.relpath(stored, source_file_path))
store = VectorCreator.create_vectorstore(
settings.VECTOR_STORE, source_id, settings.EMBEDDINGS_KEY
)
with tempfile.TemporaryDirectory() as temp_dir:
_download_source_files_to_dir(storage, source_file_path, temp_dir)
grouped: dict[str, list[tuple[object, str]]] = {}
original_doc_ids: list[object] = []
for chunk in store.get_chunks() or []:
metadata = _chunk_metadata(chunk)
rel_path = _chunk_page_path(metadata)
text = _chunk_text(chunk)
if not text.strip():
continue
if not rel_path:
skipped.append({"file": "", "reason": "missing path"})
continue
if _wiki_page_path_from_rel(rel_path) is None:
skipped.append({"file": rel_path, "reason": "invalid path"})
continue
doc_id = _chunk_doc_id(chunk)
if doc_id is not None:
original_doc_ids.append(doc_id)
grouped.setdefault(rel_path, []).append((_chunk_order_hint(metadata), text))
reader = SimpleDirectoryReader(
input_dir=temp_dir,
recursive=True,
required_exts=list(SUPPORTED_SOURCE_EXTENSIONS),
exclude_hidden=True,
errors="ignore",
file_metadata=metadata_from_filename,
)
raw_docs = reader.load_data()
texts_by_rel: dict[str, list[str]] = {}
for doc in raw_docs:
extra_info = getattr(doc, "extra_info", None) or {}
rel_path = extra_info.get("source")
if not rel_path:
continue
text = getattr(doc, "text", None) or ""
if not text.strip():
continue
texts_by_rel.setdefault(rel_path, []).append(text)
with db_session() as conn:
repo = WikiPagesRepository(conn)
for rel_path in sorted(texts_by_rel):
page_path = _wiki_page_path_from_rel(rel_path)
if page_path is None:
skipped.append({"file": rel_path, "reason": "invalid path"})
continue
content = "\n\n".join(texts_by_rel[rel_path])
title = _get_display_name(file_name_map, rel_path) or os.path.basename(
rel_path
with db_session() as conn:
repo = WikiPagesRepository(conn)
for rel_path in sorted(grouped):
page_path = _wiki_page_path_from_rel(rel_path)
entries = grouped[rel_path]
if any(hint is not None for hint, _ in entries):
entries = sorted(
enumerate(entries),
key=lambda item: (
item[1][0] if item[1][0] is not None else float("inf"),
item[0],
),
)
repo.upsert(
source_id,
page_path,
content,
title=title,
updated_by=user,
updated_via="agent",
)
created_pages.append((page_path, _content_hash(content)))
for rel_path in sorted(stored_files - set(texts_by_rel)):
ext = os.path.splitext(rel_path)[1].lower()
reason = (
"could not read/parse"
if ext in SUPPORTED_SOURCE_EXTENSIONS
else "no extractable text"
entries = [entry for _, entry in entries]
content = _join_chunk_texts([text for _, text in entries])
title = _get_display_name(file_name_map, rel_path) or os.path.basename(
rel_path
)
skipped.append({"file": rel_path, "reason": reason})
if created_pages:
with db_session() as conn:
rebuild_wiki_directory_structure(conn, source_id, user)
repo.upsert(
source_id,
page_path,
content,
title=title,
updated_by=user,
updated_via="agent",
)
created_pages.append((page_path, _content_hash(content)))
if not created_pages:
return {
@@ -2187,6 +2256,15 @@ def convert_source_to_wiki_worker(self, source_id, user):
"skipped": skipped,
}
with db_session() as conn:
rebuild_wiki_directory_structure(conn, source_id, user)
for doc_id in original_doc_ids:
try:
store.delete_chunk(doc_id)
except Exception as exc:
logging.error(f"Failed deleting original chunk {doc_id}: {exc}")
for page_path, content_hash in created_pages:
reembed_wiki_page.delay(
source_id,
+268 -156
View File
@@ -1,9 +1,10 @@
"""Tests for ``application.worker.convert_source_to_wiki_worker``.
The worker re-parses a source's stored originals into wiki pages and enqueues
a per-page re-embed. Storage, ``SimpleDirectoryReader``, and the re-embed task
are mocked so no real downloads or embeddings run; the ``sources`` /
``wiki_pages`` rows are real so the assertions read them back.
The worker reassembles wiki pages from a source's existing vector-store
chunks (grouped by ``metadata.source``) and enqueues a per-page re-embed.
``VectorCreator.create_vectorstore`` and the re-embed task are mocked so no
real store access or embeddings run; the ``sources`` / ``wiki_pages`` rows
are real so the assertions read them back.
"""
from __future__ import annotations
@@ -12,16 +13,15 @@ from unittest.mock import MagicMock
import pytest
from application.parser.schema.base import Document
from application.storage.db.repositories.sources import SourcesRepository
from application.storage.db.repositories.wiki_pages import WikiPagesRepository
def _seed_source(pg_conn, config=None, file_path="inputs/alice/doc-set"):
def _seed_source(pg_conn, config=None, file_path=""):
src = SourcesRepository(pg_conn).create(
"doc-set",
user_id="alice",
type="local",
type="crawler",
retriever="classic",
file_path=file_path,
config=config or {},
@@ -30,54 +30,48 @@ def _seed_source(pg_conn, config=None, file_path="inputs/alice/doc-set"):
return str(src["id"])
def _patch_storage(monkeypatch, stored_files):
"""Storage whose directory holds ``stored_files`` (rel paths)."""
from application import worker
base = "inputs/alice/doc-set"
storage = MagicMock(name="storage")
storage.is_directory.side_effect = lambda p: p == base
storage.list_files.return_value = [f"{base}/{f}" for f in stored_files]
storage.get_file.return_value = MagicMock(read=lambda: b"x")
monkeypatch.setattr(worker.StorageCreator, "get_storage", lambda: storage)
return storage
def _chunk(text, source=None, doc_id="d", **extra_meta):
metadata = dict(extra_meta)
if source is not None:
metadata["source"] = source
return {"doc_id": doc_id, "text": text, "metadata": metadata}
def _patch_reader(monkeypatch, docs):
from application import worker
def _patch_store(monkeypatch, chunks):
store = MagicMock(name="vectorstore")
store.get_chunks.return_value = chunks
monkeypatch.setattr(
"application.vectorstore.vector_creator.VectorCreator.create_vectorstore",
lambda *a, **kw: store,
)
return store
reader = MagicMock(name="reader")
reader.load_data.return_value = docs
monkeypatch.setattr(worker, "SimpleDirectoryReader", lambda *a, **kw: reader)
return reader
def _patch_reembed(monkeypatch):
delay = MagicMock(name="reembed_delay")
monkeypatch.setattr("application.api.user.tasks.reembed_wiki_page.delay", delay)
return delay
@pytest.mark.unit
class TestConvertSourceToWikiWorker:
def test_one_page_per_text_file(
def test_two_pages_reassembled_from_chunks(
self, pg_conn, patch_worker_db, task_self, monkeypatch
):
from application import worker
source_id = _seed_source(pg_conn)
_patch_storage(monkeypatch, ["guide.md", "notes/intro.md", "logo.png"])
_patch_reader(
_patch_store(
monkeypatch,
[
Document(text="guide body", extra_info={"source": "guide.md"}),
Document(
text="intro body", extra_info={"source": "notes/intro.md"}
),
_chunk("guide part one.", source="guide.md", doc_id="g1"),
_chunk("guide part two.", source="guide.md", doc_id="g2"),
_chunk("intro body.", source="notes/intro.md", doc_id="i1"),
],
)
delay = MagicMock(name="reembed_delay")
monkeypatch.setattr(
"application.api.user.tasks.reembed_wiki_page.delay", delay
)
delay = _patch_reembed(monkeypatch)
result = worker.convert_source_to_wiki_worker(
task_self, source_id, "alice"
)
result = worker.convert_source_to_wiki_worker(task_self, source_id, "alice")
assert result["status"] == "converted"
assert result["pages_created"] == 2
@@ -85,83 +79,82 @@ class TestConvertSourceToWikiWorker:
pages = WikiPagesRepository(pg_conn).list_for_source(source_id)
by_path = {p["path"]: p for p in pages}
assert set(by_path) == {"/guide.md", "/notes/intro.md"}
assert by_path["/guide.md"]["content"] == "guide body"
assert by_path["/guide.md"]["content"] == "guide part one.\n\nguide part two."
assert by_path["/guide.md"]["title"] == "guide.md"
assert by_path["/notes/intro.md"]["content"] == "intro body."
assert by_path["/notes/intro.md"]["title"] == "intro.md"
def test_binaries_skipped_and_reported(
self, pg_conn, patch_worker_db, task_self, monkeypatch
):
from application import worker
source_id = _seed_source(pg_conn)
_patch_storage(monkeypatch, ["guide.md", "logo.png", "scan.pdf"])
_patch_reader(
monkeypatch,
[Document(text="guide body", extra_info={"source": "guide.md"})],
)
monkeypatch.setattr(
"application.api.user.tasks.reembed_wiki_page.delay", MagicMock()
)
result = worker.convert_source_to_wiki_worker(
task_self, source_id, "alice"
)
assert result["pages_created"] == 1
skipped_files = {s["file"] for s in result["skipped"]}
assert skipped_files == {"logo.png", "scan.pdf"}
for entry in result["skipped"]:
assert entry["reason"]
def test_reembed_enqueued_per_page_with_per_page_key(
self, pg_conn, patch_worker_db, task_self, monkeypatch
):
from application import worker
source_id = _seed_source(pg_conn)
_patch_storage(monkeypatch, ["a.md", "b.md"])
_patch_reader(
monkeypatch,
[
Document(text="aaa", extra_info={"source": "a.md"}),
Document(text="bbb", extra_info={"source": "b.md"}),
],
)
delay = MagicMock(name="reembed_delay")
monkeypatch.setattr(
"application.api.user.tasks.reembed_wiki_page.delay", delay
)
worker.convert_source_to_wiki_worker(task_self, source_id, "alice")
assert delay.call_count == 2
keys = {c.kwargs["idempotency_key"] for c in delay.call_args_list}
paths = {c.args[1] for c in delay.call_args_list}
assert paths == {"/a.md", "/b.md"}
assert paths == {"/guide.md", "/notes/intro.md"}
for call in delay.call_args_list:
assert call.args[0] == source_id
assert call.kwargs["user"] == "alice"
assert call.kwargs["idempotency_key"].startswith(
f"reembed-wiki:{source_id}:"
)
assert len(keys) == 2
def test_sets_kind_wiki_and_agentic_exposure(
def test_original_chunks_deleted_after_convert(
self, pg_conn, patch_worker_db, task_self, monkeypatch
):
from application import worker
source_id = _seed_source(pg_conn)
store = _patch_store(
monkeypatch,
[
_chunk("guide part one.", source="guide.md", doc_id="g1"),
_chunk("guide part two.", source="guide.md", doc_id="g2"),
_chunk("intro body.", source="notes/intro.md", doc_id="i1"),
],
)
delay = _patch_reembed(monkeypatch)
result = worker.convert_source_to_wiki_worker(task_self, source_id, "alice")
assert result["status"] == "converted"
deleted = {c.args[0] for c in store.delete_chunk.call_args_list}
assert deleted == {"g1", "g2", "i1"}
assert delay.call_count == 2
def test_chunk_without_doc_id_not_deleted(
self, pg_conn, patch_worker_db, task_self, monkeypatch
):
from application import worker
source_id = _seed_source(pg_conn)
chunk = {"text": "body", "metadata": {"source": "a.md"}}
store = _patch_store(monkeypatch, [chunk])
_patch_reembed(monkeypatch)
result = worker.convert_source_to_wiki_worker(task_self, source_id, "alice")
assert result["status"] == "converted"
store.delete_chunk.assert_not_called()
def test_no_pages_path_deletes_nothing(
self, pg_conn, patch_worker_db, task_self, monkeypatch
):
from application import worker
source_id = _seed_source(pg_conn)
store = _patch_store(monkeypatch, [])
_patch_reembed(monkeypatch)
result = worker.convert_source_to_wiki_worker(task_self, source_id, "alice")
assert result["status"] == "no_pages"
store.delete_chunk.assert_not_called()
def test_kind_flipped_and_exposure_defaulted(
self, pg_conn, patch_worker_db, task_self, monkeypatch
):
from application import worker
from application.storage.db.source_config import SourceConfig
source_id = _seed_source(pg_conn)
_patch_storage(monkeypatch, ["a.md"])
_patch_reader(
monkeypatch,
[Document(text="aaa", extra_info={"source": "a.md"})],
)
monkeypatch.setattr(
"application.api.user.tasks.reembed_wiki_page.delay", MagicMock()
)
_patch_store(monkeypatch, [_chunk("body", source="a.md")])
_patch_reembed(monkeypatch)
worker.convert_source_to_wiki_worker(task_self, source_id, "alice")
@@ -179,14 +172,8 @@ class TestConvertSourceToWikiWorker:
source_id = _seed_source(
pg_conn, config={"retrieval": {"exposure": "agentic_tool"}}
)
_patch_storage(monkeypatch, ["a.md"])
_patch_reader(
monkeypatch,
[Document(text="aaa", extra_info={"source": "a.md"})],
)
monkeypatch.setattr(
"application.api.user.tasks.reembed_wiki_page.delay", MagicMock()
)
_patch_store(monkeypatch, [_chunk("body", source="a.md")])
_patch_reembed(monkeypatch)
worker.convert_source_to_wiki_worker(task_self, source_id, "alice")
@@ -195,7 +182,71 @@ class TestConvertSourceToWikiWorker:
assert cfg.kind == "wiki"
assert cfg.retrieval.exposure == "agentic_tool"
def test_no_pages_leaves_kind_and_structure(
def test_chunk_order_hint_respected(
self, pg_conn, patch_worker_db, task_self, monkeypatch
):
from application import worker
source_id = _seed_source(pg_conn)
_patch_store(
monkeypatch,
[
_chunk("second", source="a.md", doc_id="a2", chunk_index=1),
_chunk("first", source="a.md", doc_id="a1", chunk_index=0),
],
)
_patch_reembed(monkeypatch)
worker.convert_source_to_wiki_worker(task_self, source_id, "alice")
pages = WikiPagesRepository(pg_conn).list_for_source(source_id)
by_path = {p["path"]: p for p in pages}
assert by_path["/a.md"]["content"] == "first\n\nsecond"
def test_short_overlap_not_trimmed(
self, pg_conn, patch_worker_db, task_self, monkeypatch
):
from application import worker
source_id = _seed_source(pg_conn)
_patch_store(
monkeypatch,
[
_chunk("The end.", source="a.md", doc_id="a1"),
_chunk("The end. More", source="a.md", doc_id="a2"),
],
)
_patch_reembed(monkeypatch)
worker.convert_source_to_wiki_worker(task_self, source_id, "alice")
pages = WikiPagesRepository(pg_conn).list_for_source(source_id)
by_path = {p["path"]: p for p in pages}
assert by_path["/a.md"]["content"] == "The end.\n\nThe end. More"
def test_long_overlap_trimmed_between_chunks(
self, pg_conn, patch_worker_db, task_self, monkeypatch
):
from application import worker
overlap = "x" * 40
source_id = _seed_source(pg_conn)
_patch_store(
monkeypatch,
[
_chunk("head " + overlap, source="a.md", doc_id="a1"),
_chunk(overlap + " tail", source="a.md", doc_id="a2"),
],
)
_patch_reembed(monkeypatch)
worker.convert_source_to_wiki_worker(task_self, source_id, "alice")
pages = WikiPagesRepository(pg_conn).list_for_source(source_id)
by_path = {p["path"]: p for p in pages}
assert by_path["/a.md"]["content"] == "head " + overlap + "\n\n tail"
def test_no_chunks_leaves_kind_and_structure(
self, pg_conn, patch_worker_db, task_self, monkeypatch
):
from application import worker
@@ -203,24 +254,15 @@ class TestConvertSourceToWikiWorker:
original_structure = {"a.md": {"type": "text/markdown", "size_bytes": 1}}
source_id = _seed_source(pg_conn)
_patch_storage(monkeypatch, ["logo.png", "scan.pdf"])
_patch_reader(monkeypatch, [])
_patch_store(monkeypatch, [])
rebuild = MagicMock(name="rebuild")
monkeypatch.setattr(
worker, "rebuild_wiki_directory_structure", rebuild
)
delay = MagicMock(name="reembed_delay")
monkeypatch.setattr(
"application.api.user.tasks.reembed_wiki_page.delay", delay
)
monkeypatch.setattr(worker, "rebuild_wiki_directory_structure", rebuild)
delay = _patch_reembed(monkeypatch)
result = worker.convert_source_to_wiki_worker(
task_self, source_id, "alice"
)
result = worker.convert_source_to_wiki_worker(task_self, source_id, "alice")
assert result["status"] == "no_pages"
assert result["pages_created"] == 0
assert result["skipped"]
rebuild.assert_not_called()
delay.assert_not_called()
@@ -228,27 +270,44 @@ class TestConvertSourceToWikiWorker:
assert SourceConfig.parse(refreshed.get("config")).kind != "wiki"
assert refreshed.get("directory_structure") == original_structure
def test_traversal_filename_skipped(
def test_missing_path_chunk_skipped(
self, pg_conn, patch_worker_db, task_self, monkeypatch
):
from application import worker
source_id = _seed_source(pg_conn)
_patch_storage(monkeypatch, ["ok.md", "../evil.md"])
_patch_reader(
_patch_store(
monkeypatch,
[
Document(text="good", extra_info={"source": "ok.md"}),
Document(text="bad", extra_info={"source": "../evil.md"}),
_chunk("good", source="ok.md"),
_chunk("orphan", source=None),
],
)
monkeypatch.setattr(
"application.api.user.tasks.reembed_wiki_page.delay", MagicMock()
)
_patch_reembed(monkeypatch)
result = worker.convert_source_to_wiki_worker(
task_self, source_id, "alice"
result = worker.convert_source_to_wiki_worker(task_self, source_id, "alice")
assert result["pages_created"] == 1
pages = WikiPagesRepository(pg_conn).list_for_source(source_id)
assert {p["path"] for p in pages} == {"/ok.md"}
assert any(s["reason"] == "missing path" for s in result["skipped"])
def test_invalid_path_chunk_skipped(
self, pg_conn, patch_worker_db, task_self, monkeypatch
):
from application import worker
source_id = _seed_source(pg_conn)
_patch_store(
monkeypatch,
[
_chunk("good", source="ok.md"),
_chunk("bad", source="../evil.md"),
],
)
_patch_reembed(monkeypatch)
result = worker.convert_source_to_wiki_worker(task_self, source_id, "alice")
assert result["pages_created"] == 1
pages = WikiPagesRepository(pg_conn).list_for_source(source_id)
@@ -256,30 +315,91 @@ class TestConvertSourceToWikiWorker:
skipped = {(s["file"], s["reason"]) for s in result["skipped"]}
assert ("../evil.md", "invalid path") in skipped
def test_parse_failure_reason_distinct_from_binary(
def test_path_falls_back_to_filename_then_title(
self, pg_conn, patch_worker_db, task_self, monkeypatch
):
from application import worker
source_id = _seed_source(pg_conn)
# broken.md has a supported extension but produced no doc (parse/read
# failure); blob.bin is an unsupported/binary file.
_patch_storage(monkeypatch, ["guide.md", "broken.md", "blob.bin"])
_patch_reader(
_patch_store(
monkeypatch,
[Document(text="guide body", extra_info={"source": "guide.md"})],
)
monkeypatch.setattr(
"application.api.user.tasks.reembed_wiki_page.delay", MagicMock()
[
_chunk("from filename", source=None, filename="byname.md"),
_chunk("from title", source=None, title="bytitle.md"),
],
)
_patch_reembed(monkeypatch)
result = worker.convert_source_to_wiki_worker(
task_self, source_id, "alice"
)
result = worker.convert_source_to_wiki_worker(task_self, source_id, "alice")
reasons = {s["file"]: s["reason"] for s in result["skipped"]}
assert reasons["broken.md"] == "could not read/parse"
assert reasons["blob.bin"] == "no extractable text"
assert result["pages_created"] == 2
pages = WikiPagesRepository(pg_conn).list_for_source(source_id)
assert {p["path"] for p in pages} == {"/byname.md", "/bytitle.md"}
def test_crawler_chunk_uses_file_path_not_url(
self, pg_conn, patch_worker_db, task_self, monkeypatch
):
from application import worker
source_id = _seed_source(pg_conn)
_patch_store(
monkeypatch,
[
_chunk(
"setup body",
source="https://docs.x.com/guides/setup",
file_path="guides/setup.md",
),
],
)
_patch_reembed(monkeypatch)
result = worker.convert_source_to_wiki_worker(task_self, source_id, "alice")
assert result["status"] == "converted"
assert result["skipped"] == []
pages = WikiPagesRepository(pg_conn).list_for_source(source_id)
assert {p["path"] for p in pages} == {"/guides/setup.md"}
def test_connector_chunks_kept_separate_by_file_name(
self, pg_conn, patch_worker_db, task_self, monkeypatch
):
from application import worker
source_id = _seed_source(pg_conn)
_patch_store(
monkeypatch,
[
_chunk("page a", source="confluence", file_name="Page A", doc_id="a"),
_chunk("page b", source="confluence", file_name="Page B", doc_id="b"),
],
)
_patch_reembed(monkeypatch)
result = worker.convert_source_to_wiki_worker(task_self, source_id, "alice")
assert result["pages_created"] == 2
pages = WikiPagesRepository(pg_conn).list_for_source(source_id)
assert {p["path"] for p in pages} == {"/Page A", "/Page B"}
def test_url_only_chunk_normalized_not_skipped(
self, pg_conn, patch_worker_db, task_self, monkeypatch
):
from application import worker
source_id = _seed_source(pg_conn)
_patch_store(
monkeypatch,
[_chunk("body", source="https://x.com/a/b")],
)
_patch_reembed(monkeypatch)
result = worker.convert_source_to_wiki_worker(task_self, source_id, "alice")
assert result["status"] == "converted"
assert result["skipped"] == []
pages = WikiPagesRepository(pg_conn).list_for_source(source_id)
assert {p["path"] for p in pages} == {"/a/b"}
def test_already_wiki_returns_early(
self, pg_conn, patch_worker_db, task_self, monkeypatch
@@ -287,19 +407,11 @@ class TestConvertSourceToWikiWorker:
from application import worker
source_id = _seed_source(pg_conn, config={"kind": "wiki"})
reader = MagicMock(name="reader")
monkeypatch.setattr(
worker, "SimpleDirectoryReader", lambda *a, **kw: reader
)
delay = MagicMock(name="reembed_delay")
monkeypatch.setattr(
"application.api.user.tasks.reembed_wiki_page.delay", delay
)
store = _patch_store(monkeypatch, [_chunk("body", source="a.md")])
delay = _patch_reembed(monkeypatch)
result = worker.convert_source_to_wiki_worker(
task_self, source_id, "alice"
)
result = worker.convert_source_to_wiki_worker(task_self, source_id, "alice")
assert result["status"] == "already_wiki"
reader.load_data.assert_not_called()
store.get_chunks.assert_not_called()
delay.assert_not_called()