moonwell-client/upload_to_s3.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-

import hashlib
import json
import os
import sys
from pathlib import Path

import boto3
from botocore.config import Config
from botocore.exceptions import ClientError

S3_PREFIX = "World of Warcraft"
MANIFEST_S3_KEY = "manifest.json"


def compute_build_hash(files: list[dict]) -> str:
    lines = [
        f'{item["path"]}:{item["size"]}:{item["sha256"]}'
        for item in sorted(files, key=lambda item: item["path"])
    ]
    return hashlib.sha256("\n".join(lines).encode("utf-8")).hexdigest()


def merge_manifests(base_manifest: dict, staging_manifest: dict) -> dict:
    files_by_path = {
        item["path"]: item
        for item in base_manifest.get("files", [])
    }
    for item in staging_manifest.get("files", []):
        files_by_path[item["path"]] = item

    files = [files_by_path[path] for path in sorted(files_by_path)]
    return {
        "buildHash": compute_build_hash(files),
        "files": files,
    }


def load_dotenv(env_path: Path) -> dict[str, str]:
    values: dict[str, str] = {}
    if not env_path.exists():
        return values
    for raw_line in env_path.read_text(encoding="utf-8").splitlines():
        line = raw_line.strip()
        if not line or line.startswith("#") or "=" not in line:
            continue
        key, value = line.split("=", 1)
        key = key.strip()
        value = value.strip()
        if key.startswith("export "):
            key = key.removeprefix("export ").strip()
        if not key:
            continue
        if len(value) >= 2 and value[0] == value[-1] and value[0] in {"'", '"'}:
            value = value[1:-1]
        values[key] = value
    return values


def get_env(dotenv: dict[str, str], key: str) -> str:
    value = os.environ.get(key) or dotenv.get(key)
    if not value:
        print(f"ERROR: {key} is not set (env or .env)", file=sys.stderr)
        sys.exit(1)
    return value


def main() -> int:
    project_root = Path(__file__).resolve().parent
    dotenv = load_dotenv(project_root / ".env")

    bucket = get_env(dotenv, "AWS_BUCKET")
    endpoint = get_env(dotenv, "AWS_ENDPOINT")
    region = get_env(dotenv, "AWS_DEFAULT_REGION")
    access_key = get_env(dotenv, "AWS_ACCESS_KEY_ID")
    secret_key = get_env(dotenv, "AWS_SECRET_ACCESS_KEY")
    use_path_style = get_env(dotenv, "AWS_USE_PATH_STYLE_ENDPOINT").lower() == "true"

    s3 = boto3.client(
        "s3",
        endpoint_url=endpoint,
        region_name=region,
        aws_access_key_id=access_key,
        aws_secret_access_key=secret_key,
        config=Config(s3={"addressing_style": "path"}) if use_path_style else None,
    )

    manifest_path = project_root / "manifest.json"
    if not manifest_path.exists():
        print("ERROR: manifest.json not found", file=sys.stderr)
        return 1

    try:
        staging_manifest = json.loads(manifest_path.read_text(encoding="utf-8"))
    except (OSError, json.JSONDecodeError) as error:
        print(f"ERROR: unable to read local manifest.json: {error}", file=sys.stderr)
        return 1

    try:
        response = s3.get_object(Bucket=bucket, Key=MANIFEST_S3_KEY)
        production_manifest = json.loads(response["Body"].read())
    except ClientError as error:
        error_code = error.response.get("Error", {}).get("Code")
        if error_code not in {"NoSuchKey", "404"}:
            print(f"ERROR: unable to read production manifest: {error}", file=sys.stderr)
            return 1
        production_manifest = {"files": []}
    except (UnicodeDecodeError, json.JSONDecodeError) as error:
        print(f"ERROR: invalid production manifest: {error}", file=sys.stderr)
        return 1

    merged_manifest = merge_manifests(production_manifest, staging_manifest)
    print(
        "Manifest merge: "
        f"production={len(production_manifest.get('files', []))}, "
        f"staging={len(staging_manifest.get('files', []))}, "
        f"result={len(merged_manifest['files'])}"
    )

    dist_dir = project_root / "dist"
    if not dist_dir.exists():
        print(f"ERROR: dist/ directory not found. Run build first.", file=sys.stderr)
        return 1

    errors = False

    # Загрузка всех файлов из dist/ с сохранением относительных путей
    files = sorted(f for f in dist_dir.rglob("*") if f.is_file())
    if not files:
        print("ERROR: dist/ is empty, nothing to upload.", file=sys.stderr)
        return 1

    for local_path in files:
        rel_path = local_path.relative_to(dist_dir).as_posix()
        s3_key = f"{S3_PREFIX}/{rel_path}"
        size_mb = local_path.stat().st_size / (1024 * 1024)
        print(f"Uploading {rel_path} ({size_mb:.1f} MB) -> s3://{bucket}/{s3_key}")
        try:
            s3.upload_file(str(local_path), bucket, s3_key)
            print(f"  OK")
        except Exception as e:
            print(f"  ERROR: {e}", file=sys.stderr)
            errors = True

    # Загрузка манифеста
    if errors:
        print("\nFinished with errors; production manifest was not changed.")
        return 1

    manifest_path.write_text(
        json.dumps(merged_manifest, ensure_ascii=False, indent=2) + "\n",
        encoding="utf-8",
        newline="\n",
    )

    print(f"Uploading manifest.json -> s3://{bucket}/{MANIFEST_S3_KEY}")
    try:
        s3.upload_file(
            str(manifest_path),
            bucket,
            MANIFEST_S3_KEY,
            ExtraArgs={
                "ContentType": "application/json",
                "CacheControl": "no-cache, no-store, must-revalidate",
            },
        )
        print(f"  OK")
    except Exception as e:
        print(f"  ERROR: {e}", file=sys.stderr)
        errors = True

    if errors:
        print("\nFinished with errors.")
        return 1

    print("\nAll uploads completed successfully.")
    return 0


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