DDNS/.github/scripts/update_agents_structure.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""Compare AGENTS.md directory structure with actual files."""

import argparse
import os
import re
import sys

REPO_ROOT = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
ISSUE_BODY_FILE = os.path.join(REPO_ROOT, ".github", "issue_body.md")
# English docs in these directories mirror Chinese docs and are documented as groups.
ENGLISH_MIRROR_DIRS = ("docs/en/config", "docs/en/dev", "docs/en/providers")


def scan_files(directory, extensions):
    # type: (str, tuple) -> set
    """Scan directory for files with given extensions."""
    # Files to exclude from the agent and Q&A directory index
    exclude_files = {
        "docs/index.md",
        "docs/en/index.md",
        "docs/config/studio.md",
        "docs/en/config/studio.md",
        "docs/esa-deploy.md",
        "docs/esa.js",
        "docs/llms.txt",
        "docs/package.json",
    }

    result = set()
    base = os.path.join(REPO_ROOT, directory)
    if not os.path.isdir(base):
        return result
    for root, dirs, files in os.walk(base):
        dirs[:] = [d for d in dirs if not d.startswith(".") and d not in ("__pycache__", "node_modules")]
        for f in files:
            if not f.startswith(".") and f.endswith(extensions):
                rel_path = os.path.relpath(os.path.join(root, f), REPO_ROOT).replace(os.sep, "/")
                if rel_path not in exclude_files:
                    result.add(rel_path)
    return result


def parse_agents_md():
    # type: () -> tuple
    """Extract documented file paths and directories from AGENTS.md (Tab-indented)."""
    agents_file = os.path.join(REPO_ROOT, "AGENTS.md")
    if not os.path.exists(agents_file):
        print("Error: AGENTS.md not found")
        return set(), set()

    with open(agents_file, "r", encoding="utf-8") as f:
        content = f.read()

    match = re.search(r"### Directory Structure.*?```text\s*\n(.*?)```\s*", content, re.DOTALL)
    if not match:
        return set(), set()

    files = set()
    directories = set()
    stack = []  # type: list
    for line in match.group(1).split("\n"):
        if not line.strip():
            continue
        depth = len(line) - len(line.lstrip("\t"))
        name = line.lstrip("\t").split(":")[0].strip()
        if not name or "*" in name or name == "...":
            continue

        stack = stack[:depth]
        if name.endswith("/"):
            stack.append(name.rstrip("/"))
            path = "/".join(stack)
            if path.startswith(("ddns/", "docs/", "schema/")):
                directories.add(path)
        else:
            path = "/".join(stack + [name])
            if path.startswith(("ddns/", "docs/", "schema/")) and not path.endswith(
                (".png", ".svg", ".jpg", ".gif", ".ico")
            ):
                files.add(path)
    return files, directories


def is_documented_english_mirror(path, actual_files, documented_dirs):
    # type: (str, set, set) -> bool
    """Check whether an English doc mirrors a documented Chinese source file."""
    for directory in ENGLISH_MIRROR_DIRS:
        if directory not in documented_dirs:
            continue
        prefix = directory + "/"
        if path.startswith(prefix):
            zh_path = path.replace("docs/en/", "docs/", 1)
            return zh_path in actual_files
    return False


def structure_drift():
    # type: () -> tuple
    """Return AGENTS.md entries that are missing or undocumented."""
    actual_files = scan_files("ddns", (".py",)) | scan_files("docs", (".md",)) | scan_files("schema", (".json",))
    documented, documented_dirs = parse_agents_md()
    files_to_check = {f for f in actual_files if not is_documented_english_mirror(f, actual_files, documented_dirs)}
    return sorted(files_to_check - documented), sorted(documented - files_to_check)


def write_issue_body(added, deleted, contract_errors, docs_changes):
    # type: (list, list, list, bool) -> None
    """Write the scheduled drift report consumed by the workflow."""
    lines = ["Portable agent contracts or the AGENTS.md directory structure need attention.\n"]
    if contract_errors:
        lines.append("## Contract Errors\n")
        lines.extend("- %s" % error for error in contract_errors)
        lines.append("")
    if added:
        lines.append("## New Files\n")
        lines.extend("- `%s`" % f for f in added)
        lines.append("")
    if deleted:
        lines.append("## Missing Files\n")
        lines.extend("- `%s`" % f for f in deleted)
        lines.append("")

    required_updates = []
    if contract_errors:
        required_updates.append("Fix each contract error above")
    if added or deleted:
        required_updates.extend(("Update directory structure", "Update version/date"))
    if docs_changes:
        required_updates.append("Update docs/llms.txt")
    lines.append("## Required Updates\n")
    lines.extend("%d. %s" % (index, update) for index, update in enumerate(required_updates, 1))
    lines.append("\n---\n*Auto-generated by update-agents workflow.*")

    with open(ISSUE_BODY_FILE, "w", encoding="utf-8") as f:
        f.write("\n".join(lines))


def main(argv=None):
    # type: (object) -> int
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--check", action="store_true", help="fail without writing an issue body when structure drifts")
    arguments = parser.parse_args(argv)

    tools_dir = os.path.join(REPO_ROOT, "tools")
    if tools_dir not in sys.path:
        sys.path.insert(0, tools_dir)
    try:
        import check
    except ImportError as error:
        print("Error: unable to load tools/check.py: {}".format(error))
        sys.exit(1)

    contract_errors = check.portable_contract_errors(check.REPO_ROOT)
    if contract_errors:
        for error in contract_errors:
            print("Error: {}".format(error))
    if arguments.check and contract_errors:
        return 1

    added, deleted = structure_drift()

    # Check if any changes are in docs/ directory
    docs_changes = any(f.startswith("docs/") for f in added + deleted)

    if not added and not deleted and not contract_errors:
        print("No changes detected")
        return 0
    if arguments.check:
        print("AGENTS.md structure drift: %d new, %d missing" % (len(added), len(deleted)))
        return 1

    # Remove old issue body file only for scheduled or manually reported drift.
    if os.path.exists(ISSUE_BODY_FILE):
        os.remove(ISSUE_BODY_FILE)

    write_issue_body(added, deleted, contract_errors, docs_changes)

    print("Changes detected: %d new, %d missing, %d contract errors" % (len(added), len(deleted), len(contract_errors)))
    return 1 if contract_errors else 0


if __name__ == "__main__":
    sys.exit(main())