272 lines
14 KiB
Python
272 lines
14 KiB
Python
"""Read-only offline audit; reports aggregate sizes/hashes, never source contents.
|
|
|
|
All byte counts use UTF-8 or ZIP directory sizes. Potential savings overlap and
|
|
must not be added together. This is a measurement, not a package migration.
|
|
"""
|
|
import argparse
|
|
from collections import Counter, defaultdict
|
|
import hashlib
|
|
import io
|
|
import json
|
|
from pathlib import Path
|
|
import tarfile
|
|
import zipfile
|
|
import zlib
|
|
|
|
from viano_bug_analysis.package import PackageReader
|
|
from viano_bug_analysis.llm.gemini import GeminiProvider
|
|
|
|
|
|
def encoded(value):
|
|
return json.dumps(value, ensure_ascii=False, sort_keys=True,
|
|
separators=(',', ':')).encode('utf-8')
|
|
|
|
|
|
def digest(data):
|
|
return hashlib.sha256(data).hexdigest()
|
|
|
|
|
|
def category(name):
|
|
if name.startswith('original/'):
|
|
return 'original_archives' if name.endswith('/archive.tar.gz') else 'original_members'
|
|
if name.startswith('llm/'):
|
|
return 'llm_requests' if name.endswith('/request.json') else 'llm_other'
|
|
if '/contexts/' in name:
|
|
return 'analysis_contexts'
|
|
return name.split('/')[0]
|
|
|
|
|
|
def repeated(groups):
|
|
result = []
|
|
for entries in groups.values():
|
|
if len(entries) > 1:
|
|
result.append({'copies': len(entries),
|
|
'raw_duplicate_bytes': sum(e[1] for e in entries[1:]),
|
|
'compressed_duplicate_bytes': sum(e[2] for e in entries[1:]),
|
|
'paths': [e[0] for e in entries]})
|
|
return sorted(result, key=lambda v: v['raw_duplicate_bytes'], reverse=True)
|
|
|
|
|
|
def audit(path):
|
|
# Validate through the normal bounded reader before inspecting ZIP metadata.
|
|
with PackageReader(path) as reader:
|
|
manifest = reader.manifest
|
|
files = set(reader.files)
|
|
groups = defaultdict(list)
|
|
canonical = defaultdict(list)
|
|
categories = defaultdict(Counter)
|
|
requests = []
|
|
archive_members = Counter()
|
|
metadata_fields = Counter()
|
|
deflate_estimate = Counter()
|
|
compression_methods = Counter()
|
|
evidence_items = {}
|
|
source_lines = {}
|
|
evidence_sizes = Counter()
|
|
context_components = defaultdict(lambda: [0, 0, 0])
|
|
context_values = set()
|
|
with zipfile.ZipFile(path) as z:
|
|
infos = {v.filename: v for v in z.infolist() if not v.is_dir()}
|
|
for name, info in infos.items():
|
|
data = z.read(name)
|
|
compressor = zlib.compressobj(level=6, wbits=-15)
|
|
compressed = compressor.compress(data) + compressor.flush()
|
|
assert zlib.decompress(compressed, wbits=-15) == data
|
|
deflate_size = len(compressed)
|
|
deflate_estimate[category(name)] += deflate_size
|
|
compression_methods[str(info.compress_type)] += 1
|
|
cat = categories[category(name)]
|
|
cat.update(files=1, raw_bytes=len(data), compressed_bytes=info.compress_size)
|
|
groups[digest(data)].append((name, len(data), info.compress_size))
|
|
if name.endswith('/archive.tar.gz'):
|
|
prefix = name.removesuffix('archive.tar.gz') + 'members/'
|
|
with tarfile.open(fileobj=io.BytesIO(data), mode='r|gz') as tar:
|
|
for member in tar:
|
|
if not member.isfile():
|
|
continue
|
|
target = prefix + member.name
|
|
if target in files and member.size == infos[target].file_size:
|
|
with tar.extractfile(member) as stream:
|
|
same = digest(stream.read(member.size + 1)) == digest(z.read(target))
|
|
if same:
|
|
archive_members.update(files=1, raw_bytes=member.size,
|
|
outer_zip_member_compressed_bytes=infos[target].compress_size)
|
|
if not name.endswith('.json'):
|
|
continue
|
|
try:
|
|
doc = json.loads(data)
|
|
except (ValueError, UnicodeError):
|
|
continue
|
|
canonical[digest(encoded(doc))].append((name, len(data), info.compress_size))
|
|
if category(name) == 'analysis_contexts' and isinstance(doc, dict):
|
|
for field, value in doc.items():
|
|
b = encoded(value)
|
|
key = (field, digest(b))
|
|
context_components[field][0] += len(b)
|
|
context_components[field][2] += 1
|
|
if key not in context_values:
|
|
context_components[field][1] += len(b)
|
|
context_values.add(key)
|
|
if name.startswith('llm/operations/') and name.endswith('/request.json'):
|
|
requests.append((name, doc))
|
|
by_prompt = defaultdict(list)
|
|
system_groups = Counter()
|
|
block_groups = defaultdict(lambda: [0, 0, 0])
|
|
per_call = []
|
|
def leaves(value, at='$'):
|
|
if isinstance(value, dict):
|
|
for k, v in value.items():
|
|
if isinstance(v, (dict, list)):
|
|
yield from leaves(v, at + '.' + k)
|
|
elif isinstance(v, str) and len(v.encode('utf-8')) >= 80:
|
|
yield at + '.' + k, v
|
|
elif isinstance(value, list):
|
|
for v in value:
|
|
yield from leaves(v, at + '[]')
|
|
for name, req in requests:
|
|
messages = req.get('messages', [])
|
|
msg_bytes = sum(len(m['content'].encode('utf-8')) for m in messages)
|
|
systems = [m['content'] for m in messages if m['role'] == 'system']
|
|
for value in systems:
|
|
system_groups[(digest(value.encode()), len(value.encode()))] += 1
|
|
user_docs = []
|
|
strings = defaultdict(list)
|
|
keys = Counter()
|
|
def count_keys(value):
|
|
if isinstance(value, dict):
|
|
for k, v in value.items():
|
|
keys[k] += len(encoded(k)) + 1
|
|
count_keys(v)
|
|
elif isinstance(value, list):
|
|
for v in value:
|
|
count_keys(v)
|
|
for m in messages:
|
|
if m['role'] != 'user':
|
|
continue
|
|
try:
|
|
doc = json.loads(m['content'])
|
|
except ValueError:
|
|
continue
|
|
user_docs.append(doc)
|
|
view = doc.get('input', {}) if isinstance(doc, dict) else {}
|
|
if isinstance(view, dict):
|
|
for item in view.get('evidence', {}).values() if isinstance(view.get('evidence'), dict) else []:
|
|
evidence_items[digest(encoded(item))] = (evidence_items.get(digest(encoded(item)), (0, len(encoded(item))))[0] + 1, len(encoded(item)))
|
|
if isinstance(item, dict):
|
|
for field, value in item.items():
|
|
evidence_sizes[field] += len(encoded(value))
|
|
shared = view.get('shared_sources', {})
|
|
for source in shared.values() if isinstance(shared, dict) else []:
|
|
if not isinstance(source, dict):
|
|
continue
|
|
meta = source.get('metadata', {})
|
|
identity = digest(encoded(meta))
|
|
for block in source.get('blocks', []):
|
|
for offset, line in enumerate(block.get('lines', [])):
|
|
key = (identity, block['start_line'] + offset, digest(encoded(line)))
|
|
count, size = source_lines.get(key, (0, len(encoded(line))))
|
|
source_lines[key] = (count + 1, size)
|
|
count_keys(doc)
|
|
for at, value in leaves(doc):
|
|
strings[digest(value.encode())].append((at, len(encoded(value))))
|
|
if isinstance(doc, dict):
|
|
for k, v in doc.items():
|
|
items = v.items() if k == 'input' and isinstance(v, dict) else [(k, v)]
|
|
for field, item in items:
|
|
label = 'input.' + field if k == 'input' else field
|
|
b = encoded(item)
|
|
key = (req.get('prompt_version'), label, digest(b))
|
|
block_groups[key][0] += 1
|
|
block_groups[key][1] += len(b)
|
|
block_groups[key][2] = len(b)
|
|
dup_strings = sum(sum(n for _, n in entries[1:])
|
|
for entries in strings.values())
|
|
metadata_fields.update(keys)
|
|
partition = None
|
|
if req.get('prompt_version', '').startswith('investigation_project_'):
|
|
system, prefix, live = GeminiProvider._messages(req, stable_case_context=True)
|
|
partition = {'prefix_bytes': len(encoded([system, prefix])),
|
|
'live_bytes': len(encoded(live)),
|
|
'prefix_hash': digest(encoded([system, prefix]))}
|
|
row = {'ref': name, 'prompt': req.get('prompt_version'),
|
|
'message_utf8_bytes': msg_bytes,
|
|
'user_repeated_strings_ge80_bytes': dup_strings,
|
|
'json_key_bytes': sum(keys.values()),
|
|
'top_json_keys_by_bytes': keys.most_common(8),
|
|
'stable_partition': partition}
|
|
per_call.append(row)
|
|
by_prompt[req.get('prompt_version')].append(row)
|
|
components = defaultdict(lambda: [0, 0, 0])
|
|
for (prompt, label, _), (count, total, size) in block_groups.items():
|
|
val = components[(prompt, label)]
|
|
val[0] += total
|
|
val[1] += size
|
|
val[2] += count
|
|
exact = repeated(groups)
|
|
canon = repeated(canonical)
|
|
summary = {
|
|
'path': path.as_posix(),
|
|
'identity': {k: manifest[k] for k in ('case_id', 'package_id', 'package_generation', 'state')},
|
|
'zip_bytes': path.stat().st_size,
|
|
'raw_member_bytes': sum(v.file_size for v in infos.values()),
|
|
'compressed_member_bytes': sum(v.compress_size for v in infos.values()),
|
|
'members': len(infos),
|
|
'zip_compression_methods': dict(compression_methods),
|
|
'deflate6_estimated_payload_bytes': sum(deflate_estimate.values()),
|
|
'deflate6_by_category': dict(deflate_estimate),
|
|
'categories': dict(categories),
|
|
'stored_context_components': {k:{'total_bytes':v[0], 'unique_value_bytes':v[1],
|
|
'repeated_value_bytes':v[0]-v[1], 'occurrences':v[2]} for k,v in sorted(context_components.items(), key=lambda kv:kv[1][0], reverse=True)},
|
|
'largest_members': [{'path': v.filename, 'raw_bytes': v.file_size,
|
|
'compressed_bytes': v.compress_size} for v in sorted(infos.values(), key=lambda v:v.compress_size, reverse=True)[:12]],
|
|
'exact_duplicate_groups': len(exact),
|
|
'exact_duplicate_raw_bytes': sum(v['raw_duplicate_bytes'] for v in exact),
|
|
'exact_duplicate_compressed_bytes': sum(v['compressed_duplicate_bytes'] for v in exact),
|
|
'largest_exact_duplicate_groups': exact[:8],
|
|
'canonical_json_duplicate_raw_bytes': sum(v['raw_duplicate_bytes'] for v in canon),
|
|
'original_tar_members_also_stored_separately': dict(archive_members),
|
|
'request_count': len(requests),
|
|
'research_repetition': {
|
|
'evidence_item_total_bytes': sum(n*s for n,s in evidence_items.values()),
|
|
'evidence_item_unique_bytes': sum(s for n,s in evidence_items.values()),
|
|
'evidence_item_fields_bytes':dict(evidence_sizes),
|
|
'source_line_total_bytes':sum(n*s for n,s in source_lines.values()),
|
|
'source_line_unique_bytes':sum(s for n,s in source_lines.values()),
|
|
'source_line_occurrences':sum(n for n,s in source_lines.values()),
|
|
'source_line_unique_count':len(source_lines)},
|
|
'prompts': {p: {'calls':len(rows), 'message_utf8_bytes':sum(r['message_utf8_bytes'] for r in rows),
|
|
'within_call_repeated_strings_ge80_bytes':sum(r['user_repeated_strings_ge80_bytes'] for r in rows)} for p, rows in by_prompt.items()},
|
|
'system_message_total_bytes':sum(size*count for (_,size),count in system_groups.items()),
|
|
'system_message_repeated_bytes':sum(size*(count-1) for (_,size),count in system_groups.items()),
|
|
'top_level_components': [{'prompt':p,'field':label,'total_canonical_bytes':v[0],
|
|
'unique_value_bytes':v[1], 'repeated_value_bytes':v[0]-v[1], 'occurrences':v[2]}
|
|
for (p,label),v in sorted(components.items(), key=lambda kv:kv[1][0], reverse=True)],
|
|
'json_keys_by_bytes':metadata_fields.most_common(15),
|
|
'calls': per_call,
|
|
}
|
|
assert sum(v['raw_bytes'] for v in categories.values()) == summary['raw_member_bytes']
|
|
assert sum(v['compressed_bytes'] for v in categories.values()) == summary['compressed_member_bytes']
|
|
return summary
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument('packages', nargs='+', type=Path)
|
|
parser.add_argument('--output', required=True, type=Path)
|
|
args = parser.parse_args()
|
|
report = {'measurement':'redundancy_audit_v1', 'units':'bytes, UTF-8, decimal MB/GB',
|
|
'limitations': ['No paid API calls; no quality or token savings inferred.',
|
|
'Categories are disjoint; duplicate, archive-copy and repeated-string measures overlap.',
|
|
'Compressed duplicate sizes are an upper bound, not a compatible deletion proposal.',
|
|
'Repeated strings are counted within user messages, minimum 80 UTF-8 bytes; references and JSON framing not deducted.',
|
|
'Component repetition compares whole canonical top-level values within each prompt version, ignoring order of calls.',
|
|
'Saved requests are canonical gateway input, not literal cached HTTP payload.'],
|
|
'packages':[audit(p) for p in args.packages]}
|
|
args.output.parent.mkdir(parents=True, exist_ok=True)
|
|
args.output.write_text(json.dumps(report, ensure_ascii=False, indent=2)+'\n', encoding='utf-8')
|
|
print(json.dumps({'output':str(args.output),'packages':len(report['packages'])}))
|
|
|
|
|
|
if __name__ == '__main__':
|
|
main()
|