91 lines
3.4 KiB
Python
91 lines
3.4 KiB
Python
"""Versioned, deterministic model view; the archived context stays complete."""
|
|
import copy
|
|
import json
|
|
import re
|
|
|
|
|
|
IDENTIFIERS = {'source_ref', 'capture_ref', 'archive_ref', 'member_name', 'path',
|
|
'sha256', 'git_blob', 'assessment_ref', 'case_ref', 'case_sha256',
|
|
'snapshot_id'}
|
|
|
|
|
|
def compact_context(context):
|
|
"""Shorten storage identifiers, never source IDs or actual database row values.
|
|
|
|
Equal identifiers receive equal aliases; different identifiers remain distinct.
|
|
This is a presentation encoding, not semantic summarization or source selection.
|
|
"""
|
|
aliases = {}
|
|
|
|
def identifier(value):
|
|
if (not isinstance(value, str) or len(value) < 64
|
|
or not re.search(r"[0-9a-fA-F]{40,}", value)):
|
|
return value
|
|
if value not in aliases:
|
|
aliases[value] = f"P{len(aliases)+1:04d}"
|
|
label = value.rsplit('/', 1)[-1].rsplit('\\', 1)[-1]
|
|
# Hash-only labels add no semantic information to the model.
|
|
if len(label) >= 64 or label == value:
|
|
return aliases[value]
|
|
return aliases[value] + ':' + label
|
|
|
|
def visit(value):
|
|
if isinstance(value, list):
|
|
return [visit(item) for item in value]
|
|
if not isinstance(value, dict):
|
|
return value
|
|
result = {}
|
|
for key, item in sorted(value.items()):
|
|
if key == 'row':
|
|
# Database column names may themselves be path, sha256, etc.
|
|
result[key] = copy.deepcopy(item)
|
|
elif key in IDENTIFIERS:
|
|
result[key] = identifier(item)
|
|
else:
|
|
result[key] = visit(item)
|
|
return result
|
|
|
|
projected = visit(context)
|
|
# Repeated provenance objects are encoded once, without dropping scope,
|
|
# timestamps, authors or identifier relationships.
|
|
counts = {}
|
|
def count(value):
|
|
if isinstance(value,list):
|
|
for item in value: count(item)
|
|
elif isinstance(value,dict):
|
|
for key,item in value.items():
|
|
if key == 'row': continue
|
|
if key in {'origin','provenance'} and isinstance(item,dict):
|
|
encoded = model_json(item)
|
|
if len(encoded)>120: counts[encoded] = counts.get(encoded,0)+1
|
|
count(item)
|
|
count(projected)
|
|
shared = {}
|
|
ids = {}
|
|
def deduplicate(value):
|
|
if isinstance(value,list): return [deduplicate(item) for item in value]
|
|
if not isinstance(value,dict): return value
|
|
result = {}
|
|
for key,item in value.items():
|
|
if key == 'row':
|
|
result[key] = item
|
|
continue
|
|
encoded = model_json(item) if key in {'origin','provenance'} and isinstance(item,dict) else None
|
|
if encoded is not None and counts.get(encoded,0)>1:
|
|
if encoded not in ids:
|
|
sid = f"M{len(ids)+1:04d}"
|
|
ids[encoded] = sid
|
|
shared[sid] = copy.deepcopy(item)
|
|
result[key] = {'metadata_ref':ids[encoded]}
|
|
else:
|
|
result[key] = deduplicate(item)
|
|
return result
|
|
projected = deduplicate(projected)
|
|
if shared: projected['shared_metadata'] = shared
|
|
return projected
|
|
|
|
|
|
def model_json(value):
|
|
return json.dumps(value, sort_keys=True, ensure_ascii=False, allow_nan=False,
|
|
separators=(',', ':'))
|