# coding=utf-8
"""
@Time   :2026/6/24
@Author :tao.chen
"""
import re

from common.logging import logger

_APP_ID_RE = re.compile(r"Submitted application (\S+)")
_TRACKING_URL_RE = re.compile(r"tracking URL:\s+(\S+)")


def parse_spark_submit_output(stderr: str) -> tuple[str, str | None]:
    """Extract (application_id, tracking_url) from spark-submit stderr.

    Raises ValueError if no application_id can be found.
    """
    app_match = _APP_ID_RE.search(stderr)
    if app_match:
        application_id = app_match.group(1)
    else:
        url_match = _TRACKING_URL_RE.search(stderr)
        if not url_match:
            logger.error("Could not find application_id in spark-submit output")
            raise ValueError("Could not find application_id in spark-submit output")
        # tracking URL is of the form http://rm:8088/proxy/application_xxx/
        tracking_url = url_match.group(1)
        tail = tracking_url.rstrip("/").rsplit("/", 1)[-1]
        if not tail.startswith("application_"):
            logger.error("Could not find application_id in spark-submit output (bad tracking URL)")
            raise ValueError("Could not find application_id in spark-submit output")
        application_id = tail

    url_match = _TRACKING_URL_RE.search(stderr)
    tracking_url = url_match.group(1) if url_match else None
    logger.debug(
        f"parse_spark_submit_output -> application_id={application_id} "
        f"tracking_url={tracking_url}"
    )
    return application_id, tracking_url