112 lines
3.7 KiB
Python
112 lines
3.7 KiB
Python
"""
|
|
Orchestrate document extraction: dispatch → extract → write sidecars → manifest.
|
|
|
|
Follows the meetus WorkflowConfig / ProcessingWorkflow shape
|
|
(meetus/workflow.py) so the two pipelines feel the same.
|
|
"""
|
|
from pathlib import Path
|
|
from datetime import datetime
|
|
import logging
|
|
from typing import Dict, Any
|
|
|
|
from . import registry
|
|
from .naming import sha256_file
|
|
from .output_manager import DocOutputManager
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class DocConfig:
|
|
"""Configuration for a single document extraction."""
|
|
|
|
def __init__(self, **kwargs):
|
|
self.source_path = Path(kwargs["source"])
|
|
self.output_dir = kwargs.get("output_dir", "docs-output")
|
|
self.no_cache = kwargs.get("no_cache", False)
|
|
self.render = kwargs.get("render", False) # optional page/slide images
|
|
self.ocr = kwargs.get("ocr", False) # OCR images / scanned pdf
|
|
|
|
def options(self) -> Dict[str, Any]:
|
|
return {"render": self.render, "ocr": self.ocr}
|
|
|
|
def to_dict(self) -> Dict[str, Any]:
|
|
return {"render": self.render, "ocr": self.ocr, "no_cache": self.no_cache}
|
|
|
|
|
|
class DocWorkflow:
|
|
"""Extract one document into a run folder."""
|
|
|
|
def __init__(self, config: DocConfig):
|
|
self.config = config
|
|
self.out = DocOutputManager(
|
|
config.source_path,
|
|
config.output_dir,
|
|
use_cache=not config.no_cache,
|
|
)
|
|
|
|
def run(self) -> Dict[str, Any]:
|
|
src = self.config.source_path
|
|
if not src.exists():
|
|
raise FileNotFoundError(f"source not found: {src}")
|
|
|
|
ext = registry.ext_of(src)
|
|
family = registry.family_for(src) or "unhandled"
|
|
logger.info("Extracting %s (family: %s)", src.name, family)
|
|
|
|
result = registry.extract(src, self.config.options())
|
|
|
|
content_path = self.out.write_content(result.content)
|
|
sha = sha256_file(src)
|
|
meta = self._build_meta(src, ext, family, sha, result)
|
|
self.out.write_meta(meta)
|
|
|
|
thumb_written = False
|
|
if result.thumb:
|
|
self.out.write_thumb(result.thumb)
|
|
thumb_written = True
|
|
|
|
outputs = {
|
|
"content": content_path.name,
|
|
"meta": "meta.json",
|
|
"thumb": "thumb.jpg" if thumb_written else None,
|
|
}
|
|
self.out.save_manifest(self.config.to_dict(), outputs, sha)
|
|
|
|
if result.warnings:
|
|
for w in result.warnings:
|
|
logger.warning(" %s", w)
|
|
|
|
logger.info("✓ %s → %s", src.name, self.out.output_dir)
|
|
return {
|
|
"output_dir": str(self.out.output_dir),
|
|
"content": str(content_path),
|
|
"meta": str(self.out.get_path("meta.json")),
|
|
"family": family,
|
|
"extractor": result.extractor,
|
|
"warnings": result.warnings,
|
|
}
|
|
|
|
def _build_meta(self, src: Path, ext: str, family: str, sha: str, result) -> Dict[str, Any]:
|
|
stat = src.stat()
|
|
meta = {
|
|
"source": {
|
|
"name": src.name,
|
|
"path": str(src.absolute()),
|
|
"sha256": sha,
|
|
"bytes": stat.st_size,
|
|
"modified": datetime.fromtimestamp(stat.st_mtime).isoformat(),
|
|
},
|
|
"type": ext,
|
|
"family": family,
|
|
"extractor": result.extractor,
|
|
"extracted_at": datetime.now().isoformat(),
|
|
"word_count": len(result.content.split()),
|
|
"warnings": result.warnings,
|
|
}
|
|
# Type-specific fields (title, author, dates, pages, structure, ...) sit
|
|
# at the top level alongside the core fields.
|
|
for k, v in result.metadata.items():
|
|
if k not in meta:
|
|
meta[k] = v
|
|
return meta
|