spencermandClaude Opus 5.5
ca407a2db3
When M-Files' Effective rev differs from the rev VISUAL calls, the drawing job looks that rev up (MFilesKit DrawingLookup.revision). If M-Files has it as Effective or Retired, the job copies it into drawing_revisions and attaches it to the draft with no flag and no Clerical approval, so the draft can auto-release. Past draft, already attached or with a hand drawing: unchanged, no flag and no change order. Any other status, or no document at that rev, still flags rev_mismatch with a note saying why. Packet Station only ever attaches Effective or Retired drawings. rev_approve stays for that case only. This replaces the "Use rev V as VISUAL calls it" choice from #46: the use=called path in rev-approve and its button are removed. A packet that already kept the rev that way keeps it, since that rev is now used anyway. Co-Authored-By: Claude Opus 5.5 <[email protected]>
683 lines
27 KiB
Python
683 lines
27 KiB
Python
"""The drawing-lookup job: look drawings up in M-Files (read-only), copy released PDFs into assets, and attach,
|
|
flag, stage or open a change order on the packets that use them.
|
|
|
|
Per run: the queue (drawings/queue.py) cut to `drawing_budget`; per drawing one `Lookup.status`, a fetch when a
|
|
released file exists (drawings/fetch.py downloads only when its M-Files identity changed), the `drawing_lookups`
|
|
row, then `decide` for every live packet that uses the drawing (its own drawing, or the target of its pending feed
|
|
change order), applied through `feed.writes.system_write`. A busy packet is left for the next run.
|
|
|
|
VISUAL is the source of truth for the rev. When M-Files' Effective rev is not the rev VISUAL calls, that rev is
|
|
looked up too (`Lookup.revision`). If M-Files has it as Effective or Retired, it is copied into assets and attached
|
|
to the draft with no flag. Any other status (or no document at that rev) leaves the packet flagged rev_mismatch.
|
|
|
|
Never writes to M-Files.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
from dataclasses import dataclass
|
|
from datetime import UTC, datetime
|
|
from typing import Any
|
|
|
|
from psycopg.rows import dict_row
|
|
from psycopg.types.json import Jsonb
|
|
|
|
from ... import audit, policy
|
|
from ...domain import pipeline
|
|
from ...drawings import DRAWING_FLAGS, decide, queue
|
|
from ...drawings import lookup as lookups
|
|
from ...drawings.fetch import FetchError, FetchResult, fetch
|
|
from ...drawings.lookup import Found, Lookup
|
|
from ...feed import changes, writes
|
|
from . import Counts, JobContext, JobSpec
|
|
|
|
log = logging.getLogger(__name__)
|
|
|
|
A = policy.Action
|
|
D = decide
|
|
|
|
FEED_ORIGINS = ("visual", "drawing")
|
|
|
|
|
|
class _Stale(Exception):
|
|
"""The packet changed between the read and the lock; the next run decides again."""
|
|
|
|
|
|
def new_counts() -> Counts:
|
|
return {
|
|
"checked": 0,
|
|
"fetched": 0,
|
|
"unchanged": 0,
|
|
"attached": 0,
|
|
"flagged": {},
|
|
"lookup_failed": 0,
|
|
"cos_opened": 0,
|
|
"cos_reaimed": 0,
|
|
"staged": 0,
|
|
"budget_left": 0,
|
|
"fetch_failed": 0,
|
|
"busy": 0,
|
|
"refused": 0,
|
|
"no_file": 0,
|
|
"called_fetched": 0,
|
|
}
|
|
|
|
|
|
# ---------------------------------------------------------------- the run
|
|
|
|
|
|
def run(ctx: JobContext) -> Counts:
|
|
counts = new_counts()
|
|
fs = ctx.feed_settings
|
|
budget = int(getattr(fs, "drawing_budget", 60))
|
|
try:
|
|
lk = lookups.make(ctx.settings, ctx.stack.cache)
|
|
except Exception as e: # any failure to open M-Files skips the run
|
|
ctx.error(f"M-Files could not be opened: {type(e).__name__}: {e}")
|
|
counts["reason"] = "mfiles unavailable"
|
|
counts["budget_left"] = budget
|
|
return counts
|
|
now = datetime.now(UTC)
|
|
items = queue.queue(ctx.conn, fs, now, limit=budget)
|
|
counts["budget_left"] = max(0, budget - len(items))
|
|
counts["source"] = lk.describe()
|
|
for item in items:
|
|
try:
|
|
_drawing(ctx, lk, item, counts, now)
|
|
except Exception as e: # one bad drawing must not stop the others
|
|
log.exception("drawing %s failed", item.dwg)
|
|
ctx.error(f"drawing {item.dwg}: {type(e).__name__}: {e}")
|
|
return counts
|
|
|
|
|
|
def _drawing(ctx: JobContext, lk: Lookup, item: queue.QueueItem, counts: Counts, now: datetime) -> None:
|
|
conn = ctx.conn
|
|
try:
|
|
found = lk.status(item.dwg)
|
|
except Exception as e: # a lookup error is a failed lookup, never a crash
|
|
found = lookups.failed(item.dwg, f"M-Files lookup failed: {type(e).__name__}: {e}")
|
|
counts["checked"] += 1
|
|
fetched: FetchResult | None = None
|
|
fetch_failed = False
|
|
if found.ok and found.rev is not None:
|
|
try:
|
|
fetched = fetch(conn, ctx.settings, lk, found, dwg=item.dwg)
|
|
except FetchError as e:
|
|
fetch_failed = True
|
|
counts["fetch_failed"] += 1
|
|
ctx.error(str(e))
|
|
else:
|
|
if fetched.downloaded:
|
|
counts["fetched"] += 1
|
|
audit.append(
|
|
conn,
|
|
action="drawing.fetched",
|
|
role_snapshot="system",
|
|
subject_type="asset",
|
|
subject_id=str(fetched.asset_id),
|
|
detail={
|
|
"dwg": item.dwg,
|
|
"rev": found.rev,
|
|
"sha256": fetched.sha256.hex(),
|
|
"reused": fetched.reused,
|
|
},
|
|
)
|
|
if not found.ok:
|
|
counts["lookup_failed"] += 1
|
|
ctx.error(f"drawing {item.dwg}: {D.failed_note(found)}")
|
|
vrev_seen = item.visual_rev if item.co_id is None else own_rev(conn, item.dwg)
|
|
save_lookup(conn, item, found, fetched, now, fetch_failed=fetch_failed, visual_rev_seen=vrev_seen)
|
|
scope = [str(d) for d in (getattr(ctx.feed_settings, "departments", None) or [])]
|
|
called: dict[int, CalledCopy | None] = {} # VISUAL rev -> what M-Files holds at that rev, this drawing
|
|
for pid, relations in sharing(conn, item.dwg, scope):
|
|
_packet(ctx, lk, item.dwg, found, fetched, pid, relations, counts, called)
|
|
|
|
|
|
# ---------------------------------------------------------------- drawing_lookups
|
|
|
|
|
|
def save_lookup(
|
|
conn: Any,
|
|
item: queue.QueueItem,
|
|
found: Found,
|
|
fetched: FetchResult | None,
|
|
now: datetime,
|
|
*,
|
|
fetch_failed: bool,
|
|
visual_rev_seen: str | None,
|
|
) -> None:
|
|
"""Upsert the drawing_lookups row for this lookup.
|
|
|
|
A failed lookup keeps everything known before (rev, file, asset) and backs off. A good one stores what M-Files
|
|
has; the file identity and the asset are stored together, and cleared when there is no released file or the
|
|
copy failed (so the next run downloads again).
|
|
"""
|
|
with conn.transaction():
|
|
prev = conn.execute("SELECT fail_count FROM drawing_lookups WHERE dwg = %s FOR UPDATE", (item.dwg,)).fetchone()
|
|
if not found.ok:
|
|
fails = (int(prev[0]) if prev else 0) + 1
|
|
conn.execute(
|
|
"INSERT INTO drawing_lookups (dwg, checked_at, ok, note, fail_count, failing_since, next_check_at,"
|
|
" visual_rev_seen) VALUES (%(dwg)s, %(now)s, false, %(note)s, %(fails)s, %(now)s, %(next)s, %(vrev)s)"
|
|
" ON CONFLICT (dwg) DO UPDATE SET checked_at = EXCLUDED.checked_at, ok = false, note = EXCLUDED.note,"
|
|
" fail_count = EXCLUDED.fail_count,"
|
|
" failing_since = coalesce(drawing_lookups.failing_since, EXCLUDED.failing_since),"
|
|
" next_check_at = EXCLUDED.next_check_at, visual_rev_seen = EXCLUDED.visual_rev_seen",
|
|
{
|
|
"dwg": item.dwg,
|
|
"now": now,
|
|
"note": D.failed_note(found),
|
|
"fails": fails,
|
|
"next": now + queue.backoff(fails),
|
|
"vrev": visual_rev_seen,
|
|
},
|
|
)
|
|
return
|
|
has_file = fetched is not None
|
|
conn.execute(
|
|
"INSERT INTO drawing_lookups (dwg, checked_at, ok, rev, name, obj_id, obj_version, file_id, file_version,"
|
|
" file_size, base, draft_rev, draft_where, in_drafting, note, fail_count, failing_since, asset_id, sha256,"
|
|
" visual_rev_seen, next_check_at) VALUES (%(dwg)s, %(now)s, true, %(rev)s, %(name)s, %(obj_id)s,"
|
|
" %(obj_version)s, %(file_id)s, %(file_version)s, %(file_size)s, %(base)s, %(draft_rev)s,"
|
|
" %(draft_where)s, %(in_drafting)s, %(note)s, 0, NULL, %(asset_id)s, %(sha)s, %(vrev)s, %(next)s)"
|
|
" ON CONFLICT (dwg) DO UPDATE SET checked_at = EXCLUDED.checked_at, ok = true, rev = EXCLUDED.rev,"
|
|
" name = EXCLUDED.name, obj_id = EXCLUDED.obj_id, obj_version = EXCLUDED.obj_version,"
|
|
" file_id = EXCLUDED.file_id, file_version = EXCLUDED.file_version, file_size = EXCLUDED.file_size,"
|
|
" base = EXCLUDED.base, draft_rev = EXCLUDED.draft_rev, draft_where = EXCLUDED.draft_where,"
|
|
" in_drafting = EXCLUDED.in_drafting, note = EXCLUDED.note, fail_count = 0, failing_since = NULL,"
|
|
" asset_id = EXCLUDED.asset_id, sha256 = EXCLUDED.sha256,"
|
|
" visual_rev_seen = coalesce(EXCLUDED.visual_rev_seen, drawing_lookups.visual_rev_seen),"
|
|
" next_check_at = EXCLUDED.next_check_at",
|
|
{
|
|
"dwg": item.dwg,
|
|
"now": now,
|
|
"rev": found.rev,
|
|
"name": found.name,
|
|
"obj_id": found.obj_id if has_file else None,
|
|
"obj_version": found.obj_version if has_file else None,
|
|
"file_id": found.file_id if has_file else None,
|
|
"file_version": found.file_version if has_file else None,
|
|
"file_size": found.file_size,
|
|
"base": found.base,
|
|
"draft_rev": found.draft_rev,
|
|
"draft_where": found.draft_where or None,
|
|
"in_drafting": found.in_drafting,
|
|
"note": found.note or None,
|
|
"asset_id": fetched.asset_id if fetched else None,
|
|
"sha": fetched.sha256 if fetched else None,
|
|
"vrev": visual_rev_seen,
|
|
"next": now + queue.backoff(1) if fetch_failed else None,
|
|
},
|
|
)
|
|
|
|
|
|
def save_revision(conn: Any, dwg: str, rf: Found, fetched: FetchResult, now: datetime) -> None:
|
|
"""Upsert the drawing_revisions row for a rev `Lookup.revision` found and the copy of its file."""
|
|
conn.execute(
|
|
"INSERT INTO drawing_revisions (dwg, rev, checked_at, controlled, state, name, obj_id, obj_version, file_id,"
|
|
" file_version, file_size, base, asset_id, sha256) VALUES (%(dwg)s, %(rev)s, %(now)s, %(controlled)s,"
|
|
" %(state)s, %(name)s, %(obj_id)s, %(obj_version)s, %(file_id)s, %(file_version)s, %(file_size)s, %(base)s,"
|
|
" %(asset_id)s, %(sha)s) ON CONFLICT (dwg, rev) DO UPDATE SET checked_at = EXCLUDED.checked_at,"
|
|
" controlled = EXCLUDED.controlled, state = EXCLUDED.state, name = EXCLUDED.name, obj_id = EXCLUDED.obj_id,"
|
|
" obj_version = EXCLUDED.obj_version, file_id = EXCLUDED.file_id, file_version = EXCLUDED.file_version,"
|
|
" file_size = EXCLUDED.file_size, base = EXCLUDED.base, asset_id = EXCLUDED.asset_id, sha256 = EXCLUDED.sha256",
|
|
{
|
|
"dwg": dwg,
|
|
"rev": rf.rev,
|
|
"now": now,
|
|
"controlled": rf.controlled,
|
|
"state": rf.state or None,
|
|
"name": fetched.name or rf.name,
|
|
"obj_id": rf.obj_id,
|
|
"obj_version": rf.obj_version,
|
|
"file_id": rf.file_id,
|
|
"file_version": rf.file_version,
|
|
"file_size": rf.file_size,
|
|
"base": rf.base,
|
|
"asset_id": fetched.asset_id,
|
|
"sha": fetched.sha256,
|
|
},
|
|
)
|
|
|
|
|
|
# The M-Files statuses a rev VISUAL calls may be attached at. Anything else is never attached.
|
|
ATTACHABLE = frozenset({"effective", "retired"})
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class CalledCopy:
|
|
"""The rev VISUAL calls as M-Files holds it. `record` goes on the packet as `visual.called`: {rev, controlled,
|
|
state, name, asset_id, sha256, pages} for a copy, {rev, missing: true} when M-Files has no document at that rev,
|
|
{rev, controlled, state, refused: true} when its status may not be attached. `found` and `fetched` are set only
|
|
for a copy."""
|
|
|
|
record: dict[str, Any]
|
|
found: Found | None = None
|
|
fetched: FetchResult | None = None
|
|
|
|
|
|
def called_rev(
|
|
ctx: JobContext, lk: Lookup, dwg: str, visual_rev: str, memo: dict[int, CalledCopy | None], counts: Counts
|
|
) -> CalledCopy | None:
|
|
"""The rev VISUAL calls, as M-Files holds it, copied into assets when its status is Effective or Retired. None when
|
|
it could not be read this run (the packet keeps what it had) or the VISUAL rev is not a number."""
|
|
v = D.as_int(visual_rev)
|
|
if v is None:
|
|
return None
|
|
if v in memo:
|
|
return memo[v]
|
|
memo[v] = None
|
|
conn = ctx.conn
|
|
try:
|
|
rf = lk.revision(dwg, v)
|
|
except LookupError as e:
|
|
ctx.error(f"drawing {dwg}: rev {v}: {e}")
|
|
return None
|
|
if rf is None:
|
|
memo[v] = CalledCopy({"rev": str(v), "missing": True})
|
|
return memo[v]
|
|
if rf.controlled.casefold() not in ATTACHABLE:
|
|
memo[v] = CalledCopy({"rev": str(v), "controlled": rf.controlled, "state": rf.state or None, "refused": True})
|
|
return memo[v]
|
|
try:
|
|
fetched = fetch(conn, ctx.settings, lk, rf, dwg=dwg)
|
|
except FetchError as e:
|
|
counts["fetch_failed"] += 1
|
|
ctx.error(str(e))
|
|
return None
|
|
with conn.transaction():
|
|
save_revision(conn, dwg, rf, fetched, datetime.now(UTC))
|
|
if fetched.downloaded:
|
|
counts["called_fetched"] += 1
|
|
audit.append(
|
|
conn,
|
|
action="drawing.fetched",
|
|
role_snapshot="system",
|
|
subject_type="asset",
|
|
subject_id=str(fetched.asset_id),
|
|
detail={
|
|
"dwg": dwg,
|
|
"rev": v,
|
|
"controlled": rf.controlled,
|
|
"sha256": fetched.sha256.hex(),
|
|
"reused": fetched.reused,
|
|
},
|
|
)
|
|
record = {
|
|
"rev": str(v),
|
|
"controlled": rf.controlled,
|
|
"state": rf.state or None,
|
|
"name": fetched.name or rf.name,
|
|
"asset_id": str(fetched.asset_id),
|
|
"sha256": fetched.sha256.hex(),
|
|
"pages": fetched.pages,
|
|
}
|
|
memo[v] = CalledCopy(record, rf, fetched)
|
|
return memo[v]
|
|
|
|
|
|
# ---------------------------------------------------------------- packets
|
|
|
|
|
|
def own_rev(conn: Any, dwg: str) -> str | None:
|
|
"""The rev a live VISUAL packet with this drawing carries (for visual_rev_seen when the item came from a CO)."""
|
|
row = conn.execute(
|
|
"SELECT coalesce(btrim(rev), '') FROM packets WHERE removed_at IS NULL AND source_kind = 'visual'"
|
|
" AND status <> 'filed' AND btrim(drawing) = %s ORDER BY id LIMIT 1",
|
|
(dwg,),
|
|
).fetchone()
|
|
return None if row is None else str(row[0])
|
|
|
|
|
|
SHARING_SQL = """
|
|
SELECT p.id, 'own' AS rel FROM packets p
|
|
WHERE p.removed_at IS NULL AND p.source_kind = 'visual' AND p.status <> 'filed' AND btrim(p.drawing) = %(dwg)s
|
|
AND (cardinality(%(depts)s::text[]) = 0 OR p.dept_code = ANY(%(depts)s::text[]))
|
|
UNION
|
|
SELECT p.id, 'co' AS rel FROM change_orders c JOIN packets p ON p.id = c.packet_id
|
|
WHERE c.review = 'pending' AND c.origin IN ('visual', 'drawing') AND jsonb_typeof(c.waiting) = 'object'
|
|
AND btrim(c.waiting->>'dwg') = %(dwg)s AND p.removed_at IS NULL AND p.status NOT IN ('draft', 'filed')
|
|
AND (cardinality(%(depts)s::text[]) = 0 OR p.dept_code = ANY(%(depts)s::text[]))
|
|
ORDER BY 1
|
|
"""
|
|
|
|
|
|
def sharing(conn: Any, dwg: str, depts: list[str]) -> list[tuple[int, set[str]]]:
|
|
"""Live packets that use the drawing: as their own drawing ('own') or as a pending feed CO's target ('co')."""
|
|
out: dict[int, set[str]] = {}
|
|
for pid, rel in conn.execute(SHARING_SQL, {"dwg": dwg, "depts": depts}).fetchall():
|
|
out.setdefault(int(pid), set()).add(str(rel))
|
|
return sorted(out.items())
|
|
|
|
|
|
@dataclass
|
|
class PacketFacts:
|
|
row: dict[str, Any]
|
|
visual: dict[str, Any]
|
|
attached_rev: str | None
|
|
attached_shas: set[bytes]
|
|
co: dict[str, Any] | None
|
|
rejected_sha: bytes | None
|
|
|
|
|
|
def _sha(value: Any) -> bytes | None:
|
|
if not isinstance(value, str) or not value:
|
|
return None
|
|
try:
|
|
return bytes.fromhex(value)
|
|
except ValueError:
|
|
return None
|
|
|
|
|
|
def facts(conn: Any, pid: int) -> PacketFacts | None:
|
|
with conn.cursor(row_factory=dict_row) as cur:
|
|
row = cur.execute(
|
|
"SELECT id, status::text AS status, drawing, rev, drawing_state::text AS drawing_state, flags, visual,"
|
|
" version FROM packets WHERE id = %s AND removed_at IS NULL",
|
|
(pid,),
|
|
).fetchone()
|
|
if row is None:
|
|
return None
|
|
pages = cur.execute(
|
|
"SELECT pg.rev, a.sha256 FROM packet_pages pg LEFT JOIN assets a ON a.id = pg.asset_id"
|
|
" WHERE pg.packet_id = %s AND pg.type = 'drawing' AND pg.void_at IS NULL ORDER BY pg.n",
|
|
(pid,),
|
|
).fetchall()
|
|
co = cur.execute(
|
|
"SELECT id, co_no, origin::text AS origin, waiting, staged FROM change_orders"
|
|
" WHERE packet_id = %s AND review = 'pending' AND origin IN ('visual', 'drawing') ORDER BY id DESC LIMIT 1",
|
|
(pid,),
|
|
).fetchone()
|
|
rej = cur.execute(
|
|
"SELECT staged FROM change_orders WHERE packet_id = %s AND origin = 'drawing' AND review = 'rejected'"
|
|
" ORDER BY id DESC LIMIT 1",
|
|
(pid,),
|
|
).fetchone()
|
|
visual = row["visual"] if isinstance(row["visual"], dict) else {}
|
|
attached_rev = None
|
|
if pages:
|
|
attached_rev = str(pages[0]["rev"] or "").strip()
|
|
shas = {bytes(p["sha256"]) for p in pages if p["sha256"] is not None}
|
|
rejected = None
|
|
if rej is not None and isinstance(rej["staged"], dict):
|
|
rejected = _sha(rej["staged"].get("sha256"))
|
|
return PacketFacts(dict(row), dict(visual), attached_rev, shas, dict(co) if co else None, rejected)
|
|
|
|
|
|
def _dwg_rev(value: Any) -> tuple[str, str] | None:
|
|
if isinstance(value, dict) and value.get("dwg") and value.get("rev") is not None:
|
|
return (str(value["dwg"]).strip(), str(value["rev"]).strip())
|
|
return None
|
|
|
|
|
|
def target_for(dwg: str, pf: PacketFacts, *, own: bool) -> D.Target:
|
|
row, visual, co = pf.row, pf.visual, pf.co
|
|
waiting = co.get("waiting") if co else None
|
|
staged = co.get("staged") if co else None
|
|
co_waiting = _dwg_rev(waiting)
|
|
co_staged = _dwg_rev(staged)
|
|
packet_rev = str(row["rev"] or "").strip()
|
|
rev_ok = None
|
|
ro = visual.get("rev_ok")
|
|
# A rev_ok with use = "called" (Clerical kept the rev VISUAL calls, before that became automatic) approves no other
|
|
# rev: the rev VISUAL calls is used anyway.
|
|
if (
|
|
isinstance(ro, dict)
|
|
and ro.get("rev") is not None
|
|
and ro.get("use") != "called"
|
|
and str(ro.get("vis_rev") or "").strip() == packet_rev
|
|
):
|
|
rev_ok = str(ro["rev"]).strip()
|
|
visual_rev = packet_rev if own or co_waiting is None else co_waiting[1]
|
|
return D.Target(
|
|
dwg=dwg,
|
|
visual_rev=visual_rev,
|
|
attached_rev=pf.attached_rev,
|
|
attached_sha=next(iter(sorted(pf.attached_shas)), None),
|
|
status=str(row["status"]),
|
|
drawing_state=str(row["drawing_state"]),
|
|
has_hand_drawing=row["drawing_state"] == "hand" or bool(visual.get("manual_dwg")),
|
|
rev_ok=rev_ok,
|
|
co_waiting=co_waiting,
|
|
co_staged=co_staged,
|
|
co_pending=co is not None,
|
|
co_staged_sha=_sha(staged.get("sha256")) if isinstance(staged, dict) else None,
|
|
rejected_sha=pf.rejected_sha,
|
|
)
|
|
|
|
|
|
def appr_of(t: D.Target, f: Found, now: datetime) -> dict[str, Any]:
|
|
return {
|
|
"ok": f.ok,
|
|
"rev": f.rev,
|
|
"vrev": t.visual_rev,
|
|
"name": f.name,
|
|
"note": f.note or "",
|
|
"base": f.base,
|
|
"draft_rev": f.draft_rev,
|
|
"draft_where": f.draft_where,
|
|
"in_drafting": f.in_drafting,
|
|
"at": now.isoformat(),
|
|
}
|
|
|
|
|
|
def _same_appr(a: Any, b: dict[str, Any]) -> bool:
|
|
if not isinstance(a, dict):
|
|
return False
|
|
return {k: v for k, v in a.items() if k != "at"} == {k: v for k, v in b.items() if k != "at"}
|
|
|
|
|
|
def new_flags(flags: list[str], d: D.Decision) -> list[str]:
|
|
"""Every flag but the drawing ones, then the decision's flag."""
|
|
out = [f for f in flags if f not in DRAWING_FLAGS]
|
|
if d.flag and d.flag not in out:
|
|
out.append(d.flag)
|
|
return out
|
|
|
|
|
|
def _packet(
|
|
ctx: JobContext,
|
|
lk: Lookup,
|
|
dwg: str,
|
|
found: Found,
|
|
fetched: FetchResult | None,
|
|
pid: int,
|
|
relations: set[str],
|
|
counts: Counts,
|
|
called_memo: dict[int, CalledCopy | None],
|
|
) -> None:
|
|
conn = ctx.conn
|
|
pf = facts(conn, pid)
|
|
if pf is None:
|
|
return
|
|
own = "own" in relations
|
|
t = target_for(dwg, pf, own=own)
|
|
same_file = fetched is not None and fetched.sha256 in pf.attached_shas
|
|
cc: CalledCopy | None = None
|
|
dc: D.Called | None = None
|
|
v = D.as_int(t.visual_rev)
|
|
if own and found.ok and v is not None and v != found.rev:
|
|
cc = called_rev(ctx, lk, dwg, t.visual_rev, called_memo, counts)
|
|
if cc is not None and cc.found is not None and cc.fetched is not None:
|
|
dc = D.Called(v, cc.found.controlled, cc.fetched.sha256 in pf.attached_shas)
|
|
d = D.decide(t, found, same_file=same_file, fetched_sha=fetched.sha256 if fetched else None, called=dc)
|
|
if not own and d.kind != D.STAGE_FOR_CO:
|
|
return # only the CO's target drawing: nothing to do until it can be staged
|
|
if d.kind in (D.ATTACH, D.OPEN_CO, D.STAGE_FOR_CO) and fetched is None:
|
|
counts["no_file"] += 1 # the copy failed this run; try again next run
|
|
return
|
|
now = datetime.now(UTC)
|
|
row, visual = pf.row, pf.visual
|
|
flags = list(row["flags"] or [])
|
|
flags_after = new_flags(flags, d) if own else flags
|
|
patch: dict[str, Any] = {}
|
|
if own:
|
|
appr = appr_of(t, found, now)
|
|
if not _same_appr(visual.get("appr"), appr):
|
|
patch["appr"] = appr
|
|
if found.ok:
|
|
confirmed = D.rev_confirmed(t, found, dc)
|
|
if bool(visual.get("rev_confirmed")) != confirmed:
|
|
patch["rev_confirmed"] = confirmed
|
|
if cc is not None:
|
|
if visual.get("called") != cc.record:
|
|
patch["called"] = cc.record
|
|
elif visual.get("called") is not None and (v is None or v == found.rev):
|
|
patch["called"] = None
|
|
state_after = d.drawing_state if own else str(row["drawing_state"])
|
|
material = (
|
|
d.kind in (D.ATTACH, D.ATTACH_CALLED, D.OPEN_CO, D.STAGE_FOR_CO)
|
|
or flags_after != flags
|
|
or state_after != row["drawing_state"]
|
|
or bool(patch)
|
|
)
|
|
if not material:
|
|
counts["unchanged"] += 1
|
|
return
|
|
action = A.VIS_UPDATE_DRAFT if row["status"] == "draft" else A.VIS_OPEN_CO
|
|
label, note = "Drawing status from M-Files", d.note or None
|
|
called_now = patch.get("called") or {}
|
|
if d.kind == D.REV_MISMATCH and note and called_now.get("refused"):
|
|
note = f"{note}; rev {called_now['rev']} is {called_now['controlled']} in M-Files - not attached"
|
|
elif d.kind == D.REV_MISMATCH and note and called_now.get("missing"):
|
|
note = f"{note}; rev {called_now['rev']} is not in M-Files"
|
|
use_found, use_fetched = found, fetched
|
|
if d.kind == D.ATTACH_CALLED and cc is not None and cc.found is not None and cc.fetched is not None:
|
|
use_found, use_fetched = cc.found, cc.fetched
|
|
if d.kind in (D.ATTACH, D.ATTACH_CALLED) and use_fetched is not None:
|
|
label = "Drawing attached from M-Files"
|
|
note = "; ".join(x for x in (use_fetched.name or use_found.name or "", d.note) if x) or None
|
|
detail = {
|
|
"dwg": dwg,
|
|
"rev": use_found.rev,
|
|
"kind": d.kind,
|
|
"asset_id": str(use_fetched.asset_id) if use_fetched else None,
|
|
}
|
|
try:
|
|
with writes.system_write(conn, pid, action, label=label, note=note, detail=detail) as sw:
|
|
if int(sw.packet["version"]) != int(row["version"]):
|
|
raise _Stale
|
|
if d.kind in (D.ATTACH, D.ATTACH_CALLED) and use_fetched is not None:
|
|
_attach(sw, pid, dwg, use_found, use_fetched, d, flags_after, patch, now)
|
|
counts["attached"] += 1
|
|
else:
|
|
_status(sw, pid, state_after, flags_after, patch)
|
|
if d.flag:
|
|
fl = counts["flagged"]
|
|
fl[d.kind] = fl.get(d.kind, 0) + 1
|
|
if d.kind == D.OPEN_CO and fetched is not None:
|
|
_open_co(sw, t, dwg, found, fetched, d, counts)
|
|
elif d.kind == D.STAGE_FOR_CO and fetched is not None and pf.co is not None:
|
|
_stage(sw, pf.co, dwg, found, fetched)
|
|
counts["staged"] += 1
|
|
elif d.kind == D.UNCHANGED:
|
|
counts["unchanged"] += 1
|
|
except (_Stale, writes.Busy, writes.Missing):
|
|
counts["busy"] += 1
|
|
except writes.Refused as e:
|
|
counts["refused"] += 1
|
|
ctx.error(f"packet {pid}: {d.kind} refused: {e}", degrade=False)
|
|
|
|
|
|
def _staged(dwg: str, found: Found, fetched: FetchResult) -> dict[str, Any]:
|
|
return {
|
|
"asset_id": str(fetched.asset_id),
|
|
"name": fetched.name or found.name,
|
|
"rev": str(found.rev),
|
|
"dwg": dwg,
|
|
"sha256": fetched.sha256.hex(),
|
|
"pages": fetched.pages,
|
|
}
|
|
|
|
|
|
def _attach(
|
|
sw: Any,
|
|
pid: int,
|
|
dwg: str,
|
|
found: Found,
|
|
fetched: FetchResult,
|
|
d: D.Decision,
|
|
flags: list[str],
|
|
patch: dict[str, Any],
|
|
now: datetime,
|
|
) -> None:
|
|
conn = sw.conn
|
|
pages = max(1, int(fetched.pages))
|
|
conn.execute("DELETE FROM packet_pages WHERE packet_id = %s AND type = 'drawing'", (pid,))
|
|
n0 = pipeline.next_counter(conn, pid, "page", pages)
|
|
rev = str(found.rev)
|
|
with conn.cursor() as cur:
|
|
cur.executemany(
|
|
"INSERT INTO packet_pages (packet_id, n, type, asset_id, pn, dwg, rev) VALUES (%s, %s, 'drawing', %s, %s,"
|
|
" %s, %s)",
|
|
[(pid, n0 + i, fetched.asset_id, i + 1, dwg, rev) for i in range(pages)],
|
|
)
|
|
vault = {
|
|
"asset_id": str(fetched.asset_id),
|
|
"name": fetched.name or found.name,
|
|
"rev": found.rev,
|
|
"note": d.note,
|
|
"at": now.isoformat(),
|
|
**({"controlled": found.controlled} if found.controlled else {}),
|
|
}
|
|
merged = {**patch, "rev_confirmed": True, "vault": vault}
|
|
conn.execute(
|
|
"UPDATE packets SET drawing_state = 'attached', flags = %s, visual = (visual - 'manual_dwg') || %s"
|
|
" WHERE id = %s",
|
|
(flags, Jsonb(merged), pid),
|
|
)
|
|
sw.changed("pages", "drawing_state", "flags", "visual")
|
|
|
|
|
|
def _status(sw: Any, pid: int, state: str, flags: list[str], patch: dict[str, Any]) -> None:
|
|
sw.conn.execute(
|
|
"UPDATE packets SET drawing_state = %s::drawing_state, flags = %s, visual = visual || %s WHERE id = %s",
|
|
(state, flags, Jsonb(patch), pid),
|
|
)
|
|
sw.changed("drawing_state", "flags", "visual")
|
|
|
|
|
|
def _open_co(sw: Any, t: D.Target, dwg: str, found: Found, fetched: FetchResult, d: D.Decision, counts: Counts) -> None:
|
|
vis = {
|
|
"from_dwg": dwg,
|
|
"to_dwg": dwg,
|
|
"dwg": dwg,
|
|
"from_rev": t.attached_rev,
|
|
"to_rev": str(found.rev),
|
|
"dwg_change": True,
|
|
"changes": [d.note],
|
|
}
|
|
res = changes.open_or_reaim(
|
|
sw,
|
|
origin="drawing",
|
|
reason=d.note,
|
|
vis=vis,
|
|
waiting=None,
|
|
staged=_staged(dwg, found, fetched),
|
|
target_fp=None,
|
|
minor=False,
|
|
)
|
|
# open_or_reaim writes the packet_log label itself
|
|
counts["cos_opened" if res.fresh else "cos_reaimed"] += 1
|
|
|
|
|
|
def _stage(sw: Any, co: dict[str, Any], dwg: str, found: Found, fetched: FetchResult) -> None:
|
|
staged = _staged(dwg, found, fetched)
|
|
sw.conn.execute(
|
|
"UPDATE change_orders SET staged = %s, waiting = NULL, row_version = row_version + 1"
|
|
" WHERE id = %s AND review = 'pending'",
|
|
(Jsonb(staged), co["id"]),
|
|
)
|
|
sw.changed("change_orders")
|
|
sw.log(f"Drawing for change order {co['co_no']} ready for Clerical", staged["name"])
|
|
|
|
|
|
def _cadence(fs: Any, now: datetime) -> int | None:
|
|
return int(fs.drawing_interval_minutes) * 60
|
|
|
|
|
|
SPEC = JobSpec(name="drawing-lookup", run=run, cadence=_cadence, pausable=True)
|
|
|
|
__all__ = ["SPEC", "called_rev", "new_flags", "run", "save_lookup", "save_revision", "sharing", "target_for"]
|