Sourcecraft Publisher

config/.env

# SourceCraft Publisher — config
# Copy to .env: cp .env.example .env

# OAuth2 token for SourceCraft (with contents write access to the target repo)
# How to get: https://sourcecraft.yandex.cloud/docs/access-tokens
SOURCECRAFT_TOKEN=

# Target repository on SourceCraft (org/repo format)
SOURCECRAFT_REPO=

# Target branch
SOURCECRAFT_BRANCH=main

# Public URL of the SourceCraft Sites deployment
# Format: https://<org>.sourcecraft.site/<repo>
SOURCECRAFT_SITE_URL=

#!/usr/bin/env python3
"""Optimize oversized raster images in-place before publishing."""
from __future__ import annotations

import argparse
import shutil
import sys
from dataclasses import dataclass
from pathlib import Path

try:
    from PIL import Image, ImageOps, UnidentifiedImageError
except ImportError as exc:
    raise SystemExit(
        'Pillow is required. Run with: '
        'uv run --with pillow python3 scripts/optimize_images.py <path> --max-kb 500'
    ) from exc


IMAGE_EXTENSIONS = {'.jpg', '.jpeg', '.png', '.webp'}
DEFAULT_MAX_KB = 500
DEFAULT_MAX_DIMENSION = 1800
MIN_DIMENSION = 640
QUALITY_VALUES = range(85, 40, -5)
RESAMPLE = getattr(getattr(Image, 'Resampling', Image), 'LANCZOS')


@dataclass
class OptimizeResult:
    status: str
    original_size: int
    final_size: int
    reason: str = ''


def iter_images(root: Path):
    paths = [root] if root.is_file() else root.rglob('*')
    for path in sorted(paths):
        if path.is_file() and path.suffix.lower() in IMAGE_EXTENSIONS:
            yield path


def output_format(path: Path) -> str:
    ext = path.suffix.lower()
    if ext in {'.jpg', '.jpeg'}:
        return 'JPEG'
    if ext == '.png':
        return 'PNG'
    return 'WEBP'


def dimension_steps(width: int, height: int, max_dimension: int):
    largest = max(width, height)
    start = min(largest, max_dimension)
    if start <= MIN_DIMENSION:
        return [start]

    steps = []
    current = start
    while current >= MIN_DIMENSION:
        steps.append(current)
        current = int(current * 0.85)
    if steps[-1] != MIN_DIMENSION:
        steps.append(MIN_DIMENSION)
    return steps


def flatten_to_rgb(image: Image.Image) -> Image.Image:
    if image.mode in {'RGBA', 'LA'} or (image.mode == 'P' and 'transparency' in image.info):
        rgba = image.convert('RGBA')
        background = Image.new('RGB', rgba.size, (255, 255, 255))
        background.paste(rgba, mask=rgba.getchannel('A'))
        return background
    return image.convert('RGB')


def save_candidate(source: Image.Image, target: Path, fmt: str, max_dimension: int, quality: int | None):
    image = source.copy()
    if max(image.size) > max_dimension:
        image.thumbnail((max_dimension, max_dimension), RESAMPLE)

    if fmt == 'JPEG':
        flatten_to_rgb(image).save(
            target,
            format='JPEG',
            quality=quality,
            optimize=True,
            progressive=True,
        )
    elif fmt == 'WEBP':
        image.save(target, format='WEBP', quality=quality, method=6)
    else:
        image.save(target, format='PNG', optimize=True, compress_level=9)


def optimize_one(path: Path, max_bytes: int, max_dimension: int) -> OptimizeResult:
    original_size = path.stat().st_size
    if original_size <= max_bytes:
        return OptimizeResult('already_small', original_size, original_size)

    tmp = path.with_name(f'.{path.name}.optimize-tmp{path.suffix}')
    best = path.with_name(f'.{path.name}.optimize-best{path.suffix}')

    try:
        with Image.open(path) as opened:
            opened.load()
            if getattr(opened, 'is_animated', False):
                return OptimizeResult('skipped', original_size, original_size, 'animated image')

            fmt = output_format(path)
            source = ImageOps.exif_transpose(opened)
            qualities: list[int | None] = [None] if fmt == 'PNG' else list(QUALITY_VALUES)
            best_size = original_size

            for max_dim in dimension_steps(*source.size, max_dimension):
                for quality in qualities:
                    save_candidate(source, tmp, fmt, max_dim, quality)
                    candidate_size = tmp.stat().st_size

                    if candidate_size < best_size:
                        shutil.copy2(tmp, best)
                        best_size = candidate_size

                    if candidate_size <= max_bytes:
                        tmp.replace(path)
                        if best.exists():
                            best.unlink()
                        return OptimizeResult('optimized', original_size, candidate_size)

            if best.exists() and best_size < original_size:
                best.replace(path)
                status = 'optimized_over_target' if best_size > max_bytes else 'optimized'
                return OptimizeResult(status, original_size, best_size)

            return OptimizeResult('kept_original', original_size, original_size, 'no smaller candidate')
    except UnidentifiedImageError:
        return OptimizeResult('skipped', original_size, original_size, 'unidentified image')
    except OSError as exc:
        return OptimizeResult('skipped', original_size, original_size, str(exc))
    finally:
        if tmp.exists():
            tmp.unlink()
        if best.exists():
            best.unlink()


def human_size(size: int) -> str:
    return f'{size / 1024:.1f}KB'


def relative(path: Path, root: Path) -> str:
    try:
        return path.relative_to(root).as_posix()
    except ValueError:
        return path.as_posix()


def main():
    parser = argparse.ArgumentParser(description='Optimize oversized web images in-place')
    parser.add_argument('path', help='Image file or directory to scan')
    parser.add_argument('--max-kb', type=int, default=DEFAULT_MAX_KB, help='Target max size per image')
    parser.add_argument(
        '--max-dimension',
        type=int,
        default=DEFAULT_MAX_DIMENSION,
        help='Initial long-edge resize limit for oversized images',
    )
    args = parser.parse_args()

    root = Path(args.path).resolve()
    if not root.exists():
        raise SystemExit(f'path not found: {root}')
    if args.max_kb <= 0:
        print('IMAGE_OPTIMIZATION_DISABLED')
        return

    max_bytes = args.max_kb * 1024
    optimized = 0
    over_target = 0
    skipped = 0

    for image_path in iter_images(root):
        result = optimize_one(image_path, max_bytes, args.max_dimension)
        rel = relative(image_path, root if root.is_dir() else root.parent)

        if result.status == 'optimized':
            optimized += 1
            print(f'IMAGE_OPTIMIZED {rel} {human_size(result.original_size)} -> {human_size(result.final_size)}')
        elif result.status == 'optimized_over_target':
            optimized += 1
            over_target += 1
            print(
                f'IMAGE_OPTIMIZED_OVER_TARGET {rel} '
                f'{human_size(result.original_size)} -> {human_size(result.final_size)}',
                file=sys.stderr,
            )
        elif result.status == 'skipped':
            skipped += 1
            print(f'IMAGE_OPTIMIZATION_SKIPPED {rel}: {result.reason}', file=sys.stderr)

    if optimized or over_target or skipped:
        print(f'IMAGE_OPTIMIZATION_DONE optimized={optimized} over_target={over_target} skipped={skipped}')


if __name__ == '__main__':
    main()

#!/usr/bin/env python3
"""Publish static artifacts to a SourceCraft Sites repository.

Usage:
    python3 scripts/publish_static.py --source <dir-or-html> --slug <name> [--date YYYY-MM-DD] [--message "msg"]

Requires env vars: SOURCECRAFT_TOKEN, SOURCECRAFT_REPO, SOURCECRAFT_SITE_URL
Optional: SOURCECRAFT_BRANCH (default: main)
"""
import argparse, os, re, shutil, subprocess, sys, tempfile
from datetime import datetime
from pathlib import Path

DEFAULT_IMAGE_MAX_KB = 500
SITES_YAML = """\
site:
  root: "."
  ref: {branch}
"""


def slugify(s: str) -> str:
    s = s.strip().lower()
    s = re.sub(r'[^a-z0-9]+', '-', s)
    s = re.sub(r'-+', '-', s).strip('-')
    return s or 'page'


def run(cmd, cwd=None):
    subprocess.run(cmd, cwd=cwd, check=True)


def pillow_available() -> bool:
    return subprocess.run(
        [sys.executable, '-c', 'import PIL'],
        stdout=subprocess.DEVNULL,
        stderr=subprocess.DEVNULL,
    ).returncode == 0


def optimize_images(target: Path, max_kb: int, keep_large_images: bool):
    if keep_large_images:
        print('IMAGE_OPTIMIZATION_SKIPPED keep-large-images')
        return
    if max_kb <= 0:
        print('IMAGE_OPTIMIZATION_DISABLED')
        return

    script = Path(__file__).resolve().with_name('optimize_images.py')
    if not script.exists():
        print(f'IMAGE_OPTIMIZATION_SKIPPED optimizer not found: {script}', file=sys.stderr)
        return

    cmd = None
    if pillow_available():
        cmd = [sys.executable, str(script), str(target), '--max-kb', str(max_kb)]
    else:
        uv = shutil.which('uv')
        if uv:
            cmd = [uv, 'run', '--quiet', '--with', 'pillow', 'python3', str(script), str(target), '--max-kb', str(max_kb)]

    if cmd:
        status = subprocess.run(cmd)
        if status.returncode != 0:
            print('IMAGE_OPTIMIZATION_SKIPPED optimizer failed; publishing originals', file=sys.stderr)
        return

    print('IMAGE_OPTIMIZATION_SKIPPED Pillow unavailable; publishing originals', file=sys.stderr)


def main():
    ap = argparse.ArgumentParser(description='Publish to SourceCraft Sites')
    ap.add_argument('--source', required=True, help='Path to built static site folder or single html file')
    ap.add_argument('--slug', required=True, help='Page slug (lowercase, hyphens)')
    ap.add_argument('--date', help='ISO date, default today')
    ap.add_argument('--message', help='Commit message')
    ap.add_argument('--image-max-kb', type=int, default=DEFAULT_IMAGE_MAX_KB, help='Target max size per raster image')
    ap.add_argument('--keep-large-images', action='store_true', help='Publish original images without optimization')
    args = ap.parse_args()

    token = os.environ['SOURCECRAFT_TOKEN']
    repo = os.environ['SOURCECRAFT_REPO']
    branch = os.environ.get('SOURCECRAFT_BRANCH', 'main')
    base_url = os.environ['SOURCECRAFT_SITE_URL'].rstrip('/')

    dt = datetime.fromisoformat(args.date) if args.date else datetime.utcnow()
    year = dt.strftime('%Y')
    year_month = dt.strftime('%Y-%m')
    slug = slugify(args.slug)
    rel = Path(year) / year_month / slug

    src = Path(args.source).resolve()
    if not src.exists():
        raise SystemExit(f'source not found: {src}')

    with tempfile.TemporaryDirectory() as td:
        repo_dir = Path(td) / 'repo'
        remote = f'https://oauth2:{token}@git.sourcecraft.dev/{repo}.git'

        # Clone — try existing branch, fall back to init if repo is empty
        try:
            run(['git', 'clone', '--branch', branch, '--depth', '1', remote, str(repo_dir)])
        except subprocess.CalledProcessError:
            # Repo might be empty or branch doesn't exist
            repo_dir.mkdir(parents=True, exist_ok=True)
            run(['git', 'init'], cwd=repo_dir)
            run(['git', 'remote', 'add', 'origin', remote], cwd=repo_dir)
            run(['git', 'checkout', '-b', branch], cwd=repo_dir)

        # Ensure .sourcecraft/sites.yaml exists
        sc_dir = repo_dir / '.sourcecraft'
        sc_dir.mkdir(exist_ok=True)
        sites_yaml = sc_dir / 'sites.yaml'
        if not sites_yaml.exists():
            sites_yaml.write_text(SITES_YAML.format(branch=branch))

        # Prepare target directory
        target = repo_dir / rel
        if target.exists():
            shutil.rmtree(target)
        target.mkdir(parents=True, exist_ok=True)

        # Copy artifact
        if src.is_dir():
            for item in src.iterdir():
                dest = target / item.name
                if item.is_dir():
                    shutil.copytree(item, dest)
                else:
                    shutil.copy2(item, dest)
        else:
            shutil.copy2(src, target / 'index.html')

        optimize_images(target, args.image_max_kb, args.keep_large_images)

        if not (target / 'index.html').exists():
            raise SystemExit('index.html not found in published artifact')

        # Commit and push
        run(['git', 'config', 'user.name', 'OpenClaw Publisher'], cwd=repo_dir)
        run(['git', 'config', 'user.email', 'publisher@openclaw.local'], cwd=repo_dir)
        run(['git', 'add', '.'], cwd=repo_dir)
        msg = args.message or f'publish: {slug} -> {rel.as_posix()}'
        status = subprocess.run(['git', 'diff', '--cached', '--quiet'], cwd=repo_dir)
        if status.returncode == 0:
            print(f'NO_CHANGES {base_url}/{rel.as_posix()}/')
            return
        run(['git', 'commit', '-m', msg], cwd=repo_dir)
        # Force push — standard pattern for SourceCraft Sites
        run(['git', 'push', '--force', 'origin', branch], cwd=repo_dir)
        print(f'{base_url}/{rel.as_posix()}/')


if __name__ == '__main__':
    main()