cms-SiteForge/scripts/setup_home_blocks.py

"""Task 13b: switch the HESU Pipe homepage from template-driven to block-driven.

Writes the 8-type block sequence into the site home entry through the real
studio-draft API (login -> PUT draft -> publish -> build -> verify), so the
page renders from blocks (layout_mode=replace) with the trade-pro block
bridge instead of pages/home.html. The template file itself stays untouched.

Run AFTER the Docker image has been rebuilt with the trust_bar /
applications_band block types, then:

    PYTHONUTF8=1 .venv/Scripts/python scripts/setup_home_blocks.py

Reads SITEFORGE_ADMIN_PASSWORD / SITEFORGE_PORT from the repo .env.
"""
from __future__ import annotations
import json
import sys
import time
import uuid
from pathlib import Path

import httpx

ROOT = Path(__file__).resolve().parent.parent
sys.path.insert(0, str(ROOT))
from app import blueprints  # noqa: E402 -- local package for block defaults

SITE_ID = '483295c29ea0'          # HESU Pipe (real import site, theme trade-pro)
HOME_ENTRY_ID = 'c8ff1d1a45f3'    # route '/' home entry


def load_env() -> dict:
    values: dict[str, str] = {}
    env_path = ROOT / '.env'
    if env_path.is_file():
        for line in env_path.read_text(encoding='utf-8').splitlines():
            line = line.strip()
            if line and not line.startswith('#') and '=' in line:
                key, val = line.split('=', 1)
                values[key.strip()] = val.strip()
    return values


def real_stats(entries: list[dict]) -> list[dict]:
    """Derive stat items from actual catalogue counts - no invented numbers."""
    products = [e for e in entries if e.get('type') in ('product', 'vehicle')]
    posts = [e for e in entries if e.get('type') == 'post']

    def field(entry, key):
        fields = (entry.get('meta') or {}).get('fields') or {}
        return str(fields.get(key) or '').strip()

    categories = {field(p, 'category') for p in products if field(p, 'category')}
    brands = {field(p, 'brand') for p in products if field(p, 'brand')}
    items = [('Products', len(products)), ('Categories', len(categories)),
             ('Brands', len(brands)), ('News updates', len(posts))]
    return [{'value': str(count), 'label': label} for label, count in items if count > 0]


def build_blocks(entries: list[dict]) -> list[dict]:
    stats = real_stats(entries)
    sequence = [
        blueprints.new_block('hero', {
            'title': '{{site.name}}',
            'subtitle': 'GLOBAL SOURCING · MADE RELIABLE',
            'description': 'Reliable products, transparent specifications and responsive export service.',
            'image': '{{page.cover}}',
            'button': 'Browse products', 'href': '/products/'}),
        blueprints.new_block('trust_bar', {}),
        blueprints.new_block('image_text', {
            'title': 'One partner from inquiry to delivery.',
            'subtitle': 'ABOUT US',
            'description': 'Product specifications, quality control and export documentation in one place.',
            'image': '{{page.cover}}',
            'button': 'About our company', 'href': '/about/'}),
        blueprints.new_block('stats', {'items': stats}) if stats else None,
        blueprints.new_block('collection', {
            'title': 'Featured products', 'subtitle': 'TRADE CATALOGUE',
            'source': 'product', 'limit': 4, 'columns': 4, 'image_fit': 'cover',
            'button': 'Browse all products', 'href': '/products/'}),
        blueprints.new_block('applications_band', {}),
        blueprints.new_block('features', {'title': 'Why buyers stay with us', 'items': [
            {'title': 'Verified sourcing', 'text': 'Suppliers audited against your specifications before the first sample.'},
            {'title': 'Strict quality control', 'text': 'Pre-shipment inspection with photographic evidence for every batch.'},
            {'title': 'Export documentation', 'text': 'Certification, packing lists and customs papers prepared in-house.'},
            {'title': 'Dedicated support', 'text': 'One trade desk contact from inquiry to container loading.'}]}),
        blueprints.new_block('collection', {
            'title': 'Latest updates', 'subtitle': 'COMPANY NEWS',
            'source': 'post', 'limit': 3, 'columns': 3, 'image_fit': 'cover',
            'button': 'View all news', 'href': '/news/'}),
        blueprints.new_block('cta', {
            'title': 'Send one inquiry, get a matched quote.',
            'subtitle': 'Share your specifications and target market; our trade desk replies within one business day.',
            'button': 'Request a quote', 'href': '/contact/'}),
    ]
    return [block for block in sequence if block is not None]


def main() -> int:
    env = load_env()
    password = env.get('SITEFORGE_ADMIN_PASSWORD') or ''
    if not password:
        print('ERROR: SITEFORGE_ADMIN_PASSWORD missing from .env')
        return 1
    port = (env.get('SITEFORGE_PORT') or '8787').strip() or '8787'
    base = f'http://127.0.0.1:{port}'

    with httpx.Client(base_url=base, timeout=30) as client:
        health = client.get('/api/health')
        health.raise_for_status()
        print('server:', health.json())

        login = client.post('/api/auth/login', json={'username': 'admin', 'password': password})
        if login.status_code != 200:
            print('ERROR: login failed:', login.text[:200])
            return 1
        client.headers['X-SiteForge-CSRF'] = client.get('/api/auth/me').json()['csrf']

        blocks_catalog = client.get('/api/editor/blocks').json()['blocks']
        missing = [t for t in ('trust_bar', 'applications_band') if t not in blocks_catalog]
        if missing:
            print(f'ERROR: server does not know block types {missing} - rebuild the '
                  'Docker image with the current source before running this script.')
            return 1

        entries = client.get(f'/api/sites/{SITE_ID}/entries').json()
        entry = next((e for e in entries if e.get('route') == '/'), None)
        if not entry or entry['id'] != HOME_ENTRY_ID:
            print(f'ERROR: home entry {HOME_ENTRY_ID} (route /) not found on site {SITE_ID}')
            return 1

        blocks = build_blocks(entries)
        blueprints.validate_blocks(blocks)  # fail fast locally before touching the draft

        state = client.get(f'/api/sites/{SITE_ID}/pages/{HOME_ENTRY_ID}/studio-draft').json()
        revision = int(state.get('revision') or 0)
        print(f"draft state: {state.get('state')} revision={revision}")

        put = client.put(f'/api/sites/{SITE_ID}/pages/{HOME_ENTRY_ID}/studio-draft',
                         json={'blocks': blocks, 'page_template': 'home',
                               'layout_mode': 'replace', 'expected_revision': revision})
        if put.status_code != 200:
            print('ERROR: draft write failed:', put.status_code, put.text[:300])
            return 1
        next_revision = put.json()['revision']
        print(f'draft saved: revision={next_revision} blocks={len(blocks)} '
              f'types={[b["type"] for b in blocks]}')

        publish = client.post(f'/api/sites/{SITE_ID}/pages/{HOME_ENTRY_ID}/studio-draft/publish',
                              json={'expected_revision': next_revision})
        if publish.status_code != 200:
            print('ERROR: publish failed:', publish.status_code, publish.text[:300])
            return 1
        print('published to content library:', publish.json().get('message'))

        before = {b['id'] for b in client.get(f'/api/sites/{SITE_ID}/builds').json()}
        key = 'task13b-home-' + uuid.uuid4().hex[:12]
        job = client.post(f'/api/sites/{SITE_ID}/builds', json={'mirror_images': False},
                          headers={'Idempotency-Key': key})
        if job.status_code != 202:
            print('ERROR: build submit failed:', job.status_code, job.text[:300])
            return 1
        print('build queued:', job.json())

        build_id = None
        for _ in range(120):
            time.sleep(2)
            rows = client.get(f'/api/sites/{SITE_ID}/builds').json()
            fresh = [b for b in rows if b['id'] not in before]
            done = [b for b in fresh if b['status'] == 'complete']
            if done:
                build_id = done[0]['id']
                break
            failed = [b for b in fresh if b['status'] in ('failed', 'error')]
            if failed:
                print('ERROR: build failed:', json.dumps(failed[0], ensure_ascii=False)[:400])
                return 1
        if not build_id:
            print('ERROR: build did not complete in time')
            return 1
        print('build complete:', build_id)

        report = client.get(f'/api/sites/{SITE_ID}/builds/{build_id}/report').json()
        print('report: pages=%s errors=%s broken_links=%s missing_local_images=%s'
              % (report.get('pages'), len(report.get('errors') or []),
                 len(report.get('broken_links') or []), len(report.get('missing_local_images') or [])))
        if report.get('errors') or report.get('broken_links'):
            print('ERROR: build report is not clean')
            return 1

        page = client.get(f'/preview/{SITE_ID}/{build_id}/')
        page.raise_for_status()
        html = page.text
        # render_blocks always emits class="sf-section sf-{type}" as the opening
        # attribute, so this is an exact section-level check.
        absent = [t for t in ('hero', 'trust_bar', 'image_text', 'stats',
                              'applications_band', 'features', 'cta')
                  if f'class="sf-section sf-{t}"' not in html]
        if html.count('class="sf-section sf-collection"') < 2:
            absent.append('collection x2')
        gone = [m for m in ('sf-tp-hero', 'sf-tp-trustbar', 'sf-tp-two-col',
                            'sf-tp-applications', 'sf-tp-feature-grid', 'sf-tp-cta') if m in html]
        print('block sections missing:', absent or 'none')
        print('template markers still present:', gone or 'none')
        if absent or gone:
            print('ERROR: homepage is not block-driven as expected')
            return 1
        print('PASS: homepage is now block-driven (layout_mode=replace), build clean.')
        return 0


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