plain2code.py

import logging
import logging.config
import os
import signal
import sys
import threading
from pathlib import Path
from typing import Optional

import yaml
from liquid2.exceptions import TemplateNotFoundError

import codeplain_REST_api as codeplain_api
import file_utils
import plain_file
import plain_modules
import plain_spec
from cli_output import print_dry_run_output, print_exit_summary, print_status
from event_bus import EventBus
from module_renderer import ModuleRenderer
from partial_rendering import get_plain_module_render_state, get_render_choices
from plain2code_arguments import parse_arguments
from plain2code_console import console
from plain2code_events import RenderFailed
from plain2code_exceptions import (
    ConflictingRequirements,
    GitNotInstalledError,
    ImportedModuleWithFunctionalitiesError,
    InvalidAPIKey,
    InvalidFridArgument,
    InvalidModuleArchiveError,
    MissingAPIKey,
    MissingFunctionalitiesError,
    MissingPreviousFunctionalitiesError,
    MissingResource,
    ModuleDoesNotExistError,
    NetworkConnectionError,
    OutdatedClientVersion,
    PlainSyntaxError,
    RenderCancelledError,
    RenderingCreditBalanceTooLow,
    UnsupportedBase64Content,
    UnsupportedResourceType,
)
from plain2code_logger import (
    LOGGER_NAME,
    NARRATION_HANDLER_NAME,
    CrashLogHandler,
    ElapsedTimeFormatter,
    IndentedFormatter,
    LoggingHandler,
    dump_crash_logs,
    stop_narrating,
)
from plain2code_state import RunState
from plain2code_telemetry import capture_crash, initialize_telemetry
from system_config import system_config
from tui.plain2code_tui import Plain2CodeTUI
from tui.plain_module_render_choice_tui import PlainModuleRenderChoiceTUI

DEFAULT_TEMPLATE_DIRS = "standard_template_library"

# Search-order guidance appended to lookup failures raised by the parser, which only
# reports the missing file. The wording is the CLI's: it knows what the template
# directories are and how the user changes them.
TEMPLATE_NOT_FOUND_HINT = f"""
The required template could not be found. Templates are searched in the following order (highest to lowest precedence):

    1. The directory containing your .plain file
    2. The directory specified by --template-dir (if provided)
    3. The built-in '{DEFAULT_TEMPLATE_DIRS}' directory

Please ensure that the missing template exists in one of these locations, or specify the correct --template-dir if using custom templates.
"""

RESOURCE_NOT_FOUND_HINT = f""" Resource files are searched in the following order (highest to lowest precedence):

1. The directory containing your .plain file
2. The directory specified by --template-dir (if provided)
3. The built-in '{DEFAULT_TEMPLATE_DIRS}' directory

Please ensure that the resource exists in one of these locations, or specify the correct --template-dir if using custom templates."""


def user_error_message(e: BaseException) -> str:
    """Format an exception for the user, appending search-order guidance to lookup failures."""
    message = str(e) if str(e) else repr(e)
    if isinstance(e, TemplateNotFoundError):
        return message + TEMPLATE_NOT_FOUND_HINT
    if isinstance(e, FileNotFoundError):
        return message + RESOURCE_NOT_FOUND_HINT
    return message


RENDER_THREAD_SHUTDOWN_TIMEOUT = 0.7

# The terminal states a render can reach, as the API names them.
RENDER_OUTCOME_COMPLETED = "completed"
RENDER_OUTCOME_FAILED = "failed"
RENDER_OUTCOME_CANCELLED = "cancelled"

# Exceptions that represent expected, user-facing error conditions. They are
# reported to the user directly and must never be sent to Sentry as crashes.
EXPECTED_EXCEPTIONS = (
    InvalidFridArgument,
    FileNotFoundError,
    MissingResource,
    TemplateNotFoundError,
    PlainSyntaxError,
    ImportedModuleWithFunctionalitiesError,
    MissingFunctionalitiesError,
    MissingPreviousFunctionalitiesError,
    MissingAPIKey,
    InvalidAPIKey,
    OutdatedClientVersion,
    ConflictingRequirements,
    RenderingCreditBalanceTooLow,
    NetworkConnectionError,
    ModuleDoesNotExistError,
    UnsupportedResourceType,
    UnsupportedBase64Content,
    GitNotInstalledError,
    InvalidModuleArchiveError,
    SystemExit,
)


def setup_logging(
    args,
    event_bus: EventBus,
    run_state: RunState,
    log_to_file: bool,
    log_file_path: str,
    headless: bool = False,
):
    default_level = logging.DEBUG if args.verbose else logging.INFO

    logging.getLogger().setLevel(default_level)
    logging.getLogger(LOGGER_NAME).setLevel(default_level)
    logging.getLogger("git").setLevel(logging.WARNING)
    logging.getLogger("transitions").setLevel(logging.ERROR)
    logging.getLogger("transitions.extensions.diagrams").setLevel(logging.ERROR)

    # Try to load logging configuration from YAML file (takes precedence over --verbose)
    config_loaded = False

    if args.logging_config_path and os.path.exists(args.logging_config_path):
        try:
            with open(args.logging_config_path, "r") as f:
                config = yaml.safe_load(f)
                logging.config.dictConfig(config)
                console.info(f"Loaded logging configuration from {args.logging_config_path}")
                config_loaded = True
        except Exception as e:
            console.warning(f"Failed to load logging configuration from {args.logging_config_path}: {str(e)}")

    root_logger = logging.getLogger(LOGGER_NAME)
    configured_log_level = root_logger.level if config_loaded else default_level
    root_logger.setLevel(logging.DEBUG)  # Capture all logs; handlers will filter levels as needed

    formatter = IndentedFormatter("%(levelname)s:%(name)s:%(message)s")
    file_formatter = ElapsedTimeFormatter(run_state)

    if not headless:
        handler = LoggingHandler(event_bus, run_state)
        handler.setFormatter(formatter)
        root_logger.addHandler(handler)
    else:
        # Headless has no TUI to narrate the render and suppresses Rich output, so
        # without this the process says nothing on stdout for its entire run: a CI job
        # cannot distinguish a render wedged for hours from a healthy one until the
        # log file is collected at the end. StreamHandler flushes
        # per record, so these arrive as they happen.
        stdout_handler = logging.StreamHandler(sys.stdout)
        stdout_handler.setFormatter(file_formatter)
        stdout_handler.setLevel(configured_log_level)
        stdout_handler.set_name(NARRATION_HANDLER_NAME)
        root_logger.addHandler(stdout_handler)

    if log_to_file:
        try:
            file_handler = logging.FileHandler(log_file_path, mode="w", encoding="utf-8")
            file_handler.setFormatter(file_formatter)
            file_handler.setLevel(configured_log_level)
            root_logger.addHandler(file_handler)
        except Exception as e:
            console.warning(f"Failed to setup file logging to {log_file_path}: {str(e)}")
    else:
        crash_handler = CrashLogHandler()
        crash_handler.setFormatter(file_formatter)
        crash_handler.setLevel(configured_log_level)
        root_logger.addHandler(crash_handler)

    return logging.getLevelName(configured_log_level)


def attached_to_terminal() -> bool:
    """Whether a person at a terminal can see and drive the TUI.

    The TUI reads keys from stdin and draws on stderr; the exit summary goes to
    stdout. Started from a script or an agent, stdin is usually at end-of-file,
    and Textual's input thread then reads EOF in a tight loop at 100% CPU while
    a stopped render waits for an Enter nobody can press.
    """
    return all(stream is not None and stream.isatty() for stream in (sys.stdin, sys.stdout, sys.stderr))


def _check_connection(codeplainAPI: codeplain_api.CodeplainAPI) -> Optional[str]:
    """Check API connectivity, validate API key and client version, and return the user email."""
    response = codeplainAPI.connection_check(system_config.client_version)

    if not response.get("api_key_valid", False):
        raise InvalidAPIKey(
            "The provided API key is invalid. Please provide a valid API key using the CODEPLAIN_API_KEY environment variable "
            "or the --api-key argument.\n"
        )

    if not response.get("client_version_valid", False):
        min_version = response.get("min_client_version", "unknown")
        raise OutdatedClientVersion(
            "Outdated client version: "
            f"Your client version ({system_config.client_version}) is outdated. Minimum required version is {min_version}. "
            "Please update using: uv tool upgrade codeplain"
        )

    return response.get("user_email")


def _render_outcome(run_state: RunState, interrupted: bool = False) -> str:
    """Name the terminal state a render reached.

    Ctrl-C is the user stopping the render, so it is a cancellation. It does not
    set render_cancelled: the exit summary reads that flag, and its wording for
    an interrupt stays as it is.

    Cancellation is tested first. A cancelled render leaves render_succeeded
    False, so asking about success first would report it as a failure.
    """
    if interrupted or run_state.render_cancelled:
        return RENDER_OUTCOME_CANCELLED
    if run_state.render_succeeded:
        return RENDER_OUTCOME_COMPLETED
    return RENDER_OUTCOME_FAILED


def report_render_finished(
    codeplainAPI: codeplain_api.CodeplainAPI, run_state: RunState, interrupted: bool = False
) -> None:
    """Tell the API which terminal state this render reached.

    The API cannot work it out for itself - the state machine lives here - and
    main()'s finally is the one place that sees every ending: success, failure,
    cancellation, an unexpected crash and a keyboard interrupt alike. The state
    machine's own completed and failed hooks see only two of the five.

    Reported only once the connection check has set a user email, which is
    exactly the condition "the API knows who we are". A missing key, an invalid
    key and an outdated client all leave it unset, and none of them got as far
    as a render.

    Never raises. A report must not change the exit code, and must not mask the
    error that ended the render.
    """
    if not run_state.user_email:
        return

    try:
        codeplainAPI.render_finished(_render_outcome(run_state, interrupted), run_state)
    except Exception:
        # Deliberately silent: logging has stopped by now, and a failed report
        # is not the user's problem.
        pass


def _has_scenarios(plain_source_tree) -> bool:
    """Whether a test plain file bound a scenario to any functionality of the module."""
    for functionality in plain_source_tree.get(plain_spec.FUNCTIONAL_REQUIREMENTS) or []:
        if functionality.get(plain_spec.CONFORMANCE_TESTS) or functionality.get(plain_spec.ACCEPTANCE_TESTS):
            return True
    return False


def warn_about_functionalities_without_scenarios(plain_module) -> None:
    """Warn about every functionality whose test plain file holds no conformance scenario.

    Such a functionality is rendered, but no conformance test is rendered or run for it.
    """
    functionalities = [
        f"{module_name}:{func_id}" for module_name, func_id in plain_module.functionalities_without_scenarios
    ]
    if not functionalities:
        return

    console.warning(
        f"No conformance scenario is written for {', '.join(functionalities)}, "
        "so each one is rendered without conformance tests."
    )


def warn_if_scenarios_without_conformance_script(plain_module, args) -> None:
    """Warn when a test plain file was found but conformance tests are disabled.

    The scenarios are rendered as conformance tests, so without a conformance tests script
    none of them is rendered.
    """
    if args.conformance_tests_script:
        return

    module_names_with_scenarios = [
        module.module_name
        for module in plain_module.all_required_modules + [plain_module]
        if _has_scenarios(module.plain_source)
    ]
    if not module_names_with_scenarios:
        return

    module_names = ", ".join(module_names_with_scenarios)
    console.warning(
        f"Test scenarios were found ({module_names}) but no conformance tests script is configured, "
        "so no scenario is rendered. Set --conformance-tests-script or 'conformance_tests_script' "
        "in the config file to render them."
    )


def validate_linked_resources(plain_module) -> None:
    """Load every linked resource exactly the way rendering does, so dry-run
    surfaces the same errors (binary file, missing file, oversized base64 blob)."""
    for module in plain_module.all_required_modules + [plain_module]:
        resources_list: list[dict] = []
        plain_spec.collect_linked_resources(module.plain_source, resources_list, None, True)
        file_utils.load_linked_resources(module.template_dirs, resources_list, module.module_name)


def render(  # noqa: C901
    plain_module: plain_modules.PlainModule,
    codeplainAPI: codeplain_api.CodeplainAPI,
    args,
    run_state: RunState,
    event_bus: EventBus,
    default_log_level: str = "INFO",
):
    # Compute render range from either --render-range or --render-from
    render_range = None
    if args.render_range or args.render_from:
        render_range = plain_spec.compute_render_range(args, plain_module.plain_source)

    run_state.user_email = _check_connection(codeplainAPI)

    stop_event = threading.Event()
    enter_pause_event = threading.Event()
    signal.signal(signal.SIGTERM, lambda _signum, _frame: stop_event.set())

    warn_if_scenarios_without_conformance_script(plain_module, args)
    warn_about_functionalities_without_scenarios(plain_module)

    # A built module can be distributed as a "<module>.module" zip archive instead of an unpacked directory.
    # Extract any such module to a scratch location up front. Scratch dirs are removed in main()'s finally.
    # Skip an archive-only module that lacks conformance tests while testing is enabled: it cannot be
    # consumed as-is, so leave it unmaterialized so it is detected below as needing a re-render.
    for module in plain_module.all_required_modules + [plain_module]:
        if args.render_conformance_tests and module.is_archived_only() and not module.archive_has_conformance_tests():
            continue
        module.materialize()

    # The module_metadata file lives outside the code git repo. That means that a crash mid-render can leave it
    # claiming a functionality was implemented even thought it wasn't yet committed (because of the crash).
    # Out of precaution, this reconciles every module_metadata against the code repo.
    for module in plain_module.all_required_modules + [plain_module]:
        module.reconcile_metadata_with_git()

    render_choice = None
    if render_range is None:
        plain_module_render_state = get_plain_module_render_state(plain_module, args.render_conformance_tests)
        if plain_module_render_state is not None:
            render_choices = get_render_choices(plain_module, plain_module_render_state, args.force_render)
            ask_user = True
            render_choice = None
            if args.force_render or len(render_choices) <= 2:
                # With --force-render there is a single choice. Otherwise the last choice is Quit,
                # so the first choice is the only other actionable one.
                render_choice = render_choices[list(render_choices.keys())[0]]
                ask_user = render_choice.is_destructive and not args.force_render

            if ask_user and not args.headless:
                app = PlainModuleRenderChoiceTUI(
                    plain_module,
                    plain_module_render_state,
                    render_choices,
                    system_config.client_version,
                    run_state.render_id,
                    on_cancel=run_state.set_render_cancelled,
                    css_path="styles.css",
                )
                render_choice = app.run()
                if render_choice is None or (
                    render_choice.module is None
                    and render_choice.render_range is None
                    and render_choice.choice_type == "quit"
                ):
                    run_state.set_render_cancelled()
                    sys.exit(0)
            elif ask_user and args.headless:
                # ignore the default choice if it requires user input due to headless mode
                # fallback to --render-from
                # default choice is only used only when the action is not destructive
                render_choice = None

    if render_choice is not None and render_range is not None:
        raise Exception("Partial rendering and render range cannot be used together")

    module_renderer = ModuleRenderer(
        codeplainAPI,
        plain_module,
        render_choice,
        render_range,
        args,
        run_state,
        event_bus,
        stop_event=stop_event,
        enter_pause_event=enter_pause_event,
    )

    render_error: list[Exception] = []

    def run_render():
        try:
            module_renderer.render_module()
        except RenderCancelledError:
            run_state.set_render_cancelled()  # TUI already closed, nothing to report
        except Exception as e:
            run_state.set_render_succeeded(False)
            render_error.append(e)
            event_bus.publish(RenderFailed(error_message=user_error_message(e)))

    if args.headless:
        console.info(f"Render started. Render ID: {run_state.render_id}")
        try:
            module_renderer.render_module()
        except RenderCancelledError:
            run_state.set_render_cancelled()
        return
    else:
        render_thread = threading.Thread(target=run_render, daemon=True)
        app = Plain2CodeTUI(
            event_bus=event_bus,
            run_state=run_state,
            on_ready=render_thread.start,
            render_id=run_state.render_id,
            unittests_script=args.unittests_script,
            conformance_tests_script=args.conformance_tests_script,
            prepare_environment_script=args.prepare_environment_script,
            state_machine_version=system_config.client_version,
            enter_pause_event=enter_pause_event,
            on_cancel=run_state.set_render_cancelled,
            default_log_level=default_log_level,
            exit_when_done=args.exit_when_done,
            css_path="styles.css",
        )
        app.run()

        stop_event.set()
        render_thread.join(timeout=RENDER_THREAD_SHUTDOWN_TIMEOUT)

    if render_error:
        raise render_error[0]


def main():  # noqa: C901
    if sys.platform == "win32":
        sys.stdout.reconfigure(encoding="utf-8", errors="replace")
        sys.stderr.reconfigure(encoding="utf-8", errors="replace")

    args = parse_arguments()

    # Handle --version flag before any other initialization
    if args.version:
        console.print(f"codeplain version {system_config.client_version}")
        return

    # Handle --status flag before any other initialization
    if args.status:
        if not args.api_key:
            console.error(
                "Your API key is required. Please set the CODEPLAIN_API_KEY environment variable or provide it with the --api-key argument.\n"
            )
            sys.exit(1)

        if not args.api:
            args.api = system_config.default_api_url

        try:
            print_status(args.api_key, args.api, system_config.client_version)
        except Exception as e:
            console.error(f"Error fetching status: {str(e)}")
            sys.exit(1)
        return

    template_dirs = file_utils.get_template_directories(args.filename, args.template_dir, DEFAULT_TEMPLATE_DIRS)

    # Handle full plain early-exit (raw text dump; does not require a parsed module).
    if args.full_plain:
        try:
            module_name = Path(args.filename).stem
            plain_source = plain_file.read_module_plain_source(module_name, template_dirs)
            [full_plain_source, _] = file_utils.get_loaded_templates(template_dirs, plain_source)
            console.info("Full plain text:\n")
            console.info(full_plain_source)
        except Exception as e:
            console.error(f"Error: {user_error_message(e)}")
        return

    # Parse the plain file (and its required modules) once; reused by dry-run and rendering.
    try:
        plain_module = plain_modules.PlainModule(
            os.path.basename(args.filename),
            args.build_folder,
            template_dirs,
            args.tests_folder,
            args.render_conformance_tests,
        )
    except Exception as e:
        console.error(f"Error: {user_error_message(e)}")
        return

    if args.dry_run:
        try:
            validate_linked_resources(plain_module)
        except Exception as e:
            console.error(f"Error: {user_error_message(e)}")
            return

        console.info("Printing dry run output...\n")
        render_range = plain_spec.compute_render_range(args, plain_module.plain_source)
        print_dry_run_output(plain_module.plain_source, render_range)
        warn_if_scenarios_without_conformance_script(plain_module, args)
        warn_about_functionalities_without_scenarios(plain_module)
        return

    event_bus = EventBus()

    if not args.api:
        args.api = system_config.default_api_url

    assert args.api is not None and args.api != "", "API URL is required"
    codeplainAPI = codeplain_api.CodeplainAPI(args.api_key, console)
    codeplainAPI.api_url = args.api

    run_state = RunState(spec_filename=args.filename, replay_with=args.replay_with, job_tag=args.job_tag)

    # Without a terminal nobody can watch or drive the TUI, so render as --headless does:
    # progress on stdout, and a non-zero exit status when the render stops.
    no_terminal = not args.headless and not attached_to_terminal()
    if no_terminal:
        args.headless = True

    if args.headless:
        # Suppress Rich console output.
        console.quiet = True

    default_log_level = setup_logging(args, event_bus, run_state, args.log_to_file, args.log_file_name, args.headless)

    if no_terminal:
        console.info("No terminal attached, so rendering without the TUI, as with --headless.")

    initialize_telemetry()

    exc_info = None
    error_message = None
    interrupted = False

    try:
        # Validate API key is present
        if not args.api_key:
            raise MissingAPIKey(
                "Your API key is required. Please set the CODEPLAIN_API_KEY environment variable or provide it with the --api-key argument.\n"
            )
        render(plain_module, codeplainAPI, args, run_state, event_bus, default_log_level)
    except BaseException as e:
        if isinstance(e, KeyboardInterrupt):
            error_message = "Keyboard interrupt"
            interrupted = True
        else:
            error_message = user_error_message(e)

            if not isinstance(e, EXPECTED_EXCEPTIONS):
                exc_info = sys.exc_info()
    finally:
        if exc_info:
            dump_crash_logs(args, run_state)
            capture_crash(exc_info, run_state, args)
        stop_narrating()
        print_exit_summary(
            run_state,
            args.filename,
            error_message=error_message or run_state.render_error_message,
        )
        # Remove any scratch extractions created for archive-only ("<module>.module") modules.
        for module in plain_module.all_required_modules + [plain_module]:
            module.cleanup_scratch()
        # Last, so nothing the user is waiting for sits behind a network call.
        report_render_finished(codeplainAPI, run_state, interrupted)

    if args.headless and (exc_info is not None or not run_state.render_succeeded):
        sys.exit(1)


if __name__ == "__main__":  # noqa: C901
    main()