Files
Investidor-V2/tools/portal-segmentation-source-aliases.py
nala 287e991486
tests / ci (push) Waiting to run
oportunidades 7
2026-10-13 11:40:29 +02:00

36 lines
2.9 KiB
Python

"""Retain all locations as aliases of existing bytes, not duplicate catalogue entries."""
from pathlib import Path
import json,csv,hashlib
csv.field_size_limit(20_000_000)
root=Path(__file__).resolve().parent.parent
directory=root/'storage/app/private/investmoz-v2/portal-audit/segmentation-reconciliation-20261011'
original=json.loads((directory/'csv-edits.json').read_text(encoding='utf-8'))['files'][2]
file=root/'docs/investmoz/v2/auditoria-portal/inventario-fontes.csv'
raw=file.read_bytes();rows=list(csv.reader(file.open(encoding='utf-8-sig',newline='')));headers=rows[0]
patches=[];removals=[];aliases=[]
for added in original['append']:
same=[r for r in original['rows'][1:] if r[2]==added[2]]
if not same:continue
assert len(same)==1
canonical=same[0];assert canonical[0]!=added[0]
old_index=next(i for i,row in enumerate(rows) if row[0]==canonical[0] and row[2]==canonical[2])
new_index=next(i for i,row in enumerate(rows) if row[0]==added[0] and row[2]==added[2])
patches.append({'row':old_index,'values':{'Nota de leitura':rows[old_index][4]+'\n[Segmentação11/10/2026] Localização/cópia adicional inventariada: '+added[0]+'. MesmoSHA256; fonte única por bytes, sem nova importação. A origem anterior permanece.'}})
removals.append({'row':new_index,'sha256':added[2],'path':added[0],'canonical_path':canonical[0]})
aliases.append({'canonical_path':canonical[0],'additional_path':added[0],'sha256':added[2]})
assert len(removals)==3
packet={'proof_filename':'csv-source-alias-proof.json','files':[{'filename':'inventario-fontes.csv','input_sha256':hashlib.sha256(raw).hexdigest(),'rows':rows,'patches':patches,'remove_rows':removals,'source_hash_updates':[],'allowed_columns':['Nota de leitura']}]}
alias_dir=directory.parent/'segmentation-source-aliases-20261011';alias_dir.mkdir(exist_ok=True)
(alias_dir/'csv-edits.json').write_text(json.dumps(packet,ensure_ascii=False),encoding='utf-8')
(directory/'source-byte-aliases.json').write_text(json.dumps(aliases,ensure_ascii=False,indent=2),encoding='utf-8')
# Only entries appended by this operation are consolidated. The original registry is preserved.
inventory_path=directory.parent/'inventario-fontes.json'
inventory=json.loads(inventory_path.read_text(encoding='utf-8'))
for alias in aliases:
canonical=next(x for x in inventory['files'] if x['path']==alias['canonical_path'] and x['sha256']==alias['sha256'])
canonical.setdefault('exemplar_aliases',[]).append(alias['additional_path'])
inventory['files']=[x for x in inventory['files'] if not(x['path']==alias['additional_path'] and x['sha256']==alias['sha256'])]
inventory['segmentation_source_added'].update({'added_unique_bytes':7,'known_byte_aliases':3,'inventoried_locations':10})
inventory_path.write_text(json.dumps(inventory,ensure_ascii=False,indent=2),encoding='utf-8')
print(json.dumps({'result':'aliases_prepared','inventoried_locations':10,'new_unique_bytes':7,'existing_byte_aliases':3,'active_writes':0}))