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

from patch_layout import PATCH_LAYOUT, normalized_path

S3_PREFIX = "World of Warcraft"
MANIFEST_S3_KEY = "manifest.json"
MANAGED_MANIFEST_S3_KEY = "moonwell-managed-manifest.json"

LEGACY_MANAGED_FILES = {
    "utils/warcraftxlhost.exe",
    "utils/warcraftxlhost.log",
    "warcraftxl.before-texture-upload-fix.dll",
}

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 normalized_path_key(path: str) -> str:
    """Match client paths using Windows filesystem semantics."""
    return normalized_path(path)


def missing_repository_patches(manifest: dict) -> list[str]:
    paths = {
        normalized_path_key(item["path"])
        for item in manifest.get("files", [])
        if isinstance(item, dict) and isinstance(item.get("path"), str)
    }
    return sorted(PATCH_LAYOUT.repository_patches - paths)


def external_graphics_patches(manifest: dict) -> list[str]:
    return sorted(
        item["path"]
        for item in manifest.get("files", [])
        if isinstance(item, dict)
        and isinstance(item.get("path"), str)
        and PATCH_LAYOUT.is_external_graphics_patch(item["path"])
    )


def is_legacy_managed_path(path: str) -> bool:
    key = normalized_path_key(path)
    return (
        key in LEGACY_MANAGED_FILES
        or key in {"wow.exe", "d3d9.dll", "warcraftxl.dll", "utils/d3d9-native.dll"}
        or key.startswith("extensions/")
        or key.startswith("data/patch-wxl.mpq/")
    )


def merge_manifests(
    base_manifest: dict,
    staging_manifest: dict,
    previous_managed_paths: list[str] | None = None,
) -> dict:
    previous_managed_keys = {
        normalized_path_key(path) for path in (previous_managed_paths or [])
    }
    files_by_path = {
        normalized_path_key(item["path"]): item
        for item in base_manifest.get("files", [])
        if not (
            normalized_path_key(item["path"]) in previous_managed_keys
            or is_legacy_managed_path(item["path"])
            or PATCH_LAYOUT.is_repository_patch(item["path"])
        )
        or PATCH_LAYOUT.is_external_graphics_patch(item["path"])
    }
    for item in staging_manifest.get("files", []):
        if PATCH_LAYOUT.is_external_graphics_patch(item["path"]):
            continue
        files_by_path[normalized_path_key(item["path"])] = item

    files = sorted(files_by_path.values(), key=lambda item: item["path"].casefold())
    return {
        "buildHash": compute_build_hash(files),
        "files": files,
    }


def find_stale_managed_paths(
    base_manifest: dict,
    staging_manifest: dict,
    previous_managed_paths: list[str] | None = None,
) -> list[str]:
    previous_managed_keys = {
        normalized_path_key(path) for path in (previous_managed_paths or [])
    }
    staging_keys = {
        normalized_path_key(item["path"])
        for item in staging_manifest.get("files", [])
    }
    return sorted(
        (
            item["path"]
            for item in base_manifest.get("files", [])
            if (
                (
                    normalized_path_key(item["path"]) in previous_managed_keys
                    or is_legacy_managed_path(item["path"])
                    or PATCH_LAYOUT.is_repository_patch(item["path"])
                )
                and not PATCH_LAYOUT.is_external_graphics_patch(item["path"])
            )
            and normalized_path_key(item["path"]) not in staging_keys
        ),
        key=str.casefold,
    )


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

    missing_repository = missing_repository_patches(staging_manifest)
    if missing_repository:
        print(
            "ERROR: local manifest is missing repository patches: "
            + ", ".join(missing_repository),
            file=sys.stderr,
        )
        return 1

    unexpected_graphics = external_graphics_patches(staging_manifest)
    if unexpected_graphics:
        print(
            "ERROR: managed manifest contains externally managed graphics patches: "
            + ", ".join(unexpected_graphics),
            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

    try:
        response = s3.get_object(Bucket=bucket, Key=MANAGED_MANIFEST_S3_KEY)
        previous_managed_manifest = json.loads(response["Body"].read())
        previous_managed_paths = [
            item["path"] for item in previous_managed_manifest.get("files", [])
        ]
    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 managed manifest: {error}", file=sys.stderr)
            return 1
        previous_managed_paths = []
    except (UnicodeDecodeError, json.JSONDecodeError, KeyError) as error:
        print(f"ERROR: invalid managed manifest: {error}", file=sys.stderr)
        return 1

    merged_manifest = merge_manifests(
        production_manifest,
        staging_manifest,
        previous_managed_paths,
    )
    stale_managed_paths = find_stale_managed_paths(
        production_manifest,
        staging_manifest,
        previous_managed_paths,
    )
    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()
        if PATCH_LAYOUT.is_external_graphics_patch(rel_path):
            print(f"Skipping externally managed graphics patch: {rel_path}")
            continue
        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

    json_content_type = "application/json"
    no_cache = "no-cache, no-store, must-revalidate"

    print(
        f"Uploading managed package inventory -> "
        f"s3://{bucket}/{MANAGED_MANIFEST_S3_KEY}"
    )
    try:
        s3.put_object(
            Bucket=bucket,
            Key=MANAGED_MANIFEST_S3_KEY,
            Body=(json.dumps(staging_manifest, ensure_ascii=False, indent=2) + "\n").encode("utf-8"),
            ContentType=json_content_type,
            CacheControl=no_cache,
        )
        print("  OK")
    except Exception as e:
        print(f"  ERROR: {e}", file=sys.stderr)
        return 1

    print(f"Uploading manifest.json -> s3://{bucket}/{MANIFEST_S3_KEY}")
    try:
        s3.put_object(
            Bucket=bucket,
            Key=MANIFEST_S3_KEY,
            Body=(json.dumps(merged_manifest, ensure_ascii=False, indent=2) + "\n").encode("utf-8"),
            ContentType=json_content_type,
            CacheControl=no_cache,
        )
        print(f"  OK")
    except Exception as e:
        print(f"  ERROR: {e}", file=sys.stderr)
        return 1

    if stale_managed_paths:
        print(f"Deleting {len(stale_managed_paths)} stale managed object(s)...")
        for relative_path in stale_managed_paths:
            s3_key = f"{S3_PREFIX}/{relative_path.replace(chr(92), '/')}"
            try:
                s3.delete_object(Bucket=bucket, Key=s3_key)
                print(f"  Deleted {s3_key}")
            except Exception as e:
                print(f"  ERROR deleting {s3_key}: {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())