[PYTHON] To people who are "recruiting but not recruiting"

Import

abe.py

# -*- coding:utf-8 -*-

import os
import urllib.request
import json
import configparser
import codecs
import csv
import sys
import sqlite3
from collections import namedtuple
import types

COTOHA

abe.py

#/_/_/_/_/_/_/_/_/_/_/_/_/COTOHA_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/
#The code for this part is taken from here.
# https://qiita.com/gossy5454/items/83072418fb0c5f3e269f

class CotohaApi:
    #Initialization
    def __init__(self, client_id, client_secret, developer_api_base_url, access_token_publish_url):
        self.client_id = client_id
        self.client_secret = client_secret
        self.developer_api_base_url = developer_api_base_url
        self.access_token_publish_url = access_token_publish_url
        self.getAccessToken()

    #Get access token
    def getAccessToken(self):
        #Access token acquisition URL specification
        url = self.access_token_publish_url

        #Header specification
        headers={
            "Content-Type": "application/json;charset=UTF-8"
        }

        #Request body specification
        data = {
            "grantType": "client_credentials",
            "clientId": self.client_id,
            "clientSecret": self.client_secret
        }
        #Encode request body specification to JSON
        data = json.dumps(data).encode()

        #Request generation
        req = urllib.request.Request(url, data, headers)

        #Send a request and receive a response
        res = urllib.request.urlopen(req)

        #Get response body
        res_body = res.read()

        #Decode the response body from JSON
        res_body = json.loads(res_body)

        #Get an access token from the response body
        self.access_token = res_body["access_token"]


    #Parsing API
    def parse(self, sentence):
        #Parsing API URL specification
        url = self.developer_api_base_url + "v1/parse"
        #Header specification
        headers={
            "Authorization": "Bearer " + self.access_token,
            "Content-Type": "application/json;charset=UTF-8",
        }
        #Request body specification
        data = {
            "sentence": sentence
        }
        #Encode request body specification to JSON
        data = json.dumps(data).encode()
        #Request generation
        req = urllib.request.Request(url, data, headers)
        #Send a request and receive a response
        try:
            res = urllib.request.urlopen(req)
        #What to do if an error occurs in the request
        except urllib.request.HTTPError as e:
            #If the status code is 401 Unauthorized, get the access token again and request again
            if e.code == 401:
                print ("get access token")
                self.access_token = getAccessToken(self.client_id, self.client_secret)
                headers["Authorization"] = "Bearer " + self.access_token
                req = urllib.request.Request(url, data, headers)
                res = urllib.request.urlopen(req)
            #Show cause for errors other than 401
            else:
                print ("<Error> " + e.reason)

        #Get response body
        res_body = res.read()
        #Decode the response body from JSON
        res_body = json.loads(res_body)
        #Get analysis result from response body
        return res_body


    #Similarity calculation API
    def similarity(self, s1, s2):
        #Similarity calculation API URL specification
        url = self.developer_api_base_url + "v1/similarity"
        #Header specification
        headers={
            "Authorization": "Bearer " + self.access_token,
            "Content-Type": "application/json;charset=UTF-8",
        }
        #Request body specification
        data = {
            "s1": s1,
            "s2": s2
        }
        #Encode request body specification to JSON
        data = json.dumps(data).encode()
        #Request generation
        req = urllib.request.Request(url, data, headers)
        #Send a request and receive a response
        try:
            res = urllib.request.urlopen(req)
        #What to do if an error occurs in the request
        except urllib.request.HTTPError as e:
            #If the status code is 401 Unauthorized, get the access token again and request again
            if e.code == 401:
                print ("get access token")
                self.access_token = getAccessToken(self.client_id, self.client_secret)
                headers["Authorization"] = "Bearer " + self.access_token
                req = urllib.request.Request(url, data, headers)
                res = urllib.request.urlopen(req)
            #Show cause for errors other than 401
            else:
                print ("<Error> " + e.reason)

        #Get response body
        res_body = res.read()
        #Decode the response body from JSON
        res_body = json.loads(res_body)
        #Get analysis result from response body
        return res_body

Convert to continuous connection

abe.py

#/_/_/_/_/_/_/_/_/_/_/_/_/CONVERSION_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/

def convert(word):
    file_name = "./data/Verb.csv"
    with open(file_name,"r") as f:
        handler = csv.reader(f)
        for row in handler:
            if word == row[10]: #Part of speech discovery
                if "Continuous connection" in row[9]: #Utilization discovery
                    return row[0]
    return None

Synonyms

abe.py

#/_/_/_/_/_/_/_/_/_/_/_/_/SYNONYM_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/
#The code for this part is taken from here.
# https://www.yoheim.net/blog.php?q=20160201

conn = sqlite3.connect("./data/wnjpn.db")

Word = namedtuple('Word', 'wordid lang lemma pron pos')

def getWords(lemma):
  cur = conn.execute("select * from word where lemma=?", (lemma,))
  return [Word(*row) for row in cur]


Sense = namedtuple('Sense', 'synset wordid lang rank lexid freq src')

def getSenses(word):
  cur = conn.execute("select * from sense where wordid=?", (word.wordid,))
  return [Sense(*row) for row in cur]

Synset = namedtuple('Synset', 'synset pos name src')

def getSynset(synset):
  cur = conn.execute("select * from synset where synset=?", (synset,))
  return Synset(*cur.fetchone())

def getWordsFromSynset(synset, lang):
  cur = conn.execute("select word.* from sense, word where synset=? and word.lang=? and sense.wordid = word.wordid;", (synset,lang))
  return [Word(*row) for row in cur]

def getWordsFromSenses(sense, lang="jpn"):
  synonym = {}
  for s in sense:
    lemmas = []
    syns = getWordsFromSynset(s.synset, lang)
    for sy in syns:
      lemmas.append(sy.lemma)
    synonym[getSynset(s.synset).name] = lemmas
  return synonym

def getSynonym (word):
    synonym = {}
    words = getWords(word)
    if words:
        for w in words:
            sense = getSenses(w)
            s = getWordsFromSenses(sense)
            synonym = dict(list(synonym.items()) + list(s.items()))
    return synonym

Main

abe.py

#/_/_/_/_/_/_/_/_/_/_/_/_/MAIN_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/
if __name__ == '__main__':
    #Get the location of the source file
    APP_ROOT = os.path.dirname(os.path.abspath( __file__)) + "/"

    #Get set value
    config = configparser.ConfigParser()
    config.read(APP_ROOT + "config.ini")
    CLIENT_ID = config.get("COTOHA API", "Developer Client id")
    CLIENT_SECRET = config.get("COTOHA API", "Developer Client secret")
    DEVELOPER_API_BASE_URL = config.get("COTOHA API", "Developer API Base URL")
    ACCESS_TOKEN_PUBLISH_URL = config.get("COTOHA API", "Access Token Publish URL")

    #COTOHA API instantiation
    cotoha_api = CotohaApi(CLIENT_ID, CLIENT_SECRET, DEVELOPER_API_BASE_URL, ACCESS_TOKEN_PUBLISH_URL)

    #Analysis target sentence
    if len(sys.argv) >= 2:
        sentence = sys.argv[1]
    else:
        raise TypeError

    #Take a verb from the original sentence and convert it to a continuous form connection
    result = cotoha_api.parse(sentence)
    ret = ""
    verb = ""
    for chunk in result["result"]:
        for token in chunk["tokens"]:
            if token["pos"] == "Verb stem":
                verb = token["lemma"]
                form = token["form"]
                conv_verb = convert(verb)
                if conv_verb==None:
                    ret += form
                else:
                    ret += conv_verb

                if ret[-1] == "Hmm":
                    ret += "But"
                else:
                    ret += "Yes, but"
                break
            else:
                ret += token["form"]

    #Take synonyms for verbs
    synonym = getSynonym(verb)
    noun = ""
    sim = 0.

    #Extract synonyms for the most similar nouns
    for syns in synonym.values():
        for syn in syns:
            result = cotoha_api.parse(syn)['result'][0]['tokens'][0]
            if result['pos'] == 'noun':
                cand = result['form']
                cand_sim = cotoha_api.similarity(sentence, cand+'To do')['result']['score']
                if cand_sim > sim:
                    noun = result['form']
                    sim = cand_sim
    ret += noun
    ret += "Not done"

    #Final output
    print(ret)

config.ini

config.ini

#To use the COTOHA API, register with the COTOHA API to get an ID and SECRET, and
# config.You need to create an ini file.
# https://api.ce-cotoha.com/contents/index.html

[COTOHA API]
Developer API Base URL: https://api.ce-cotoha.com/api/dev/nlp/
Developer Client id: IDIDIDIDIDIDIDIDIDIDIDIDIDIDIDI
Developer Client secret: SECRETSECRETSECRETSECRET
Access Token Publish URL: https://api.ce-cotoha.com/v1/oauth/accesstokens

$ python abe.py "Stay at the hotel"
I'm staying at a hotel, but I'm not staying

$ python abe.py "answer the questions"
Answered the question but not answered

$ python abe.py "Sleep at night"
I sleep at night, but I don't sleep

$ python abe.py "Walk outside"
I'm walking outside, but I'm not walking

$ python abe.py "View the net"
I'm looking at the net, but I haven't checked

$ python abe.py "Buy meat"
I'm buying meat, but not

$ python abe.py "Burn the fire"
I'm burning a fire, but I'm not burning