justvitamin/scripts/validate-template-schema.py

from __future__ import annotations

import csv
import json
from pathlib import Path
import sys
from datetime import UTC, datetime

try:
    import openpyxl
except ImportError as exc:
    raise SystemExit('openpyxl is required and is already available in the Codex Python environment') from exc

ROOT = Path(__file__).resolve().parents[1]
WORK_ROOT = ROOT.parent
TEMPLATE = Path(r'C:\Users\uldvs\Downloads\Just Vitamins - Content Population Templates (1).xlsx')
DASH_OUT = ROOT / 'content_population_exports'
MIG_OUT = WORK_ROOT / 'JV Migration to shopify' / 'content_population_exports'

META_SHEETS = [
    'Metaobject - Dietary Tag',
    'Metaobject - Health Goals',
    'Metaobject - Key Ingredients',
    'Metaobject - Benefits',
    'Metaobject - Clinically Shown t',
    'Metaobject - Collection Promo C',
]

CHECKS = [
    {
        'name': 'products_dashboard',
        'template_sheet': 'Products',
        'csv': DASH_OUT / 'products_content_matrixify.csv',
    },
    {
        'name': 'products_migration',
        'template_sheet': 'Products',
        'csv': MIG_OUT / 'products_content_v5.csv',
    },
    {
        'name': 'smart_collections_dashboard',
        'template_sheet': 'Smart Collections',
        'csv': DASH_OUT / 'smart_collections_matrixify.csv',
    },
    {
        'name': 'smart_collections_migration',
        'template_sheet': 'Smart Collections',
        'csv': MIG_OUT / 'smart_collections_promo_cards.csv',
    },
    {
        'name': 'metaobjects_dashboard',
        'template_sheet': 'Metaobject - Dietary Tag',
        'csv': DASH_OUT / 'metaobjects_matrixify.csv',
    },
    {
        'name': 'metaobjects_migration_all',
        'template_sheet': 'Metaobject - Dietary Tag',
        'csv': MIG_OUT / 'metaobjects_all.csv',
    },
]


def template_headers(wb, sheet: str) -> list[str]:
    return [str(value) for value in next(wb[sheet].iter_rows(min_row=1, max_row=1, values_only=True)) if value is not None]


def csv_headers(path: Path) -> list[str]:
    with path.open(newline='', encoding='utf-8-sig') as f:
        return next(csv.reader(f))


def csv_count(path: Path) -> int:
    with path.open(newline='', encoding='utf-8-sig') as f:
        return max(sum(1 for _ in csv.reader(f)) - 1, 0)


def main() -> int:
    if not TEMPLATE.exists():
        raise SystemExit(f'Template not found: {TEMPLATE}')
    wb = openpyxl.load_workbook(TEMPLATE, read_only=True, data_only=True)
    checks = []
    for item in CHECKS:
        expected = template_headers(wb, item['template_sheet'])
        path = item['csv']
        exists = path.exists()
        actual = csv_headers(path) if exists else []
        missing = [header for header in expected if header not in actual]
        extra = [header for header in actual if header not in expected]
        order_match = actual == expected
        checks.append({
            'name': item['name'],
            'templateSheet': item['template_sheet'],
            'csv': str(path),
            'exists': exists,
            'rowCount': csv_count(path) if exists else 0,
            'expectedHeaders': expected,
            'actualHeaders': actual,
            'orderMatch': order_match,
            'missingHeaders': missing,
            'extraHeaders': extra,
            'status': 'pass' if exists and order_match else 'fail',
        })

    # Confirm all template metaobject sheets use the same long-format header.
    meta_headers = {sheet: template_headers(wb, sheet) for sheet in META_SHEETS}
    canonical_meta = meta_headers[META_SHEETS[0]]
    meta_header_consistent = all(headers == canonical_meta for headers in meta_headers.values())

    manifest = {
        'generatedAt': datetime.now(UTC).replace(microsecond=0).isoformat().replace('+00:00', 'Z'),
        'template': str(TEMPLATE),
        'metaHeaderConsistent': meta_header_consistent,
        'checks': checks,
        'ready': meta_header_consistent and all(check['status'] == 'pass' for check in checks),
    }
    DASH_OUT.mkdir(parents=True, exist_ok=True)
    (DASH_OUT / 'template_schema_validation.json').write_text(json.dumps(manifest, indent=2) + '\n', encoding='utf-8')

    rows = [[
        'Name', 'Status', 'Template Sheet', 'CSV', 'Rows', 'Order Match', 'Missing Headers', 'Extra Headers'
    ]]
    for check in checks:
        rows.append([
            check['name'], check['status'], check['templateSheet'], check['csv'], check['rowCount'],
            check['orderMatch'], ' | '.join(check['missingHeaders']), ' | '.join(check['extraHeaders'])
        ])
    with (DASH_OUT / 'template_schema_validation.csv').open('w', newline='', encoding='utf-8') as f:
        writer = csv.writer(f)
        writer.writerows(rows)

    print(f"Template schema validation ready: {manifest['ready']}")
    for check in checks:
        print(f"{check['name']}: {check['status']} rows={check['rowCount']}")
    return 0 if manifest['ready'] else 1


if __name__ == '__main__':
    raise SystemExit(main())