Eliteforge New Java Maven Model

interface:
  display_name: "Java Maven Model 模块"
  short_description: "按仓库约定生成并同步 Maven 模块骨架"
  default_prompt: "使用 $eliteforge-new-java-maven-model 为当前仓库创建并同步一个新的 Maven 模块。"

#!/usr/bin/env python3
"""Scaffold a Maven model module that follows repository naming conventions."""

from __future__ import annotations

import argparse
import re
import sys
import xml.etree.ElementTree as ET
from dataclasses import dataclass
from pathlib import Path

MAVEN_NS = "http://maven.apache.org/POM/4.0.0"
NS = {"m": MAVEN_NS}

ET.register_namespace("", MAVEN_NS)

DESCRIPTION_KEYS = {
    "公司名称": "company_name",
    "产品名称": "product_name",
    "服务名称": "service_name",
}


@dataclass(frozen=True)
class MavenProject:
    """A minimal view of a Maven project's coordinates."""

    path: Path
    group_id: str
    artifact_id: str
    packaging: str
    java_version: str | None


@dataclass(frozen=True)
class NamingContext:
    """The resolved names used to generate the module."""

    company_name: str
    product_name: str
    service_name: str
    model_name: str

    @property
    def module_name(self) -> str:
        """Build the module directory name."""

        return f"{to_hyphen(self.service_name)}-{to_hyphen(self.model_name)}"

    @property
    def package_name(self) -> str:
        """Build the Java package name."""

        return ".".join(
            [
                "cn",
                to_dot(self.company_name),
                to_dot(self.product_name),
                to_dot(self.service_name),
                to_dot(self.model_name),
            ]
        )

    @property
    def group_id(self) -> str:
        """Build the Maven groupId."""

        return ".".join(["cn", to_dot(self.company_name), to_dot(self.product_name)])

    @property
    def artifact_id(self) -> str:
        """Build the Maven artifactId."""

        return "-".join(
            [
                "app",
                to_hyphen(self.company_name),
                to_hyphen(self.product_name),
                to_hyphen(self.service_name),
                to_hyphen(self.model_name),
            ]
        )

    @property
    def module_artifact_prefix(self) -> str:
        """Build the artifactId prefix used by sibling internal modules."""

        return "-".join(
            [
                "app",
                to_hyphen(self.company_name),
                to_hyphen(self.product_name),
                to_hyphen(self.service_name),
            ]
        ) + "-"


def split_segments(raw: str) -> list[str]:
    """Normalize text into lowercase identifier segments."""

    lowered = raw.strip().lower()
    normalized = re.sub(r"[^a-z0-9]+", "-", lowered)
    normalized = normalized.strip("-")
    if not normalized:
        raise ValueError(f"无法从输入生成合法标识: {raw!r}")
    return [segment for segment in normalized.split("-") if segment]


def to_hyphen(raw: str) -> str:
    """Convert a value to hyphen-case."""

    return "-".join(split_segments(raw))


def to_dot(raw: str) -> str:
    """Convert a value to dot.case."""

    return ".".join(split_segments(raw))


def parse_args() -> argparse.Namespace:
    """Parse command-line arguments."""

    parser = argparse.ArgumentParser(
        description="为 Java 多模块 Maven 仓库创建新的 model 模块骨架。",
    )
    parser.add_argument("model_name", help="model 名称，例如 reasoning-engine")
    parser.add_argument(
        "--repo-root",
        help="仓库根目录；默认从当前目录向上自动识别",
    )
    parser.add_argument(
        "--dry-run",
        action="store_true",
        help="只打印计划执行的动作，不写入文件",
    )
    return parser.parse_args()


def read_xml(pom_path: Path) -> ET.Element:
    """Read a pom.xml document and return its root element."""

    try:
        return ET.parse(pom_path).getroot()
    except ET.ParseError as exc:
        raise ValueError(f"解析 pom.xml 失败: {pom_path}") from exc


def child_text(node: ET.Element, xpath: str) -> str | None:
    """Return stripped child text for a namespaced Maven XPath."""

    target = node.find(xpath, NS)
    if target is None or target.text is None:
        return None
    value = target.text.strip()
    return value or None


def parse_maven_project(pom_path: Path) -> MavenProject:
    """Parse the coordinates of a Maven project."""

    root = read_xml(pom_path)
    group_id = child_text(root, "m:groupId") or child_text(root, "m:parent/m:groupId")
    artifact_id = child_text(root, "m:artifactId")
    packaging = child_text(root, "m:packaging") or "jar"
    java_version = child_text(root, "m:properties/m:java.version")
    if not group_id or not artifact_id:
        raise ValueError(f"pom.xml 缺少 groupId 或 artifactId: {pom_path}")
    return MavenProject(
        path=pom_path,
        group_id=group_id,
        artifact_id=artifact_id,
        packaging=packaging,
        java_version=java_version,
    )


def parse_root_description(root_pom: Path) -> dict[str, str]:
    """Extract naming fields from the root pom description."""

    root = read_xml(root_pom)
    description = root.find("m:description", NS)
    if description is None:
        raise ValueError(f"根 pom.xml 缺少 description: {root_pom}")

    text = " ".join(segment.strip() for segment in description.itertext() if segment.strip())
    matches = dict(re.findall(r"(公司名称|产品名称|服务名称)\s*[:：]\s*([A-Za-z0-9._-]+)", text))

    result: dict[str, str] = {}
    for label, field in DESCRIPTION_KEYS.items():
        value = matches.get(label)
        if not value:
            raise ValueError(f"根 pom.xml 的 description 缺少 {label}: {root_pom}")
        result[field] = value
    return result


def discover_repo_root(start: Path) -> Path:
    """Find the nearest ancestor pom.xml that contains the required description keys."""

    current = start.resolve()
    search_points = [current, *current.parents]
    for candidate in search_points:
        pom_path = candidate / "pom.xml"
        if not pom_path.is_file():
            continue
        try:
            parse_root_description(pom_path)
            return candidate
        except ValueError:
            continue
    raise ValueError(f"未找到包含公司/产品/服务名称描述的根 pom.xml: {start}")


def find_parent_project(module_dir: Path, repo_root: Path) -> MavenProject:
    """Locate the nearest Maven aggregator parent for the target module directory."""

    current = module_dir.parent.resolve()
    repo_root = repo_root.resolve()

    while True:
        pom_path = current / "pom.xml"
        if pom_path.is_file():
            project = parse_maven_project(pom_path)
            if project.packaging == "pom":
                return project
        if current == repo_root:
            break
        if current.parent == current:
            break
        current = current.parent

    raise ValueError(f"未识别到可用的父聚合 pom.xml: {module_dir}")


def render_pom(parent: MavenProject, naming: NamingContext) -> str:
    """Render the new module pom.xml."""

    java_version = parent.java_version or "25"
    return f"""<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
  xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
  xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
  <modelVersion>4.0.0</modelVersion>
  <parent>
    <groupId>{parent.group_id}</groupId>
    <artifactId>{parent.artifact_id}</artifactId>
    <version>${{revision}}</version>
  </parent>

  <groupId>{naming.group_id}</groupId>
  <artifactId>{naming.artifact_id}</artifactId>

  <properties>
    <java.version>{java_version}</java.version>
    <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
    <project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding>
  </properties>

</project>
"""


def insert_module(parent_pom: Path, module_name: str) -> bool:
    """Ensure the parent pom contains the target module entry."""

    content = parent_pom.read_text()
    module_tag = f"<module>{module_name}</module>"
    if module_tag in content:
        return False

    modules_match = re.search(
        r"(?P<indent>^[ \t]*)<modules>\n(?P<body>.*?)(?P=indent)</modules>",
        content,
        re.MULTILINE | re.DOTALL,
    )
    if modules_match:
        indent = modules_match.group("indent")
        body = modules_match.group("body")
        module_indent_match = re.search(r"^([ \t]*)<module>", body, re.MULTILINE)
        module_indent = module_indent_match.group(1) if module_indent_match else f"{indent}  "
        insertion = f"{module_indent}<module>{module_name}</module>\n"
        updated = f"{indent}<modules>\n{body}{insertion}{indent}</modules>"
        parent_pom.write_text(content[:modules_match.start()] + updated + content[modules_match.end():])
        return True

    block = f"  <modules>\n    <module>{module_name}</module>\n  </modules>\n"
    anchor_match = re.search(
        r"^[ \t]*<(properties|dependencyManagement|profiles|build)\b",
        content,
        re.MULTILINE,
    )
    if anchor_match:
        updated = content[:anchor_match.start()] + block + content[anchor_match.start():]
    else:
        updated = content.replace("</project>", f"{block}</project>", 1)
    parent_pom.write_text(updated)
    return True


def dependency_management_sections(
    content: str,
) -> tuple[re.Match[str] | None, re.Match[str] | None]:
    """Find the dependencyManagement and nested dependencies sections."""

    dependency_management_match = re.search(
        r"(?P<indent>^[ \t]*)<dependencyManagement>\n(?P<body>.*?)(?P=indent)</dependencyManagement>",
        content,
        re.MULTILINE | re.DOTALL,
    )
    if not dependency_management_match:
        return None, None

    dependencies_match = re.search(
        r"(?P<indent>^[ \t]*)<dependencies>\n(?P<body>.*?)(?P=indent)</dependencies>",
        dependency_management_match.group("body"),
        re.MULTILINE | re.DOTALL,
    )
    return dependency_management_match, dependencies_match


def plan_managed_dependency(parent_pom: Path, naming: NamingContext) -> str:
    """Describe whether the parent pom should receive a managed dependency."""

    content = parent_pom.read_text()
    dependency_management_match, dependencies_match = dependency_management_sections(content)
    if not dependency_management_match:
        return "skipped(no-dependencyManagement)"
    if not dependencies_match:
        return "skipped(no-dependencies)"

    dependencies_body = dependencies_match.group("body")
    artifact_tag = f"<artifactId>{naming.artifact_id}</artifactId>"
    if artifact_tag in dependencies_body:
        return "already-present"

    dependency_pattern = re.compile(
        r"(?P<indent>^[ \t]*)<dependency>\n(?P<body>.*?)(?P=indent)</dependency>\n?",
        re.MULTILINE | re.DOTALL,
    )
    group_tag = f"<groupId>{naming.group_id}</groupId>"
    artifact_prefix = f"<artifactId>{naming.module_artifact_prefix}"

    for match in dependency_pattern.finditer(dependencies_body):
        dependency_body = match.group("body")
        if group_tag in dependency_body and artifact_prefix in dependency_body:
            return "will-insert"

    return "skipped(no-managed-module-convention)"


def insert_managed_dependency(parent_pom: Path, naming: NamingContext) -> str:
    """Insert the managed dependency when the parent pom already manages sibling modules."""

    action = plan_managed_dependency(parent_pom, naming)
    if action != "will-insert":
        return action

    content = parent_pom.read_text()
    dependency_management_match, dependencies_match = dependency_management_sections(content)
    if not dependency_management_match or not dependencies_match:
        return "skipped(no-dependencyManagement)"

    dependencies_body = dependencies_match.group("body")
    dependency_pattern = re.compile(
        r"(?P<indent>^[ \t]*)<dependency>\n(?P<body>.*?)(?P=indent)</dependency>\n?",
        re.MULTILINE | re.DOTALL,
    )
    group_tag = f"<groupId>{naming.group_id}</groupId>"
    artifact_prefix = f"<artifactId>{naming.module_artifact_prefix}"

    last_internal_match: re.Match[str] | None = None
    for match in dependency_pattern.finditer(dependencies_body):
        dependency_body = match.group("body")
        if group_tag in dependency_body and artifact_prefix in dependency_body:
            last_internal_match = match

    if last_internal_match is None:
        return "skipped(no-managed-module-convention)"

    dependency_indent = last_internal_match.group("indent")
    value_indent = f"{dependency_indent}  "
    managed_dependency = (
        f"{dependency_indent}<dependency>\n"
        f"{value_indent}<groupId>{naming.group_id}</groupId>\n"
        f"{value_indent}<artifactId>{naming.artifact_id}</artifactId>\n"
        f"{value_indent}<version>${{project.version}}</version>\n"
        f"{dependency_indent}</dependency>\n"
    )
    updated_dependencies = (
        dependencies_body[: last_internal_match.end()]
        + managed_dependency
        + dependencies_body[last_internal_match.end() :]
    )
    updated_dependencies_block = (
        f"{dependencies_match.group('indent')}<dependencies>\n"
        f"{updated_dependencies}"
        f"{dependencies_match.group('indent')}</dependencies>"
    )
    updated_dependency_management = (
        f"{dependency_management_match.group('indent')}<dependencyManagement>\n"
        f"{dependency_management_match.group('body')[: dependencies_match.start()]}"
        f"{updated_dependencies_block}"
        f"{dependency_management_match.group('body')[dependencies_match.end() :]}"
        f"{dependency_management_match.group('indent')}</dependencyManagement>"
    )
    parent_pom.write_text(
        content[: dependency_management_match.start()]
        + updated_dependency_management
        + content[dependency_management_match.end() :]
    )
    return "inserted"


def ensure_template_hint(repo_root: Path, service_name: str) -> str | None:
    """Return an optional sibling module that can be used as a layout reference."""

    template_dir = repo_root / f"{to_hyphen(service_name)}-runtime-plane"
    return str(template_dir) if template_dir.exists() else None


def create_layout(module_dir: Path, package_name: str, pom_content: str) -> None:
    """Create the Maven module directory layout."""

    package_path = Path(*package_name.split("."))
    main_package_dir = module_dir / "src" / "main" / "java" / package_path
    resources_dir = module_dir / "src" / "main" / "resources"
    test_java_dir = module_dir / "src" / "test" / "java"

    main_package_dir.mkdir(parents=True, exist_ok=False)
    resources_dir.mkdir(parents=True, exist_ok=True)
    test_java_dir.mkdir(parents=True, exist_ok=True)
    (module_dir / "pom.xml").write_text(pom_content)
    (main_package_dir / ".gitkeep").write_text("")
    (resources_dir / ".gitkeep").write_text("")
    (test_java_dir / ".gitkeep").write_text("")


def print_summary(
    repo_root: Path,
    parent: MavenProject,
    naming: NamingContext,
    module_dir: Path,
    template_dir: str | None,
    managed_dependency: str,
    dry_run: bool,
) -> None:
    """Print the resolved scaffold summary."""

    print(f"repoRoot={repo_root}")
    print(f"parentPom={parent.path}")
    print(f"moduleDir={module_dir}")
    print(f"moduleName={naming.module_name}")
    print(f"packageName={naming.package_name}")
    print(f"groupId={naming.group_id}")
    print(f"artifactId={naming.artifact_id}")
    print(f"managedDependency={managed_dependency}")
    if template_dir:
        print(f"referenceModule={template_dir}")
    else:
        print("referenceModule=<not found, use built-in minimal layout>")
    print(f"mode={'dry-run' if dry_run else 'apply'}")


def main() -> int:
    """Program entry point."""

    args = parse_args()
    start = Path(args.repo_root).resolve() if args.repo_root else Path.cwd().resolve()

    try:
        repo_root = discover_repo_root(start)
        root_names = parse_root_description(repo_root / "pom.xml")
        naming = NamingContext(
            company_name=root_names["company_name"],
            product_name=root_names["product_name"],
            service_name=root_names["service_name"],
            model_name=args.model_name,
        )
        module_dir = repo_root / naming.module_name
        if module_dir.exists():
            raise ValueError(f"模块目录已存在: {module_dir}")

        parent = find_parent_project(module_dir, repo_root)
        template_dir = ensure_template_hint(repo_root, naming.service_name)
        managed_dependency = plan_managed_dependency(parent.path, naming)
        print_summary(
            repo_root,
            parent,
            naming,
            module_dir,
            template_dir,
            managed_dependency,
            args.dry_run,
        )

        if args.dry_run:
            return 0

        pom_content = render_pom(parent, naming)
        create_layout(module_dir, naming.package_name, pom_content)
        inserted = insert_module(parent.path, naming.module_name)
        managed_dependency_result = insert_managed_dependency(parent.path, naming)
        print(f"created={module_dir}")
        print(f"moduleEntry={'inserted' if inserted else 'already-present'}")
        print(f"managedDependency={managed_dependency_result}")
        return 0
    except ValueError as exc:
        print(f"[ERROR] {exc}", file=sys.stderr)
        return 1


if __name__ == "__main__":
    raise SystemExit(main())