PacketStation/server/packetstation/worker/jobs/visual_feed.py

"""The visual-feed job: read VISUAL (or the fixture files), record the read in the cache, build the candidates
from the cache and apply them.

Per run: the feed settings; the tracked keys (every work order of a live VISUAL packet that is not filed, plus the
go-live rows Clerical pulled in); the window from today in site time; one read through the source (labor at most
every 30 minutes); `cache.write_batch`; `cache.rows_for_apply`; `candidates.build`; `apply.apply`. The run is
degraded when a read section failed or the batch is suspect.

VisualKit: one `Visual.from_env()` per run, closed after, and the stack guard checked against its settings before
the first read. Read-only: VisualKit enforces it.
"""

from __future__ import annotations

from datetime import UTC, datetime, timedelta
from typing import Any

from ... import guard
from ...feed import apply as feed_apply
from ...feed import cache, candidates, config
from ...feed.config import FeedSettings
from ...feed.source import FeedSource, FixtureSource, VisualKitSource
from ...feed.types import FeedData, WoKey
from . import Counts, JobContext, JobSpec

LABOR_EVERY = timedelta(minutes=30)


def tracked_keys(conn: Any) -> list[WoKey]:
    """Work orders of live VISUAL packets that are not filed, plus pulled-in go-live rows."""
    rows = conn.execute(
        "SELECT DISTINCT w.wo_base, w.lot, w.split, w.sub FROM packet_work_orders w JOIN packets p"
        " ON p.id = w.packet_id WHERE p.removed_at IS NULL AND p.status <> 'filed' AND p.source_kind = 'visual'"
        " UNION SELECT wo_base, lot, split, sub FROM visual_ignored_wos WHERE pulled_in_at IS NOT NULL"
    ).fetchall()
    return sorted({WoKey(r[0], r[1], r[2], r[3]) for r in rows}, key=lambda k: k.as_tuple())


def resource_map(conn: Any) -> dict[str, str]:
    return {r[0]: r[1] for r in conn.execute("SELECT resource, target FROM visual_resource_map")}


def _labor_due(conn: Any, fs: FeedSettings) -> bool:
    if not fs.labor_tickets:
        return False
    last = cache.last_labor_read(conn)
    return last is None or datetime.now(UTC) - last >= LABOR_EVERY


def _read(source: FeedSource, ctx: JobContext, tracked: list[WoKey], window: tuple[Any, Any], labor: bool) -> FeedData:
    return source.read(
        site=ctx.settings.site_id,
        since=window[0],
        until=window[1],
        tracked=tracked,
        with_materials=True,
        with_orders=True,
        with_labor=labor,
        labor_since=None,
    )


def run(ctx: JobContext) -> Counts:
    conn = ctx.conn
    fs = ctx.feed_settings
    today = ctx.today_site()
    window = config.window(fs, today)
    tracked = tracked_keys(conn)
    labor_due = _labor_due(conn, fs)

    if ctx.settings.feed_source == "fixture":
        fixture = FixtureSource(ctx.settings.fixture_dir)
        describe = fixture.describe()
        data = _read(fixture, ctx, tracked, window, labor_due)
    else:
        from visualkit import Visual

        with Visual.from_env() as v:
            # The stack guard against this run's own VisualKit settings, before the first read.
            result = guard.check_stack(ctx.settings, v.settings)
            ctx.visual_target = ctx.visual_target or result.target
            vk = VisualKitSource(v)
            describe = vk.describe()
            data = _read(vk, ctx, tracked, window, labor_due)
    for e in data.errors:
        ctx.error(e)

    batch = cache.write_batch(conn, ctx.run_id, data, describe, window)
    if batch.suspect:
        ctx.degrade(f"suspect read: {batch.reason}; nothing marked gone")

    in_play = {r.key for r in data.rows} | set(tracked)
    rows, mats, mat_legs = cache.rows_for_apply(conn, in_play)
    if data.material_legs:
        mat_legs |= data.material_legs
    built = candidates.build(
        rows,
        mats,
        mat_legs,
        resource_map=resource_map(conn),
        window=window,
        golive=fs.golive_date,
        departments=fs.departments,
        tracked_keys=set(tracked),
        today=today,
    )
    bases = {r.wo for r in rows}
    projects = {candidates.project_of(b) for b in bases} | {c.project for c in built.candidates.values()}
    orders = cache.orders(conn, bases=bases | projects, projects=projects)
    counts = feed_apply.apply(
        conn,
        ctx=ctx,
        batch=batch,
        feed_settings=fs,
        built=built,
        orders=orders,
        labor=cache.labor_hours(conn),
        today=today,
        seen={r.key: r.st for r in data.rows},
        missing=() if batch.suspect else data.tracked_missing,
    )
    counts["wos_window"] = data.window_count
    counts["wos_tracked"] = len(tracked)
    counts["rows"] = len(data.rows)
    counts["materials"] = len(data.materials) if data.materials is not None else 0
    counts["orders"] = int(batch.counts.get("orders", 0))
    counts["labor_legs"] = int(batch.counts.get("labor_legs", 0))
    counts["batch_id"] = batch.id
    if batch.suspect:
        counts["suspect_reason"] = batch.reason
    cache.finish_batch(conn, batch.id, counts)
    return counts


def _cadence(fs: FeedSettings, now_site: datetime) -> int | None:
    return config.interval_seconds(fs, now_site)


SPEC = JobSpec(name="visual-feed", run=run, cadence=_cadence, pausable=True, chain=("drawing-lookup",))

__all__ = ["SPEC", "resource_map", "run", "tracked_keys"]