197 lines
15 KiB
Python
197 lines
15 KiB
Python
"""Reconcile the 50 captured identities; public GETs only, no DB/publication writes."""
|
|
from pathlib import Path
|
|
from html.parser import HTMLParser
|
|
from concurrent.futures import ThreadPoolExecutor
|
|
import collections, csv, datetime, hashlib, json, re, subprocess, sys, urllib.request
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
OUT = ROOT / 'outputs/legacy-correspondence-20261022'
|
|
OUT.mkdir(parents=True, exist_ok=True)
|
|
AUDIT = ROOT / 'docs/investmoz/v2/auditoria-portal'
|
|
reconciliation = json.loads((ROOT / 'outputs/investmoz-legado-inventario-reconciliado-20261015/RECONCILIACAO-LEGADO.json').read_text(encoding='utf-8'))
|
|
inventory = json.loads((ROOT / 'storage/app/private/investmoz-v2/portal-audit/legacy-discovery/inventory.json').read_text(encoding='utf-8'))
|
|
config = json.loads(subprocess.check_output(['php', '-r', "$c=require 'config/legacy_content.php';echo json_encode($c);"], cwd=ROOT))
|
|
mapped = {int(re.search(r'post-(\d+)-', item['external_id'])[1]): (slug, item) for slug, item in config['articles'].items() if item['publishable']}
|
|
assert len(mapped) == 20 and len(reconciliation['articles']) == 50
|
|
|
|
class Page(HTMLParser):
|
|
data = None
|
|
reading = False
|
|
chunks = None
|
|
def handle_starttag(self, tag, attrs):
|
|
attrs = dict(attrs)
|
|
if tag == 'script' and attrs.get('type') == 'application/json' and 'data-page' in attrs:
|
|
self.reading = True; self.chunks = []
|
|
elif 'data-page' in attrs and attrs['data-page'].startswith('{'):
|
|
self.data = json.loads(attrs['data-page'])
|
|
def handle_data(self, value):
|
|
if self.reading: self.chunks.append(value)
|
|
def handle_endtag(self, tag):
|
|
if tag == 'script' and self.reading:
|
|
self.data = json.loads(''.join(self.chunks)); self.reading = False
|
|
|
|
def fetch(pair):
|
|
legacy_id, (slug, identity) = pair
|
|
url = 'https://teste-investidor.tempo.co.mz/' + slug
|
|
try:
|
|
request = urllib.request.Request(url, headers={'User-Agent': 'Mozilla/5.0 InvesteMoz-Legacy-Correspondence', 'Cache-Control': 'no-cache'})
|
|
with urllib.request.urlopen(request, timeout=35) as response:
|
|
raw = response.read(); status = response.status
|
|
page = Page(); page.feed(raw.decode('utf-8'))
|
|
article = page.data['props']['article']
|
|
assert page.data['component'] == 'portal/legacy' and article['legacy_id'] == legacy_id
|
|
assert article['source_url'] == identity['source_url'] and article['html'].strip()
|
|
return legacy_id, {'url': url, 'status': status, 'identity_verified': True,
|
|
'resource_id': article['id'], 'document_sha256': article['sha256'],
|
|
'current_title': article['title'], 'image_url': article['image']['href'] if article.get('image') else None,
|
|
'response_sha256': hashlib.sha256(raw).hexdigest()}
|
|
except Exception as error:
|
|
return legacy_id, {'url': url, 'identity_verified': False, 'error': type(error).__name__, 'status': getattr(error, 'code', None)}
|
|
|
|
if '--reuse-http' in sys.argv:
|
|
cached = json.loads((OUT / 'public-pages.json').read_text(encoding='utf-8'))
|
|
checks = {int(k): v for k, v in cached['pages'].items()}; stamp = cached['checked_utc']
|
|
assert set(checks) == set(mapped)
|
|
else:
|
|
with ThreadPoolExecutor(max_workers=4) as pool:
|
|
checks = dict(pool.map(fetch, mapped.items()))
|
|
stamp = datetime.datetime.now(datetime.timezone.utc).isoformat()
|
|
(OUT / 'public-pages.json').write_text(json.dumps({'checked_utc': stamp, 'method': 'Anonymous HTTP GET; Inertia article identity/source/content checked; no visual acceptance or new publication', 'pages': checks}, ensure_ascii=False, indent=2), encoding='utf-8')
|
|
|
|
captures = {}
|
|
for record in inventory['records']:
|
|
post = record.get('public_props', {}).get('post')
|
|
if post:
|
|
captures.setdefault(int(post['id']), []).append(record)
|
|
staged = collections.defaultdict(list)
|
|
prepared = {}
|
|
for article in reconciliation['articles']:
|
|
legacy_id = article['legacy_id']
|
|
if legacy_id in mapped or article['group'] in ['mercado_financeiro', 'identidade_incompleta']:
|
|
continue
|
|
record = next(r for r in captures[legacy_id] if hashlib.sha256(r['public_props']['post']['content'].encode()).hexdigest() == article['body_sha256'])
|
|
original = ROOT / record['private_file']
|
|
assert original.is_file() and hashlib.sha256(original.read_bytes()).hexdigest() == record['sha256']
|
|
post = record['public_props']['post']; seo = record['public_props']['seo']
|
|
group = article['group']; directory = OUT / 'private-preparation' / group; directory.mkdir(parents=True, exist_ok=True)
|
|
payload = {'format': 'investmoz-legacy-article-v1', 'legacy_id': legacy_id, 'title': article['title'],
|
|
'author': seo.get('author', 'tempo'), 'legacy_category': article['category'],
|
|
'source_url': record['url'], 'published_at': seo.get('publishedTime', ''),
|
|
'modified_at': article['modified_at'], 'content_html': post['content'],
|
|
'image_source_url': post.get('thumbnail'), 'image_sha256': None,
|
|
'references': article['references']}
|
|
data = (json.dumps(payload, ensure_ascii=False, indent=2) + '\n').encode()
|
|
filename = f"post-{legacy_id}-{article['modified_at'][:10].replace('-', '')}.txt"
|
|
path = directory / filename
|
|
if path.exists():
|
|
assert path.read_bytes() == data, 'Prepared version changed; preserve prior bytes'
|
|
else:
|
|
path.write_bytes(data)
|
|
item = {'legacy_id': legacy_id, 'title': article['title'], 'source_url': record['url'],
|
|
'external_id': f"legacy-tempo-post-{legacy_id}-{article['modified_at'][:10].replace('-', '')}",
|
|
'file_name': filename, 'relative_path': f'{group}/{filename}',
|
|
'sha256': hashlib.sha256(data).hexdigest(), 'source_capture_sha256': record['sha256'],
|
|
'original_body_sha256': article['body_sha256'], 'bytes': len(data), 'published': False,
|
|
'scan': 'not_run_on_prepared_derivative', 'dependencies': article['dependencies'],
|
|
'note': 'Private preparation only. Not imported, scanned, published or assigned to a current opportunity. Image unresolved, never substituted by generic image.'}
|
|
staged[group].append(item); prepared[legacy_id] = item
|
|
assert len(prepared) == 28
|
|
qa_receipt_path = OUT / 'qa-import-receipt.json'
|
|
qa_rows = {}
|
|
if qa_receipt_path.exists():
|
|
proof = json.loads(qa_receipt_path.read_text(encoding='utf-8'))
|
|
assert proof['qa_database'] == 'simulator_fiscal_delta_20261021_test' and proof['port'] == 55441
|
|
assert proof['imported_private'] == proof['clean_actual_scans'] == proof['idempotent_replay_reused'] == 28
|
|
qa_rows = {int(re.search(r'post-(\d+)-', r['external_id'])[1]): r for r in proof['resources']}
|
|
assert set(qa_rows) == set(prepared)
|
|
assert all(not r['published'] and r['malware_scan_status'] == 'clean' and r['sha256'] == prepared[i]['sha256'] for i, r in qa_rows.items())
|
|
for group, items in staged.items():
|
|
(OUT / 'private-preparation' / group / 'preparation.json').write_text(json.dumps({'importable_manifest': False, 'group': group, 'items': items}, ensure_ascii=False, indent=2), encoding='utf-8')
|
|
|
|
path = AUDIT / 'legado-conteudos-reconciliados.csv'
|
|
with path.open(encoding='utf-8-sig', newline='') as f:
|
|
reader = csv.DictReader(f); columns = reader.fieldnames; rows = list(reader)
|
|
editorial_receipt = AUDIT / 'legado-editorial-qa-20261022.json'
|
|
editorial_rows = {}
|
|
if editorial_receipt.exists():
|
|
editorial = json.loads(editorial_receipt.read_text(encoding='utf-8'))
|
|
assert editorial['qa_only'] and not editorial['destination_imported'] and not editorial['destination_published']
|
|
editorial_rows = {int(re.search(r'post-(\d+)-', r['external_id'])[1]): r for r in editorial['rows']}
|
|
assert len(editorial_rows) == 28 and all(r['private'] for r in editorial_rows.values())
|
|
extra = ['Endereço no novo portal', 'Estado actual consolidado', 'Pendência actual', 'Conferência HTTP UTC', 'Grupo de preparação QA']
|
|
destination_receipt = ROOT / 'docs/investmoz/unification/legacy-editorial-installation-receipt-20261022.json'
|
|
destination_rows = {}
|
|
if destination_receipt.exists():
|
|
destination = json.loads(destination_receipt.read_text(encoding='utf-8'))
|
|
assert destination['new_publications'] == 0
|
|
destination_rows = {r['legacy_id']: r for r in destination['rows']}
|
|
assert len(destination_rows) == 28 and all(r['private'] for r in destination_rows.values())
|
|
for col in extra:
|
|
if col not in columns: columns.append(col)
|
|
articles = {a['legacy_id']: a for a in reconciliation['articles']}
|
|
for row in rows:
|
|
legacy_id = int(row['ID legado']); article = articles[legacy_id]; check = checks.get(legacy_id)
|
|
row['Endereço no novo portal'] = check['url'] if check else ''
|
|
row['Conferência HTTP UTC'] = stamp if check else ''
|
|
row['Grupo de preparação QA'] = article['group'] if legacy_id in prepared else ''
|
|
if check:
|
|
row['Estado actual consolidado'] = 'Publicado; identidade e conteúdo consultáveis na homologação' if check['identity_verified'] else 'Publicação documentada; consulta actual não confirmada'
|
|
row['Pendência actual'] = 'Não foi feita aceitação visual ou conferência pela Tempo nesta passagem.' if check['identity_verified'] else str(check)
|
|
if check['identity_verified']:
|
|
row['Situação'] = 'publicado'; row['Recurso ativo'] = str(check['resource_id'])
|
|
row['Grupo restante'] = 'publicados'
|
|
row['Dependência concreta'] = 'Sem pendência de integração desta identidade; mantém-se a revisão editorial corrente.'
|
|
row['Próxima ação'] = 'Preservar a publicação; gerir edições e versões pela biblioteca existente.'
|
|
row['Imagem atual'] = json.dumps({'scope': 'homologação', 'href': check['image_url'], 'status': 'referência na ficha; binário não reensaiado'}, ensure_ascii=False)
|
|
elif legacy_id in prepared:
|
|
row['Estado actual consolidado'] = 'Importado privado em QA; scan real limpo; por integrar na homologação' if legacy_id in qa_rows else 'Por integrar; exemplar derivado preparado privadamente, sem importação'
|
|
if legacy_id in qa_rows: row['Recurso QA'] = str(qa_rows[legacy_id]['id'])
|
|
row['Pendência actual'] = 'Fechar edição/destino de leitura e publicação por lote na homologação. Dependências históricas de 15/10, por confrontar com o estado actual (não são novos bloqueios): ' + ' | '.join(article['dependencies'])
|
|
if legacy_id in qa_rows:
|
|
row['Dependência concreta'] = 'Rascunho QA analisado; falta fechar leitor/edição e entrega/publicação na homologação. Referências externas e imagens não recolhidas continuam discriminadas no inventário.'
|
|
row['Próxima ação'] = 'Continuar por lote, preservando o carácter histórico/editorial; não converter em oportunidade vigente nem em regra fiscal.'
|
|
elif article['group'] == 'mercado_financeiro':
|
|
row['Estado actual consolidado'] = 'Original vazio'; row['Pendência actual'] = 'Fornecer corpo editorial pela Tempo; imagem e Fintech não substituem o original. Sem novo endereço público.'
|
|
else:
|
|
row['Estado actual consolidado'] = 'Original sem título'; row['Pendência actual'] = 'Identificação editorial pela Tempo para o post 74364; corpo com 21 caracteres úteis preservado. Sem título ou endereço novo inventado.'
|
|
row['Evidência atual'] = 'outputs/legacy-correspondence-20261022/public-pages.json; qa-import-receipt.json; editorial-gaps-current.json; inventário histórico de 15/10 preservado'
|
|
if legacy_id in editorial_rows:
|
|
assert editorial_rows[legacy_id]['original_sha256'] == prepared[legacy_id]['sha256']
|
|
row['Estado actual consolidado'] = 'Privado em QA; edição, leitura e prévia disponíveis; entrega privada preparada'
|
|
row['Pendência actual'] = 'Revisão editorial pela equipa; integração Git e transporte privado autenticado para homologação; publicação separada pelo portal. Imagens não verificadas mantêm referência/aviso.'
|
|
if legacy_id == 62413:
|
|
row['Pendência actual'] += ' Rever/corrigir a nota de demonstração QA antes de publicar.'
|
|
row['Dependência concreta'] = row['Pendência actual']
|
|
row['Próxima ação'] = 'Seguir o guia e a sequência de entrega privada do manual; não repetir importação QA ou provas dos 20 aceites.'
|
|
row['Evidência atual'] = 'legado-editorial-qa-20261022.json; manual: Legado — edição e leitura dos 28 rascunhos em QA; inventário e referências históricas preservados'
|
|
if legacy_id in destination_rows:
|
|
item = destination_rows[legacy_id]
|
|
assert item['sha256'] == prepared[legacy_id]['sha256']
|
|
row['Endereço no novo portal'] = f"https://teste-investidor.tempo.co.mz/management/portal-services/resources/{item['resource_id']}/article/preview"
|
|
row['Recurso ativo'] = str(item['resource_id'])
|
|
row['Estado actual consolidado'] = 'Rascunho privado na homologação; pesquisa, editor, prévia e recuperação conferidos'
|
|
row['Pendência actual'] = 'Revisão editorial e publicação separada pela equipa no portal. As referências sem exemplar de imagem verificado permanecem explícitas.'
|
|
if legacy_id == 62413:
|
|
row['Pendência actual'] += ' Rever/corrigir a nota de demonstração QA antes de publicar.'
|
|
row['Dependência concreta'] = row['Pendência actual']
|
|
row['Próxima ação'] = 'Rever pela gestão; corrigir e pré-visualizar; publicar expressamente quando autorizado, sem comandos.'
|
|
row['Evidência atual'] = 'docs/investmoz/unification/legacy-editorial-installation-receipt-20261022.json; preservação dos registos anteriores conferida; provas QA reutilizadas'
|
|
assert len(rows) == 50 and len({row['ID legado'] for row in rows}) == 50
|
|
with path.open('w', encoding='utf-8-sig', newline='') as f:
|
|
writer = csv.DictWriter(f, columns); writer.writeheader(); writer.writerows(rows)
|
|
|
|
def esc(value): return str(value).replace('|', '\\|').replace('\n', ' ')
|
|
table = ['# Correspondência integral dos 50 artigos do legado', '',
|
|
'Conferência HTTP individual em ' + stamp + '. Lista por identidade editorial; imagens, recursos da biblioteca e variantes de URL não são artigos adicionais. Não altera publicações. Endereço em branco significa que não foi demonstrada uma página nova, não uma ligação a criar por suposição.', '',
|
|
'| ID | Título original | Endereço original | Endereço no novo portal | Estado | Pendência |', '| --- | --- | --- | --- | --- | --- |']
|
|
for row in rows:
|
|
table.append('| ' + ' | '.join(esc(row[k]) for k in ['ID legado', 'Título', 'URLs observadas', 'Endereço no novo portal', 'Estado actual consolidado', 'Pendência actual']) + ' |')
|
|
(AUDIT / 'legado-correspondencia-integral.md').write_text('\n'.join(table) + '\n', encoding='utf-8')
|
|
summary = {'checked_utc': stamp, 'distinct_articles': 50, 'publications_documented': 20,
|
|
'public_pages_identity_verified': sum(p['identity_verified'] for p in checks.values()),
|
|
'pending_content': 28, 'private_preparation': {k: len(v) for k, v in staged.items()},
|
|
'empty': 1, 'untitled': 1, 'database_writes_by_correspondence_script': 0, 'active_database_writes': 0, 'new_publications': 0,
|
|
'qa_imports': len(qa_rows), 'qa_clean_scans': len(qa_rows), 'remaining_integration_complete': False}
|
|
(OUT / 'receipt.json').write_text(json.dumps(summary, indent=2), encoding='utf-8')
|
|
print(json.dumps(summary))
|