ape2e/tools/finalize_instructors_v2.py

"""Verify promoted identities and clean up only this run's temporary avatars."""
import hashlib
import json
import shlex

from promote_instructors_v2 import connect, run, RELEASE
from retrain_instructors_v2 import ROOT, REPORT, IDS, FILES, API, SERVICE, HTTP, api_session, save


def main():
    jobs = json.loads((REPORT / 'jobs.json').read_text('utf-8'))
    checks = json.loads((REPORT / 'generated-validation.json').read_text('utf-8'))
    playback = json.loads((REPORT / 'stable-viewer-validation.json').read_text('utf-8'))
    catalog = json.loads((REPORT / 'catalog-validation.json').read_text('utf-8'))
    assert len(playback) == catalog['count'] == 3
    assert all(not item['errors'] and item['driver']['active'] > 30 for item in playback)
    session = api_session()
    response = session.get(API + '/api/v1/avatars', params={'pageSize': 200}, timeout=30)
    response.raise_for_status()
    after = response.json()
    save('after-platform.json', after)
    before = json.loads((REPORT / 'before-platform.json').read_text('utf-8'))
    # No full catalog sync: it would re-import previously soft-deleted legacy entries.
    assert before['data']['total'] == after['data']['total'] == 3
    before_items = {item['id']: item for item in before['data']['items']}
    after_items = {item['id']: item for item in after['data']['items']}
    assert set(before_items) == set(after_items) == set(IDS)
    preserved_fields = ('id', 'name', 'status', 'default', 'voice', 'voiceCapabilityId', 'ownerId', 'visibility', 'viewerUrl')
    for ident in IDS:
        assert all(before_items[ident][field] == after_items[ident][field] for field in preserved_fields)
    service_after = []
    for check in checks:
        ident = check['stable_id']
        metadata = HTTP.get(SERVICE + '/api/v1/avatars/' + ident, timeout=30)
        metadata.raise_for_status()
        service_after.append(metadata.json())
        assert metadata.json()['status'] == 'ready'
        for filename in FILES:
            response = HTTP.get(SERVICE + f'/api/v1/avatars/{ident}/assets/{filename}', timeout=45)
            response.raise_for_status()
            assert hashlib.sha256(response.content).hexdigest() == check['sha256'][filename]
    save('after-service.json', service_after)

    client = connect()
    try:
        expected = {row['stable_id']: row['sha256'] for row in checks}
        code = 'from pathlib import Path;import json,hashlib;expected=' + repr(expected) + ";print(json.dumps({ident:{name:hashlib.sha256((Path('/app/builtin_avatars')/ident/'assets'/name).read_bytes()).hexdigest() for name in files} for ident,files in expected.items()}))"
        actual = json.loads(run(client, 'docker run --rm --network none --entrypoint python digital-human-service:instructors-v2-20260914 -c ' + shlex.quote(code)))
        assert actual == expected
        save('image-validation.json', actual)
        # The list uses source-delivered PNG covers, so replace those exact assets too.
        run(client, f'mkdir -p {RELEASE}/backup-covers/source {RELEASE}/backup-covers/published')
        with client.open_sftp() as sftp:
            for number in (1, 2, 3):
                name = f'equipment-instructor-{number}.png'
                local = ROOT / 'ai_person/ai_person_web/public/avatars/builtin' / name
                for group, base in (
                    ('source', '/home/ai_person/ai_person_web/public/avatars/builtin'),
                    ('published', '/home/ai_person/html_ap/avatars/builtin'),
                ):
                    target = base + '/' + name
                    backup = RELEASE + '/backup-covers/' + group + '/' + name
                    run(client, f'test -f {target} && test ! -e {backup} && cp {target} {backup}')
                    sftp.put(str(local), target + '.v2-stage')
                    sftp.posix_rename(target + '.v2-stage', target)
                    with sftp.open(target, 'rb') as stream:
                        assert hashlib.sha256(stream.read()).digest() == hashlib.sha256(local.read_bytes()).digest()
        # Keep service-side generation logs in the rollback/audit report before deleting temporary jobs.
        with client.open_sftp() as sftp:
            for job in jobs:
                sftp.get('/home/digital-human-service/data/jobs/' + job['job_id'] + '.log', str(REPORT / f"generation-{job['number']}.log"))
    finally:
        client.close()
    removed = []
    for job in jobs:
        assert job['avatar_id'] not in IDS
        response = HTTP.delete(SERVICE + '/api/v1/avatars/' + job['avatar_id'], timeout=45)
        assert response.status_code == 204, response.status_code
        removed.append(job['avatar_id'])
    save('final-validation.json', {
        'identitiesPreserved': True, 'defaultInstructor': 2, 'voicesPreserved': True,
        'platformCount': 3, 'runtimeAssetHashesMatch': True, 'imageAssetHashesMatch': True,
        'localAndPublishedCoversUpdated': True, 'temporaryAvatarsRemoved': removed,
        'oldLectureVideosAndAudioPreserved': True,
        'scope': 'Resource regeneration and runtime playback verified; phoneme-level lip sync quality not accepted by this check.',
    })
    print('Verified: three original identities, default instructor, voices, runtime/image assets and covers. Temporary generation entries removed.')


if __name__ == '__main__':
    main()