# -*- coding: utf-8 -*- """Compare two issues of the same shop ticket and say what changed. WHY THIS MODULE EXISTS A revised ticket that does not say what moved forces the plant to diff it by eye, and they will not. Everything else in this repo answers "is this piece right?"; this answers "what changed since the last issue?", which is the question a consultancy is actually exposed to. WHAT "CHANGED" MEANS HERE, AND WHAT IT DELIBERATELY DOES NOT A pixel diff is useless -- every ticket is replotted, so every pixel moves. A line diff is nearly as bad: it reports the plot timestamp on all twelve tickets and buries the one anchor that moved. So the comparison is per REGION, each with its own notion of sameness: title block field by field, by printed label bill of matl row by row, keyed on the part mark handling dimension by dimension within DUNNAGE / PICKING / ERECTION dimensions chain by chain, in printed order, with the chain total notes paragraph by paragraph callouts as a multiset of strings, located but not positioned THE FAILURE MODE THIS IS BUILT AGAINST False positives kill this class of tool. A diff that reports forty spurious changes per ticket will not be used, and then the one real change goes unread. Three specific defences: * dimensions compare by VALUE, not by string -- the same length prints as 1' - 0" on the bed plan and 1'-0" in the handling diagrams, and a string compare would fire on every ticket in the set; * position-only movement is NOT a change -- a dimension whose text is identical but whose witness line shifted two points is a replot, not a revision; * anything a hazard scan touched is reported as UNCERTAIN, never as a change and never dropped. "Could not judge" is not "judged and found nothing". Severity is about who needs to act, not about how big the number is: MATERIAL changes what gets cast or embedded -- geometry, quantities, embed marks, handling stations ADVISORY changes what the sheet says without changing the piece -- notes, callout wording HOUSEKEEP issue stamps, plot times, sheet-status blocks UNCERTAIN a difference the extraction cannot stand behind Pure Python, no pymupdf and no Revit, so it is testable on plain CPython. """ import difflib import re MATERIAL = 'MATERIAL' ADVISORY = 'ADVISORY' HOUSEKEEP = 'HOUSEKEEP' UNCERTAIN = 'UNCERTAIN' _ORDER = {MATERIAL: 0, UNCERTAIN: 1, ADVISORY: 2, HOUSEKEEP: 3} def finding(kind, severity, where, what, old=None, new=None, note=None, loc_old=None, loc_new=None): """One reported difference. `loc_old` / `loc_new` are {'page','x','y'} on each issue when the finding has a place on the sheet. They are what lets the runner render the two crops a reviewer confirms the change on -- the text layer is not the sheet, and every number this tool prints should be checkable by eye without hunting for it. """ return {'kind': kind, 'severity': severity, 'where': where, 'what': what, 'old': old, 'new': new, 'note': note, 'loc_old': loc_old, 'loc_new': loc_new} def _loc(s): if not s: return None return {'page': s['page'], 'x': s['x'], 'y': s['y']} # -------------------------------------------------------------------------- # pairing files across two issues # -------------------------------------------------------------------------- _MARK_RE = re.compile(r'(?:^|[_\-\s])(WP\d+[A-Za-z]?)(?:[_\-\s]|\.|$)', re.I) _CTRL_RE = re.compile(r'^(\d{4,6})[-_](\d{4,6})[_\-]') def parse_name(filename): """Pull job / control / mark out of a ticket filename. The two issues of this set are named differently -- the old by mark (WP199.pdf), the new by job and control number (25881-20132_WP199.pdf). That is itself a reportable fact: a set whose file naming changed between issues cannot be paired by filename alone, and the pairing that IS possible is one-to-many, because this plant issues one ticket per control number and three identical WP215 panels means three files against one old file. """ base = filename if base.lower().endswith('.pdf'): base = base[:-4] base = base.strip() out = {'file': filename, 'job': None, 'control': None, 'mark': None, 'revision': None} m = _CTRL_RE.match(base) if m: out['job'], out['control'] = m.group(1), m.group(2) m = _MARK_RE.search(base) if m: out['mark'] = m.group(1).upper() m = re.search(r'[_\-](R\d+)$', base, re.I) if m: out['revision'] = m.group(1).upper() return out def pair_sets(old_files, new_files): """Match old-issue files to new-issue files. Returns {'pairs': [(old, [new, ...])], 'old_only': [...], 'new_only': [...], 'naming': {...}}. Pairing is on the MARK, because it is the only identifier both namings carry. Everything that pairing cannot do is returned rather than hidden: a mark present in one issue only is an added or withdrawn ticket, which is a bigger deal than any dimension inside one. """ o = [parse_name(f) for f in old_files] n = [parse_name(f) for f in new_files] old_by_mark = {} for r in o: old_by_mark.setdefault(r['mark'], []).append(r) new_by_mark = {} for r in n: new_by_mark.setdefault(r['mark'], []).append(r) pairs = [] for mark in sorted(k for k in old_by_mark if k in new_by_mark): for orec in old_by_mark[mark]: pairs.append((orec, sorted(new_by_mark[mark], key=lambda r: (r['control'] or '', r['file'])))) naming = { 'old_has_control': sum(1 for r in o if r['control']), 'new_has_control': sum(1 for r in n if r['control']), 'old_count': len(o), 'new_count': len(n), 'changed': (bool([r for r in o if r['control']]) != bool([r for r in n if r['control']])), 'fan_out': {m: len(new_by_mark[m]) for m in sorted(new_by_mark) if m in old_by_mark and len(new_by_mark[m]) > 1}, } return { 'pairs': pairs, 'old_only': sorted(r['file'] for r in o if r['mark'] not in new_by_mark), 'new_only': sorted(r['file'] for r in n if r['mark'] not in old_by_mark), 'naming': naming, } # -------------------------------------------------------------------------- # helpers # -------------------------------------------------------------------------- # -------------------------------------------------------------------------- # region comparisons # -------------------------------------------------------------------------- # Fields that only ever record when the sheet was plotted or issued. Their # changing is expected on every reissue and says nothing about the piece. _HOUSEKEEPING_FIELDS = ('DRAWN BY:', 'ENG BY:') # Fields whose value changing means the piece or its paperwork identity # changed. SEQUENCE #, PROJECT NAME and the like are identity, not geometry, # but a changed one still needs a human. _MATERIAL_FIELDS = ( 'LENGTH:', 'WIDTH:', 'THICKNESS:', 'WEIGHT (lbs.):', 'QUANITY:', 'QUANTITY:', "CONCRETE (f'c):", "CONCRETE (f'ci):", 'FACE MIX #', 'BASE MIX#', 'FACE MIX (C.Y.)', 'BASE MIX (C.Y.)', 'PRODUCT CODE:', "ADD'L PRODUCTION:", "ADD'L FINISH:", ) def compare_title_block(a, b): out = [] keys = sorted(set(a.get('title_block', {})) | set(b.get('title_block', {}))) for k in keys: fa = a.get('title_block', {}).get(k) fb = b.get('title_block', {}).get(k) va = fa['value'] if fa else None vb = fb['value'] if fb else None if va == vb: continue if fa and not fa['ok'] or fb and not fb['ok']: out.append(finding( 'TITLE', UNCERTAIN, 'title block, %s' % k.rstrip(':'), 'the value does not look like a %s, so this pair is not ' 'compared' % k.rstrip(':').lower(), va, vb, 'extraction unreliable for this cell')) continue sev = HOUSEKEEP if k in _HOUSEKEEPING_FIELDS else ( MATERIAL if k in _MATERIAL_FIELDS else ADVISORY) out.append(finding('TITLE', sev, 'title block, %s' % k.rstrip(':'), 'title-block value changed', va, vb)) return out def _bom_is_tainted(doc): """True when a text-layer hazard lands inside this ticket's BOM table. Measured on a real reissue in this job (a ticket and its R1): the earlier file draws the strand row's mark and description ON TOP OF the bar row above it, so get_text returns SR5602's mark for the strand row's quantity. Diffed against a clean R1 that produced two phantom "item added / item removed" pairs and a quantity going from 376 to 4. Every one of those numbers looked completely plausible. So when the hazard scan finds overlaid text inside the table, the table's findings are reported as UNCERTAIN rather than as changes -- and they are still reported. "Could not judge" is not "judged and found nothing". """ box = doc.get('bom_bbox') if not box: return None for h in doc.get('hazards', []): if h['mode'] not in ('overlaid-text', 'thin-text-layer'): continue if h['page'] != box['page']: continue if h.get('x') is None: return h['mode'] if (box['x0'] <= h['x'] <= box['x1'] and box['y0'] <= h['y'] <= box['y1']): return h['mode'] return None def compare_bom(a, b): """Row-by-row on the part mark. A quantity change is a material change.""" out = [] taint = _bom_is_tainted(a) or _bom_is_tainted(b) ra = {} for r in a.get('bom', []): ra.setdefault(r['mark'], []).append(r) rb = {} for r in b.get('bom', []): rb.setdefault(r['mark'], []).append(r) for mark in sorted(set(ra) | set(rb)): la, lb = ra.get(mark, []), rb.get(mark, []) if not la: for r in lb: out.append(finding('BOM', MATERIAL, 'bill of material, %s' % mark, 'item added to the bill of material', None, '%s x %s -- %s' % (r['qty'], mark, r['desc']))) continue if not lb: for r in la: out.append(finding('BOM', MATERIAL, 'bill of material, %s' % mark, 'item removed from the bill of material', '%s x %s -- %s' % (r['qty'], mark, r['desc']), None)) continue for i in range(max(len(la), len(lb))): x = la[i] if i < len(la) else None y = lb[i] if i < len(lb) else None if x is None or y is None: out.append(finding('BOM', MATERIAL, 'bill of material, %s' % mark, 'the number of rows for this mark changed', '%d row(s)' % len(la), '%d row(s)' % len(lb))) break if x['qty'] != y['qty']: out.append(finding('BOM', MATERIAL, 'bill of material, %s' % mark, 'quantity changed', x['qty'], y['qty'], x['desc'] or None, loc_old=_loc(x), loc_new=_loc(y))) if _norm(x['desc']) != _norm(y['desc']): out.append(finding('BOM', ADVISORY, 'bill of material, %s' % mark, 'description changed', x['desc'], y['desc'], loc_old=_loc(x), loc_new=_loc(y))) return _downgrade(out, taint, 'bill of material') def _downgrade(findings, taint, region): if not taint: return findings for f in findings: f['severity'] = UNCERTAIN f['note'] = ('a text-layer hazard (%s) lands inside the %s, so this ' 'difference may be an artefact of reading the PDF rather ' 'than a change to the ticket -- check it against the ' 'sheet before acting on it' % (taint, region)) return findings def _norm(s): return ' '.join((s or '').split()) def compare_handling(a, b, parse_dim, format_inches): """The three little diagrams, dimension by dimension. Station order is printed order, left to right. It is not mirrored and not sorted by value: reading the two insets the other way round was measured to turn one engineer's CORRECTED dunnage ticket back into a FAIL, which is the precise way a diff tool loses its reader. """ out = [] ha, hb = a.get('handling', {}), b.get('handling', {}) for name in sorted(set(ha) | set(hb)): da = ha.get(name, []) db = hb.get(name, []) if name not in ha or name not in hb: out.append(finding('HANDLING', MATERIAL, '%s diagram' % name.lower(), 'the diagram itself is present on one issue only', 'present' if name in ha else 'absent', 'present' if name in hb else 'absent')) continue if len(da) != len(db): out.append(finding( 'HANDLING', MATERIAL, '%s diagram' % name.lower(), 'the number of printed dimensions changed', ' / '.join(s['text'] for s in da) or '(none)', ' / '.join(s['text'] for s in db) or '(none)', loc_old=_loc(da[0] if da else None), loc_new=_loc(db[0] if db else None))) continue for i, (x, y) in enumerate(zip(da, db)): vx, vy = parse_dim(x['text']), parse_dim(y['text']) if vx is None or vy is None or abs(vx - vy) > 1e-6: out.append(finding( 'HANDLING', MATERIAL, '%s diagram, dimension %d of %d (left to right)' % (name.lower(), i + 1, len(da)), 'handling station moved', x['text'], y['text'], loc_old=_loc(x), loc_new=_loc(y))) return out def _chain_values(chain, parse_dim): return [parse_dim(s['text']) for s in chain['items']] def _chain_key(chain, parse_dim): return tuple(round(v, 4) if v is not None else None for v in _chain_values(chain, parse_dim)) def _similarity(ka, kb): """How much two chains share, as a ratio.""" return difflib.SequenceMatcher(a=ka, b=kb).ratio() def match_chains(ca, cb, parse_dim): """Pair old chains to new chains across a REPLOT, not just a revision. The hard fact this is built around, measured on WP194: adding one note to the sheet pushed the whole bed plan 89 points up the page, so every dimension chain on it changed absolute position while none of them changed value. An absolute-position matcher reported 40 material changes on a ticket whose only real change was that note. That is exactly the output that gets a diff tool switched off, and then the one real change on some other ticket goes unread. So matching is content-first: 1. chains whose value sequences are IDENTICAL are paired, nearest first; 2. the median position offset of those pairs is the sheet's replot shift -- measured on this pair of files, not assumed; 3. remaining chains are paired on content similarity, with the measured shift applied to the position tolerance. Returns (pairs, unmatched_old_indices, unmatched_new_indices). """ keys_a = [_chain_key(g, parse_dim) for g in ca] keys_b = [_chain_key(h, parse_dim) for h in cb] used_a, used_b = set(), set() pairs = [] def bucket(g): return (g['page'], g['vertical']) cands = [] for i, g in enumerate(ca): for j, h in enumerate(cb): if bucket(g) != bucket(h) or keys_a[i] != keys_b[j]: continue cands.append((abs(g['pos'] - h['pos']), i, j)) for _, i, j in sorted(cands): if i in used_a or j in used_b: continue used_a.add(i) used_b.add(j) pairs.append((i, j)) shifts, alongs = {}, {} for i, j in pairs: shifts.setdefault(bucket(ca[i]), []).append(cb[j]['pos'] - ca[i]['pos']) alongs.setdefault(bucket(ca[i]), []).append( cb[j].get('along', 0.0) - ca[i].get('along', 0.0)) shift, along_shift = {}, {} for k, v in shifts.items(): v.sort() shift[k] = v[len(v) // 2] for k, v in alongs.items(): v.sort() along_shift[k] = v[len(v) // 2] cands = [] for i, g in enumerate(ca): if i in used_a: continue for j, h in enumerate(cb): if j in used_b or bucket(g) != bucket(h): continue if 'along' not in g or 'along' not in h: g.setdefault('along', 0.0) h.setdefault('along', 0.0) sim = _similarity(keys_a[i], keys_b[j]) if sim < 0.34: continue off = abs((h['pos'] - g['pos']) - shift.get(bucket(g), 0.0)) if off > 25.0: continue off2 = abs((h['along'] - g['along']) - along_shift.get(bucket(g), 0.0)) if off2 > 40.0: continue cands.append((-sim, off, i, j)) for _, _, i, j in sorted(cands): if i in used_a or j in used_b: continue used_a.add(i) used_b.add(j) pairs.append((i, j)) # 4 -- a chain where MOST of the segments changed shares too little # content to pass step 3, and a short one can share none at all: a # three-segment run with two segments edited scores 0.33. So a last pass # pairs what is left on position alone, but only at the measured replot # shift, within a few points, and with the same number of segments. That # is a much stronger claim than similarity -- it says the two chains are # drawn in the same place on the same run -- and the page-level value gate # downstream still has to agree before anything is reported. cands = [] for i, g in enumerate(ca): if i in used_a: continue for j, h in enumerate(cb): if j in used_b or bucket(g) != bucket(h): continue if len(g['items']) != len(h['items']): continue off = abs((h['pos'] - g['pos']) - shift.get(bucket(g), 0.0)) off2 = abs((h.get('along', 0.0) - g.get('along', 0.0)) - along_shift.get(bucket(g), 0.0)) if off > 6.0 or off2 > 10.0: continue cands.append((off + off2, i, j)) for _, i, j in sorted(cands): if i in used_a or j in used_b: continue used_a.add(i) used_b.add(j) pairs.append((i, j)) return (pairs, [i for i in range(len(ca)) if i not in used_a], [j for j in range(len(cb)) if j not in used_b]) def _page_net(ca, cb, parse_dim): """Per page, the dimension values one issue prints and the other does not. Returns (net_old, net_new) as {page: {value: surplus count}}. A value printed the same number of times on both sides appears in neither, which is exactly the point: it did not change. """ def count(chains): d = {} for g in chains: for sp in g['items']: v = parse_dim(sp['text']) v = round(v, 4) if v is not None else None d.setdefault(g['page'], {}) d[g['page']][v] = d[g['page']].get(v, 0) + 1 return d da, db = count(ca), count(cb) net_a, net_b = {}, {} for page in set(da) | set(db): ra, rb = da.get(page, {}), db.get(page, {}) net_a[page] = {} net_b[page] = {} for v in set(ra) | set(rb): diff = ra.get(v, 0) - rb.get(v, 0) if diff > 0: net_a[page][v] = diff elif diff < 0: net_b[page][v] = -diff return net_a, net_b def compare_chains(a, b, parse_dim, format_inches, locate): """Dimension chains on the drawing views. A dimension chain is what a detailer actually edits: a run of segments along one line adding up to an overall. Comparing the run IN ORDER, and reporting the chain total alongside, is the difference between "two numbers changed" and "the anchors were redistributed, the panel did not get longer" -- the second is what a precast engineer would say, and it is what tells him whether he has to look further. Two tiers, and the second exists purely to keep false positives down: tier 1 matched chains, compared segment by segment with difflib, so a chain that gained or lost a segment still reports the segments that actually moved rather than "the chain changed"; tier 2 chains that could not be matched at all are NOT reported as chains. Their dimensions are pooled per page and compared as a multiset of VALUES. Measured reason: the same run of dimensions is sometimes grouped as one chain on one issue and two on the other, purely because the text sat a point further apart. That is a grouping artefact of this reader, not a revision, and reporting it as a revision would be the tool lying about its own limits. """ out = [] ca = a.get('chains', []) cb = b.get('chains', []) titles_a = a.get('view_titles', []) titles_b = b.get('view_titles', []) pairs, only_a, only_b = match_chains(ca, cb, parse_dim) leftover_a, leftover_b = [], [] # THE GATE. Before any chain is compared, count every dimension VALUE # printed on each page of each issue and take the net difference. # # If a page prints the same multiset of dimension values on both issues, # then no dimension on that page changed value -- whatever happened to the # chains is a rearrangement, and this reader's chain grouping is not stable # enough across a replot to be trusted to describe it. Measured: WP196 and # WP412 carry byte-identical dimension text on page 2, and the section # details were nonetheless re-laid-out far enough that chain matching # paired a dimension in SECTION A against one in SECTION B. Without this # gate that was five confident, wrong, material findings on two tickets # that had not changed at all. # # So nothing is reported unless the page's set of printed values actually # moved. The cost is stated rather than hidden: a dimension that swaps # places with another dimension of the SAME value is invisible here. net_a, net_b = _page_net(ca, cb, parse_dim) for i, j in sorted(pairs): g, h = ca[i], cb[j] ta = [s['text'] for s in g['items']] tb_ = [s['text'] for s in h['items']] va = _chain_key(g, parse_dim) vb = _chain_key(h, parse_dim) if va == vb: continue where = locate(titles_b, h['page'], h['items'][0]['x'], h['items'][0]['y']) sum_a = sum(v for v in va if v is not None) sum_b = sum(v for v in vb if v is not None) if len(va) == len(vb) and abs(sum_a - sum_b) < 1e-6: note = ('chain total unchanged at %s -- a redistribution along the ' 'same run, not a size change' % format_inches(sum_a)) else: note = ('chain total %s -> %s' % (format_inches(sum_a), format_inches(sum_b))) sm = difflib.SequenceMatcher(a=va, b=vb) for tag, i1, i2, j1, j2 in sm.get_opcodes(): if tag == 'equal': continue if tag == 'replace' and (i2 - i1) == (j2 - j1): for k in range(i2 - i1): pg = g['items'][i1 + k]['page'] ov, nv = va[i1 + k], vb[j1 + k] if (net_a.get(pg, {}).get(ov, 0) <= 0 and net_b.get(pg, {}).get(nv, 0) <= 0): # Both values are still printed elsewhere on the page # in the same numbers: the chain grouping moved, the # dimension did not. leftover_a.append(g['items'][i1 + k]) leftover_b.append(h['items'][j1 + k]) continue if net_a.get(pg, {}).get(ov, 0) > 0: net_a[pg][ov] -= 1 if net_b.get(pg, {}).get(nv, 0) > 0: net_b[pg][nv] -= 1 out.append(finding( 'DIM', MATERIAL, '%s, segment %d of %d (%s chain)' % (where, i1 + k + 1, len(ta), 'vertical' if g['vertical'] else 'horizontal'), 'dimension changed', ta[i1 + k], tb_[j1 + k], note, loc_old=_loc(g['items'][i1 + k]), loc_new=_loc(h['items'][j1 + k]))) else: # A chain that gained or lost segments is NOT reported as a # chain. Its odd segments go into the leftover pool below and # are only reported if they do not cancel against the other # issue. See the pool's comment for the measurement. leftover_a.extend(g['items'][i1:i2]) leftover_b.extend(h['items'][j1:j2]) for i in only_a: leftover_a.extend(ca[i]['items']) for j in only_b: leftover_b.extend(cb[j]['items']) # THE POOL, and why it is the whole defence against grouping noise. # # Chains are grouped by the printed line the dimensions sit on. That # grouping is not stable across a replot: on WP196 and WP412 the section # details on page 2 carry byte-identical dimension text on both issues, # and the reader still split them into different chains because one string # sat a point further from its neighbour. Reported as chains, that was 11 # material findings across two tickets that had not changed at all. # # So every dimension a chain comparison could not account for is pooled by # page and compared as a MULTISET OF VALUES. Regrouping cancels exactly. # A dimension that genuinely appears or disappears does not, and is still # reported -- without chain context, which is the honest price. def pool(items): d = {} for sp in items: v = parse_dim(sp['text']) d.setdefault((sp['page'], round(v, 4) if v is not None else None), []).append(sp) return d # Whatever the chain comparison could not account for is reported against # what is LEFT of the page's net difference, so a regrouping cancels # exactly and a genuine appearance or disappearance still surfaces -- # without chain context, which is the honest price of not having been able # to match its chain. pa, pb = pool(leftover_a), pool(leftover_b) for k in sorted(set(pa) | set(pb), key=lambda k: (k[0], k[1] or 0)): page, val = k tail = ('this reader could not match the dimension chain it belongs to ' 'across the two issues, so it is reported on its own, without ' 'its neighbours') n_new = min(len(pb.get(k, [])), net_b.get(page, {}).get(val, 0)) n_old = min(len(pa.get(k, [])), net_a.get(page, {}).get(val, 0)) if n_new > 0: sp = pb[k][0] net_b[page][val] -= n_new out.append(finding( 'DIM', MATERIAL, locate(titles_b, sp['page'], sp['x'], sp['y']), 'a dimension appears in the new issue that is not in the old' + ('' if n_new == 1 else ' (%d of them)' % n_new), None, sp['text'], tail, loc_new=_loc(sp))) if n_old > 0: sp = pa[k][0] net_a[page][val] -= n_old out.append(finding( 'DIM', MATERIAL, locate(titles_a, sp['page'], sp['x'], sp['y']), 'a dimension in the old issue is not in the new one' + ('' if n_old == 1 else ' (%d of them)' % n_old), sp['text'], None, tail, loc_old=_loc(sp))) return out def compare_notes(a, b, locate): """Paragraph-level. A note is added, removed, or reworded -- never 'line 3'.""" out = [] na = {_norm(n['text']): n for n in a.get('notes', [])} nb = {_norm(n['text']): n for n in b.get('notes', [])} titles_b = b.get('view_titles', []) titles_a = a.get('view_titles', []) only_a = [k for k in na if k not in nb] only_b = [k for k in nb if k not in na] # A reworded note is a removal and an addition that share a heading; pair # them so the reader sees one finding with two versions, not two findings. paired = [] for ka in list(only_a): head_a = ka.split(':')[0] for kb in list(only_b): if kb.split(':')[0] == head_a and _similar(ka, kb): paired.append((ka, kb)) only_a.remove(ka) only_b.remove(kb) break for ka, kb in paired: n = nb[kb] out.append(finding('NOTE', ADVISORY, locate(titles_b, n['page'], n['x'], n['y']), 'note reworded', na[ka]['text'], nb[kb]['text'], loc_old=_loc(na[ka]), loc_new=_loc(n))) for k in sorted(only_a): n = na[k] out.append(finding('NOTE', ADVISORY, locate(titles_a, n['page'], n['x'], n['y']), 'note removed', n['text'], None, loc_old=_loc(n))) for k in sorted(only_b): n = nb[k] out.append(finding('NOTE', ADVISORY, locate(titles_b, n['page'], n['x'], n['y']), 'note added', None, n['text'], loc_new=_loc(n))) return out def _similar(x, y): xs, ys = set(x.upper().split()), set(y.upper().split()) if not xs or not ys: return False return len(xs & ys) / float(len(xs | ys)) >= 0.4 _TS_RE = re.compile(r'^\d{1,2}/\d{1,2}/\d{2,4}(\s+\d{1,2}:\d{2}(:\d{2})?\s*(AM|PM)?)?$', re.I) # Plot stamps and page numbers change on every reissue and mean nothing about # the piece. They are still reported -- under --all -- because "the sheet was # replotted and nothing else changed" is a useful thing for a reviewer to be # able to confirm. _FURNITURE_RE = re.compile(r'^PAGE\s+\d+\s+OF\s+\d+$', re.I) def _is_housekeeping_text(t): return bool(_TS_RE.match(t) or _FURNITURE_RE.match(t)) _PART_RE = re.compile(r'\b((?:P|SR|SP|S)\d+)\b') def compare_callouts(a, b, locate): """Loose annotation text, as a multiset. Position is a locator here, not a comparison key. Callouts drift by a point or two on every replot; comparing positions would fire on all of them, while the multiset fires only when a string genuinely appears or disappears. Added and removed strings that name the SAME PART are then paired into one finding and raised to MATERIAL. That pairing is the difference between removed P106 (S.I.F.)(E.S.) added P106 (S.I.F.) -- two advisory lines a reader skims past -- and "the P106 callout lost its (E.S.)", which is an instruction about where an embed goes and is the kind of thing this whole tool exists to surface. """ out = [] ca, cb = a.get('callouts', []), b.get('callouts', []) titles_a, titles_b = a.get('view_titles', []), b.get('view_titles', []) def bag(items): d = {} for s in items: d.setdefault(s['text'], []).append(s) return d ba, bb = bag(ca), bag(cb) removed, added = [], [] for t in sorted(set(ba) | set(bb)): na, nb = len(ba.get(t, [])), len(bb.get(t, [])) if na == nb: continue if nb > na: added.append((t, nb - na, bb[t][0])) else: removed.append((t, na - nb, ba[t][0])) def parts(t): return tuple(_PART_RE.findall(t)) used_add = set() for i, (ta, na, sa) in enumerate(removed): pa = parts(ta) if not pa: continue for j, (tb_, nb, sb) in enumerate(added): if j in used_add or parts(tb_) != pa: continue used_add.add(j) removed[i] = None out.append(finding( 'CALLOUT', MATERIAL, locate(titles_b, sb['page'], sb['x'], sb['y']), 'the callout for %s was reworded' % ', '.join(pa), ta, tb_, loc_old=_loc(sa), loc_new=_loc(sb))) break for entry in removed: if entry is None: continue t, n, s = entry sev = HOUSEKEEP if _is_housekeeping_text(t) else ADVISORY out.append(finding( 'CALLOUT', sev, locate(titles_a, s['page'], s['x'], s['y']), 'text removed from the sheet%s' % ('' if n == 1 else ' (%d times)' % n), t, None, loc_old=_loc(s))) for j, (t, n, s) in enumerate(added): if j in used_add: continue sev = HOUSEKEEP if _is_housekeeping_text(t) else ADVISORY out.append(finding( 'CALLOUT', sev, locate(titles_b, s['page'], s['x'], s['y']), 'text added to the sheet%s' % ('' if n == 1 else ' (%d times)' % n), None, t, loc_new=_loc(s))) return out def compare_hazards(a, b): """Every hazard on either side, reported as uncertainty, never as silence.""" out = [] for doc, side in ((a, 'old'), (b, 'new')): for h in doc.get('hazards', []): loc = 'page %d' % (h['page'] + 1) if h.get('x') is not None: loc += ' near x=%.0f y=%.0f' % (h['x'], h['y']) out.append(finding( 'HAZARD', UNCERTAIN, '%s issue, %s' % (side, loc), 'text-layer hazard (%s): %s' % (h['mode'], h['detail']), None, ' | '.join(h.get('texts') or []) or None)) return out # -------------------------------------------------------------------------- # top level # -------------------------------------------------------------------------- def compare(a, b, parse_dim, format_inches, locate): """Two extracted tickets -> a list of findings, most material first. parse_dim / format_inches / locate are injected rather than imported so this module stays free of pymupdf and testable on plain dicts. """ out = [] out += compare_title_block(a, b) out += compare_bom(a, b) out += compare_handling(a, b, parse_dim, format_inches) out += compare_chains(a, b, parse_dim, format_inches, locate) out += compare_notes(a, b, locate) out += compare_callouts(a, b, locate) out += compare_hazards(a, b) if a.get('pages') != b.get('pages'): out.append(finding('SHEET', MATERIAL, 'the ticket as a whole', 'page count changed', a.get('pages'), b.get('pages'))) out.sort(key=lambda f: (_ORDER[f['severity']], f['kind'], f['where'] or '')) return out def summarise(findings): """Counts per severity, for a one-line verdict.""" d = {MATERIAL: 0, UNCERTAIN: 0, ADVISORY: 0, HOUSEKEEP: 0} for f in findings: d[f['severity']] += 1 return d