microsoft--promptflow/scripts/readme/readme.py

# Generate Readme file for the examples folder
import json
from pathlib import Path
import workflow_generator
import readme_generator
from jinja2 import Environment, FileSystemLoader
from ghactions_driver.readme_step import ReadmeStepsManage
from operator import itemgetter
import argparse
import sys
import os
import re

BRANCH = "main"


def get_notebook_readme_description(notebook) -> str:
    """
    Set each ipynb metadata description at .metadata.description
    """
    try:
        # read in notebook
        with open(notebook, "r", encoding="utf-8") as f:
            data = json.load(f)
        return data["metadata"]["description"]
    except Exception:
        print(f"{notebook} metadata description not set")
        return ""


def get_notebook_buildDoc_description(notebook) -> str:
    """
    Set each ipynb metadata description at .metadata.description
    """
    try:
        # read in notebook
        with open(notebook, "r", encoding="utf-8") as f:
            data = json.load(f)
        return data["metadata"]["build_doc"]
    except Exception:
        print(f"{notebook} metadata build_doc not set")
        return {}


def get_readme_description_first_sentence(readme) -> str:
    """
    Get each readme first sentence of first paragraph
    """
    try:
        with open(readme, "r", encoding="utf-8") as f:
            # read first line
            line = f.readline()
            sentence = ""
            while True:
                line = f.readline()
                if line.startswith("#"):
                    line = ""
                # skip metadata section
                if (
                    line.startswith("---")
                    or line.startswith("resources:")
                    or line.startswith("title:")
                    or line.startswith("cloud:")
                    or line.startswith("category:")
                    or line.startswith("weight:")
                ):
                    line = ""
                if line.strip() == "" and sentence != "":
                    break
                elif "." in line:
                    sentence += " " + line.split(".")[0].strip()
                    break
                else:
                    if sentence == "":
                        sentence += line.strip()
                    elif line.strip() != "":
                        sentence += " " + line.strip()
            return sentence
    except Exception:
        print(f"Error during reading {readme}")
        return ""


def write_readme(workflow_telemetries, readme_telemetries):
    global BRANCH

    ReadmeStepsManage.git_base_dir()
    readme_file = Path(ReadmeStepsManage.git_base_dir()) / "examples/README.md"

    quickstarts = {
        "readmes": [],
        "notebooks": [],
    }
    tutorials = {
        "readmes": [],
        "notebooks": [],
    }
    flex_flows = {
        "readmes": [],
        "notebooks": [],
    }
    prompty = {
        "readmes": [],
        "notebooks": [],
    }
    flows = {
        "readmes": [],
        "notebooks": [],
    }
    evaluations = {
        "readmes": [],
        "notebooks": [],
    }
    chats = {
        "readmes": [],
        "notebooks": [],
    }
    toolusecases = {
        "readmes": [],
        "notebooks": [],
    }
    connections = {
        "readmes": [],
        "notebooks": [],
    }

    for workflow_telemetry in workflow_telemetries:
        notebook_name = f"{workflow_telemetry.name}.ipynb"
        gh_working_dir = workflow_telemetry.gh_working_dir
        pipeline_name = workflow_telemetry.workflow_name
        yaml_name = f"{pipeline_name}.yml"

        # For workflows, open ipynb as raw json and
        # setup description at .metadata.description
        description = get_notebook_readme_description(workflow_telemetry.notebook)
        build_doc = get_notebook_buildDoc_description(workflow_telemetry.notebook)
        notebook_path = gh_working_dir.replace("examples/", "") + f"/{notebook_name}"
        default_workflow_item = {
            "name": notebook_name,
            "path": notebook_path,
            "pipeline_name": pipeline_name,
            "yaml_name": yaml_name,
            "description": description,
            "build_doc": build_doc,
            "title": workflow_telemetry.title.capitalize()
            if hasattr(workflow_telemetry, "title")
            else "Empty title",
            "cloud": workflow_telemetry.cloud.capitalize()
            if hasattr(workflow_telemetry, "cloud")
            else "NOT DEFINED",
            "category": workflow_telemetry.category.capitalize()
            if hasattr(workflow_telemetry, "category")
            else "General",
            "weight": workflow_telemetry.weight
            if hasattr(workflow_telemetry, "weight")
            else 0,
        }
        if gh_working_dir.startswith("examples/flows/standard"):
            flows["notebooks"].append(default_workflow_item)
        elif gh_working_dir.startswith("examples/connections"):
            connections["notebooks"].append(default_workflow_item)
        elif gh_working_dir.startswith("examples/flows/evaluation"):
            evaluations["notebooks"].append(default_workflow_item)
        elif gh_working_dir.startswith("examples/tutorials"):
            if "quickstart" in notebook_name:
                quickstarts["notebooks"].append(default_workflow_item)
            else:
                tutorials["notebooks"].append(default_workflow_item)
        elif gh_working_dir.startswith("examples/flows/chat"):
            chats["notebooks"].append(default_workflow_item)
        elif gh_working_dir.startswith("examples/flex-flows"):
            flex_flows["notebooks"].append(default_workflow_item)
        elif gh_working_dir.startswith("examples/prompty"):
            prompty["notebooks"].append(default_workflow_item)
        elif gh_working_dir.startswith("examples/tools/use-cases"):
            toolusecases["notebooks"].append(default_workflow_item)
        else:
            print(f"Unknown workflow type: {gh_working_dir}")

        # Adjust tutorial names:

        no_workflow_readmes = []

    for readme_telemetry in readme_telemetries:
        if readme_telemetry.readme_name.endswith("README.md"):
            notebook_name = readme_telemetry.readme_folder.split("/")[-1]
        else:
            notebook_name = readme_telemetry.readme_name.split("/")[-1].replace(
                ".md", ""
            )
        notebook_path = readme_telemetry.readme_name.replace("examples/", "")
        if not hasattr(readme_telemetry, "workflow_name"):
            no_workflow_readme_item = {
                "name": notebook_name,
                "path": notebook_path,
                "description": get_readme_description_first_sentence(
                    readme_telemetry.readme_name
                ),
                "title": readme_telemetry.title.capitalize()
                if hasattr(readme_telemetry, "title")
                else "Empty title",
                "cloud": readme_telemetry.cloud.capitalize()
                if hasattr(readme_telemetry, "cloud")
                else "NOT DEFINED",
                "category": readme_telemetry.category.capitalize()
                if hasattr(readme_telemetry, "category")
                else "General",
                "weight": readme_telemetry.weight
                if hasattr(readme_telemetry, "weight")
                else 0,
            }
            no_workflow_readmes.append(no_workflow_readme_item)
            continue

        pipeline_name = readme_telemetry.workflow_name
        yaml_name = f"{readme_telemetry.workflow_name}.yml"
        description = get_readme_description_first_sentence(
            readme_telemetry.readme_name
        )
        readme_folder = readme_telemetry.readme_folder

        default_readme_item = {
            "name": notebook_name,
            "path": notebook_path,
            "pipeline_name": pipeline_name,
            "yaml_name": yaml_name,
            "description": description,
            "title": readme_telemetry.title.capitalize()
            if hasattr(readme_telemetry, "title")
            else "Empty title",
            "cloud": readme_telemetry.cloud.capitalize()
            if hasattr(readme_telemetry, "cloud")
            else "NOT DEFINED",
            "category": readme_telemetry.category.capitalize()
            if hasattr(readme_telemetry, "category")
            else "General",
            "weight": readme_telemetry.weight
            if hasattr(readme_telemetry, "weight")
            else 0,
        }
        if readme_folder.startswith("examples/flows/standard"):
            flows["readmes"].append(default_readme_item)
        elif readme_folder.startswith("examples/connections"):
            connections["readmes"].append(default_readme_item)
        elif readme_folder.startswith("examples/flows/evaluation"):
            evaluations["readmes"].append(default_readme_item)
        elif readme_folder.startswith("examples/tutorials"):
            if "quickstart" in notebook_name:
                quickstarts["readmes"].append(default_readme_item)
            else:
                tutorials["readmes"].append(default_readme_item)
        elif readme_folder.startswith("examples/flows/chat"):
            chats["readmes"].append(default_readme_item)
        elif readme_folder.startswith("examples/flex-flows"):
            flex_flows["readmes"].append(default_readme_item)
        elif readme_folder.startswith("examples/prompty"):
            prompty["readmes"].append(default_readme_item)
        elif readme_folder.startswith("examples/tools/use-cases"):
            toolusecases["readmes"].append(default_readme_item)
        else:
            print(f"Unknown workflow type: {readme_folder}")

    quickstarts["notebooks"] = sorted(
        quickstarts["notebooks"],
        key=itemgetter("name"),
        reverse=True,
    )

    # Debug this replacement to check if generated correctly
    replacement = {
        "branch": BRANCH,
        "tutorials": tutorials,
        "flex_flows": flex_flows,
        "prompty": prompty,
        "flows": flows,
        "evaluations": evaluations,
        "chats": chats,
        "toolusecases": toolusecases,
        "connections": connections,
        "quickstarts": quickstarts,
    }

    print("writing README.md...")
    env = Environment(
        loader=FileSystemLoader(
            Path(ReadmeStepsManage.git_base_dir())
            / "scripts/readme/ghactions_driver/readme_templates"
        )
    )
    template = env.get_template("README.md.jinja2")
    with open(readme_file, "w") as f:
        f.write(template.render(replacement))
    print(f"finished writing {str(readme_file)}")

    # Build a table out of replacement
    # |Area|Cloud|Category|Sample|Description|
    new_items = []
    for row in replacement.keys():
        if row == "branch":
            continue
        for item in replacement[row]["notebooks"]:
            item[
                "url"
            ] = f"https://github.com/microsoft/promptflow/blob/main/examples/{item['path']}"
            item["area"] = "SDK"
            if "azure" in item["name"].lower():
                item["weight"] += 1000
            new_items.append(item)
        for item in replacement[row]["readmes"]:
            if item.get("category", "General") == "General":
                print(
                    f"Tutorial Index: Skipping {item['path']} for not having a category"
                )
                continue
            item[
                "url"
            ] = f"https://github.com/microsoft/promptflow/blob/main/examples/{item['path']}"
            item["area"] = "CLI"
            new_items.append(item)
    for item in no_workflow_readmes:
        if not item["path"].startswith("tutorials"):
            print(f"Tutorial Index: Skipping {item['path']} for not being in tutorials")
            continue
        if item.get("category", "General") == "General":
            print(f"Tutorial Index: Skipping {item['path']} for not having a category")
            continue
        item[
            "url"
        ] = f"https://github.com/microsoft/promptflow/blob/main/examples/{item['path']}"
        item["area"] = "CLI"
        new_items.append(item)

    # sort new_items by category
    tracing_category = sorted(
        [item for item in new_items if item["category"] == "Tracing"],
        key=lambda x: x["weight"],
    )
    prompty_category = sorted(
        [item for item in new_items if item["category"] == "Prompty"],
        key=lambda x: x["weight"],
    )
    flow_category = sorted(
        [item for item in new_items if item["category"] == "Flow"],
        key=lambda x: x["weight"],
    )
    deployment_category = sorted(
        [item for item in new_items if item["category"] == "Deployment"],
        key=lambda x: x["weight"],
    )
    rag_category = sorted(
        [item for item in new_items if item["category"] == "Rag"],
        key=lambda x: x["weight"],
    )

    real_new_items = [
        *tracing_category,
        *prompty_category,
        *flow_category,
        *deployment_category,
        *rag_category,
    ]
    tutorial_items = {"items": real_new_items}
    tutorial_index_file = (
        Path(ReadmeStepsManage.git_base_dir()) / "docs/tutorials/index.md"
    )
    template_tutorial = env.get_template("tutorial_index.md.jinja2")
    with open(tutorial_index_file, "w") as f:
        f.write(template_tutorial.render(tutorial_items))
    print(f"Tutorial Index: finished writing {str(tutorial_index_file)}")


def main(check):
    if check:
        # Disable print
        sys.stdout = open(os.devnull, "w")

    input_glob = ["examples/**/*.ipynb"]
    workflow_telemetry = []
    workflow_generator.main(input_glob, workflow_telemetry, check=check)

    input_glob_readme = [
        "examples/flows/**/README.md",
        "examples/flex-flows/**/README.md",
        "examples/prompty/**/README.md",
        "examples/connections/**/README.md",
        "examples/tutorials/**/*.md",
        "examples/tools/use-cases/**/README.md",
    ]
    # exclude the readme since this is 3p integration folder, pipeline generation is not included
    input_glob_readme_exclude = ["examples/flows/integrations/**/README.md"]
    readme_telemetry = []
    readme_generator.main(
        input_glob_readme, input_glob_readme_exclude, readme_telemetry
    )

    write_readme(workflow_telemetry, readme_telemetry)

    if check:
        output_object = {}
        for workflow in workflow_telemetry:
            workflow_items = re.split(r"\[|,| |\]", workflow.path_filter)
            workflow_items = list(filter(None, workflow_items))
            output_object[workflow.workflow_name] = []
            for item in workflow_items:
                if item == "examples/*requirements.txt":
                    output_object[workflow.workflow_name].append(
                        "examples/requirements.txt"
                    )
                    output_object[workflow.workflow_name].append(
                        "examples/dev_requirements.txt"
                    )
                    continue
                output_object[workflow.workflow_name].append(item)
        for readme in readme_telemetry:
            if not hasattr(readme, "workflow_name"):
                continue
            output_object[readme.workflow_name] = []
            readme_items = re.split(r"\[|,| |\]", readme.path_filter)
            readme_items = list(filter(None, readme_items))
            for item in readme_items:
                if item == "examples/*requirements.txt":
                    output_object[readme.workflow_name].append(
                        "examples/requirements.txt"
                    )
                    output_object[readme.workflow_name].append(
                        "examples/dev_requirements.txt"
                    )
                    continue
                output_object[readme.workflow_name].append(item)
        # enable output
        sys.stdout = sys.__stdout__
        return output_object
    else:
        return ""


if __name__ == "__main__":
    # setup argparse
    parser = argparse.ArgumentParser()
    parser.add_argument(
        "-c", "--check", action="store_true", help="Check what file is affected"
    )
    args = parser.parse_args()
    output = main(args.check)
    print(json.dumps(output))