"""Repositorio sobre PostgreSQL (tablas digitador_* + sync de tasks)."""
from __future__ import annotations

from datetime import datetime, timedelta, timezone
from typing import Any, Dict, List, Optional

from sqlalchemy import select

from domain.models import (
    Job,
    JobEventRecord,
    JobStatus,
    Worker,
    WorkerStatus,
    utc_now,
)
from infrastructure.database import (
    DigitadorJobEventRow,
    DigitadorJobRow,
    DigitadorWorkerRow,
    Task,
    _iso,
    init_db,
    session_scope,
    utcnow,
)
from infrastructure.storage_minio import reports_prefix

HEARTBEAT_OFFLINE_SECONDS = 45
_FRESH_WORKER_GRACE_SECONDS = 45


def _parse_iso(value: str | None) -> datetime | None:
    if not value:
        return None
    dt = datetime.fromisoformat(value)
    if dt.tzinfo is None:
        dt = dt.replace(tzinfo=timezone.utc)
    return dt


class JobStore:
    """API de persistencia compatible con el patrón bot_descarga."""

    def __init__(self, db_path: str | None = None):
        # db_path ignorado (compat); usa DATABASE_URL.
        init_db()
        self._fresh_until: dict[str, datetime] = {}

    def _row_worker(self, row: DigitadorWorkerRow) -> Worker:
        worker = Worker(
            id=row.id,
            hostname=row.hostname or "",
            max_concurrent=int(row.max_concurrent or 1),
            status=row.status,
            last_heartbeat=_iso(row.last_heartbeat) or utc_now(),
            current_job_id=row.current_job_id,
        )
        return self._refresh_worker_status(worker)

    def _row_job(self, row: DigitadorJobRow) -> Job:
        return Job(
            id=row.id,
            worker_id=row.worker_id,
            status=row.status,
            title=row.title or "",
            csv_key=row.csv_key or "",
            reports_prefix=row.reports_prefix or "",
            original_filename=row.original_filename or "",
            cancel_requested=bool(row.cancel_requested),
            result=dict(row.result or {}),
            error=row.error,
            created_at=_iso(row.created_at) or utc_now(),
            started_at=_iso(row.started_at),
            finished_at=_iso(row.finished_at),
            job_id=row.job_id,
            task_id=row.task_id,
            parent_digitador_job_id=row.parent_digitador_job_id,
            shard_index=row.shard_index,
            shard_count=row.shard_count,
        )

    def _refresh_worker_status(self, worker: Worker) -> Worker:
        try:
            hb = datetime.fromisoformat(worker.last_heartbeat)
            if hb.tzinfo is None:
                hb = hb.replace(tzinfo=timezone.utc)
            age = datetime.now(timezone.utc) - hb.astimezone(timezone.utc)
            if age > timedelta(seconds=HEARTBEAT_OFFLINE_SECONDS):
                worker.status = WorkerStatus.OFFLINE.value
            elif worker.current_job_id:
                worker.status = WorkerStatus.BUSY.value
            else:
                worker.status = WorkerStatus.ONLINE.value
        except Exception:
            worker.status = WorkerStatus.OFFLINE.value
        return worker

    def register_worker(
        self,
        worker_id: str,
        hostname: str = "",
        max_concurrent: int = 1,
    ) -> Worker:
        with session_scope() as session:
            row = session.get(DigitadorWorkerRow, worker_id)
            now = utcnow()
            if row:
                row.hostname = hostname
                row.max_concurrent = max_concurrent
                row.status = WorkerStatus.ONLINE.value
                row.last_heartbeat = now
                row.current_job_id = None
            else:
                row = DigitadorWorkerRow(
                    id=worker_id,
                    hostname=hostname,
                    max_concurrent=max_concurrent,
                    status=WorkerStatus.ONLINE.value,
                    last_heartbeat=now,
                    current_job_id=None,
                )
                session.add(row)
            session.flush()
            parent_ids = self._abandon_assigned_jobs(
                session,
                worker_id,
                now,
                "Agent reconectado: job anterior abandonado. Lanza una corrida nueva desde el desktop.",
            )
            session.flush()
            worker = self._row_worker(row)
        self._fresh_until[worker_id] = now + timedelta(seconds=_FRESH_WORKER_GRACE_SECONDS)
        for parent_id in parent_ids:
            try:
                self.refresh_parent_from_shards(parent_id)
            except Exception:
                pass
        return worker

    def _abandon_assigned_jobs(
        self, session, worker_id: str, now: datetime, message: str
    ) -> list[str]:
        """Cancela pending/running de este worker para que un agent nuevo arranque idle."""
        rows = session.scalars(
            select(DigitadorJobRow).where(
                DigitadorJobRow.worker_id == worker_id,
                DigitadorJobRow.status.in_((JobStatus.PENDING.value, JobStatus.RUNNING.value)),
            )
        ).all()
        parent_ids: set[str] = set()
        for row in rows:
            is_parent = (
                row.parent_digitador_job_id is None
                and row.shard_count is not None
                and int(row.shard_count) > 1
            )
            if is_parent:
                continue
            row.status = JobStatus.CANCELLED.value
            row.cancel_requested = True
            row.finished_at = now
            row.error = message
            self._sync_task(session, row, status=JobStatus.CANCELLED.value, error=message)
            session.add(
                DigitadorJobEventRow(
                    digitador_job_id=row.id,
                    event_type="warning",
                    message=message,
                    timestamp=now,
                    metadata_json={"abandoned_on_register": True, "worker_id": worker_id},
                )
            )
            if row.parent_digitador_job_id:
                parent_ids.add(row.parent_digitador_job_id)
                session.add(
                    DigitadorJobEventRow(
                        digitador_job_id=row.parent_digitador_job_id,
                        event_type="warning",
                        message=f"Shard {row.shard_index} abandonado (agent reconectado)",
                        timestamp=now,
                        metadata_json={"shard_id": row.id, "worker_id": worker_id},
                    )
                )
        return list(parent_ids)

    def heartbeat(self, worker_id: str) -> Optional[Worker]:
        with session_scope() as session:
            row = session.get(DigitadorWorkerRow, worker_id)
            if not row:
                return None
            row.last_heartbeat = utcnow()
            row.status = (
                WorkerStatus.BUSY.value if row.current_job_id else WorkerStatus.ONLINE.value
            )
            session.flush()
            worker = self._row_worker(row)
        try:
            self.reclaim_stale_running_jobs()
        except Exception:
            pass
        return worker

    def get_worker(self, worker_id: str) -> Optional[Worker]:
        with session_scope() as session:
            row = session.get(DigitadorWorkerRow, worker_id)
            if not row:
                return None
            return self._row_worker(row)

    def list_workers(self) -> List[Worker]:
        try:
            self.reclaim_stale_running_jobs()
        except Exception:
            pass
        with session_scope() as session:
            rows = session.scalars(
                select(DigitadorWorkerRow).order_by(DigitadorWorkerRow.id)
            ).all()
            return [self._row_worker(r) for r in rows]

    def list_online_worker_ids(self) -> List[str]:
        return [
            w.id
            for w in self.list_workers()
            if w.status in (WorkerStatus.ONLINE.value, WorkerStatus.BUSY.value)
        ]

    def is_worker_online(self, worker_id: str) -> bool:
        worker = self.get_worker(worker_id)
        if not worker:
            return False
        return worker.status in (WorkerStatus.ONLINE.value, WorkerStatus.BUSY.value)

    def update_csv_key(self, job_id: str, csv_key: str) -> Optional[Job]:
        with session_scope() as session:
            row = session.get(DigitadorJobRow, job_id)
            if not row:
                return None
            row.csv_key = csv_key
            session.flush()
            return self._row_job(row)

    def create_job(
        self,
        worker_id: Optional[str],
        csv_key: str,
        original_filename: str = "",
        title: str = "",
        *,
        job_id: Optional[str] = None,
        task_id: Optional[str] = None,
        reports_prefix_override: Optional[str] = None,
        parent_digitador_job_id: Optional[str] = None,
        shard_index: Optional[int] = None,
        shard_count: Optional[int] = None,
        result: Optional[Dict[str, Any]] = None,
        status: Optional[str] = None,
        digitador_id: Optional[str] = None,
    ) -> Job:
        import uuid as uuid_mod

        digitador_id = digitador_id or str(uuid_mod.uuid4())
        prefix = reports_prefix_override or reports_prefix(digitador_id)
        with session_scope() as session:
            row = DigitadorJobRow(
                id=digitador_id,
                job_id=job_id,
                task_id=task_id,
                worker_id=worker_id,
                parent_digitador_job_id=parent_digitador_job_id,
                shard_index=shard_index,
                shard_count=shard_count,
                status=status or JobStatus.PENDING.value,
                title=title or f"Digitador {original_filename}",
                csv_key=csv_key,
                reports_prefix=prefix,
                original_filename=original_filename,
                cancel_requested=False,
                result=result or {},
                created_at=utcnow(),
            )
            session.add(row)
            session.flush()
            return self._row_job(row)

    def get_job(self, job_id: str) -> Optional[Job]:
        with session_scope() as session:
            row = session.get(DigitadorJobRow, job_id)
            if not row:
                return None
            return self._row_job(row)

    def list_jobs(
        self,
        limit: int = 100,
        *,
        job_id: Optional[str] = None,
        parents_only: bool = True,
    ) -> List[Job]:
        with session_scope() as session:
            stmt = select(DigitadorJobRow).order_by(DigitadorJobRow.created_at.desc())
            if job_id:
                stmt = stmt.where(DigitadorJobRow.job_id == job_id)
            if parents_only:
                stmt = stmt.where(DigitadorJobRow.parent_digitador_job_id.is_(None))
            stmt = stmt.limit(limit)
            rows = session.scalars(stmt).all()
            return [self._row_job(r) for r in rows]

    def list_shards(self, parent_id: str) -> List[Job]:
        with session_scope() as session:
            rows = session.scalars(
                select(DigitadorJobRow)
                .where(DigitadorJobRow.parent_digitador_job_id == parent_id)
                .order_by(DigitadorJobRow.shard_index.asc())
            ).all()
            return [self._row_job(r) for r in rows]

    def claim_pending(self, worker_id: str) -> Optional[Job]:
        try:
            self.reclaim_stale_running_jobs()
        except Exception:
            pass
        with session_scope() as session:
            worker = session.get(DigitadorWorkerRow, worker_id)
            if not worker or worker.current_job_id:
                return None

            rows = session.scalars(
                select(DigitadorJobRow)
                .where(
                    DigitadorJobRow.worker_id == worker_id,
                    DigitadorJobRow.status == JobStatus.PENDING.value,
                )
                .order_by(DigitadorJobRow.created_at.asc())
            ).all()
            row = None
            for candidate in rows:
                is_parent = (
                    candidate.parent_digitador_job_id is None
                    and candidate.shard_count is not None
                    and int(candidate.shard_count) > 1
                )
                if is_parent:
                    continue
                row = candidate
                break
            if not row:
                return None

            now = utcnow()
            row.status = JobStatus.RUNNING.value
            row.started_at = now
            worker.current_job_id = row.id
            worker.status = WorkerStatus.BUSY.value
            worker.last_heartbeat = now

            task_id = row.task_id
            if row.parent_digitador_job_id:
                parent = session.get(DigitadorJobRow, row.parent_digitador_job_id)
                if parent:
                    task_id = parent.task_id
                    if parent.status == JobStatus.PENDING.value:
                        parent.status = JobStatus.RUNNING.value
                        parent.started_at = parent.started_at or now

            if task_id:
                task = session.get(Task, task_id)
                if task and task.status == JobStatus.PENDING.value:
                    task.status = JobStatus.RUNNING.value
                    task.started_at = now
                    task.updated_at = now

            session.flush()
            return self._row_job(row)

    def request_cancel(self, job_id: str) -> Optional[Job]:
        with session_scope() as session:
            row = session.get(DigitadorJobRow, job_id)
            if not row:
                return None
            if row.status in (
                JobStatus.COMPLETED.value,
                JobStatus.FAILED.value,
                JobStatus.CANCELLED.value,
            ):
                return self._row_job(row)

            now = utcnow()
            row.cancel_requested = True
            if row.status == JobStatus.PENDING.value:
                row.status = JobStatus.CANCELLED.value
                row.finished_at = now
                self._sync_task(session, row, status=JobStatus.CANCELLED.value)

            if row.parent_digitador_job_id is None and row.shard_count and int(row.shard_count) > 1:
                children = session.scalars(
                    select(DigitadorJobRow).where(
                        DigitadorJobRow.parent_digitador_job_id == job_id
                    )
                ).all()
                for child in children:
                    if child.status in (
                        JobStatus.COMPLETED.value,
                        JobStatus.FAILED.value,
                        JobStatus.CANCELLED.value,
                    ):
                        continue
                    child.cancel_requested = True
                    if child.status == JobStatus.PENDING.value:
                        child.status = JobStatus.CANCELLED.value
                        child.finished_at = now
                        worker = (
                            session.get(DigitadorWorkerRow, child.worker_id)
                            if child.worker_id
                            else None
                        )
                        if worker and worker.current_job_id == child.id:
                            worker.current_job_id = None
                            worker.status = WorkerStatus.ONLINE.value

            session.flush()
            return self._row_job(row)

    def is_cancel_requested(self, job_id: str) -> bool:
        job = self.get_job(job_id)
        if not job:
            return False
        if job.cancel_requested:
            return True
        if job.parent_digitador_job_id:
            parent = self.get_job(job.parent_digitador_job_id)
            if parent and parent.cancel_requested:
                return True
        return False

    def _sync_task(
        self,
        session,
        row: DigitadorJobRow,
        *,
        status: Optional[str] = None,
        result: Optional[Dict[str, Any]] = None,
        error: Optional[str] = None,
    ) -> None:
        target = row
        if row.parent_digitador_job_id:
            parent = session.get(DigitadorJobRow, row.parent_digitador_job_id)
            if parent:
                target = parent
        if not target.task_id:
            return
        task = session.get(Task, target.task_id)
        if not task:
            return
        now = utcnow()
        if status:
            task.status = status
        if result is not None:
            task.result = {
                **dict(task.result or {}),
                "digitador_job_id": target.id,
                **result,
            }
        if error is not None:
            task.error = error
        task.updated_at = now
        if status in (
            JobStatus.COMPLETED.value,
            JobStatus.FAILED.value,
            JobStatus.CANCELLED.value,
        ):
            task.finished_at = now

    def patch_job(
        self,
        job_id: str,
        *,
        status: Optional[str] = None,
        result: Optional[Dict[str, Any]] = None,
        error: Optional[str] = None,
    ) -> Optional[Job]:
        parent_id: Optional[str] = None
        with session_scope() as session:
            row = session.get(DigitadorJobRow, job_id)
            if not row:
                return None

            new_status = status or row.status
            if result is not None:
                row.result = result
            if error is not None:
                row.error = error
            row.status = new_status

            finished = False
            if new_status in (
                JobStatus.COMPLETED.value,
                JobStatus.FAILED.value,
                JobStatus.CANCELLED.value,
            ):
                row.finished_at = row.finished_at or utcnow()
                finished = True

            if not row.parent_digitador_job_id:
                self._sync_task(
                    session,
                    row,
                    status=new_status if status else None,
                    result=result,
                    error=error,
                )

            if finished:
                worker = session.get(DigitadorWorkerRow, row.worker_id) if row.worker_id else None
                if worker and worker.current_job_id == job_id:
                    worker.current_job_id = None
                    worker.status = WorkerStatus.ONLINE.value

            parent_id = row.parent_digitador_job_id
            session.flush()
            job = self._row_job(row)

        if parent_id:
            self.refresh_parent_from_shards(parent_id)
            refreshed = self.get_job(job_id)
            return refreshed or job
        return job

    def refresh_parent_from_shards(self, parent_id: str) -> Optional[Job]:
        with session_scope() as session:
            parent = session.get(DigitadorJobRow, parent_id)
            if not parent:
                return None
            children = list(
                session.scalars(
                    select(DigitadorJobRow).where(
                        DigitadorJobRow.parent_digitador_job_id == parent_id
                    )
                ).all()
            )
            if not children:
                return self._row_job(parent)

            statuses = [c.status for c in children]
            terminal = {
                JobStatus.COMPLETED.value,
                JobStatus.FAILED.value,
                JobStatus.CANCELLED.value,
            }
            any_active = any(
                s in (JobStatus.PENDING.value, JobStatus.RUNNING.value) for s in statuses
            )
            all_terminal = all(s in terminal for s in statuses)

            total = 0
            exitosos = 0
            fallidos = 0
            omitidos = 0
            timeouts = 0
            existentes = 0
            uploaded = 0
            for c in children:
                r = dict(c.result or {})
                total += int(r.get("total") or 0)
                # Los agents actuales publican los nombres en español. Se
                # conservan los aliases anteriores para jobs históricos.
                exitosos += int(r.get("exitosos") or r.get("ok") or 0)
                fallidos += int(r.get("fallidos") or r.get("error") or 0)
                omitidos += int(r.get("omitidos") or 0)
                timeouts += int(r.get("timeouts") or r.get("timeout") or 0)
                existentes += int(r.get("existentes") or 0)
                uploaded += int(r.get("uploaded_count") or 0)

            processed = 0
            for c in children:
                r = dict(c.result or {})
                if c.status in terminal:
                    # fallidos ya incluye omitidos + timeouts; no se suman de
                    # nuevo para evitar que el progreso supere el total.
                    processed += int(
                        r.get("processed")
                        or (
                            int(r.get("exitosos") or r.get("ok") or 0)
                            + int(r.get("fallidos") or r.get("error") or 0)
                            + int(r.get("existentes") or 0)
                        )
                    )
                else:
                    processed += int(r.get("processed") or 0)

            shard_ids = [c.id for c in children]
            parent_result = dict(parent.result or {})
            parent_result.update(
                {
                    "mode": "distributed",
                    "shard_ids": shard_ids,
                    "total": total or int(parent_result.get("total_rows") or 0),
                    "total_rows": int(parent_result.get("total_rows") or total or 0),
                    "exitosos": exitosos,
                    "fallidos": fallidos,
                    "omitidos": omitidos,
                    "timeouts": timeouts,
                    "existentes": existentes,
                    "processed": processed,
                    "uploaded_count": uploaded,
                    "reports_prefix": parent.reports_prefix,
                    "shards": [
                        {
                            "id": c.id,
                            "worker_id": c.worker_id,
                            "status": c.status,
                            "shard_index": c.shard_index,
                            "result": dict(c.result or {}),
                            "error": c.error,
                        }
                        for c in children
                    ],
                }
            )
            parent.result = parent_result

            now = utcnow()
            if any_active:
                parent.status = JobStatus.RUNNING.value
                parent.started_at = parent.started_at or now
                parent.finished_at = None
                parent.error = None
                self._sync_task(
                    session,
                    parent,
                    status=JobStatus.RUNNING.value,
                    result=parent_result,
                )
            elif all_terminal:
                any_failed = any(s == JobStatus.FAILED.value for s in statuses)
                any_cancelled = any(s == JobStatus.CANCELLED.value for s in statuses)
                all_cancelled = all(s == JobStatus.CANCELLED.value for s in statuses)
                if all_cancelled:
                    parent.status = JobStatus.CANCELLED.value
                elif any_failed:
                    parent.status = JobStatus.FAILED.value
                    parent.error = next(
                        (c.error for c in children if c.error),
                        "Uno o más shards fallaron",
                    )
                elif any_cancelled and not any_failed:
                    parent.status = JobStatus.COMPLETED.value
                else:
                    parent.status = JobStatus.COMPLETED.value
                parent.finished_at = parent.finished_at or now
                self._sync_task(
                    session,
                    parent,
                    status=parent.status,
                    result=parent_result,
                    error=parent.error,
                )

            session.flush()
            return self._row_job(parent)

    def reclaim_stale_running_jobs(self) -> int:
        """Jobs running cuyo worker está offline → pending + reasignar a otro online."""
        reclaimed = 0
        with session_scope() as session:
            now = utcnow()
            cutoff = now - timedelta(seconds=HEARTBEAT_OFFLINE_SECONDS)
            workers = list(session.scalars(select(DigitadorWorkerRow)).all())
            offline_ids: set[str] = set()
            online_free: List[DigitadorWorkerRow] = []
            for w in workers:
                hb = w.last_heartbeat
                if hb is None:
                    offline_ids.add(w.id)
                    continue
                if hb.tzinfo is None:
                    hb = hb.replace(tzinfo=timezone.utc)
                if hb < cutoff:
                    offline_ids.add(w.id)
                    if w.current_job_id:
                        w.current_job_id = None
                    w.status = WorkerStatus.OFFLINE.value
                elif not w.current_job_id:
                    fresh_until = self._fresh_until.get(w.id)
                    if fresh_until is not None and now < fresh_until:
                        continue
                    online_free.append(w)

            if not offline_ids:
                session.flush()
                return 0

            running = session.scalars(
                select(DigitadorJobRow).where(
                    DigitadorJobRow.status == JobStatus.RUNNING.value,
                    DigitadorJobRow.worker_id.in_(list(offline_ids)),
                )
            ).all()

            rr = 0
            for row in running:
                is_parent = (
                    row.parent_digitador_job_id is None
                    and row.shard_count is not None
                    and int(row.shard_count) > 1
                )
                if is_parent:
                    continue

                row.status = JobStatus.PENDING.value
                row.started_at = None
                row.cancel_requested = False

                if online_free:
                    target = online_free[rr % len(online_free)]
                    rr += 1
                    row.worker_id = target.id

                reclaimed += 1
                session.add(
                    DigitadorJobEventRow(
                        digitador_job_id=row.id,
                        event_type="warning",
                        message=(
                            "Worker offline; job reclaimed"
                            + (f" → {row.worker_id}" if row.worker_id else "")
                        ),
                        timestamp=now,
                        metadata_json={"reclaimed": True, "worker_id": row.worker_id},
                    )
                )
                if row.parent_digitador_job_id:
                    session.add(
                        DigitadorJobEventRow(
                            digitador_job_id=row.parent_digitador_job_id,
                            event_type="warning",
                            message=f"Shard {row.shard_index} reclamado (worker offline)",
                            timestamp=now,
                            metadata_json={
                                "shard_id": row.id,
                                "worker_id": row.worker_id,
                            },
                        )
                    )

            if online_free:
                pending_orphan = session.scalars(
                    select(DigitadorJobRow).where(
                        DigitadorJobRow.status == JobStatus.PENDING.value,
                        DigitadorJobRow.worker_id.in_(list(offline_ids)),
                    )
                ).all()
                for row in pending_orphan:
                    is_parent = (
                        row.parent_digitador_job_id is None
                        and row.shard_count is not None
                        and int(row.shard_count) > 1
                    )
                    if is_parent:
                        continue
                    target = online_free[rr % len(online_free)]
                    rr += 1
                    if row.worker_id != target.id:
                        row.worker_id = target.id
                        reclaimed += 1

            session.flush()

        return reclaimed

    def add_event(
        self,
        job_id: str,
        event_type: str,
        message: str,
        *,
        radicado: Optional[str] = None,
        nit: Optional[str] = None,
        index: Optional[int] = None,
        total: Optional[int] = None,
        metadata: Optional[Dict[str, Any]] = None,
        timestamp: Optional[str] = None,
    ) -> JobEventRecord:
        with session_scope() as session:
            ts = _parse_iso(timestamp) or utcnow()
            ev = DigitadorJobEventRow(
                digitador_job_id=job_id,
                event_type=event_type,
                message=message,
                timestamp=ts,
                radicado=radicado,
                nit=nit,
                idx=index,
                total=total,
                metadata_json=metadata or {},
            )
            session.add(ev)
            session.flush()
            row = session.get(DigitadorJobRow, job_id)
            if row and row.parent_digitador_job_id:
                session.add(
                    DigitadorJobEventRow(
                        digitador_job_id=row.parent_digitador_job_id,
                        event_type=event_type,
                        message=message,
                        timestamp=ts,
                        radicado=radicado,
                        nit=nit,
                        idx=index,
                        total=total,
                        metadata_json={
                            **(metadata or {}),
                            "shard_id": job_id,
                            "shard_index": row.shard_index,
                            "worker_id": row.worker_id,
                        },
                    )
                )
                session.flush()
            return JobEventRecord(
                id=int(ev.id),
                job_id=job_id,
                event_type=event_type,
                message=message,
                timestamp=_iso(ts) or utc_now(),
                radicado=radicado,
                nit=nit,
                index=index,
                total=total,
                metadata=metadata or {},
            )

    def list_events(self, job_id: str, since_id: int = 0) -> List[JobEventRecord]:
        with session_scope() as session:
            rows = session.scalars(
                select(DigitadorJobEventRow)
                .where(
                    DigitadorJobEventRow.digitador_job_id == job_id,
                    DigitadorJobEventRow.id > since_id,
                )
                .order_by(DigitadorJobEventRow.id.asc())
            ).all()
            return [
                JobEventRecord(
                    id=int(r.id),
                    job_id=r.digitador_job_id,
                    event_type=r.event_type,
                    message=r.message,
                    timestamp=_iso(r.timestamp) or utc_now(),
                    radicado=r.radicado,
                    nit=r.nit,
                    index=r.idx,
                    total=r.total,
                    metadata=dict(r.metadata_json or {}),
                )
                for r in rows
            ]

    def create_digitador_task(
        self,
        *,
        parent_job_id: str,
        digitador_job_id: str,
        worker_id: Optional[str],
        title: str,
        original_filename: str,
        worker_ids: Optional[List[str]] = None,
        mode: str = "single",
        source_type: Optional[str] = None,
    ) -> str:
        """Crea Task(type=digitador) y la asocia al digitador_job. Devuelve task_id."""
        with session_scope() as session:
            now = utcnow()
            request: Dict[str, Any] = {
                "worker_id": worker_id,
                "worker_ids": worker_ids or ([worker_id] if worker_id else []),
                "original_filename": original_filename,
                "digitador_job_id": digitador_job_id,
                "mode": mode,
            }
            result: Dict[str, Any] = {"digitador_job_id": digitador_job_id, "mode": mode}
            if source_type:
                request["source_type"] = source_type
                result["source_type"] = source_type
            task = Task(
                job_id=parent_job_id,
                type="digitador",
                status=JobStatus.PENDING.value,
                title=title or f"Bot digitador ({original_filename})",
                request=request,
                result=result,
                created_at=now,
                updated_at=now,
            )
            session.add(task)
            session.flush()
            row = session.get(DigitadorJobRow, digitador_job_id)
            if row:
                row.task_id = task.id
                row.job_id = parent_job_id
            session.flush()
            return task.id

    def parent_job_exists(self, parent_job_id: str) -> bool:
        from infrastructure.database import Job as IpsJob

        with session_scope() as session:
            return session.get(IpsJob, parent_job_id) is not None