def picoKPE()

in src/plugins/utils/kpe.py [0:0]

• 46 lines of code
• 10 McCabe index (conditional complexity)

def picoKPE(KibbleBit, bodies):
    """ KPE using picoAPI Text Analysis """
    if 'picoapi' in KibbleBit.config:
        headers = {
            'Content-Type': 'application/json',
            'PicoAPI-Key': KibbleBit.config['picoapi']['key']
        }
        
        
        js = {
            "texts": []
          }
        
        # For each body...
        a = 0
        KPEs = []
        for body in bodies:
            body = trimBody(body)
            
            doc = {
                "id": str(a),
                "body": body
              }
            js['texts'].append(doc)
            KPEs.append({}) # placeholder for each doc, to be replaced
            a += 1
        try:
            rv = requests.post(
                "https://v1.picoapi.com/api/text/keyphrase",
                headers = headers,
                data = json.dumps(js)
            )
            jsout = rv.json()
        except:
            jsout = {} # borked sentiment analysis?
        
        if 'results' in jsout and len(jsout['results']) > 0:
            for doc in jsout['results']:
                phrases = []
                # This is a bit different than Azure, in that it has a weighting score
                # So we need to just extract key phrases above a certain level.
                # Grab up o 5 key phrases per text
                MINIMUM_WEIGHT = 0.02
                for element in doc['keyphrases']:
                    if element['score'] > MINIMUM_WEIGHT:
                        phrases.append(element['phrase'])
                    if len(phrases) == 5:
                        break
                KPEs[int(doc['id'])] = phrases # Replace KPEs[X] with the actual phrases
                
        else:
            KibbleBit.pprint("Failed to analyze email body.")
            print(jsout)
            # 403 returned on invalid key, 429 on rate exceeded.
            # If we see a code return, let's just stop for now.
            # Later scans can pick up the slack.
            if 'code' in jsout:
                KibbleBit.pprint("Possible rate limiting in place, stopping for now.")
                return False
        return KPEs