sync_processview.py

import argparse
import filecmp
import os
import shutil
import subprocess
import sys
from collections import defaultdict
from dataclasses import dataclass
from datetime import datetime
from pathlib import Path
from typing import Optional


SCRIPT_DIR = Path(__file__).resolve().parent
ROOT_DIR = SCRIPT_DIR.parent
DEFAULT_SOURCE_DIR = ROOT_DIR / "SOURCE FILES"
DEFAULT_JSON_DIR = ROOT_DIR / "JSON FILES"
DEFAULT_TRANSLATIONS_SUBDIR = Path("hmi") / "translations"
FALLBACK_LANGUAGE_FILE = "en.json"
SUPPORTED_LANGUAGE_FILES = {
    "bg.json",
    "da.json",
    "de.json",
    "en.json",
    "fr.json",
    "nl.json",
    "uk.json",
}

PROJECT_TARGETS = {
    "AVA": "ava",
    "CAR": "crevin",
    "DSM": "dsm",
    "UFA": "ufa",
    "VILO": "vilo",
    "SUR": "sursee",
}


@dataclass(frozen=True)
class ExportBundle:
    bundle_name: str
    project_name: str
    source_csv: Path
    json_dir: Path
    target_dir: Path
    target_rel: Path


@dataclass(frozen=True)
class RepoPreparation:
    effective_skip_pull: bool
    temporary_stash_ref: Optional[str] = None


def parse_args():
    parser = argparse.ArgumentParser(
        description="Generate JSON files and sync them into the ProcessView translations repo."
    )
    parser.add_argument(
        "--processview-repo",
        help="Path to the local processview git repository. Defaults to PROCESSVIEW_REPO or common clone locations.",
    )
    parser.add_argument(
        "--source-dir",
        default=str(DEFAULT_SOURCE_DIR),
        help="Directory that contains the source CSV files.",
    )
    parser.add_argument(
        "--json-dir",
        default=str(DEFAULT_JSON_DIR),
        help="Directory that contains the generated JSON folders.",
    )
    parser.add_argument(
        "--translations-subdir",
        default=str(DEFAULT_TRANSLATIONS_SUBDIR),
        help="Path inside the processview repo where translation folders live.",
    )
    parser.add_argument(
        "--project",
        action="append",
        help="Restrict the sync to one project code. Repeat the flag for multiple projects.",
    )
    parser.add_argument(
        "--skip-generate",
        action="store_true",
        help="Skip running JsonGenerator.py before syncing.",
    )
    parser.add_argument(
        "--skip-pull",
        action="store_true",
        help="Skip git pull --ff-only before copying translations.",
    )
    push_group = parser.add_mutually_exclusive_group()
    push_group.add_argument(
        "--push",
        dest="push",
        action="store_true",
        help="Stage, commit and push the updated translation folders after syncing. This is the default.",
    )
    push_group.add_argument(
        "--no-push",
        dest="push",
        action="store_false",
        help="Sync translation files without staging, committing or pushing.",
    )
    parser.set_defaults(push=True)
    parser.add_argument(
        "--commit-message",
        help="Commit message to use together with --push.",
    )
    parser.add_argument(
        "--allow-dirty",
        action="store_true",
        help="Allow copying into a repo that already has local changes. Pull is still blocked on a dirty repo.",
    )
    parser.add_argument(
        "--prune",
        action="store_true",
        help="Remove supported language files from the target folder when they are missing in the generated JSON folder.",
    )
    parser.add_argument(
        "--dry-run",
        action="store_true",
        help="Preview the sync without running the generator, pulling, copying or pushing.",
    )
    return parser.parse_args()


def resolve_processview_repo(cli_value):
    candidates = []

    if cli_value:
        candidates.append(Path(cli_value))

    env_value = os.environ.get("PROCESSVIEW_REPO")
    if env_value:
        candidates.append(Path(env_value))

    home = Path.home()
    candidates.extend(
        [
            home / "Documents" / "GitHub" / "processview",
            home / "Documents" / "Git" / "processview",
            home / "Documents" / "Git" / "Proccesview",
        ]
    )

    for candidate in candidates:
        repo_path = candidate.expanduser().resolve()
        if (repo_path / ".git").exists():
            return repo_path

    searched = "\n".join(f" - {candidate.expanduser()}" for candidate in candidates)
    raise RuntimeError(
        "Could not locate the processview repository.\n"
        "Use --processview-repo or set PROCESSVIEW_REPO.\n"
        f"Searched:\n{searched}"
    )


def run_command(command, cwd=None, dry_run=False, env=None):
    printable = " ".join(f'"{part}"' if " " in str(part) else str(part) for part in command)
    if dry_run:
        print(f"[dry-run] {printable}")
        return ""

    completed = subprocess.run(
        command,
        cwd=cwd,
        capture_output=True,
        text=True,
        env=env,
    )
    if completed.returncode != 0:
        raise RuntimeError(
            f"Command failed ({completed.returncode}): {printable}\n"
            f"stdout:\n{completed.stdout}\n"
            f"stderr:\n{completed.stderr}"
        )
    return completed.stdout.strip()


def get_repo_status(repo_path):
    return run_command(["git", "-C", str(repo_path), "status", "--porcelain"])


def get_repo_status_entries(repo_path):
    status = get_repo_status(repo_path)
    if not status:
        return []

    entries = []
    for line in status.splitlines():
        state = line[:2]
        path = line[3:]
        if " -> " in path:
            path = path.split(" -> ", 1)[1]
        entries.append((state, Path(path)))
    return entries


def format_dirty_paths(paths, max_items=12):
    visible_paths = [path.as_posix() for path in paths[:max_items]]
    lines = [f" - {path}" for path in visible_paths]
    remaining = len(paths) - len(visible_paths)
    if remaining > 0:
        lines.append(f" - ... and {remaining} more")
    return "\n".join(lines)


def is_path_in_targets(path, target_dirs):
    posix_path = path.as_posix()
    for target_dir in target_dirs:
        target_prefix = target_dir.as_posix().rstrip("/")
        if posix_path == target_prefix or posix_path.startswith(target_prefix + "/"):
            return True
    return False


def get_upstream_state(repo_path):
    try:
        run_command(["git", "-C", str(repo_path), "rev-parse", "--verify", "@{upstream}"])
    except RuntimeError:
        return None

    run_command(["git", "-C", str(repo_path), "fetch", "--quiet"])
    counts = run_command(
        ["git", "-C", str(repo_path), "rev-list", "--left-right", "--count", "@{upstream}...HEAD"]
    )
    behind, ahead = map(int, counts.split())
    return {"behind": behind, "ahead": ahead}


def stash_target_changes(repo_path, bundles):
    target_paths = sorted({bundle.target_rel.as_posix() for bundle in bundles})
    timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
    stash_message = f"sync_processview auto-stash {timestamp}"
    run_command(
        [
            "git",
            "-C",
            str(repo_path),
            "stash",
            "push",
            "--include-untracked",
            "-m",
            stash_message,
            "--",
            *target_paths,
        ]
    )
    stash_ref = run_command(
        ["git", "-C", str(repo_path), "stash", "list", "-1", "--format=%gd"]
    )
    if not stash_ref:
        raise RuntimeError(
            "The script tried to auto-stash the translation changes, but git did not return a stash reference."
        )

    print(
        "Temporarily stashed local translation changes to allow pulling the latest ProcessView updates "
        f"({stash_ref})."
    )
    return stash_ref


def drop_stash(repo_path, stash_ref):
    run_command(["git", "-C", str(repo_path), "stash", "drop", stash_ref])
    print(f"Removed temporary stash {stash_ref}.")


def prepare_repo_for_sync(repo_path, bundles, allow_dirty, skip_pull, dry_run):
    if dry_run:
        return RepoPreparation(effective_skip_pull=skip_pull)

    status_entries = get_repo_status_entries(repo_path)
    if not status_entries:
        return RepoPreparation(effective_skip_pull=skip_pull)

    dirty_paths = [path for _, path in status_entries]
    target_dirs = [bundle.target_rel for bundle in bundles]
    dirty_in_targets_only = all(
        is_path_in_targets(path, target_dirs) for path in dirty_paths
    )

    effective_skip_pull = skip_pull
    temporary_stash_ref = None
    if not effective_skip_pull:
        if not dirty_in_targets_only:
            raise RuntimeError(
                "The processview repo has local changes outside the translation folders selected for sync, "
                "so git pull --ff-only would be unsafe.\n"
                f"Changed files:\n{format_dirty_paths(dirty_paths)}\n"
                "Commit or stash those changes, or rerun with --skip-pull --allow-dirty after reviewing them."
            )

        upstream_state = get_upstream_state(repo_path)
        if upstream_state and upstream_state["behind"] == 0:
            print(
                "ProcessView repo has local changes only in the selected translation folders; "
                "skipping git pull because the branch is not behind upstream."
            )
            effective_skip_pull = True
        elif upstream_state and upstream_state["behind"] > 0:
            temporary_stash_ref = stash_target_changes(repo_path, bundles)
        else:
            behind_note = ""
            if upstream_state is None:
                behind_note = "The upstream status could not be verified automatically."
            else:
                behind_note = (
                    f"The local branch is behind upstream by {upstream_state['behind']} commit(s)."
                )
            raise RuntimeError(
                "The processview repo has local changes, so git pull --ff-only would be unsafe.\n"
                f"{behind_note}\n"
                f"Changed files:\n{format_dirty_paths(dirty_paths)}\n"
                "Commit or stash those changes, or rerun with --skip-pull --allow-dirty after reviewing them."
            )

    if not allow_dirty and not dirty_in_targets_only:
        raise RuntimeError(
            "The processview repo has local changes.\n"
            "Rerun with --allow-dirty if you intentionally want to copy only the translation updates."
        )

    return RepoPreparation(
        effective_skip_pull=effective_skip_pull,
        temporary_stash_ref=temporary_stash_ref,
    )


def discover_bundles(source_dir, json_dir, repo_path, translations_subdir, selected_projects):
    grouped = defaultdict(list)

    for source_csv in sorted(source_dir.glob("*.csv")):
        bundle_name = source_csv.stem
        project_name = bundle_name.split("_", 1)[0].upper()

        if selected_projects and project_name not in selected_projects:
            continue

        target_name = PROJECT_TARGETS.get(project_name)
        if not target_name:
            print(f"Skipping {bundle_name}: no target folder mapping for project {project_name}.")
            continue

        target_rel = translations_subdir / target_name
        bundle = ExportBundle(
            bundle_name=bundle_name,
            project_name=project_name,
            source_csv=source_csv,
            json_dir=json_dir / bundle_name,
            target_dir=repo_path / target_rel,
            target_rel=target_rel,
        )
        grouped[project_name].append(bundle)

    duplicates = {project: bundles for project, bundles in grouped.items() if len(bundles) > 1}
    if duplicates:
        lines = ["Found multiple CSV files for the same project. Keep only one active export per project:"]
        for project, bundles in sorted(duplicates.items()):
            names = ", ".join(bundle.bundle_name for bundle in bundles)
            lines.append(f" - {project}: {names}")
        raise RuntimeError("\n".join(lines))

    bundles = sorted(
        (bundle_list[0] for bundle_list in grouped.values()),
        key=lambda bundle: (bundle.project_name, bundle.bundle_name),
    )

    if selected_projects:
        found_projects = {bundle.project_name for bundle in bundles}
        missing_projects = sorted(selected_projects - found_projects)
        if missing_projects:
            raise RuntimeError(
                "No active CSV export found for: " + ", ".join(missing_projects)
            )

    if not bundles:
        raise RuntimeError(f"No CSV files found in {source_dir}.")

    return bundles


def build_copy_plan(source_dir, target_dir, prune):
    all_source_files = {
        path.name: path for path in sorted(source_dir.glob("*.json")) if path.is_file()
    }
    source_files = {
        filename: path
        for filename, path in all_source_files.items()
        if filename in SUPPORTED_LANGUAGE_FILES
    }

    if not source_files:
        raise RuntimeError(f"No JSON files found in {source_dir}.")

    ignored_source_files = sorted(set(all_source_files) - set(source_files))
    effective_source_files = dict(source_files)
    fallback_files = []

    if FALLBACK_LANGUAGE_FILE in source_files:
        fallback_source = source_files[FALLBACK_LANGUAGE_FILE]
        for filename in sorted(SUPPORTED_LANGUAGE_FILES):
            if filename not in effective_source_files:
                effective_source_files[filename] = fallback_source
                fallback_files.append(filename)

    target_files = {
        path.name: path
        for path in sorted(target_dir.glob("*.json"))
        if path.is_file() and path.name in SUPPORTED_LANGUAGE_FILES
    }

    files_to_copy = []
    unchanged_files = []
    files_to_remove = (
        sorted(target_files.keys() - effective_source_files.keys()) if prune else []
    )

    for filename, source_file in sorted(effective_source_files.items()):
        target_file = target_dir / filename
        if target_file.exists() and filecmp.cmp(source_file, target_file, shallow=False):
            unchanged_files.append(filename)
            continue
        files_to_copy.append(filename)

    return files_to_copy, files_to_remove, unchanged_files, ignored_source_files, fallback_files, effective_source_files


def sync_bundle(bundle, dry_run, prune):
    if not bundle.json_dir.exists():
        raise RuntimeError(
            f"Generated JSON folder not found for {bundle.bundle_name}: {bundle.json_dir}"
        )

    bundle.target_dir.mkdir(parents=True, exist_ok=True)

    (
        files_to_copy,
        files_to_remove,
        unchanged_files,
        ignored_source_files,
        fallback_files,
        effective_source_files,
    ) = build_copy_plan(
        bundle.json_dir, bundle.target_dir, prune=prune
    )

    print(f"{bundle.bundle_name} -> {bundle.target_rel.as_posix()}")
    print(
        f"  copy: {len(files_to_copy)} file(s), "
        f"remove: {len(files_to_remove)} file(s), "
        f"unchanged: {len(unchanged_files)} file(s)"
    )
    if ignored_source_files:
        print(f"  ignored unsupported language file(s): {', '.join(ignored_source_files)}")
    if fallback_files:
        print(f"  fallback from {FALLBACK_LANGUAGE_FILE}: {', '.join(fallback_files)}")

    if dry_run:
        for filename in files_to_copy:
            source_label = effective_source_files[filename].name
            if source_label == filename:
                print(f"  [dry-run] copy {filename}")
            else:
                print(f"  [dry-run] copy {filename} from {source_label}")
        for filename in files_to_remove:
            print(f"  [dry-run] remove {filename}")
        return bool(files_to_copy or files_to_remove)

    for filename in files_to_remove:
        (bundle.target_dir / filename).unlink()

    for filename in files_to_copy:
        shutil.copy2(effective_source_files[filename], bundle.target_dir / filename)

    return bool(files_to_copy or files_to_remove)


def build_commit_message(bundles):
    names = ", ".join(bundle.bundle_name for bundle in bundles)
    return f"Update ProcessView translations: {names}"


def stage_commit_and_push(repo_path, bundles, commit_message, dry_run):
    target_paths = sorted({bundle.target_rel.as_posix() for bundle in bundles})
    status = run_command(
        ["git", "-C", str(repo_path), "status", "--porcelain", "--", *target_paths]
    )

    if not status:
        print("No git changes detected in the translation folders.")
        return

    if dry_run:
        print("[dry-run] git add/commit/push would run for:")
        for target_path in target_paths:
            print(f"  - {target_path}")
        return

    run_command(["git", "-C", str(repo_path), "add", "--", *target_paths])

    staged = run_command(
        ["git", "-C", str(repo_path), "diff", "--cached", "--name-only", "--", *target_paths]
    )
    if not staged:
        print("Nothing was staged after git add; skipping commit.")
        return

    final_message = commit_message or build_commit_message(bundles)
    run_command(["git", "-C", str(repo_path), "commit", "-m", final_message])
    run_command(["git", "-C", str(repo_path), "push"])
    print("Changes committed and pushed.")


def run_generator(dry_run, source_dir, json_dir):
    generator_path = SCRIPT_DIR / "JsonGenerator.py"
    env = os.environ.copy()
    env["JSON_GENERATOR_SOURCE_DIR"] = str(source_dir)
    env["JSON_GENERATOR_OUTPUT_DIR"] = str(json_dir)
    if dry_run:
        print(
            f"[dry-run] JSON_GENERATOR_SOURCE_DIR={source_dir} "
            f"JSON_GENERATOR_OUTPUT_DIR={json_dir} "
            f"{sys.executable} {generator_path}"
        )
        return

    print("Running JsonGenerator.py...")
    run_command([sys.executable, str(generator_path)], cwd=SCRIPT_DIR, env=env)


def main():
    args = parse_args()

    source_dir = Path(args.source_dir).expanduser().resolve()
    json_dir = Path(args.json_dir).expanduser().resolve()
    translations_subdir = Path(args.translations_subdir)
    selected_projects = {project.upper() for project in args.project or []}

    if not source_dir.exists():
        raise RuntimeError(f"Source directory not found: {source_dir}")

    if not json_dir.exists() and (args.skip_generate or args.dry_run):
        raise RuntimeError(f"JSON directory not found: {json_dir}")

    repo_path = resolve_processview_repo(args.processview_repo)

    print(f"ProcessView repo: {repo_path}")
    print(f"Source CSV dir:   {source_dir}")
    print(f"JSON output dir:  {json_dir}")

    if args.dry_run:
        print("Dry-run mode: generation, pull, copy and push will be skipped.")

    if not args.skip_generate:
        run_generator(dry_run=args.dry_run, source_dir=source_dir, json_dir=json_dir)

    bundles = discover_bundles(
        source_dir=source_dir,
        json_dir=json_dir,
        repo_path=repo_path,
        translations_subdir=translations_subdir,
        selected_projects=selected_projects,
    )

    print("Bundles to sync:")
    for bundle in bundles:
        print(f" - {bundle.bundle_name} ({bundle.project_name})")

    prep = prepare_repo_for_sync(
        repo_path=repo_path,
        bundles=bundles,
        allow_dirty=args.allow_dirty,
        skip_pull=args.skip_pull,
        dry_run=args.dry_run,
    )

    run_succeeded = False
    drop_temporary_stash = False
    try:
        if not prep.effective_skip_pull and not args.dry_run:
            print("Running git pull --ff-only...")
            run_command(["git", "-C", str(repo_path), "pull", "--ff-only"])
        elif args.dry_run and not prep.effective_skip_pull:
            print("[dry-run] git pull --ff-only")

        changed_bundles = []
        for bundle in bundles:
            if sync_bundle(bundle, dry_run=args.dry_run, prune=args.prune):
                changed_bundles.append(bundle)

        if not changed_bundles:
            print("All translation folders are already up to date.")
            run_succeeded = True
            return

        if args.push:
            stage_commit_and_push(
                repo_path=repo_path,
                bundles=changed_bundles,
                commit_message=args.commit_message,
                dry_run=args.dry_run,
            )

        drop_temporary_stash = True
        run_succeeded = True
    finally:
        if prep.temporary_stash_ref and not args.dry_run:
            if run_succeeded and drop_temporary_stash:
                drop_stash(repo_path, prep.temporary_stash_ref)
            elif run_succeeded:
                print(
                    "No new translation changes were produced after pulling ProcessView. "
                    f"The temporary stash is still available at {prep.temporary_stash_ref}."
                )
            else:
                print(
                    "The sync stopped after creating a temporary stash. "
                    f"Your previous translation changes are still available at {prep.temporary_stash_ref}.",
                    file=sys.stderr,
                )


if __name__ == "__main__":
    try:
        main()
    except Exception as exc:
        print(f"ERROR: {exc}", file=sys.stderr)
        sys.exit(1)