41 lines
1.8 KiB
Python
41 lines
1.8 KiB
Python
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass
|
|
from typing import Any
|
|
|
|
from .card import AGENT_CARD, AgentCard
|
|
from .kab_ingestion.connectors.github import GitHubSCMConnector
|
|
from .kab_ingestion.connectors.sharepoint import SharePointCMSConnector
|
|
from .kab_ingestion.domain.models import IngestionRequest, IngestionResult
|
|
from .publication import publish_document
|
|
from .settings import Settings, get_settings
|
|
from .validation import validate_document
|
|
|
|
|
|
@dataclass
|
|
class IngestionService:
|
|
settings: Settings
|
|
scm: Any
|
|
cms: Any = None
|
|
|
|
def ingest(self, request: IngestionRequest) -> list[IngestionResult]:
|
|
if request.source not in self.settings.allowed_sources:
|
|
return []
|
|
results = []
|
|
for document in self.scm.fetch(request):
|
|
report = validate_document(document)
|
|
if not report.valid:
|
|
from .kab_ingestion.domain.models import IngestionResult
|
|
results.append(IngestionResult(document, False, "invalid", "; ".join(report.errors)))
|
|
continue
|
|
results.append(publish_document(document, self.cms, self.settings.publish_enabled, self.settings.require_governance_approval))
|
|
return results
|
|
|
|
|
|
def create_agent_app(settings: Settings | None = None) -> dict[str, Any]:
|
|
config = settings or get_settings()
|
|
scm = GitHubSCMConnector(config.github_token, config.github_api_base)
|
|
cms = SharePointCMSConnector(config.sharepoint_site_url, config.sharepoint_access_token) if config.sharepoint_site_url and config.sharepoint_access_token else None
|
|
service = IngestionService(config, scm, cms)
|
|
return {"card": AGENT_CARD, "service": service, "routes": {"/health": "health", "/card": "card", "/ingest": "ingest"}, "triggers": {"github": "github_trigger"}}
|