promptflow

1
import argparse
2
from pathlib import Path
3
from functools import reduce
4

5
from ghactions_driver.readme_workflow_generate import write_readme_workflow
6
from ghactions_driver.readme_step import ReadmeStepsManage, ReadmeSteps
7
from ghactions_driver.readme_parse import readme_parser
8

9
from ghactions_driver.telemetry_obj import Telemetry
10

11

12
def local_filter(callback, array: [Path]):
13
    results = []
14
    backups = []
15
    for index, item in enumerate(array):
16
        result = callback(item, index, array)
17
        # if returned true, append item to results
18
        if result:
19
            results.append(item)
20
        else:
21
            backups.append(item)
22
    return results, backups
23

24

25
def no_readme_generation_filter(item: Path, index, array) -> bool:
26
    """
27
    If there is no steps in the readme, then no generation
28
    """
29
    try:
30
        if 'build' in str(item):  # skip build folder
31
            return False
32

33
        full_text = readme_parser(item.relative_to(ReadmeStepsManage.git_base_dir()))
34
        if full_text == "":
35
            return False
36
        else:
37
            return True
38
    except Exception as error:
39
        print(error)
40
        return False  # generate readme
41

42

43
def main(input_glob, exclude_glob=[], output_files=[]):
44
    def set_add(p, q):
45
        return p | q
46

47
    def set_difference(p, q):
48
        return p - q
49

50
    globs = reduce(set_add, [set(Path(ReadmeStepsManage.git_base_dir()).glob(p)) for p in input_glob], set())
51
    globs_exclude = reduce(set_difference,
52
                           [set(Path(ReadmeStepsManage.git_base_dir()).glob(p)) for p in exclude_glob],
53
                           globs)
54
    readme_items = sorted([i for i in globs_exclude])
55

56
    readme_items, no_generation_files = local_filter(no_readme_generation_filter, readme_items)
57
    for readme in readme_items:
58
        readme_telemetry = Telemetry()
59
        workflow_name = readme.relative_to(ReadmeStepsManage.git_base_dir())
60
        # Deal with readme
61
        write_readme_workflow(workflow_name.resolve(), readme_telemetry)
62
        ReadmeSteps.cleanup()
63
        output_files.append(readme_telemetry)
64
    for readme in no_generation_files:
65
        readme_telemetry = Telemetry()
66
        from ghactions_driver.resource_resolver import resolve_tutorial_resource
67
        try:
68
            resolve_tutorial_resource(
69
                "TEMP", readme.resolve(), readme_telemetry
70
            )
71
        except Exception:
72
            pass
73
        readme_telemetry.readme_name = str(readme.relative_to(ReadmeStepsManage.git_base_dir()))
74
        readme_telemetry.readme_folder = str(readme.relative_to(ReadmeStepsManage.git_base_dir()).parent)
75
        output_files.append(readme_telemetry)
76

77

78
if __name__ == "__main__":
79
    # setup argparse
80
    parser = argparse.ArgumentParser()
81
    parser.add_argument(
82
        "-g",
83
        "--input-glob",
84
        nargs="+",
85
        help="Input Readme.md glob example 'examples/flows/**/Readme.md'",
86
    )
87
    args = parser.parse_args()
88

89
    # call main
90
    main(args.input_glob)
91