Dam and libary/archive metadata

<record xmlns="http://www.loc.gov/MARC21/slim">
  <controlfield tag="001">123456</controlfield>

  <datafield tag="100" ind1="1" ind2=" ">
    <subfield code="a">Austen, Jane</subfield>
  </datafield>

  <datafield tag="245" ind1="1" ind2="0">
    <subfield code="a">Pride and prejudice</subfield>
  </datafield>

  <datafield tag="264" ind1=" " ind2="1">
    <subfield code="a">London :</subfield>
    <subfield code="b">T. Egerton,</subfield>
    <subfield code="c">1813.</subfield>
  </datafield>

  <datafield tag="650" ind1=" " ind2="0">
    <subfield code="a">Courtship</subfield>
    <subfield code="v">Fiction.</subfield>
  </datafield>
</record>

Collection
  └── Series
      └── Box
          └── Folder
              └── Item

from __future__ import annotations

import csv
from dataclasses import dataclass
from datetime import date, datetime
from pathlib import Path
from typing import Optional


VALID_RIGHTS = {
    "public domain": "public_domain",
    "copyright held by organization": "copyright_organization",
    "all rights reserved": "all_rights_reserved",
}


@dataclass
class NormalizedAsset:
    source_id: str
    original_filename: str
    title: str
    creator: Optional[str]
    created_date: Optional[date]
    subjects: list[str]
    rights_code: str


@dataclass
class ValidationIssue:
    source_id: str
    field: str
    message: str
    severity: str = "error"


def parse_date(value: str) -> Optional[date]:
    """Parse known legacy date formats.

    In production, date parsing rules should be agreed with the customer.
    Avoid silently guessing when a date is ambiguous.
    """
    value = value.strip()

    if not value:
        return None

    formats = [
        "%Y/%m/%d",  # 1924/06/15
        "%Y-%m-%d",  # 1924-06-15
        "%B %Y",     # June 1932
        "%Y",        # 1932
    ]

    for format_string in formats:
        try:
            return datetime.strptime(value, format_string).date()
        except ValueError:
            pass

    raise ValueError(f"Unsupported date format: {value!r}")


def normalize_subjects(value: str) -> list[str]:
    """Split legacy pipe-delimited subjects and normalize whitespace."""
    return [
        subject.strip()
        for subject in value.split("|")
        if subject.strip()
    ]


def normalize_rights(value: str) -> str:
    normalized = value.strip().lower()

    try:
        return VALID_RIGHTS[normalized]
    except KeyError as exc:
        raise ValueError(f"Unknown rights statement: {value!r}") from exc


def normalize_row(row: dict[str, str]) -> NormalizedAsset:
    return NormalizedAsset(
        source_id=row["legacy_id"].strip(),
        original_filename=row["file_name"].strip(),
        title=row["title"].strip(),
        creator=row["creator"].strip() or None,
        created_date=parse_date(row["created_date"]),
        subjects=normalize_subjects(row["subjects"]),
        rights_code=normalize_rights(row["rights"]),
    )


def validate_asset(asset: NormalizedAsset) -> list[ValidationIssue]:
    issues: list[ValidationIssue] = []

    if not asset.source_id:
        issues.append(
            ValidationIssue("", "legacy_id", "A legacy identifier is required.")
        )

    if not asset.original_filename:
        issues.append(
            ValidationIssue(
                asset.source_id,
                "file_name",
                "An original filename is required.",
            )
        )

    if not asset.title:
        issues.append(
            ValidationIssue(asset.source_id, "title", "A title is required.")
        )

    if not asset.subjects:
        issues.append(
            ValidationIssue(
                asset.source_id,
                "subjects",
                "No subjects supplied; record may be less discoverable.",
                severity="warning",
            )
        )

    return issues


def load_assets(csv_path: Path) -> tuple[list[NormalizedAsset], list[ValidationIssue]]:
    assets: list[NormalizedAsset] = []
    issues: list[ValidationIssue] = []
    seen_source_ids: set[str] = set()

    with csv_path.open(newline="", encoding="utf-8-sig") as file:
        reader = csv.DictReader(file)

        required_columns = {
            "legacy_id",
            "file_name",
            "title",
            "creator",
            "created_date",
            "subjects",
            "rights",
        }

        missing_columns = required_columns - set(reader.fieldnames or [])
        if missing_columns:
            raise ValueError(
                f"Input CSV is missing required columns: {sorted(missing_columns)}"
            )

        for line_number, row in enumerate(reader, start=2):
            try:
                asset = normalize_row(row)

                if asset.source_id in seen_source_ids:
                    issues.append(
                        ValidationIssue(
                            asset.source_id,
                            "legacy_id",
                            f"Duplicate legacy ID on CSV line {line_number}.",
                        )
                    )
                    continue

                seen_source_ids.add(asset.source_id)

                row_issues = validate_asset(asset)
                issues.extend(row_issues)

                if not any(issue.severity == "error" for issue in row_issues):
                    assets.append(asset)

            except ValueError as exc:
                issues.append(
                    ValidationIssue(
                        source_id=row.get("legacy_id", f"line:{line_number}"),
                        field="row",
                        message=str(exc),
                    )
                )

    return assets, issues


if __name__ == "__main__":
    assets, issues = load_assets(Path("legacy_assets.csv"))

    print(f"Valid assets ready to migrate: {len(assets)}")
    for asset in assets:
        print(asset)

    print("\nValidation issues:")
    for issue in issues:
        print(
            f"[{issue.severity.upper()}] "
            f"{issue.source_id} - {issue.field}: {issue.message}"
        )

from django.db import transaction
from django.utils import timezone

from catalog.models import Asset, MigrationRun


BATCH_SIZE = 1_000


def import_assets(
    assets: list[NormalizedAsset],
    migration_run: MigrationRun,
) -> int:
    """Create only assets not already imported from this source.

    Assumes Asset has a unique constraint on:
        (source_system, legacy_id)
    """
    source_system = "legacy_dam"

    existing_ids = set(
        Asset.objects.filter(
            source_system=source_system,
            legacy_id__in=[asset.source_id for asset in assets],
        ).values_list("legacy_id", flat=True)
    )

    to_create = [
        Asset(
            source_system=source_system,
            legacy_id=item.source_id,
            original_filename=item.original_filename,
            title=item.title,
            creator=item.creator,
            created_date=item.created_date,
            rights_code=item.rights_code,
            subjects=item.subjects,  # Could instead be a ManyToMany relation.
            migration_run=migration_run,
            imported_at=timezone.now(),
        )
        for item in assets
        if item.source_id not in existing_ids
    ]

    with transaction.atomic():
        Asset.objects.bulk_create(
            to_create,
            batch_size=BATCH_SIZE,
        )

        migration_run.imported_count += len(to_create)
        migration_run.save(update_fields=["imported_count"])

    return len(to_create)