Files
shopdb-flask/shopdb/core/services/dualpath.py
cproudlock cd353b6432
Some checks failed
CI / backend (push) Has been cancelled
CI / naming (push) Has been cancelled
CI / frontend (push) Has been cancelled
Review safe-polish: docs accuracy, dead imports, no-emoji, geenforce robustness
From the full multi-agent review (0 high, 7 medium, 17 low findings). Applies
the mechanical, low-risk items; design/policy findings left for a decision.

Docs accuracy: CLAUDE.md contract 0.10.0 -> 0.11.0 and both stale Alembic head
citations -> 7d24_customfield_searchable / 31 migrations; Dockerfile bundled-
plugin comment fixed (drop nonexistent "equipment", add machines +
measuringtools, count eleven).

Style/naming (LOCKED rules): remove a CSS-escaped pushpin emoji before location
search results (no-emoji policy); rename ManifestEditor shareRoot -> shareroot
(variable mirrors the API field verbatim).

Dead code: remove confirmed-unused imports across ~20 modules (require_role/
require_permission scaffold residue, stray db/Vendor/Model/current_user/Optional/
error_response); drop unused build_scope import + a stale GEENFORCE_API_KEY
docstring clause in geenforce. Migration files left untouched.

Correctness: geenforce ingest robustness - record_enforcement_report now 400s
on a non-dict counts / non-list results instead of 500; _apply_app_link ignores
a non-numeric appid per its docstring instead of 500. Regression tests added.

Backend query.get sweep finished: auth.py refresh -> db.session.get (last one).

910 backend tests pass; pyflakes clean; naming green; frontend build green.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-07-13 08:02:43 -04:00

116 lines
4.5 KiB
Python

"""Dualpath single-machine collapse resolution.
A Dualpath relationship pair is ONE physical dual-bay machine (single
controller, one bay-selector switch) recorded as two asset rows. Most
facilities want lists, counts, and the map to show that pair as a single
machine; the data model always keeps both rows. This module resolves the
pairs so the consumers (machines list, dashboard/report counts, floor map)
can collapse them, and is exposed via the plugin contract surface
(shopdb.api) so the machines plugin can reach it contract-purely.
PRIMARY = the pair member with the lower natural-sort assetnumber; the other
member is SECONDARY and is the one hidden when collapsing. Pairs are derived
from active assetrelationships rows whose type is named 'Dualpath' (either
direction; mirrored rows in both directions collapse to one pair). Only pairs
where BOTH assets are active are resolved.
"""
import re
from collections import namedtuple
from shopdb.core.models import Asset, AssetRelationship, RelationshipType
# secondaryassetids: set of the non-primary bay asset ids (hide when collapsing)
# partnerbyasset: {assetid -> {'assetid', 'assetnumber'}} for EVERY pair member,
# primary and secondary alike, so detail pages can show a sibling
# banner from whichever bay you land on.
DualpathCollapse = namedtuple('DualpathCollapse', ['secondaryassetids', 'partnerbyasset'])
DUALPATH_TYPE_NAME = 'Dualpath'
def _naturalkey(assetnumber):
# split into digit/non-digit chunks so 2007 sorts before 2008 and before 10a
parts = re.split(r'(\d+)', assetnumber or '')
return [int(p) if p.isdigit() else p.lower() for p in parts]
def resolve_dualpath_pairs():
"""Resolve active Dualpath pairs into a DualpathCollapse.
Direction-blind and dedup-safe: rows stored in either direction (or both)
for the same two assets collapse to one pair. Ignores the site toggle;
callers gate on dualpath_single_machine_enabled() where the collapse should
only apply when the setting is on (the detail-page banner shows always).
"""
reltype = RelationshipType.query.filter_by(
relationshiptype=DUALPATH_TYPE_NAME).first()
if not reltype:
return DualpathCollapse(set(), {})
rows = AssetRelationship.query.filter(
AssetRelationship.relationshiptypeid == reltype.relationshiptypeid,
AssetRelationship.isactive == True,
).all()
if not rows:
return DualpathCollapse(set(), {})
# batch-load the involved assets (assetnumber + active flag) in one query
involved = set()
for row in rows:
involved.add(row.sourceassetid)
involved.add(row.targetassetid)
assetbyid = {
a.assetid: a
for a in Asset.query.filter(Asset.assetid.in_(involved)).all()
}
secondaryassetids = set()
partnerbyasset = {}
seenpairs = set()
for row in rows:
aid, bid = row.sourceassetid, row.targetassetid
if aid == bid:
continue # defensive: no self-pairs
pairkey = frozenset((aid, bid))
if pairkey in seenpairs:
continue # mirrored row already handled
seenpairs.add(pairkey)
aone = assetbyid.get(aid)
atwo = assetbyid.get(bid)
# collapse only affects visible/counted (active) assets
if not aone or not atwo or not aone.isactive or not atwo.isactive:
continue
# PRIMARY = lower natural-sort assetnumber; SECONDARY is the other bay
if _naturalkey(aone.assetnumber) <= _naturalkey(atwo.assetnumber):
primary, secondary = aone, atwo
else:
primary, secondary = atwo, aone
secondaryassetids.add(secondary.assetid)
partnerbyasset[primary.assetid] = {
'assetid': secondary.assetid,
'assetnumber': secondary.assetnumber,
}
partnerbyasset[secondary.assetid] = {
'assetid': primary.assetid,
'assetnumber': primary.assetnumber,
}
return DualpathCollapse(secondaryassetids, partnerbyasset)
def dualpath_single_machine_enabled():
"""True when the site treats Dualpath pairs as one machine (default true).
Reads the cached settings; the row is absent on an un-seeded site, in which
case the default (true - 'most places' consider a dual-bay pair one machine)
applies.
"""
# imported here to avoid a settings<->api import cycle at module load
from shopdb.core.api.settings import get_cached_settings
settings = get_cached_settings()
return bool(settings.get('dualpath_single_machine', True))