app/dda/auto_detect.py

"""Periodic automatic detection for identified same-area pairs.

Manual Change Detection is unchanged (user clicks Run). Automatic mode also
keeps Run-now; this module queues the same identified Before/After pairs on a
cadence (default 10 days) and saves reports without anyone opening the UI.

The FastAPI process must stay running for the schedule to fire.
"""
from __future__ import annotations

import hashlib
import json
import logging
import os
import threading
from datetime import datetime, timedelta, timezone
from pathlib import Path
from typing import Any, Optional

from sqlalchemy.orm import Session

from ..database import SessionLocal
from ..models import User
from .models import AutoDetectSettings, AutoPairSchedule, DetectionJob

logger = logging.getLogger(__name__)

AUTO_SOURCE = "auto_schedule"

_scheduler_lock = threading.Lock()
_tick_lock = threading.Lock()
_scheduler_started = False


def _utcnow() -> datetime:
    return datetime.now(timezone.utc)


def _aware(dt: Optional[datetime]) -> Optional[datetime]:
    if dt is None:
        return None
    if dt.tzinfo is None:
        return dt.replace(tzinfo=timezone.utc)
    return dt


_IST = timezone(timedelta(hours=5, minutes=30))


def parse_run_at(value: str) -> str:
    raw = (value or "02:00").strip()
    try:
        parts = raw.split(":")
        hour = int(parts[0])
        minute = int(parts[1]) if len(parts) > 1 else 0
    except (TypeError, ValueError):
        hour, minute = 2, 0
    hour = max(0, min(23, hour))
    minute = max(0, min(59, minute))
    return f"{hour:02d}:{minute:02d}"


def clamp_interval_days(value: Any, default: int = 10) -> int:
    try:
        days = int(value)
    except (TypeError, ValueError):
        days = default
    return max(1, min(365, days))


def get_or_create_settings(db: Session) -> AutoDetectSettings:
    row = db.query(AutoDetectSettings).filter(AutoDetectSettings.id == 1).first()
    if row:
        return row
    row = AutoDetectSettings(
        id=1,
        running=False,
        interval_days=clamp_interval_days(os.environ.get("AUTO_DETECT_INTERVAL_DAYS", "10")),
        run_at=parse_run_at(os.environ.get("AUTO_DETECT_RUN_AT", "02:00")),
        next_run_at=None,
    )
    db.add(row)
    db.commit()
    db.refresh(row)
    return row


def scheduler_is_armed() -> bool:
    from .config import IS_DDA_MODE
    if not IS_DDA_MODE:
        return False
    raw = os.environ.get("AUTO_DETECT_ENABLED", "1").strip().lower()
    if raw in ("0", "false", "no", "off"):
        return False
    db = SessionLocal()
    try:
        return bool(get_or_create_settings(db).running)
    except Exception:
        return False
    finally:
        db.close()


def auto_detect_enabled() -> bool:
    """Feature exists in DDA mode unless explicitly disabled in env.

    The background *queue* only runs when the operator clicks Start
    (``scheduler_is_armed``). This flag only means the Automatic tab controls
    are available.
    """
    from .config import IS_DDA_MODE
    if not IS_DDA_MODE:
        return False
    raw = os.environ.get("AUTO_DETECT_ENABLED", "1").strip().lower()
    return raw not in ("0", "false", "no", "off")


def get_auto_detect_interval_days() -> int:
    db = SessionLocal()
    try:
        return clamp_interval_days(get_or_create_settings(db).interval_days)
    except Exception:
        return clamp_interval_days(os.environ.get("AUTO_DETECT_INTERVAL_DAYS", "10"))
    finally:
        db.close()


def next_run_at_from_clock(run_at: str, now: Optional[datetime] = None) -> datetime:
    """Next future IST clock time for HH:MM. Never returns 'now' (no launch burst)."""
    ist_now = (now or datetime.now(_IST)).astimezone(_IST)
    hhmm = parse_run_at(run_at)
    hour, minute = (int(p) for p in hhmm.split(":"))
    candidate = ist_now.replace(hour=hour, minute=minute, second=0, microsecond=0)
    if candidate <= ist_now + timedelta(seconds=45):
        candidate += timedelta(days=1)
    return candidate.astimezone(timezone.utc)


def cancel_queued_auto_jobs(db: Session, reason: str) -> int:
    cancelled = 0
    pending = (
        db.query(DetectionJob)
        .filter(DetectionJob.status == "queued")
        .all()
    )
    for job in pending:
        if not is_auto_schedule_job(job):
            continue
        job.status = "failed"
        job.error_message = reason[:2000]
        job.completed_at = _utcnow()
        cancelled += 1
    if cancelled:
        db.commit()
        logger.info("Cancelled %d queued auto-detect job(s)", cancelled)
    return cancelled


def pair_key_for(before_path: str, after_path: str) -> str:
    a = (before_path or "").replace("\\", "/").strip().lower()
    b = (after_path or "").replace("\\", "/").strip().lower()
    return hashlib.sha256(f"{a}|{b}".encode("utf-8")).hexdigest()


def job_source(job: DetectionJob) -> str:
    try:
        return str(json.loads(job.params_json or "{}").get("source") or "")
    except json.JSONDecodeError:
        return ""


def is_auto_schedule_job(job: Optional[DetectionJob]) -> bool:
    return bool(job) and job_source(job) == AUTO_SOURCE


def is_auto_schedule_run(db: Session, run_id: int) -> bool:
    if not run_id:
        return False
    job = (
        db.query(DetectionJob)
        .filter(DetectionJob.run_id == run_id)
        .order_by(DetectionJob.id.desc())
        .first()
    )
    return is_auto_schedule_job(job)


def user_can_access_run(db: Session, run_id: int, user_id: int) -> bool:
    from ..models import DetectionRun
    run = db.query(DetectionRun).filter(DetectionRun.id == run_id).first()
    if not run:
        return False
    if run.user_id == user_id:
        return True
    return is_auto_schedule_run(db, run_id)


def _norm_path(path: str) -> str:
    return (path or "").replace("\\", "/").strip()


def _paths_match(job: DetectionJob, before: str, after: str) -> bool:
    try:
        params = json.loads(job.params_json or "{}")
    except json.JSONDecodeError:
        return False
    return (
        _norm_path(params.get("base_path", "")) == _norm_path(before)
        and _norm_path(params.get("comparison_path", "")) == _norm_path(after)
    )


def _schedule_owner_id(db: Session) -> int:
    admin = (
        db.query(User)
        .filter(User.role == "admin")
        .order_by(User.id.asc())
        .first()
    )
    if admin:
        return admin.id
    user = (
        db.query(User)
        .filter(User.email.notlike("__guest__%"))
        .order_by(User.id.asc())
        .first()
    )
    if user:
        return user.id
    from ..auth import get_or_create_guest_user
    return get_or_create_guest_user(db).id


def _load_identified_pairs(db: Session) -> list[dict[str, Any]]:
    from .tree.area_groups import load_cached_area_groups

    payload = load_cached_area_groups(db)
    pairs = []
    for group in payload.get("groups") or []:
        before = group.get("beforePath") or (group.get("suggestedBefore") or {}).get("path")
        after = group.get("afterPath") or (group.get("suggestedAfter") or {}).get("path")
        if not before or not after or _norm_path(before) == _norm_path(after):
            continue
        pairs.append({
            "group_id": group.get("id") or "",
            "label": group.get("label") or Path(before).name,
            "before_path": _norm_path(before),
            "after_path": _norm_path(after),
            "before_date": group.get("beforeDate"),
            "after_date": group.get("afterDate"),
        })
    return pairs


def _sync_schedules(db: Session, pairs: list[dict[str, Any]]) -> None:
    interval = get_auto_detect_interval_days()
    live_keys = set()
    now = _utcnow()
    for pair in pairs:
        key = pair_key_for(pair["before_path"], pair["after_path"])
        live_keys.add(key)
        row = db.query(AutoPairSchedule).filter(AutoPairSchedule.pair_key == key).first()
        if row:
            row.group_label = pair["label"]
            row.before_path = pair["before_path"]
            row.after_path = pair["after_path"]
            row.interval_days = interval
            row.enabled = True
            row.updated_at = now
        else:
            db.add(AutoPairSchedule(
                pair_key=key,
                group_label=pair["label"],
                before_path=pair["before_path"],
                after_path=pair["after_path"],
                interval_days=interval,
                enabled=True,
            ))
    stale = (
        db.query(AutoPairSchedule)
        .filter(AutoPairSchedule.enabled.is_(True))
        .all()
    )
    for row in stale:
        if row.pair_key not in live_keys:
            row.enabled = False
            row.updated_at = now
    db.commit()


def _pending_job_for_paths(db: Session, before: str, after: str) -> Optional[DetectionJob]:
    pending = (
        db.query(DetectionJob)
        .filter(DetectionJob.status.in_(("queued", "running")))
        .order_by(DetectionJob.id.desc())
        .limit(80)
        .all()
    )
    for job in pending:
        if _paths_match(job, before, after):
            return job
    return None


def _latest_success_for_paths(db: Session, before: str, after: str) -> Optional[datetime]:
    done = (
        db.query(DetectionJob)
        .filter(DetectionJob.status == "completed", DetectionJob.completed_at.isnot(None))
        .order_by(DetectionJob.completed_at.desc())
        .limit(80)
        .all()
    )
    for job in done:
        if _paths_match(job, before, after):
            return _aware(job.completed_at)
    return None


def _is_due(row: AutoPairSchedule, last_success: Optional[datetime], now: datetime) -> bool:
    interval = timedelta(days=max(1, int(row.interval_days or get_auto_detect_interval_days())))
    last = _aware(row.last_completed_at) or last_success
    if last and now < last + interval:
        return False
    last_enq = _aware(row.last_enqueued_at)
    if row.last_error and last_enq and now < last_enq + timedelta(hours=12):
        return False
    if last_enq and last and now < last_enq + timedelta(minutes=30):
        return False
    return True


def _enqueue_pair(db: Session, row: AutoPairSchedule) -> Optional[DetectionJob]:
    from .job_runner import create_local_folder_job
    from .local_routes import safe_resolve

    try:
        safe_resolve(row.before_path)
        safe_resolve(row.after_path)
    except Exception as exc:
        row.last_error = f"Library path missing: {exc}"[:2000]
        db.commit()
        logger.warning("Auto-detect skip %s: %s", row.group_label, exc)
        return None

    title = f"[Auto] {row.group_label}: {Path(row.before_path).name} vs {Path(row.after_path).name}"
    job = create_local_folder_job(
        db,
        base_path=row.before_path,
        comparison_path=row.after_path,
        title=title,
        created_by=_schedule_owner_id(db),
        extra_params={
            "source": AUTO_SOURCE,
            "pair_key": row.pair_key,
            "group_label": row.group_label,
        },
    )
    row.last_enqueued_at = _utcnow()
    row.last_job_id = job.id
    row.last_error = ""
    db.commit()
    logger.info(
        "Auto-detect queued job %d for %s (%s → %s)",
        job.id, row.group_label, Path(row.before_path).name, Path(row.after_path).name,
    )
    return job


def tick_auto_detect() -> int:
    """Sync identified pairs and enqueue those that are due. Returns jobs queued."""
    if not scheduler_is_armed():
        return 0
    if not _tick_lock.acquire(blocking=False):
        return 0
    queued = 0
    db = SessionLocal()
    try:
        pairs = _load_identified_pairs(db)
        _sync_schedules(db, pairs)
        now = _utcnow()
        rows = (
            db.query(AutoPairSchedule)
            .filter(AutoPairSchedule.enabled.is_(True))
            .order_by(AutoPairSchedule.id.asc())
            .all()
        )
        for row in rows:
            if _pending_job_for_paths(db, row.before_path, row.after_path):
                continue
            last_success = _latest_success_for_paths(db, row.before_path, row.after_path)
            if last_success and (
                row.last_completed_at is None or _aware(row.last_completed_at) < last_success
            ):
                row.last_completed_at = last_success
                db.commit()
            if not _is_due(row, last_success, now):
                continue
            if _enqueue_pair(db, row):
                queued += 1
        if queued:
            logger.info("Auto-detect tick queued %d pair(s)", queued)
    except Exception:
        logger.exception("Auto-detect tick failed")
        db.rollback()
        queued = 0
    finally:
        db.close()
        _tick_lock.release()
    if queued:
        from .job_runner import start_next_queued_job
        start_next_queued_job()
    return queued


def mark_schedule_from_job(job_id: int, *, status: str, run_id: Optional[int], error: str = "") -> None:
    db = SessionLocal()
    try:
        job = db.query(DetectionJob).filter(DetectionJob.id == job_id).first()
        if not is_auto_schedule_job(job):
            # Manual success for the same paths still postpones the next auto run.
            if status == "completed" and job:
                try:
                    params = json.loads(job.params_json or "{}")
                except json.JSONDecodeError:
                    return
                key = pair_key_for(params.get("base_path", ""), params.get("comparison_path", ""))
                row = db.query(AutoPairSchedule).filter(AutoPairSchedule.pair_key == key).first()
                if row:
                    row.last_completed_at = _utcnow()
                    row.last_job_id = job_id
                    if run_id:
                        row.last_run_id = run_id
                    row.last_error = ""
                    db.commit()
            return
        try:
            params = json.loads(job.params_json or "{}")
        except json.JSONDecodeError:
            params = {}
        key = params.get("pair_key") or pair_key_for(
            params.get("base_path", ""), params.get("comparison_path", ""))
        row = db.query(AutoPairSchedule).filter(AutoPairSchedule.pair_key == key).first()
        if not row:
            return
        row.last_job_id = job_id
        if status == "completed":
            row.last_completed_at = _utcnow()
            row.last_run_id = run_id
            row.last_error = ""
        else:
            row.last_error = (error or "")[:2000]
        db.commit()
    except Exception:
        logger.exception("Could not update auto-detect schedule for job %s", job_id)
        db.rollback()
    finally:
        db.close()


def refresh_schedule_rows() -> None:
    """Create/update schedule rows from identified pairs without enqueueing jobs."""
    db = SessionLocal()
    try:
        pairs = _load_identified_pairs(db)
        _sync_schedules(db, pairs)
    except Exception:
        logger.exception("Auto-detect schedule sync failed")
        db.rollback()
    finally:
        db.close()


def schedule_status_payload(db: Session) -> dict:
    settings = get_or_create_settings(db)
    interval = clamp_interval_days(settings.interval_days)
    feature_on = auto_detect_enabled()
    armed = bool(settings.running) and feature_on
    next_run = _aware(settings.next_run_at)
    rows = db.query(AutoPairSchedule).order_by(AutoPairSchedule.group_label.asc()).all()
    out = []
    for row in rows:
        last = _aware(row.last_completed_at)
        job = db.query(DetectionJob).filter(DetectionJob.id == row.last_job_id).first() if row.last_job_id else None
        out.append({
            "pairKey": row.pair_key,
            "label": row.group_label,
            "beforePath": row.before_path,
            "afterPath": row.after_path,
            "enabled": bool(row.enabled) and armed,
            "intervalDays": interval,
            "lastCompletedAt": last.isoformat() if last else None,
            "nextDueAt": next_run.isoformat() if next_run and armed else None,
            "dueNow": False,
            "lastJobId": row.last_job_id,
            "lastRunId": row.last_run_id,
            "lastJobStatus": job.status if job else None,
            "lastError": row.last_error or "",
        })
    return {
        "enabled": feature_on,
        "running": armed,
        "intervalDays": interval,
        "runAt": parse_run_at(settings.run_at),
        "nextRunAt": next_run.isoformat() if next_run and armed else None,
        "pairs": out,
    }


def start_schedule(*, interval_days: int, run_at: str) -> dict:
    if not auto_detect_enabled():
        raise ValueError("Automatic detection is disabled on the server.")
    db = SessionLocal()
    try:
        refresh_schedule_rows()
        settings = get_or_create_settings(db)
        settings.interval_days = clamp_interval_days(interval_days)
        settings.run_at = parse_run_at(run_at)
        settings.running = True
        settings.next_run_at = next_run_at_from_clock(settings.run_at)
        settings.updated_at = _utcnow()
        db.commit()
        payload = schedule_status_payload(db)
        logger.info(
            "Auto-detect started: interval=%d day(s) at %s IST; first queue %s (not now)",
            settings.interval_days, settings.run_at, settings.next_run_at,
        )
        return payload
    finally:
        db.close()


def stop_schedule() -> dict:
    db = SessionLocal()
    try:
        settings = get_or_create_settings(db)
        settings.running = False
        settings.next_run_at = None
        settings.updated_at = _utcnow()
        db.commit()
        cancel_queued_auto_jobs(
            db,
            "Cancelled because automatic schedule was stopped.",
        )
        logger.info("Auto-detect stopped; queued auto jobs cancelled")
        return schedule_status_payload(db)
    finally:
        db.close()


def _maybe_fire_scheduled_tick() -> None:
    if not scheduler_is_armed():
        return
    db = SessionLocal()
    try:
        settings = get_or_create_settings(db)
        due_at = _aware(settings.next_run_at)
        if not due_at or _utcnow() < due_at:
            return
        slot = due_at
        settings.next_run_at = slot + timedelta(days=clamp_interval_days(settings.interval_days))
        db.commit()
    except Exception:
        logger.exception("Auto-detect schedule clock failed")
        db.rollback()
        return
    finally:
        db.close()
    tick_auto_detect()


def _scheduler_loop() -> None:
    while True:
        threading.Event().wait(20)
        try:
            _maybe_fire_scheduled_tick()
        except Exception:
            logger.exception("Auto-detect waiter failed")


def reset_schedule_on_process_start() -> None:
    """Disarm the queue and cancel leftover auto jobs. Safe to call twice."""
    db = SessionLocal()
    try:
        settings = get_or_create_settings(db)
        settings.running = False
        settings.next_run_at = None
        db.commit()
        cancel_queued_auto_jobs(
            db,
            "Cancelled on server start. Automatic detection does not run until you click Start.",
        )
    except Exception:
        logger.exception("Could not reset auto-detect on launch")
        db.rollback()
    finally:
        db.close()


def start_auto_detect_scheduler() -> None:
    """Start the idle waiter. Never enqueues jobs on launch."""
    global _scheduler_started
    reset_schedule_on_process_start()

    if not auto_detect_enabled():
        logger.info("Automatic pair detection feature disabled")
        return
    with _scheduler_lock:
        if _scheduler_started:
            return
        _scheduler_started = True
    logger.info(
        "Automatic detection is idle. Set interval/time on the Automatic tab and click Start."
    )
    threading.Thread(target=_scheduler_loop, daemon=True, name="dda-auto-detect").start()