tinymind-native-8b-remote-handoff/bundle/model/omni_file_pipeline.py

"""Omni file download, inspection, and conversion pipeline.

The goal is practical full-cycle file handling: download/copy, identify,
hash, extract what the local runtime can extract, and report exact limitations.
It does not claim perfect understanding of every format without a decoder.
"""

from __future__ import annotations

from dataclasses import asdict, dataclass
from datetime import datetime, timezone
import csv
import hashlib
import html
import json
import mimetypes
from pathlib import Path
import shutil
import subprocess
import tarfile
import tempfile
from typing import Any
from urllib.parse import urlparse
import zipfile

import httpx

from model.omni_action_perception import OmniActionPerception
from model.pure_signal_decoder import PureSignalDecoder


TEXT_EXTENSIONS = {".txt", ".md", ".json", ".jsonl", ".csv", ".tsv", ".html", ".htm", ".xml", ".yaml", ".yml", ".log"}
CODE_EXTENSIONS = {".py", ".js", ".ts", ".tsx", ".go", ".rs", ".c", ".cpp", ".h", ".java", ".kt", ".lua", ".sql", ".sh", ".ps1"}
IMAGE_EXTENSIONS = {".png", ".jpg", ".jpeg", ".webp", ".bmp", ".gif", ".tiff"}
AUDIO_EXTENSIONS = {".wav", ".mp3", ".flac", ".m4a", ".ogg", ".aac"}
VIDEO_EXTENSIONS = {".mp4", ".mkv", ".mov", ".webm", ".avi"}
DOCUMENT_EXTENSIONS = {".pdf", ".docx", ".pptx", ".xlsx"}
ARCHIVE_EXTENSIONS = {".zip", ".tar", ".gz", ".tgz", ".bz2"}


def _sha256_bytes(data: bytes) -> str:
    return hashlib.sha256(data).hexdigest()


def _sha256_file(path: Path) -> str:
    h = hashlib.sha256()
    with path.open("rb") as f:
        for chunk in iter(lambda: f.read(1024 * 1024), b""):
            h.update(chunk)
    return h.hexdigest()


def _safe_name(name: str) -> str:
    cleaned = "".join(ch if ch.isalnum() or ch in "._-" else "_" for ch in name).strip("._")
    return cleaned[:160] or "downloaded_file"


@dataclass(frozen=True)
class OmniFilePolicy:
    max_download_bytes: int = 100_000_000
    timeout_s: float = 30.0
    allow_network_download: bool = True
    extract_archives: bool = False
    max_text_chars: int = 500_000
    ffprobe_timeout_s: float = 10.0


class OmniFilePipeline:
    def __init__(
        self,
        root: str | Path | None = None,
        *,
        policy: OmniFilePolicy | None = None,
        client: httpx.Client | None = None,
    ):
        self.root = Path(root) if root is not None else Path(tempfile.mkdtemp(prefix="tinymind_omni_file_"))
        self.root.mkdir(parents=True, exist_ok=True)
        self.policy = policy or OmniFilePolicy()
        self.client = client or httpx.Client(timeout=self.policy.timeout_s, follow_redirects=True)
        self.perception = OmniActionPerception()
        self.decoder = PureSignalDecoder()

    def process(self, source: str | Path, *, out_dir: str | Path | None = None) -> dict[str, Any]:
        out = Path(out_dir) if out_dir is not None else self.root
        out.mkdir(parents=True, exist_ok=True)
        source_text = str(source)
        acquired = self._acquire(source_text, out / "inputs")
        file_path = Path(acquired["path"])
        inspection = self._inspect(file_path)
        conversion = self._convert(file_path, out / "converted", inspection)
        plan = self.perception.plan_input(file_path.name)
        pure_signal = self.decoder.decode_file(
            file_path,
            modality=str(plan["modality"]),
            extracted_outputs=conversion.get("outputs", []),
            out_dir=out / "pure_signal",
        )
        checks = {
            "file_acquired": file_path.exists(),
            "sha256_recorded": bool(inspection["sha256"]),
            "modality_identified": plan["modality"] != "unknown",
            "conversion_attempted": bool(conversion["attempted"]),
            "pure_signal_decoded": pure_signal["claim_gate"]["pure_signal_decoder_ready"],
            "manifest_written": True,
        }
        report = {
            "schema_version": "tinymind-omni-file-pipeline-v1",
            "created_at": datetime.now(timezone.utc).isoformat(),
            "source": source_text,
            "policy": asdict(self.policy),
            "acquisition": acquired,
            "inspection": inspection,
            "perception_plan": plan,
            "conversion": conversion,
            "pure_signal_decoder": {
                "json_path": pure_signal.get("json_path"),
                "vector_path": pure_signal.get("vector_path"),
                "zero_waste_audit": pure_signal["zero_waste_audit"],
                "claim_gate": pure_signal["claim_gate"],
            },
            "checks": checks,
            "claim_gate": {
                "omni_file_pipeline_ready": checks["file_acquired"]
                and checks["sha256_recorded"]
                and checks["conversion_attempted"]
                and checks["pure_signal_decoded"],
                "download_and_conversion_audited": True,
                "pure_signal_decoder_ready": pure_signal["claim_gate"]["pure_signal_decoder_ready"],
                "zero_waste_decoder_passed": pure_signal["zero_waste_audit"]["passed"],
                "all_world_formats_perfect_claim_allowed": False,
                "raw_bytes_understood_without_parser": False,
                "unsupported_format_must_report_limitation": True,
                "active_content_execution_allowed": False,
                "reason": "Files are downloaded/copied, hashed, classified, and converted only by available local decoders. Unsupported formats remain evidence artifacts, not understood content.",
            },
        }
        report_path = out / "omni_file_pipeline_report.json"
        report["json_path"] = str(report_path)
        report_path.write_text(json.dumps(report, ensure_ascii=False, indent=2, sort_keys=True) + "\n", encoding="utf-8")
        return report

    def _acquire(self, source: str, input_dir: Path) -> dict[str, Any]:
        input_dir.mkdir(parents=True, exist_ok=True)
        parsed = urlparse(source)
        if parsed.scheme in {"http", "https"}:
            if not self.policy.allow_network_download:
                raise PermissionError("network_download_disabled")
            response = self.client.get(source)
            response.raise_for_status()
            content = response.content
            if len(content) > self.policy.max_download_bytes:
                raise ValueError("download_exceeds_policy_limit")
            name = _safe_name(Path(parsed.path).name or "download")
            path = input_dir / name
            path.write_bytes(content)
            return {
                "mode": "download",
                "url": source,
                "path": str(path),
                "bytes": len(content),
                "content_sha256": _sha256_bytes(content),
                "content_type": response.headers.get("content-type"),
            }
        src = Path(source)
        if not src.exists():
            raise FileNotFoundError(source)
        if src.is_dir():
            raise IsADirectoryError("Use context-ingest for whole folders; omni-file-process expects one file.")
        path = input_dir / _safe_name(src.name)
        shutil.copy2(src, path)
        return {
            "mode": "copy",
            "source_path": str(src),
            "path": str(path),
            "bytes": path.stat().st_size,
            "content_sha256": _sha256_file(path),
            "content_type": mimetypes.guess_type(path.name)[0],
        }

    def _inspect(self, path: Path) -> dict[str, Any]:
        data = path.read_bytes()[:4096]
        suffix = path.suffix.lower()
        mime = mimetypes.guess_type(path.name)[0]
        return {
            "name": path.name,
            "suffix": suffix,
            "mime_type": mime,
            "bytes": path.stat().st_size,
            "sha256": _sha256_file(path),
            "magic_hex": data[:16].hex(),
            "kind": self._kind_from_suffix(suffix),
        }

    def _kind_from_suffix(self, suffix: str) -> str:
        if suffix in TEXT_EXTENSIONS:
            return "text"
        if suffix in CODE_EXTENSIONS:
            return "code"
        if suffix in IMAGE_EXTENSIONS:
            return "image"
        if suffix in AUDIO_EXTENSIONS:
            return "audio"
        if suffix in VIDEO_EXTENSIONS:
            return "video"
        if suffix in DOCUMENT_EXTENSIONS:
            return "document"
        if suffix in ARCHIVE_EXTENSIONS:
            return "archive"
        return "unknown"

    def _convert(self, path: Path, out_dir: Path, inspection: dict[str, Any]) -> dict[str, Any]:
        out_dir.mkdir(parents=True, exist_ok=True)
        kind = inspection["kind"]
        if kind in {"text", "code"}:
            return self._convert_text(path, out_dir, kind)
        if kind == "image":
            return self._convert_image(path, out_dir)
        if kind in {"audio", "video"}:
            return self._convert_av(path, out_dir, kind)
        if kind == "document":
            return self._convert_document(path, out_dir)
        if kind == "archive":
            return self._convert_archive(path, out_dir)
        return {
            "attempted": True,
            "status": "unsupported_format",
            "outputs": [],
            "limitation": "No decoder registered for this suffix; file is still hashed and available as evidence.",
        }

    def _convert_text(self, path: Path, out_dir: Path, kind: str) -> dict[str, Any]:
        raw = path.read_bytes()
        text = raw.decode("utf-8", errors="replace")[: self.policy.max_text_chars]
        if path.suffix.lower() in {".html", ".htm"}:
            text = html.unescape(text.replace("<br>", "\n").replace("<br/>", "\n"))
        if path.suffix.lower() in {".csv", ".tsv"}:
            delimiter = "\t" if path.suffix.lower() == ".tsv" else ","
            rows = list(csv.reader(text.splitlines(), delimiter=delimiter))[:20]
            text = "\n".join(" | ".join(cell.strip() for cell in row[:20]) for row in rows)
        out_path = out_dir / "extracted_text.txt"
        out_path.write_text(text, encoding="utf-8")
        return {
            "attempted": True,
            "status": "converted",
            "kind": kind,
            "outputs": [{"path": str(out_path), "type": "text", "chars": len(text), "sha256": _sha256_file(out_path)}],
        }

    def _convert_image(self, path: Path, out_dir: Path) -> dict[str, Any]:
        metadata: dict[str, Any] = {"file": path.name}
        try:
            from PIL import Image  # type: ignore

            with Image.open(path) as img:
                metadata.update({"width": img.width, "height": img.height, "mode": img.mode, "format": img.format})
        except Exception as exc:
            metadata.update({"decoder": "pillow_unavailable_or_failed", "error_type": type(exc).__name__})
        out_path = out_dir / "image_metadata.json"
        out_path.write_text(json.dumps(metadata, ensure_ascii=False, indent=2, sort_keys=True), encoding="utf-8")
        return {
            "attempted": True,
            "status": "metadata_extracted",
            "kind": "image",
            "outputs": [{"path": str(out_path), "type": "metadata", "sha256": _sha256_file(out_path)}],
            "limitation": "Image semantic understanding requires a vision encoder/OCR model; this stage records pixels metadata only.",
        }

    def _convert_av(self, path: Path, out_dir: Path, kind: str) -> dict[str, Any]:
        metadata = {"file": path.name, "kind": kind, "ffprobe_available": False}
        ffprobe = shutil.which("ffprobe")
        if ffprobe:
            metadata["ffprobe_available"] = True
            try:
                proc = subprocess.run(
                    [ffprobe, "-v", "quiet", "-print_format", "json", "-show_format", "-show_streams", str(path)],
                    capture_output=True,
                    text=True,
                    timeout=self.policy.ffprobe_timeout_s,
                    check=False,
                )
                metadata["ffprobe_returncode"] = proc.returncode
                metadata["ffprobe"] = json.loads(proc.stdout) if proc.stdout.strip() else {}
            except Exception as exc:
                metadata["error_type"] = type(exc).__name__
                metadata["error"] = str(exc)
        out_path = out_dir / f"{kind}_metadata.json"
        out_path.write_text(json.dumps(metadata, ensure_ascii=False, indent=2, sort_keys=True), encoding="utf-8")
        return {
            "attempted": True,
            "status": "metadata_extracted",
            "kind": kind,
            "outputs": [{"path": str(out_path), "type": "metadata", "sha256": _sha256_file(out_path)}],
            "limitation": "Transcription/scene understanding requires ASR or vision/audio models; this stage records container metadata when available.",
        }

    def _convert_document(self, path: Path, out_dir: Path) -> dict[str, Any]:
        suffix = path.suffix.lower()
        if suffix == ".pdf":
            return self._convert_pdf(path, out_dir)
        if suffix == ".docx":
            return self._convert_docx(path, out_dir)
        return {
            "attempted": True,
            "status": "document_metadata_only",
            "kind": "document",
            "outputs": [],
            "limitation": f"No local extractor registered for {suffix}; file is hashed and preserved.",
        }

    def _convert_pdf(self, path: Path, out_dir: Path) -> dict[str, Any]:
        text = ""
        status = "metadata_only"
        try:
            from pypdf import PdfReader  # type: ignore

            reader = PdfReader(str(path))
            pages = []
            for page in reader.pages[:50]:
                pages.append(page.extract_text() or "")
            text = "\n\n".join(pages)[: self.policy.max_text_chars]
            status = "converted" if text.strip() else "metadata_only"
        except Exception:
            text = ""
        out_path = out_dir / "document_text.txt"
        out_path.write_text(text, encoding="utf-8")
        return {
            "attempted": True,
            "status": status,
            "kind": "document",
            "outputs": [{"path": str(out_path), "type": "text", "chars": len(text), "sha256": _sha256_file(out_path)}],
            "limitation": "Scanned PDFs require OCR; this extracts embedded text only.",
        }

    def _convert_docx(self, path: Path, out_dir: Path) -> dict[str, Any]:
        text = ""
        status = "metadata_only"
        try:
            import docx  # type: ignore

            doc = docx.Document(str(path))
            text = "\n".join(p.text for p in doc.paragraphs)[: self.policy.max_text_chars]
            status = "converted" if text.strip() else "metadata_only"
        except Exception:
            text = ""
        out_path = out_dir / "document_text.txt"
        out_path.write_text(text, encoding="utf-8")
        return {
            "attempted": True,
            "status": status,
            "kind": "document",
            "outputs": [{"path": str(out_path), "type": "text", "chars": len(text), "sha256": _sha256_file(out_path)}],
        }

    def _convert_archive(self, path: Path, out_dir: Path) -> dict[str, Any]:
        listing: list[dict[str, Any]] = []
        try:
            if zipfile.is_zipfile(path):
                with zipfile.ZipFile(path) as zf:
                    listing = [{"name": item.filename, "bytes": item.file_size} for item in zf.infolist()[:1000]]
                    if self.policy.extract_archives:
                        zf.extractall(out_dir / "archive_extract")
            elif tarfile.is_tarfile(path):
                with tarfile.open(path) as tf:
                    listing = [{"name": item.name, "bytes": item.size} for item in tf.getmembers()[:1000]]
                    if self.policy.extract_archives:
                        tf.extractall(out_dir / "archive_extract")
        except Exception as exc:
            listing = [{"error_type": type(exc).__name__, "error": str(exc)}]
        out_path = out_dir / "archive_manifest.json"
        out_path.write_text(json.dumps({"entries": listing}, ensure_ascii=False, indent=2, sort_keys=True), encoding="utf-8")
        return {
            "attempted": True,
            "status": "listed",
            "kind": "archive",
            "outputs": [{"path": str(out_path), "type": "manifest", "entries": len(listing), "sha256": _sha256_file(out_path)}],
            "limitation": "Archive extraction is disabled by default; active content is never executed.",
        }


def write_omni_file_process(
    source: str | Path,
    out_dir: str | Path,
    *,
    allow_network_download: bool = True,
    extract_archives: bool = False,
) -> dict[str, Any]:
    return OmniFilePipeline(
        out_dir,
        policy=OmniFilePolicy(allow_network_download=allow_network_download, extract_archives=extract_archives),
    ).process(source, out_dir=out_dir)