vyta/sync/vyta-sync/sync.py

import json
import os
import subprocess
import sys
import urllib.error
import urllib.request
from datetime import datetime, timezone


BASE_DIR = os.path.dirname(
    os.path.abspath(__file__)
)

COLLECTOR = os.path.join(
    BASE_DIR,
    "collector.py",
)

NORMALIZER = os.path.join(
    BASE_DIR,
    "normalizer.py",
)


LOG_DIR = os.path.join(
    BASE_DIR,
    "logs",
)

LOG_FILE = os.path.join(
    LOG_DIR,
    "sync.log",
)

STATUS_FILE = os.environ.get(
    "VYTA_SYNC_STATUS_PATH",
    "/data/sync-status.json",
)


def utc_now():
    return datetime.now(
        timezone.utc
    ).isoformat()


def log(message: str):
    os.makedirs(
        LOG_DIR,
        exist_ok=True,
    )

    timestamp = datetime.now().strftime(
        "%Y-%m-%d %H:%M:%S"
    )

    line = f"[{timestamp}] {message}"

    print(
        line,
        flush=True,
    )

    with open(
        LOG_FILE,
        "a",
        encoding="utf-8",
    ) as file:
        file.write(
            line + "\n"
        )


def read_status():
    try:
        with open(
            STATUS_FILE,
            "r",
            encoding="utf-8",
        ) as file:
            value = json.load(
                file
            )

        return (
            value
            if isinstance(
                value,
                dict,
            )
            else {}
        )

    except (
        FileNotFoundError,
        json.JSONDecodeError,
    ):
        return {}


def write_status(
    **updates,
):
    existing = read_status()

    result = {
        **existing,
        **updates,
    }

    directory = os.path.dirname(
        STATUS_FILE
    )

    if directory:
        os.makedirs(
            directory,
            exist_ok=True,
        )

    temporary_file = (
        STATUS_FILE +
        ".tmp"
    )

    with open(
        temporary_file,
        "w",
        encoding="utf-8",
    ) as file:
        json.dump(
            result,
            file,
            ensure_ascii=False,
            indent=2,
        )

    os.replace(
        temporary_file,
        STATUS_FILE,
    )


def run_script(
    script_path: str
):
    name = os.path.basename(
        script_path
    )

    log(
        f"Startar {name}"
    )

    result = subprocess.run(
        [
            sys.executable,
            script_path,
        ],
        cwd=BASE_DIR,
        stdin=None,
        stdout=None,
        stderr=None,
    )

    if result.returncode != 0:
        raise RuntimeError(
            f"{name} misslyckades "
            f"med exit code "
            f"{result.returncode}"
        )

    log(
        f"{name} klar"
    )




def notify_vyta():
    web_url = os.environ.get(
        "VYTA_WEB_URL",
        "",
    ).strip().rstrip("/")

    secret = os.environ.get(
        "VYTA_INTERNAL_SYNC_SECRET",
        "",
    ).strip()

    if not web_url:
        log(
            "Notifiering hoppas över: "
            "VYTA_WEB_URL saknas"
        )
        return

    if not secret:
        log(
            "Notifiering hoppas över: "
            "VYTA_INTERNAL_SYNC_SECRET saknas"
        )
        return

    url = (
        web_url +
        "/api/notifications/evaluate"
    )

    request = urllib.request.Request(
        url,
        data=b"{}",
        method="POST",
        headers={
            "Authorization":
                f"Bearer {secret}",
            "Content-Type":
                "application/json",
        },
    )

    try:
        with urllib.request.urlopen(
            request,
            timeout=15,
        ) as response:
            body = (
                response
                .read()
                .decode(
                    "utf-8",
                    errors="replace",
                )
            )

            log(
                "Notifieringsutvärdering klar "
                f"(HTTP {response.status}): "
                f"{body}"
            )

    except urllib.error.HTTPError as error:
        body = (
            error
            .read()
            .decode(
                "utf-8",
                errors="replace",
            )
        )

        log(
            "VARNING: "
            "Notifieringsutvärdering "
            f"gav HTTP {error.code}: "
            f"{body}"
        )

    except Exception as error:
        log(
            "VARNING: "
            "Notifieringsutvärdering "
            f"misslyckades: {error}"
        )


def main():
    started_at = utc_now()

    write_status(
        status="syncing",
        startedAt=started_at,
        error=None,
    )

    log(
        "=" * 60
    )

    log(
        "VYTA SYNC START"
    )

    try:
        run_script(
            COLLECTOR
        )

        run_script(
            NORMALIZER
        )

        notify_vyta()

        completed_at = utc_now()

        write_status(
            status="idle",
            startedAt=started_at,
            completedAt=completed_at,
            lastSuccessfulSync=completed_at,
            error=None,
        )

        log(
            "VYTA SYNC KLAR"
        )

    except KeyboardInterrupt:
        write_status(
            status="idle",
            error="Sync avbruten",
        )

        log(
            "VYTA SYNC AVBRUTEN"
        )

        sys.exit(
            130
        )

    except Exception as error:
        write_status(
            status="error",
            completedAt=utc_now(),
            error=str(
                error
            ),
        )

        log(
            f"SYNC MISSLYCKADES: "
            f"{error}"
        )

        sys.exit(
            1
        )

    finally:
        log(
            "=" * 60
        )


if __name__ == "__main__":
    main()