feat: add CAD workflow for DWG/DXF file translation

.gitignore

@@ -6,6 +6,12 @@ dist/
 wheels/
 *.egg-info
 tests/
+!tests/
+tests/*
+!tests/test_workflow/
+tests/test_workflow/*
+!tests/test_workflow/test_cad_workflow.py
+!tests/test_workflow/test_cad_simple.dxf
 tests/resource/
 tests/test_data/
 htmlcov/

docutranslate/cad/__init__.py

@@ -0,0 +1,2 @@
+# SPDX-FileCopyrightText: 2025 QinHan
+# SPDX-License-Identifier: MPL-2.0

docutranslate/cad/dwg_converter.py

@@ -0,0 +1,157 @@
+# SPDX-FileCopyrightText: 2025 QinHan
+# SPDX-License-Identifier: MPL-2.0
+"""DWG ↔ DXF converter using external backends.
+
+Detects and invokes user-installed converters (LibreDWG, HaoChen, AutoCAD).
+DocuTranslate does NOT bundle converters — users install them separately.
+"""
+from __future__ import annotations
+
+import shutil
+import subprocess
+from dataclasses import dataclass
+from pathlib import Path
+from typing import Optional
+
+
+@dataclass
+class ConverterResult:
+    success: bool
+    input_path: str
+    output_path: str | None = None
+    backend_used: str = ""
+    message: str = ""
+
+
+class DwgConverter:
+    """Detect and call external DWG ↔ DXF converters."""
+
+    BACKENDS = {
+        "libredwg": "dwg2dxf",
+        "haochen": "haochen_com",
+        "autocad": "autocad_com",
+    }
+
+    def __init__(self, backend: str = "auto"):
+        self.backend = backend.strip().lower()
+
+    # ------------------------------------------------------------------
+    # Detection
+    # ------------------------------------------------------------------
+
+    def detect_available(self) -> dict[str, bool]:
+        """Return which backends are available on this system."""
+        results: dict[str, bool] = {}
+        for name, binary in self.BACKENDS.items():
+            results[name] = shutil.which(binary) is not None
+        return results
+
+    def _select_backend(self) -> str:
+        if self.backend and self.backend != "auto":
+            return self.backend
+        available = self.detect_available()
+        for name in ("libredwg", "haochen", "autocad"):
+            if available.get(name):
+                return name
+        return ""
+
+    # ------------------------------------------------------------------
+    # Conversion
+    # ------------------------------------------------------------------
+
+    def dwg_to_dxf(self, input_path: str, output_dir: str, backend: str = "") -> ConverterResult:
+        """Convert DWG to DXF. If input is already DXF, copy it."""
+        src = Path(input_path)
+        if not src.exists():
+            return ConverterResult(False, input_path, message=f"File not found: {input_path}")
+
+        if src.suffix.lower() == ".dxf":
+            out = Path(output_dir) / src.name
+            out.parent.mkdir(parents=True, exist_ok=True)
+            if src.resolve() != out.resolve():
+                import shutil as _shutil
+                _shutil.copy2(str(src), str(out))
+            return ConverterResult(True, input_path, str(out), "dxf_only", "Input is already DXF")
+
+        be = backend or self._select_backend()
+        if not be:
+            return ConverterResult(
+                False, input_path,
+                message="No DWG converter available. Install LibreDWG (dwg2dxf), HaoChen CAD, or AutoCAD.",
+            )
+
+        out_dir = Path(output_dir)
+        out_dir.mkdir(parents=True, exist_ok=True)
+
+        if be == "libredwg":
+            return self._run_libredwg(src, out_dir)
+        elif be in ("haochen", "autocad"):
+            return self._run_com_backend(src, out_dir, be)
+        else:
+            return ConverterResult(False, input_path, message=f"Unknown backend: {be}")
+
+    def dxf_to_dwg(self, input_path: str, output_dir: str, backend: str = "") -> ConverterResult:
+        """Convert DXF to DWG (only supported by some backends)."""
+        src = Path(input_path)
+        if not src.exists():
+            return ConverterResult(False, input_path, message=f"File not found: {input_path}")
+
+        be = backend or self._select_backend()
+        if be in ("haochen", "autocad"):
+            return self._run_com_backend_dxf2dwg(src, Path(output_dir), be)
+
+        return ConverterResult(False, input_path, message="DXF→DWG conversion requires HaoChen or AutoCAD backend")
+
+    # ------------------------------------------------------------------
+    # Backend implementations
+    # ------------------------------------------------------------------
+
+    def _run_libredwg(self, src: Path, out_dir: Path) -> ConverterResult:
+        exe = shutil.which("dwg2dxf")
+        if not exe:
+            return ConverterResult(False, str(src), message="dwg2dxf not found in PATH")
+        out = out_dir / f"{src.stem}.dxf"
+        try:
+            subprocess.run(
+                [exe, "-o", str(out), str(src)],
+                check=True, capture_output=True, timeout=120,
+            )
+            return ConverterResult(True, str(src), str(out), "libredwg")
+        except subprocess.CalledProcessError as e:
+            return ConverterResult(False, str(src), message=f"LibreDWG failed: {e.stderr.decode(errors='replace')}")
+        except subprocess.TimeoutExpired:
+            return ConverterResult(False, str(src), message="LibreDWG conversion timed out")
+
+    def _run_com_backend(self, src: Path, out_dir: Path, backend: str) -> ConverterResult:
+        """Invoke HaoChen/AutoCAD COM automation via Python script."""
+        exe = shutil.which(f"{backend}_converter")
+        if not exe:
+            return ConverterResult(False, str(src), message=f"{backend}_converter not found in PATH")
+        out = out_dir / f"{src.stem}.dxf"
+        try:
+            subprocess.run(
+                [exe, "--input", str(src), "--output", str(out)],
+                check=True, capture_output=True, timeout=300,
+            )
+            return ConverterResult(True, str(src), str(out), backend)
+        except subprocess.CalledProcessError as e:
+            return ConverterResult(False, str(src), message=f"{backend} failed: {e.stderr.decode(errors='replace')}")
+        except subprocess.TimeoutExpired:
+            return ConverterResult(False, str(src), message=f"{backend} conversion timed out")
+
+    def _run_com_backend_dxf2dwg(self, src: Path, out_dir: Path, backend: str) -> ConverterResult:
+        exe = shutil.which(f"{backend}_converter")
+        if not exe:
+            return ConverterResult(False, str(src), message=f"{backend}_converter not found in PATH")
+        out = out_dir / f"{src.stem}.dwg"
+        try:
+            subprocess.run(
+                [exe, "--input", str(src), "--output", str(out), "--format", "dwg"],
+                check=True, capture_output=True, timeout=300,
+            )
+            return ConverterResult(True, str(src), str(out), backend)
+        except subprocess.CalledProcessError as e:
+            return ConverterResult(False, str(src), message=f"{backend} failed: {e.stderr.decode(errors='replace')}")
+        except subprocess.TimeoutExpired:
+            return ConverterResult(False, str(src), message=f"{backend} conversion timed out")
+

docutranslate/cad/text_applier.py

@@ -0,0 +1,171 @@
+# SPDX-FileCopyrightText: 2025 QinHan
+# SPDX-License-Identifier: MPL-2.0
+"""Apply translated text back to DXF files using ezdxf."""
+from __future__ import annotations
+
+import logging
+import math
+from dataclasses import dataclass
+from pathlib import Path
+
+logger = logging.getLogger(__name__)
+
+SUPPORTED_ENTITY_TYPES = ("TEXT", "MTEXT", "ATTDEF", "ATTRIB")
+
+
+@dataclass
+class ApplyResult:
+    success: bool
+    translated_count: int = 0
+    message: str = ""
+
+
+class CadTextApplier:
+    """Apply translations back to DXF files."""
+
+    def apply(
+        self,
+        dxf_path: str,
+        output_path: str,
+        translation_map: dict[str, str],
+        mode: str = "replace",
+        font_name: str = "Times New Roman",
+        font_size_reduction: int = 2,
+    ) -> ApplyResult:
+        try:
+            import ezdxf
+        except ImportError:
+            return ApplyResult(
+                False, message="ezdxf not installed. Install with: pip install docutranslate[cad]"
+            )
+
+        src = Path(dxf_path)
+        out = Path(output_path)
+        if not src.exists():
+            return ApplyResult(False, message=f"File not found: {dxf_path}")
+
+        try:
+            doc = ezdxf.readfile(str(src))
+        except Exception as e:
+            return ApplyResult(False, message=f"Cannot read DXF: {e}")
+
+        replace_mode = mode == "replace"
+        translated_count = 0
+
+        def _process_space(space):
+            nonlocal translated_count
+            for entity in list(space):
+                try:
+                    if self._translate_entity(space, entity, translation_map, font_name, replace_mode, font_size_reduction, doc):
+                        translated_count += 1
+                except Exception as e:
+                    logger.debug("Entity translate failed: %s", e)
+
+        _process_space(doc.modelspace())
+        for layout in doc.layouts:
+            if layout.name != "Model":
+                _process_space(layout)
+        try:
+            for block in doc.blocks:
+                if not block.name.startswith("*"):
+                    _process_space(block)
+        except Exception as e:
+            logger.debug("Block translate failed: %s", e)
+
+        out.parent.mkdir(parents=True, exist_ok=True)
+        doc.saveas(str(out))
+        logger.info("Applied %d translations to %s", translated_count, out.name)
+        return ApplyResult(True, translated_count, f"Translated {translated_count} entities")
+
+    def _smart_match(self, text: str, translation_map: dict[str, str]) -> str | None:
+        if text in translation_map and translation_map[text].strip():
+            return translation_map[text]
+        strategies = [
+            lambda x: re.sub(r"\s+", "", x),
+            lambda x: re.sub(r"\s+", " ", x.strip()),
+            lambda x: x.strip(),
+        ]
+        for strategy in strategies:
+            src = strategy(text)
+            for orig, trans in translation_map.items():
+                if strategy(orig) == src and trans.strip():
+                    return trans
+        return None
+
+    def _set_font(self, entity, font_name: str, doc) -> None:
+        try:
+            style_name = f"TStyle_{font_name.replace(' ', '_')}"
+            if style_name not in doc.styles:
+                style = doc.styles.add(style_name, font=font_name)
+                style.dxf.bigfont = ""
+            entity.dxf.style = style_name
+        except Exception as e:
+            logger.debug("Set font failed: %s", e)
+
+    def _translate_entity(self, owner, entity, translation_map, font_name, replace_mode, font_size_reduction, doc) -> bool:
+        entity_type = entity.dxftype()
+        if entity_type not in SUPPORTED_ENTITY_TYPES:
+            return False
+
+        try:
+            if entity_type in ("TEXT", "MTEXT"):
+                original_text = entity.dxf.text
+            else:
+                original_text = getattr(entity.dxf, "text", None) or getattr(entity.dxf, "tag", None)
+
+            if not original_text or not original_text.strip():
+                return False
+
+            translated = self._smart_match(original_text.strip(), translation_map)
+            if not translated:
+                return False
+
+            height = float(getattr(entity.dxf, "height", None) or getattr(entity.dxf, "char_height", 2.5))
+
+            if replace_mode:
+                if entity_type in ("TEXT", "ATTDEF", "ATTRIB"):
+                    entity.dxf.text = translated
+                    entity.dxf.height = max(1.0, height - font_size_reduction)
+                elif entity_type == "MTEXT":
+                    entity.dxf.text = translated
+                    entity.dxf.char_height = max(1.0, height - font_size_reduction)
+                self._set_font(entity, font_name, doc)
+            else:
+                self._add_text_below(owner, entity, translated, font_name, height, font_size_reduction, doc)
+
+            return True
+        except Exception as e:
+            logger.debug("Translate entity failed: %s", e)
+            return False
+
+    def _add_text_below(self, owner, original_entity, translated_text, font_name, original_height, font_size_reduction, doc):
+        try:
+            insert_point = getattr(original_entity.dxf, "insert", (0, 0, 0))
+            layer = getattr(original_entity.dxf, "layer", "0")
+            rotation = float(getattr(original_entity.dxf, "rotation", 0))
+
+            offset_y = -original_height * 1.2
+            rotation_rad = rotation * (math.pi / 180.0)
+            dx = offset_y * math.sin(rotation_rad)
+            dy = offset_y * math.cos(rotation_rad)
+
+            new_x = float(insert_point[0]) + dx
+            new_y = float(insert_point[1]) + dy
+            new_z = float(insert_point[2]) if len(insert_point) > 2 else 0.0
+
+            style_name = f"TStyle_{font_name.replace(' ', '_')}"
+            if style_name not in doc.styles:
+                s = doc.styles.add(style_name, font=font_name)
+                s.dxf.bigfont = ""
+
+            attribs = {
+                "insert": (new_x, new_y, new_z),
+                "height": max(1.0, original_height - font_size_reduction),
+                "layer": layer,
+                "rotation": rotation,
+                "color": 1,
+                "style": style_name,
+            }
+            owner.add_text(translated_text, dxfattribs=attribs)
+        except Exception as e:
+            logger.debug("Add text below failed: %s", e)