from __future__ import annotations

import argparse
import importlib
import json
from typing import Any, Callable


def _row_counts(results: Any) -> list[int | None]:
    if not isinstance(results, tuple):
        results = (results,)
    return [len(result) if hasattr(result, "__len__") else None for result in results]


def _import_callable(callable_path: str) -> Callable[..., Any]:
    module_name, separator, attr_name = callable_path.partition(":")
    if not separator or not module_name or not attr_name:
        raise ValueError(f"callable path must be '<module>:<attribute>', got {callable_path!r}")
    module = importlib.import_module(module_name)
    value = getattr(module, attr_name)
    if not callable(value):
        raise TypeError(f"{callable_path!r} resolved to a non-callable object")
    return value


def _skip_reason_from_exception(exc: Exception) -> str | None:
    if exc.__class__.__name__ == "NoOutputProductsError":
        return str(exc)
    return None


def main() -> None:
    parser = argparse.ArgumentParser(description="Run a configured Python asset job.")
    parser.add_argument("--payload-json", required=True)
    args = parser.parse_args()

    payload = json.loads(args.payload_json)
    run = _import_callable(payload["callable_path"])
    try:
        results = run(**dict(payload.get("run_kwargs") or {}))
    except Exception as exc:
        skip_reason = _skip_reason_from_exception(exc)
        if skip_reason:
            print(json.dumps({"skip_reason": skip_reason}, sort_keys=True))
            return
        raise
    print(json.dumps({"row_counts": _row_counts(results)}, sort_keys=True))


if __name__ == "__main__":
    main()