diff --git a/gknowledge/core/ingestion.py b/gknowledge/core/ingestion.py index a2e17ef..5443ce2 100644 --- a/gknowledge/core/ingestion.py +++ b/gknowledge/core/ingestion.py @@ -9,6 +9,8 @@ from __future__ import annotations +import json +import re import shutil import threading from dataclasses import dataclass @@ -20,7 +22,8 @@ from . import parsers from .extraction import ExtractionPipeline from .hashing import hash_file -from .models import Job, JobState, Source, SourceStatus, new_id, now +from .knowledge import KnowledgeWriter +from .models import Job, JobState, PageKind, Source, SourceStatus, new_id, now from .project import Project # progress(event: str, job: Job, message: str) -> None @@ -100,6 +103,118 @@ def import_folder(self, folder: str | Path) -> ImportOutcome: imported.append(src) return ImportOutcome(imported=imported, skipped=skipped) + def reingest_source(self, source_id: str) -> Source | None: + """Clear stale provenance and queue a fresh extraction for a source.""" + if self.is_running(): + raise RuntimeError("Cannot re-ingest a source while ingestion is running.") + source = self.store.get_source(source_id) + if source is None: + return None + source_path = self._managed_path(self.project.sources_dir, source.rel_path) + if source_path is None or not source_path.is_file(): + raise FileNotFoundError(f"Stored source file is missing: {source.rel_path}") + + writer = self._detach_source(source) + self.store.delete_jobs_for_source(source.id) + self.store.delete_reviews_for_source(source.id) + source.status = SourceStatus.PENDING + self.store.upsert_source(source) + self.store.upsert_job(Job(source_id=source.id, state=JobState.QUEUED)) + self.store.log("reingest", source.title) + writer.rebuild_index() + writer.rebuild_overview() + writer.append_log("re-ingest", source.title) + return source + + def remove_source(self, source_id: str) -> Source | None: + """Remove the source and clean up knowledge no longer backed by it.""" + if self.is_running(): + raise RuntimeError("Cannot remove a source while ingestion is running.") + source = self.store.get_source(source_id) + if source is None: + return None + + source_path = self._managed_path(self.project.sources_dir, source.rel_path) + writer = self._detach_source(source) + if source_path is not None and source_path.is_file(): + source_path.unlink() + self.store.delete_source(source.id) + self.store.log("source.remove", source.title) + writer.rebuild_index() + writer.rebuild_overview() + writer.append_log("remove source", source.title) + return source + + def _detach_source(self, source: Source) -> KnowledgeWriter: + """Remove a source's provenance, deleting only pages it solely supported.""" + writer = KnowledgeWriter(self.project) + pages = self.store.pages_for_source(source.id) + source_page = next((p for p in pages if p.kind == PageKind.SOURCE), None) + source_slug = source_page.slug if source_page else None + + for page in pages: + if page.kind == PageKind.SOURCE: + self._delete_page(page) + continue + + page.sources = [sid for sid in page.sources if sid != source.id] + if not page.sources: + self._delete_page(page) + continue + + page.body = writer.read_body(page) + if source_slug: + page.body = re.sub( + rf"(?m)^[ \t]*-\s*\[\[{re.escape(source_slug)}(?:\|[^\]]+)?\]\][ \t]*(?:\r?\n|$)", + "", + page.body, + ) + page.links = [slug for slug in page.links if slug != source_slug] + writer.write_page(page) + + self._remove_source_mentions(source.id) + return writer + + def _delete_page(self, page) -> None: + page_path = self._managed_path(self.project.knowledge_dir, page.rel_path) + if page_path is not None and page_path.is_file(): + page_path.unlink() + self.store.delete_page(page.slug) + + @staticmethod + def _managed_path(directory: str | Path, rel_path: str) -> Path | None: + root = Path(directory).resolve() + path = (root / rel_path).resolve() + return path if path != root and path.is_relative_to(root) else None + + def _remove_source_mentions(self, source_id: str) -> None: + path = self.project.cache_dir / "mentions.json" + try: + mentions = json.loads(path.read_text("utf-8")) + except (OSError, ValueError): + return + if not isinstance(mentions, dict): + return + changed = False + for key, entry in list(mentions.items()): + if not isinstance(entry, dict): + continue + sources = entry.get("sources", []) + if not isinstance(sources, list): + continue + remaining = [sid for sid in sources if sid != source_id] + if remaining == sources: + continue + changed = True + if remaining: + entry["sources"] = remaining + else: + del mentions[key] + if changed: + path.write_text( + json.dumps(mentions, indent=1, ensure_ascii=False), "utf-8" + ) + def queue_all_pending_sources(self) -> int: """Ensure every non-ingested source has a queued job. Returns count.""" existing_job_sources = {j.source_id for j in self.store.list_jobs()} diff --git a/gknowledge/core/store.py b/gknowledge/core/store.py index a57c959..6e34cb1 100644 --- a/gknowledge/core/store.py +++ b/gknowledge/core/store.py @@ -204,6 +204,8 @@ def list_sources(self) -> list[Source]: return [Source.from_row(dict(r)) for r in cur.fetchall()] def delete_source(self, source_id: str) -> None: + self.delete_jobs_for_source(source_id) + self.delete_reviews_for_source(source_id) self._conn.execute("DELETE FROM sources WHERE id=?", (source_id,)) self._conn.commit() @@ -279,6 +281,10 @@ def upsert_review(self, item: ReviewItem) -> None: ) self._conn.commit() + def delete_reviews_for_source(self, source_id: str) -> None: + self._conn.execute("DELETE FROM reviews WHERE source_id=?", (source_id,)) + self._conn.commit() + def list_reviews(self, status: str | None = None) -> list[ReviewItem]: if status: cur = self._conn.execute( @@ -307,6 +313,10 @@ def upsert_job(self, j: Job) -> None: ) self._conn.commit() + def delete_jobs_for_source(self, source_id: str) -> None: + self._conn.execute("DELETE FROM jobs WHERE source_id=?", (source_id,)) + self._conn.commit() + def pending_jobs(self) -> list[Job]: cur = self._conn.execute( "SELECT * FROM jobs WHERE state IN (?,?) ORDER BY created_at", diff --git a/gknowledge/ui/window.py b/gknowledge/ui/window.py index 3ee7afd..8460d6b 100644 --- a/gknowledge/ui/window.py +++ b/gknowledge/ui/window.py @@ -78,6 +78,7 @@ def __init__(self, app: Adw.Application): self.app = app self.project: Project | None = None self.ingestion: IngestionManager | None = None + self._selected_source_id: str | None = None self.provider = None self.set_title(APP_NAME) @@ -278,6 +279,7 @@ def _add_section(self, name, title, icon, sidebar: Gtk.Widget, content: Gtk.Widg def set_project(self, project: Project): if self.project: self.project.close() + self._selected_source_id = None self.project = project self.provider = build_provider(project.settings.as_dict()) self.ingestion = IngestionManager(project, self.provider) @@ -340,9 +342,20 @@ def _build_sources_pane(self): self.ingest_btn = Gtk.Button(label="Ingest Queued") self.ingest_btn.add_css_class("suggested-action") self.ingest_btn.set_action_name("win.ingest") + self.reingest_btn = Gtk.Button(label="Re-ingest") + self.reingest_btn.set_sensitive(False) + self.reingest_btn.connect("clicked", self._on_reingest_source) + self.remove_source_btn = Gtk.Button(label="Remove") + self.remove_source_btn.add_css_class("destructive-action") + self.remove_source_btn.set_sensitive(False) + self.remove_source_btn.connect("clicked", self._on_remove_source) + source_actions = Gtk.Box(orientation=Gtk.Orientation.HORIZONTAL, spacing=8) + source_actions.append(self.reingest_btn) + source_actions.append(self.remove_source_btn) self.progress = Gtk.ProgressBar(show_text=True, visible=False) bottom = Gtk.Box(orientation=Gtk.Orientation.VERTICAL, spacing=8, margin_top=8, margin_bottom=8, margin_start=8, margin_end=8) + bottom.append(source_actions) bottom.append(self.progress) bottom.append(self.ingest_btn) @@ -360,8 +373,11 @@ def _build_sources_pane(self): def _refresh_sources(self): _clear_listbox(self.sources_list) if not self.project: + self._selected_source_id = None + self._update_source_action_buttons() return sources = self.project.store.list_sources() + selected_row = None if not sources: self.sources_list.append(_placeholder_row("No sources yet. Import some.")) for s in sources: @@ -370,12 +386,22 @@ def _refresh_sources(self): row.add_prefix(Gtk.Image.new_from_icon_name("text-x-generic-symbolic")) row.source = s self.sources_list.append(row) + if s.id == self._selected_source_id: + selected_row = row + if selected_row: + self.sources_list.select_row(selected_row) + self._on_source_selected(self.sources_list, selected_row) + else: + self._selected_source_id = None + self.source_stack.set_visible_child_name("empty") + self._update_source_action_buttons() self._update_ingest_button() def _on_source_selected(self, _list, row): source: Source | None = getattr(row, "source", None) if not source or not self.project: return + self._selected_source_id = source.id pages = self.project.store.pages_for_source(source.id) writer = KnowledgeWriter(self.project) summary = next((p for p in pages if p.kind == PageKind.SOURCE), None) @@ -388,6 +414,17 @@ def _on_source_selected(self, _list, row): body = (f"# {source.title}\n\n(Not yet ingested.)\n\n" + parsers.parse(fp)[:4000]) self._show_detail(self.source_stack, self.source_detail, body) + self._update_source_action_buttons() + + def _update_source_action_buttons(self): + available = bool( + self._selected_source_id and self.project and self.ingestion + and not self.ingestion.is_running() + ) + if hasattr(self, "reingest_btn"): + self.reingest_btn.set_sensitive(available) + if hasattr(self, "remove_source_btn"): + self.remove_source_btn.set_sensitive(available) def _update_ingest_button(self): if not self.project: @@ -397,6 +434,7 @@ def _update_ingest_button(self): n = len(pending) self.ingest_btn.set_label(f"Ingest Queued ({n})" if n else "Ingest Queued") self.ingest_btn.set_sensitive(n > 0 and not self.ingestion.is_running()) + self._update_source_action_buttons() # -- knowledge pane ------------------------------------------------------ def _build_knowledge_pane(self): @@ -636,6 +674,64 @@ def _do_import_folder(self, folder): self.toast(f"Imported {len(outcome.imported)}, " f"skipped {len(outcome.skipped)} duplicate(s)") + def _on_reingest_source(self, *_a): + if not (self.project and self.ingestion and self._selected_source_id): + return + try: + source = self.ingestion.reingest_source(self._selected_source_id) + except (OSError, RuntimeError, ValueError) as exc: + self.toast(f"Could not re-ingest source: {exc}") + return + if source is None: + self.toast("That source no longer exists.") + self.refresh_all() + return + self.refresh_all() + self.toast(f"Re-ingesting {source.title}") + self._on_ingest() + + def _on_remove_source(self, *_a): + if not (self.project and self.ingestion and self._selected_source_id): + return + source = self.project.store.get_source(self._selected_source_id) + if source is None: + self.toast("That source no longer exists.") + self.refresh_all() + return + dialog = Adw.MessageDialog( + transient_for=self, + heading="Remove source?", + body=f"Remove “{source.title}” and detach it from generated knowledge pages?", + ) + dialog.add_response("cancel", "Cancel") + dialog.add_response("remove", "Remove") + dialog.set_default_response("cancel") + dialog.set_close_response("cancel") + dialog.set_response_appearance("remove", Adw.ResponseAppearance.DESTRUCTIVE) + dialog.connect( + "response", + lambda _dialog, response: self._remove_source(source.id) + if response == "remove" else None, + ) + dialog.present() + + def _remove_source(self, source_id: str): + if not (self.project and self.ingestion): + return + try: + source = self.ingestion.remove_source(source_id) + except (OSError, RuntimeError, ValueError) as exc: + self.toast(f"Could not remove source: {exc}") + return + if source is None: + self.toast("That source no longer exists.") + self.refresh_all() + return + self._selected_source_id = None + self.source_stack.set_visible_child_name("empty") + self.refresh_all() + self.toast(f"Removed source: {source.title}") + def _on_ingest(self, *_a): if not self.ingestion or self.ingestion.is_running(): return @@ -667,6 +763,7 @@ def done(): GLib.idle_add(done) self.ingestion.start(progress=progress, on_finish=finished) + self._update_source_action_buttons() # -- helpers ------------------------------------------------------------- def _warn(self, text: str): diff --git a/tests/test_core.py b/tests/test_core.py index d8caf48..3c7c8f2 100644 --- a/tests/test_core.py +++ b/tests/test_core.py @@ -9,7 +9,8 @@ sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) from gknowledge.core.ingestion import IngestionManager -from gknowledge.core.models import PageKind, SourceStatus +from gknowledge.core.knowledge import KnowledgeWriter +from gknowledge.core.models import Page, PageKind, Source, SourceStatus from gknowledge.core.project import Project, Settings from gknowledge.core.query import QueryEngine from gknowledge.providers import build_provider @@ -101,6 +102,119 @@ def test_import_ingest_and_dedup(tmp_path): proj.close() +def test_reingest_clears_stale_pages_and_rebuilds_source(tmp_path): + proj = make_project(tmp_path) + src_file = tmp_path / "solar.md" + src_file.write_text(SAMPLE, "utf-8") + mgr = IngestionManager(proj, build_provider(proj.settings.as_dict())) + source = mgr.import_file(src_file) + assert source is not None + assert mgr.process_pending() == 1 + + source = proj.store.get_source(source.id) + source_page = next( + p for p in proj.store.pages_for_source(source.id) + if p.kind == PageKind.SOURCE + ) + other = Source(title="Second source", rel_path="second.md", + status=SourceStatus.INGESTED) + proj.store.upsert_source(other) + writer = KnowledgeWriter(proj) + stale = writer.write_page(Page( + kind=PageKind.CONCEPT, + title="Stale only source", + slug="stale-only-source", + summary="Only supported by the selected source.", + body="# Stale only source\n\nOnly supported by the selected source.", + sources=[source.id], + )) + shared = writer.write_page(Page( + kind=PageKind.CONCEPT, + title="Shared lifecycle concept", + slug="shared-lifecycle-concept", + summary="Supported by two sources.", + body=(f"# Shared lifecycle concept\n\n## Sources\n\n" + f"- [[{source_page.slug}]]\n- [[second-source]]"), + sources=[source.id, other.id], + )) + + queued = mgr.reingest_source(source.id) + assert queued is not None + assert queued.status == SourceStatus.PENDING + assert proj.store.get_page(source_page.slug) is None + assert proj.store.get_page(stale.slug) is None + detached = proj.store.get_page(shared.slug) + assert detached is not None + assert detached.sources == [other.id] + shared_md = (proj.knowledge_dir / detached.rel_path).read_text("utf-8") + assert f"[[{source_page.slug}]]" not in shared_md + assert "[[second-source]]" in shared_md + + assert mgr.process_pending() == 1 + refreshed = proj.store.get_source(source.id) + assert refreshed is not None + assert refreshed.status == SourceStatus.INGESTED + assert any( + p.kind == PageKind.SOURCE and source.id in p.sources + for p in proj.store.list_pages() + ) + assert proj.store.get_page(stale.slug) is None + proj.close() + + +def test_remove_source_detaches_shared_pages_and_deletes_orphans(tmp_path): + proj = make_project(tmp_path) + src_file = tmp_path / "solar.md" + src_file.write_text(SAMPLE, "utf-8") + mgr = IngestionManager(proj, build_provider(proj.settings.as_dict())) + source = mgr.import_file(src_file) + assert source is not None + assert mgr.process_pending() == 1 + + source_page = next( + p for p in proj.store.pages_for_source(source.id) + if p.kind == PageKind.SOURCE + ) + source_file = proj.sources_dir / source.rel_path + other = Source(title="Second source", rel_path="second.md", + status=SourceStatus.INGESTED) + proj.store.upsert_source(other) + writer = KnowledgeWriter(proj) + orphan = writer.write_page(Page( + kind=PageKind.ENTITY, + title="Orphan entity", + slug="orphan-entity", + summary="Supported only by the selected source.", + body="# Orphan entity\n\n## Sources\n\n" + f"- [[{source_page.slug}]]", + sources=[source.id], + )) + shared = writer.write_page(Page( + kind=PageKind.CONCEPT, + title="Shared lifecycle concept", + slug="shared-lifecycle-concept", + summary="Supported by two sources.", + body=(f"# Shared lifecycle concept\n\n## Sources\n\n" + f"- [[{source_page.slug}]]\n- [[second-source]]"), + sources=[source.id, other.id], + )) + + removed = mgr.remove_source(source.id) + assert removed is not None + assert not source_file.exists() + assert proj.store.get_source(source.id) is None + assert proj.store.get_page(source_page.slug) is None + assert proj.store.get_page(orphan.slug) is None + remaining = proj.store.get_page(shared.slug) + assert remaining is not None + assert remaining.sources == [other.id] + shared_md = (proj.knowledge_dir / remaining.rel_path).read_text("utf-8") + assert source_page.slug not in shared_md + assert "[[second-source]]" in shared_md + assert mgr.remove_source(source.id) is None + proj.close() + + def test_query_returns_citations(tmp_path): proj = make_project(tmp_path) src_file = tmp_path / "solar.md" @@ -155,7 +269,9 @@ def test_import_and_ingest_epub(tmp_path): for name, fn in (("a", test_project_layout), ("b", test_import_ingest_and_dedup), ("c", test_query_returns_citations), - ("d", test_import_and_ingest_epub)): + ("d", test_import_and_ingest_epub), + ("e", test_reingest_clears_stale_pages_and_rebuilds_source), + ("f", test_remove_source_detaches_shared_pages_and_deletes_orphans)): d = tmp / name d.mkdir(parents=True, exist_ok=True) fn(d)