Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
117 changes: 116 additions & 1 deletion gknowledge/core/ingestion.py
Original file line number Diff line number Diff line change
Expand Up @@ -9,6 +9,8 @@

from __future__ import annotations

import json
import re
import shutil
import threading
from dataclasses import dataclass
Expand All @@ -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
Expand Down Expand Up @@ -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()}
Expand Down
10 changes: 10 additions & 0 deletions gknowledge/core/store.py
Original file line number Diff line number Diff line change
Expand Up @@ -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()

Expand Down Expand Up @@ -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(
Expand Down Expand Up @@ -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",
Expand Down
97 changes: 97 additions & 0 deletions gknowledge/ui/window.py
Original file line number Diff line number Diff line change
Expand Up @@ -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)
Expand Down Expand Up @@ -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)
Expand Down Expand Up @@ -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)

Expand All @@ -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:
Expand All @@ -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)
Expand All @@ -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:
Expand All @@ -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):
Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -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):
Expand Down
Loading