ci: migrate GitHub actions to Gitea actions

shared/tools/monorepo-cli/src/monorepo/knowledge/ingestion/pipeline.py

@@ -127,11 +127,13 @@ def register_default_strategies() -> None:
     from monorepo.knowledge.sources.jira import JiraSource
     from monorepo.knowledge.sources.markdown import MarkdownSource
     from monorepo.knowledge.sources.pdf import PDFSource
+    from monorepo.knowledge.sources.docx import DocxSource
     from monorepo.knowledge.sources.wissensdatenbank import WissensdatenbankSource
 
     defaults: dict[str, type[SourceStrategy]] = {
         "markdown": MarkdownSource,
         "pdf": PDFSource,
+        "docx": DocxSource,
         "confluence": ConfluenceSource,
         "jira": JiraSource,
         "email": EmailSource,

shared/tools/monorepo-cli/src/monorepo/knowledge/sources/docx.py

@@ -0,0 +1,151 @@
+"""DOCX-Quellstrategie für die ETL-Pipeline.
+
+Implementiert die Extraktion von Text aus DOCX-Dateien.
+"""
+
+from __future__ import annotations
+
+import logging
+from datetime import date, datetime
+from pathlib import Path
+
+from monorepo.knowledge.artifact import Artifact, ArtifactMetadata, compute_content_hash
+from monorepo.knowledge.sources.base import (
+    ExtractionResult,
+    SourceConfig,
+    SourceError,
+    SourceStrategy,
+)
+
+logger = logging.getLogger(__name__)
+
+DOCX_EXTENSIONS = {".docx"}
+
+
+def _extract_docx_text(file_path: Path) -> str:
+    try:
+        import docx  # type: ignore[import-untyped]
+    except ImportError:
+        raise ImportError(
+            "Keine DOCX-Bibliothek verfügbar. "
+            "Installiere 'python-docx': pip install python-docx"
+        )
+
+    doc = docx.Document(str(file_path))
+    full_text = []
+    for para in doc.paragraphs:
+        full_text.append(para.text)
+    
+    return "\n\n".join(full_text)
+
+
+class DocxSource(SourceStrategy):
+    """Quellstrategie für DOCX-Dateien."""
+
+    def extract(self, config: SourceConfig, context: str) -> ExtractionResult:
+        result = ExtractionResult()
+
+        source_path = self._resolve_source_path(config)
+        if source_path is None:
+            result.errors.append(
+                SourceError(
+                    source_name=config.name,
+                    error_type="parse",
+                    message="Kein gültiger Pfad in config.params['path'] oder config.params['directory']",
+                    timestamp=datetime.now().isoformat(),
+                    retry=False,
+                )
+            )
+            return result
+
+        files = self._collect_files(source_path)
+        if not files:
+            logger.info("Keine DOCX-Dateien gefunden in: %s", source_path)
+            return result
+
+        for file_path in files:
+            try:
+                artifact = self._process_file(file_path, config, context)
+                if artifact is not None:
+                    result.artifacts.append(artifact)
+            except Exception as e:
+                logger.error("Fehler bei der Verarbeitung von %s: %s", file_path, e)
+                result.errors.append(
+                    SourceError(
+                        source_name=config.name,
+                        error_type="parse",
+                        message=f"Fehler bei {file_path.name}: {e}",
+                        timestamp=datetime.now().isoformat(),
+                        retry=True,
+                        artifact_path=str(file_path),
+                    )
+                )
+
+        return result
+
+    def supports_incremental(self) -> bool:
+        return True
+
+    def _resolve_source_path(self, config: SourceConfig) -> Path | None:
+        path_str = config.params.get("path") or config.params.get("directory") or ""
+        if not path_str:
+            return None
+        source_path = Path(path_str)
+        if not source_path.exists():
+            return None
+        return source_path
+
+    def _collect_files(self, source_path: Path) -> list[Path]:
+        if source_path.is_file():
+            if source_path.suffix.lower() in DOCX_EXTENSIONS:
+                return [source_path]
+            return []
+
+        files: list[Path] = []
+        for ext in DOCX_EXTENSIONS:
+            files.extend(source_path.rglob(f"*{ext}"))
+        return sorted(files)
+
+    def _process_file(
+        self, file_path: Path, config: SourceConfig, context: str
+    ) -> Artifact | None:
+        text = ""
+        try:
+            text = _extract_docx_text(file_path)
+        except Exception as e:
+            logger.warning("DOCX Extraktion fehlgeschlagen für %s: %s", file_path, e)
+
+        return self._create_artifact(file_path, text, context)
+
+    def _create_artifact(
+        self, file_path: Path, text: str, context: str
+    ) -> Artifact:
+        title = file_path.stem.replace("-", " ").replace("_", " ").strip()
+        if not title:
+            title = file_path.name
+
+        content_hash = compute_content_hash(text) if text else ""
+
+        metadata = ArtifactMetadata(
+            type="reference",
+            title=title,
+            tags=["docx"],
+            source_context=context,
+            created=date.today(),
+            updated=date.today(),
+            shareable=False,
+            content_hash=content_hash,
+            source={
+                "type": "docx",
+                "file": str(file_path),
+            },
+            category="references",
+        )
+
+        content = f"# {title}\n\n{text}\n" if text else f"# {title}\n\n"
+
+        return Artifact(
+            metadata=metadata,
+            content=content,
+            file_path=file_path,
+        )