Self-hosted knowledge management for SMEs: a split-screen Markdown editor whose sections an LLM refines while you write, and RAG question answering over the documents that result. FastAPI + Postgres/pgvector on the back, SvelteKit on the front, everything OpenAI-compatible and self-hostable. Squashed into a single commit; the development history stays local. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01CA43ZJda8Rbp2hKXNy8f6b
98 lines
3.3 KiB
Python
98 lines
3.3 KiB
Python
"""Built-in help documents: Markdown files → documents table.
|
|
|
|
The help pages that describe Pablan itself ship with the product and live in
|
|
the repo-level help/ directory (product content, not code). They are
|
|
re-imported on every start, so a release always carries the current
|
|
documentation, and they are flagged `is_builtin` so the API refuses to edit
|
|
or delete them.
|
|
"""
|
|
|
|
import logging
|
|
from pathlib import Path
|
|
|
|
import yaml
|
|
from sqlalchemy import select
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.config import get_settings
|
|
from app.ingestion.handlers import INDEX_DOCUMENT
|
|
from app.ingestion.queue import enqueue
|
|
from app.models import (
|
|
Document,
|
|
DocumentStatus,
|
|
DocumentVisibility,
|
|
)
|
|
|
|
logger = logging.getLogger("pablan.help")
|
|
|
|
FRONTMATTER_SEPARATOR = "---"
|
|
META_KEY = "help_key"
|
|
|
|
|
|
class HelpImportError(Exception):
|
|
"""Malformed help file — a packaging bug, never user input."""
|
|
|
|
|
|
def parse_help_document(source: str) -> tuple[str, str, str]:
|
|
"""Split the `key`/`title` frontmatter from the Markdown body."""
|
|
if not source.startswith(FRONTMATTER_SEPARATOR):
|
|
raise HelpImportError("Help document must start with YAML frontmatter.")
|
|
_, frontmatter, body = source.split(FRONTMATTER_SEPARATOR, 2)
|
|
try:
|
|
meta = yaml.safe_load(frontmatter)
|
|
except yaml.YAMLError as exc:
|
|
raise HelpImportError(f"Invalid frontmatter: {type(exc).__name__}") from None
|
|
if not isinstance(meta, dict) or not meta.get("key") or not meta.get("title"):
|
|
raise HelpImportError("Help frontmatter needs at least 'key' and 'title'.")
|
|
return str(meta["key"]), str(meta["title"]), body.strip()
|
|
|
|
|
|
async def import_help_documents(db: AsyncSession) -> int:
|
|
"""Upsert every help/*.md by its key. Returns the number re-indexed."""
|
|
directory = Path(get_settings().help_dir)
|
|
if not directory.is_dir():
|
|
logger.warning("help directory missing", extra={"event": "help_import_skipped"})
|
|
return 0
|
|
|
|
reindexed = 0
|
|
for path in sorted(directory.glob("*.md")):
|
|
key, title, body = parse_help_document(path.read_text())
|
|
existing = (
|
|
await db.execute(
|
|
select(Document).where(
|
|
Document.is_builtin.is_(True),
|
|
Document.meta[META_KEY].astext == key,
|
|
)
|
|
)
|
|
).scalar_one_or_none()
|
|
|
|
if existing is None:
|
|
document = Document(
|
|
title=title,
|
|
status=DocumentStatus.published,
|
|
# Help is for everyone; it has no author and no department.
|
|
visibility=DocumentVisibility.public,
|
|
content_md=body,
|
|
meta={META_KEY: key},
|
|
is_builtin=True,
|
|
)
|
|
db.add(document)
|
|
await db.flush()
|
|
elif existing.content_md == body and existing.title == title:
|
|
continue # unchanged — no need to re-embed
|
|
else:
|
|
existing.title = title
|
|
existing.content_md = body
|
|
existing.meta = {**existing.meta, META_KEY: key}
|
|
document = existing
|
|
|
|
await enqueue(db, INDEX_DOCUMENT, {"document_id": str(document.id)})
|
|
reindexed += 1
|
|
|
|
await db.commit()
|
|
logger.info(
|
|
"help documents imported",
|
|
extra={"event": "help_import", "reindexed": reindexed},
|
|
)
|
|
return reindexed
|