#
# Licensed to the Apache Software Foundation (ASF) under one or more
# contributor license agreements.  See the NOTICE file distributed with
# this work for additional information regarding copyright ownership.
# The ASF licenses this file to You under the Apache License, Version 2.0
# (the "License"); you may not use this file except in compliance with
# the License.  You may obtain a copy of the License at
#
#    http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
#

import re
import os
import sys
import inspect
# # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # #  
#  _____ _ _   _       __     __    _ _     _       _   _                 #
# |_   _(_) |_| | ___  \ \   / /_ _| (_) __| | __ _| |_(_) ___  _ __      #  
#   | | | | __| |/ _ \  \ \ / / _` | | |/ _` |/ _` | __| |/ _ \| '_ \     #
#   | | | | |_| |  __/   \ V / (_| | | | (_| | (_| | |_| | (_) | | | |    #
#   |_| |_|\__|_|\___|    \_/ \__,_|_|_|\__,_|\__,_|\__|_|\___/|_| |_|    #
#                                                                         # 
# # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # #                                                                 


#validator for titles
def validate_title(title: str):
    return len(re.findall('(\[HUDI\-[0-9]{1,}\]|\[MINOR\])',title)) == 1

#runs an individual title test
#
#   PARAMS
# name: str - the name of the test
# title: str - the title to test
# isTrue: bool - is the title valid
#
#   RETURN
# bool - True if the test passed, False if it failed
def run_title_test(name: str, title: str, isTrue: bool):
    if isTrue != validate_title(title):
        print(f"{name} - FAILED")
        return False
    print(f"{name} - PASSED")
    return True

#tests for title validation
#
#   RETURN
# bool - True if all tests passed, False if any tests fail
def test_title():
    test_return = True
    #test that position doesn't matter for issue
    test_return = run_title_test("issue at front", "[HUDI-1324] my fake pr", True) and test_return
    test_return =  run_title_test("issue in middle", " my [HUDI-1324] fake pr", True) and test_return
    test_return =  run_title_test("issue at end", " my fake pr [HUDI-1324]", True)  and test_return

    #test position doesn't matter for minor
    test_return = run_title_test("minor at front", "[MINOR] my fake pr", True) and test_return
    test_return = run_title_test("minor in middle", " my [MINOR] fake pr", True) and test_return
    test_return = run_title_test("minor at end", " my fake pr [MINOR]", True) and test_return

    #test that more than 4 nums is also ok
    test_return = run_title_test("more than 4 nums in issue", "[HUDI-12345] my fake pr", True) and test_return

    #test that 1 nums is also ok
    test_return = run_title_test("1 num in issue", "[HUDI-1] my fake pr", True) and test_return

    #no nums not ok
    test_return = run_title_test("no nums in issue", "[HUDI-] my fake pr", False) and test_return

    #no brackets not ok
    test_return = run_title_test("no brackets around issue", "HUDI-1234 my fake pr", False) and test_return
    test_return = run_title_test("no brackets around minor", "MINOR my fake pr", False) and test_return
    
    #lowercase not ok
    test_return = run_title_test("lowercase hudi", "[hudi-1234] my fake pr", False) and test_return
    test_return = run_title_test("lowercase minor", "[minor] my fake pr", False) and test_return
    
    #duplicate not ok
    test_return = run_title_test("duplicate issue", "[HUDI-1324][HUDI-1324] my fake pr", False) and test_return
    test_return = run_title_test("duplicate minor", "[MINOR] my fake pr [MINOR]", False) and test_return
    
    #hudi and minor not ok
    test_return = run_title_test("issue and minor", "[HUDI-1324] my [MINOR]fake pr", False) and test_return
    print("*****")
    if test_return:
        print("All title tests passed")
    else:
        print("Some title tests failed")
    print("*****")

    return test_return


# # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # #
#  ____            _        __     __    _ _     _       _   _                #
# | __ )  ___   __| |_   _  \ \   / /_ _| (_) __| | __ _| |_(_) ___  _ __     #
# |  _ \ / _ \ / _` | | | |  \ \ / / _` | | |/ _` |/ _` | __| |/ _ \| '_ \    #
# | |_) | (_) | (_| | |_| |   \ V / (_| | | | (_| | (_| | |_| | (_) | | | |   #
# |____/ \___/ \__,_|\__, |    \_/ \__,_|_|_|\__,_|\__,_|\__|_|\___/|_| |_|   #
#                    |___/                                                    #
#                                                                             #
# # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # #                                            
    
#Enums for the the outcome of parsing a single line
class Outcomes:
    #error was found so we should stop parsing and exit with error
    ERROR = 0

    #continue to parse the next line
    CONTINUE = 1

    #All requirements for the current section have been met so we should start 
    #parsing the next section
    NEXTSECTION = 2

    #parsing has concluded succesfully, exit with no error
    SUCCESS = 3


#Holds the data for a section
#   PARAMS
# name: str - name of the parse section
# identifier: str - line that signifies the start of a section
# linesAfter: set of str - default lines in the template that we ignore when 
#                          verifying that the user filled out the section
class ParseSectionData:
    def __init__(self, name: str, identifier: str, linesAfter: str):
        self.name = name
        self.identifier = identifier
        self.linesAfter = linesAfter
        self.prevSection = ""
        self.nextSection = ""
        
        
    #returns true if line matches the identifier
    def identify(self, line: str):
        return line == self.identifier
    
    #returns true if user has added new text to the section
    def identifyAfter(self, line: str):
        return line not in self.linesAfter


#Special holder of data for risk level because the identifier line is modified
#by the user
class RiskLevelData(ParseSectionData):
    def __init__(self, name: str, identifier: str, linesAfter):
        super().__init__(name, identifier, linesAfter)
    
    #we check that the line start with the identifier because the identifier 
    #line will be modified when filled out correctly
    def identify(self, line: str):
        return line.startswith(self.identifier)

#Holds all the parse section data in a map
#   PARAMS
# psd: list of ParseSectionData - a list of the data for all the parse sections
class ParseSections:
    def __init__(self, psd):
        self.sections = {}
        assert len(psd) > 0
        for i in range(len(psd)):
            prevI = i - 1
            nextI = i + 1
            if prevI < 0:
                psd[i].prevSection = "START"
            else:
                psd[i].prevSection = psd[prevI].name
            
            if nextI >= len(psd):
                psd[i].nextSection = "END"
            else:
                psd[i].nextSection = psd[nextI].name
            
            self.sections[psd[i].name] = psd[i]
            

    #returns true if line is an identifier for a section that is not value
    #   PARAMS
    # line: str - the line that we are parsing
    # value: str - the name of the parse section that we are parsing
    def validateOthers(self, line: str, value: str):
        for name in self.sections:
            if name != value:
                if self.sections[name].identify(line):
                    return True
        return False    
    
    #gets the name of the section identified in the line
    #   PARAMS
    # line: str - the line that we are parsing
    #   RETURN
    # string - name of the section if the identifier is found, else none
    def getSectionName(self, line: str):
        for name in self.sections:
            if self.sections[name].identify(line):
                return name
        return None    


    #returns the ParseSectionData that is named name
    def get(self, name):
        return self.sections.get(name)


#Main class used to parse a section
class ParseSection:
    def __init__(self, data: ParseSectionData, sections: ParseSections, debug=False):
        #set to true when the sections identifier is found
        self.found = False

        #true if debug messages should be printed
        self.debug = debug

        #the data for this parse section
        self.data = data

        #data of all the parse sections
        self.sections = sections

    #prints error message if debug is set to true
    def error(self, line: str, lineno: str, message: str):
        if self.debug:
            pyline = inspect.getframeinfo(inspect.stack()[1][0]).lineno
            print(f"::error file=pr_compliance.py,line={pyline}::{message}")
            if lineno != "" and line != "":
                print(f"::error file=pr_compliance.py,line={pyline}::found on line {lineno}: {line}")
            print(f"::debug::state: {self.data.name}, found: {self.found},")

    #returns the name of the next section
    def nextSection(self):
        return self.data.nextSection

    #Returns true if we have already found the section identifier and line is 
    #not in the default template
    def validateAfter(self, line):
        return self.found and self.data.identifyAfter(line)

    #Decides what outcome occurs when the section identifier is found
    def processIdentify(self, line, lineno):
        if self.found:
            #since we have already found the identifier, this means that we have
            #found a duplicate of the identifier
            self.error(line, lineno, f"Duplicate {self.data.name} section found")
            return Outcomes.ERROR
        self.found = True
        return Outcomes.CONTINUE

    def makeValidateOthersErrorMessage(self, line):
        if self.found:
            if self.nextSection() != "END" and self.sections.sections[self.nextSection()].identify(line):
                #we found the next identifier but haven't found a description
                #yet for this section
                return f"Section {self.data.name} is missing a description"
            #we found a section other than the next section
            return f"Section {self.data.name} should be followed by section {self.data.nextSection}"

        #section identifier has not been found yet
        sectionFound = self.sections.getSectionName(line)
        if sectionFound is None:
            print("ERROR: none found even though validateOthers returned True")
            exit(1)
        elif sectionFound == self.data.prevSection:
            #we have not found the current section identifier but we found the
            #previous section identifier again
            return f"Duplicate {self.data.prevSection} section found"
         
        if self.data.prevSection == "START":
            return f"Section {self.data.name} should be the first section"
        if sectionFound == self.data.nextSection:
            return f"Missing section {self.data.name} between {self.data.prevSection} and {self.data.nextSection}"
        return f"Section {self.data.name} was expected after section {self.data.prevSection}"
    
    #Decides the outcome state by processing line
    def validateLine(self,line,lineno):
        if self.data.identify(line):
            #we have found the identifier so we should decide what to do
            return self.processIdentify(line,lineno)
        elif self.sections.validateOthers(line, self.data.name):
            #we have found the identifier for another section
            #figure out what the error is
            self.error(line,lineno,self.makeValidateOthersErrorMessage(line))
            return Outcomes.ERROR
        elif self.validateAfter(line):
            #the pr author has added new text to this section so we consider it
            #to be filled out
            if self.nextSection() == "END":
                #if next section is "END" then there are no more sections 
                #to process
                return Outcomes.SUCCESS
            return Outcomes.NEXTSECTION
        return Outcomes.CONTINUE

#We do not check this section for data
#currently just used for the checklist where they just need to check boxes
class NoDataSection(ParseSection):
    def __init__(self, data: ParseSectionData, sections: ParseSections, debug=False):
        super().__init__(data, sections, debug)

    #After finding the identifier we don't need to look for anything else so we
    #can just go to the next section or terminate if this is the last
    def processIdentify(self, line, lineno):
        o = super().processIdentify(line, lineno)
        if o  == Outcomes.CONTINUE:
            if self.nextSection() == "END":
                return Outcomes.SUCCESS
            else:
                return Outcomes.NEXTSECTION
        return o

#Class that orchestrates the validation of the entire body
class ValidateBody:
    def __init__(self, body: str, firstSection: str, sections: ParseSections, debug=False):
        #the body of the pr post
        self.body = body

        #true if debug messages should be printed
        self.debug = debug

        #the name of the first section of the post
        self.firstSection = firstSection

        #the current section we are processing
        self.section = None

        #ParseSections which holds the data for all the sections
        self.sections = sections

    #Updates self.section to the next section
    def nextSection(self):
        #get the name of the section to parse
        sectionName = self.firstSection
        if self.section is not None:
            sectionName = self.section.nextSection()

        #get the data for that section
        data = self.sections.get(sectionName)
        if data is None:
            print(f"ERROR with your parse section setup. Parse section {sectionName} not found")
            exit(-3)
        
        #create the section
        if data.name == "CHECKLIST":
            self.section = NoDataSection(data=data, sections=self.sections, debug=self.debug)
        else:
            self.section = ParseSection(data=data, sections=self.sections, debug=self.debug)

    #Returns true if the body complies with the validation rules else false
    def validate(self):
        #instantiate self.section since it starts null
        self.nextSection()

        #validate each line
        for lineno, line in enumerate(self.body.splitlines(), 1):
            #ignore empty lines
            if len(line) == 0:
                continue

            #run the parse section validation
            o = self.section.validateLine(line, lineno)
            
            #decide what to do based on outcome
            if o == Outcomes.ERROR:
                return False
            elif o == Outcomes.SUCCESS:
                return True
            elif o == Outcomes.NEXTSECTION:
                self.nextSection()
        #if we get through all the lines without a success outcome, then the 
        #body does not comply
        if self.section.data.nextSection == "END":
            if self.section.found:
                self.section.error("","",f"Section {self.section.data.name} is missing a description")
                return False
            self.section.error("","",f"Missing section {self.section.data.name} at the end")
            return False
        self.section.error("","", "Please make sure you have filled out the template correctly. You can find a blank template in /.github/PULL_REQUEST_TEMPLATE.md")
        return False
        
#Generate the validator for the current template.
#needs to be manually updated
def make_default_validator(body, debug=False):
    changelogs = ParseSectionData("CHANGELOGS",
        "### Change Logs",
        {"_Describe context and summary for this change. Highlight if any code was copied._"})
    impact = ParseSectionData("IMPACT",
        "### Impact",
        {"_Describe any public API or user-facing feature change or any performance impact._"})
    risklevel = RiskLevelData("RISKLEVEL",
        "### Risk level",
        {"_If medium or high, explain what verification was done to mitigate the risks._"})
    checklist = ParseSectionData("CHECKLIST",
        "### Contributor's checklist",
        {})
    parseSections = ParseSections([changelogs, impact, risklevel, checklist])

    return ValidateBody(body, "CHANGELOGS", parseSections, debug)


#takes a list of strings and returns a string of those lines separated by \n
def joinLines(lines):
    return "\n".join(lines)

#runs a test for parsing the body
#   PARAMS
# name: str - the name of the test
# body: str - the body to parse
# isTrue: bool - True if the body complies with our validation rules
# debug: bool - True if we want to print debug information
def run_test(name: str, body: str, isTrue: bool, debug: bool):
    validator = make_default_validator(body, debug)
    if isTrue != validator.validate():
        print(f"{name} - FAILED")
        return False
    print(f"{name} - PASSED")
    return True

# Given a list of sections which are lists of strings, it combines them into one
# giant string that is the body to be parsed
def build_body(sections):
    res = ""
    for s in sections:
        res += joinLines(s) + "\n"
    return res

# Tests for validating the body of a pr. Returns true if all tests pass
def test_body():
    DEBUG_MESSAGES = False
    #Create sections that we will combine to create bodies to test validation on
    template_changelogs = [
         "### Change Logs",
        "",
        "_Describe context and summary for this change. Highlight if any code was copied._",
        ""
    ]

    good_changelogs = template_changelogs.copy()
    good_changelogs[1] = "changelogs description"

    template_impact = [
        "### Impact",
        "",
        "_Describe any public API or user-facing feature change or any performance impact._",
        ""
    ]

    good_impact = template_impact.copy()
    good_impact[1] = "impact description"

    template_risklevel = [
        "### Risk level (write none, low medium or high below)",
        "",
        "_If medium or high, explain what verification was done to mitigate the risks._",
        ""
    ]

    good_risklevel = template_risklevel.copy()
    good_risklevel[1] = "none"

    template_checklist = [
        "### Contributor's checklist",
        "",
        "- [ ] Read through [contributor's guide](https://hudi.apache.org/contribute/how-to-contribute)",
        "- [ ] Change Logs and Impact were stated clearly",
        "- [ ] Adequate tests were added if applicable",
        "- [ ] CI passed"
    ]

    #list of sections that when combined form a valid body
    good_sections = [good_changelogs, good_impact, good_risklevel, template_checklist]

    #list of sections that when combined form the template
    template_sections = [template_changelogs, template_impact, template_risklevel, template_checklist]

    tests_passed = True
    #Test section not filled out
    #no need to test checklist section
    for i in range(len(good_sections)-1):
        test_sections = []
        for j in range(len(good_sections)):
            if j != i:
                test_sections.append(good_sections[j].copy())
            else:
                test_sections.append(template_sections[j].copy())
        tests_passed = run_test(f"template section not filled out: {i}", build_body(test_sections), False, DEBUG_MESSAGES) and tests_passed

    #Test duplicate section
    for i in range(len(good_sections)-1):
        test_sections = []
        for j in range(len(good_sections)):
            test_sections.append(good_sections[j].copy())
            if j == i:
                test_sections.append(good_sections[j].copy())
        tests_passed = run_test(f"duplicate section: {i}", build_body(test_sections), False, DEBUG_MESSAGES) and tests_passed
    
    #Test out of order section
    for i in range(len(good_sections)-1):
        test_sections = []
        for j in range(len(good_sections)):
            test_sections.append(good_sections[j].copy())
        for k in range(i+1,len(good_sections)):
            test_sections[i], test_sections[k] = test_sections[k],test_sections[i]
            tests_passed = run_test(f"Swapped sections: {i}, {k}", build_body(test_sections), False, DEBUG_MESSAGES) and tests_passed

    #Test missing section
    for i in range(len(good_sections)):
        test_sections = []
        for j in range(len(good_sections)):
            if i != j:
                test_sections.append(good_sections[j].copy())
        tests_passed = run_test(f"Missing Section: {i}", build_body(test_sections), False, DEBUG_MESSAGES) and tests_passed
    
    #Test good body:
    tests_passed = run_test("good documentation", build_body(good_sections), True, DEBUG_MESSAGES) and tests_passed

    print("*****")
    if tests_passed:
        print("All body tests passed")
    else:
        print("Some body tests failed")
    print("*****")
    
    return tests_passed


if __name__ == '__main__':
    if len(sys.argv) > 1:
        title_tests = test_title()
        body_tests = test_body()
        if title_tests and body_tests:
            exit(0)
        else:
            exit(-1)


    title = os.getenv("REQUEST_TITLE")
    body = os.getenv("REQUEST_BODY")

    if title is None:
        print("no title")
        exit(-1)

    if not validate_title(title):
        print("invalid title")
        exit(-1)

    if body is None:
        print("no pr body")
        exit(-1)

    validator = make_default_validator(body,True)
    if not validator.validate():
        exit(-1)
    exit(0)