ragflow_sync_cli/commands/sync_confluence.py - master - dictator/ragflow_sync

ragflow_sync_cli/commands/sync_confluence.py
165 строк5 KB
Developer
make document parsing enabled by default after upload
15 май 2026, 12:36
15 май 2026, 12:3696c1c05
"""
Sync Confluence to RAGFlow command.
"""

import logging
import sys
from typing import Optional

from ragflow_uploader.client import RAGFlowClient
from ragflow_uploader.uploader import Document, DocumentUploader
from jira_confluence_retriever.config import ConfluenceConfig
from jira_confluence_retriever.retriever import ConfluencePage
from cleanup.clean import preprocess_text


logger = logging.getLogger(__name__)


def sync_confluence(
    client: RAGFlowClient,
    config: ConfluenceConfig,
    dataset_name: str,
    spaces: Optional[list[str]] = None,
    query: Optional[str] = None,
    page_ids: Optional[list[str]] = None,
    limit: int = 100,
    create_dataset: bool = True,
    recursive: bool = True,
    cleanup: bool = True,
    parse: bool = True,
    wait_for_parsing: bool = False,
) -> None:
    """Sync Confluence pages to RAGFlow."""
    from jira_confluence_retriever.retriever import ConfluenceRetriever

    logger.info(f"Syncing Confluence to dataset: {dataset_name}")

    try:
        # Initialize retriever
        retriever = ConfluenceRetriever(config)

        # Get or create dataset
        uploader = DocumentUploader(client)
        if create_dataset:
            dataset_id = uploader.get_or_create_dataset(
                dataset_name,
                description="Confluence wiki pages",
            )
        else:
            # Find existing dataset
            datasets = client.list_datasets(name=dataset_name)
            if not datasets:
                logger.error(f"Dataset '{dataset_name}' not found")
                sys.exit(1)
            dataset_id = datasets[0]["id"]

        # Fetch pages from Confluence
        pages = retriever.fetch_data(
            space_keys=spaces,
            query=query,
            page_ids=page_ids,
            limit=limit,
            recursive=recursive,
        )

        logger.info(f"Fetched {len(pages)} pages from Confluence")

        # Convert to documents
        documents = _convert_pages_to_documents(pages, cleanup=cleanup)

        # Fetch existing documents for deduplication
        try:
            existing_docs = client.list_documents(dataset_id, page_size=1000)
        except Exception as e:
            logger.warning(f"Failed to list existing documents: {e}")
            existing_docs = []

        # Upload to RAGFlow
        result = uploader.upload_documents(
            dataset_id=dataset_id,
            documents=documents,
            existing_documents=existing_docs,
            dedup_key="id",
            parse=parse,
            wait_for_parsing=wait_for_parsing,
        )

        # Print results
        _print_sync_results(result)

        if not result.success:
            sys.exit(1)

    except Exception as e:
        logger.error(f"Confluence sync failed: {e}")
        sys.exit(1)


def _convert_pages_to_documents(
    pages: list[ConfluencePage], cleanup: bool = True
) -> list[Document]:
    """Convert ConfluencePage objects to Document objects."""
    documents = []

    for page in pages:
        # Build document content with metadata
        content_parts = []

        # Add metadata header
        content_parts.append(f"# {page.title}\n")
        content_parts.append(f"**Space:** {page.space}\n")
        if page.url:
            content_parts.append(f"**URL:** {page.url}\n")
        content_parts.append(f"**Version:** {page.version}\n")
        content_parts.append("\n---\n\n")

        # Add page content
        content_parts.append(page.content)

        # Combine all parts
        full_content = "".join(content_parts)

        # Apply cleanup if enabled
        if cleanup:
            full_content = preprocess_text(full_content)

        # Create document
        doc = Document(
            title=page.title,
            content=full_content,
            metadata={
                "source": "confluence",
                "page_id": page.id,
                "space": page.space,
                "url": page.url,
                "version": page.version,
            },
        )

        documents.append(doc)

    return documents


def _print_sync_results(result) -> None:
    """Print sync results to stdout."""
    print("\n" + "=" * 50)
    print("Sync Results:")
    print("=" * 50)
    print(f"Status: {'✓ Success' if result.success else '✗ Failed'}")
    print(f"Documents created: {result.documents_created}")
    print(f"Documents updated: {result.documents_updated}")
    print(f"Documents failed: {result.documents_failed}")

    if result.documents_parsed > 0:
        print(f"Documents parsed: {result.documents_parsed}")
    if result.documents_parse_failed > 0:
        print(f"Documents parse failed: {result.documents_parse_failed}")

    if result.errors:
        print("\nErrors:")
        for error in result.errors:
            print(f"  - {error}")

    print()