archival.ArchiveBox/archivebox/core/migrations/0024_auto_20240513_1143.py

# Generated by Django 5.0.6 on 2024-05-13 11:43

from django.db import migrations
from datetime import datetime
from abid_utils.abid import abid_from_values, DEFAULT_ABID_URI_SALT


def calculate_abid(self):
    """
    Return a freshly derived ABID (assembled from attrs defined in ABIDModel.abid_*_src).
    """
    prefix = self.abid_prefix
    ts = eval(self.abid_ts_src)
    uri = eval(self.abid_uri_src)
    subtype = eval(self.abid_subtype_src)
    rand = eval(self.abid_rand_src)

    if (not prefix) or prefix == 'obj_':
        suggested_abid = self.__class__.__name__[:3].lower()
        raise Exception(f'{self.__class__.__name__}.abid_prefix must be defined to calculate ABIDs (suggested: {suggested_abid})')

    if not ts:
        ts = datetime.utcfromtimestamp(0)
        print(f'[!] WARNING: Generating ABID with ts=0000000000 placeholder because {self.__class__.__name__}.abid_ts_src={self.abid_ts_src} is unset!', ts.isoformat())

    if not uri:
        uri = str(self)
        print(f'[!] WARNING: Generating ABID with uri=str(self) placeholder because {self.__class__.__name__}.abid_uri_src={self.abid_uri_src} is unset!', uri)

    if not subtype:
        subtype = self.__class__.__name__
        print(f'[!] WARNING: Generating ABID with subtype={subtype} placeholder because {self.__class__.__name__}.abid_subtype_src={self.abid_subtype_src} is unset!', subtype)

    if not rand:
        rand = getattr(self, 'uuid', None) or getattr(self, 'id', None) or getattr(self, 'pk')
        print(f'[!] WARNING: Generating ABID with rand=self.id placeholder because {self.__class__.__name__}.abid_rand_src={self.abid_rand_src} is unset!', rand)

    abid = abid_from_values(
        prefix=prefix,
        ts=ts,
        uri=uri,
        subtype=subtype,
        rand=rand,
        salt=DEFAULT_ABID_URI_SALT,
    )
    assert abid.ulid and abid.uuid and abid.typeid, f'Failed to calculate {prefix}_ABID for {self.__class__.__name__}'
    return abid


def copy_snapshot_uuids(apps, schema_editor):
    print('   Copying snapshot.id -> snapshot.uuid...')
    Snapshot = apps.get_model("core", "Snapshot")
    for snapshot in Snapshot.objects.all():
        snapshot.uuid = snapshot.id
        snapshot.save(update_fields=["uuid"])

def generate_snapshot_abids(apps, schema_editor):
    print('   Generating snapshot.abid values...')
    Snapshot = apps.get_model("core", "Snapshot")
    for snapshot in Snapshot.objects.all():
        snapshot.abid_prefix = 'snp_'
        snapshot.abid_ts_src = 'self.added'
        snapshot.abid_uri_src = 'self.url'
        snapshot.abid_subtype_src = '"01"'
        snapshot.abid_rand_src = 'self.uuid'

        snapshot.abid = calculate_abid(snapshot)
        snapshot.uuid = snapshot.abid.uuid
        snapshot.save(update_fields=["abid", "uuid"])

def generate_archiveresult_abids(apps, schema_editor):
    print('   Generating ArchiveResult.abid values... (may take an hour or longer for large collections...)')
    ArchiveResult = apps.get_model("core", "ArchiveResult")
    Snapshot = apps.get_model("core", "Snapshot")
    for result in ArchiveResult.objects.all():
        result.abid_prefix = 'res_'
        result.snapshot = Snapshot.objects.get(pk=result.snapshot_id)
        result.snapshot_added = result.snapshot.added
        result.snapshot_url = result.snapshot.url
        result.abid_ts_src = 'self.snapshot_added'
        result.abid_uri_src = 'self.snapshot_url'
        result.abid_subtype_src = 'self.extractor'
        result.abid_rand_src = 'self.id'

        result.abid = calculate_abid(result)
        result.uuid = result.abid.uuid
        result.save(update_fields=["abid", "uuid"])


class Migration(migrations.Migration):

    dependencies = [
        ('core', '0023_alter_archiveresult_options_archiveresult_abid_and_more'),
    ]

    operations = [
        migrations.RunPython(copy_snapshot_uuids, reverse_code=migrations.RunPython.noop),
        migrations.RunPython(generate_snapshot_abids, reverse_code=migrations.RunPython.noop),
        migrations.RunPython(generate_archiveresult_abids, reverse_code=migrations.RunPython.noop),
    ]