iFLYTEK_Chinese_Semantic_Checker.ipynb

{
 "cells": [
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# iFLYTEK Chinese Semantic Checker\n",
    "\n",
    "This notebook is used to check the semantic of Chinese language unit. It is based on iFLYTEK's API.\n",
    "\n",
    "Author: Zexin Xu, Zilu Zhang"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [],
   "source": [
    "import pandas as pd"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## iFLYTEK API\n",
    "\n",
    "This snippet invloves the iFLYTEK API. `API_key` and `Secret_key` are deleted for security reasons. Please use your own API key and secret key. For more details of the usage of iFLYTEK API, please refer to [iFLYTEK API](https://www.xfyun.cn/doc/nlp/textCorrection/API.html#%E8%BF%94%E5%9B%9E%E7%BB%93%E6%9E%9C). If you are a English user, there is a language switch button on the top right corner."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [],
   "source": [
    "# -*- coding:utf-8 -*-\n",
    "from datetime import datetime\n",
    "from wsgiref.handlers import format_date_time\n",
    "from time import mktime\n",
    "import hashlib\n",
    "import base64\n",
    "import hmac\n",
    "from urllib.parse import urlencode\n",
    "import json\n",
    "import requests\n",
    "\n",
    "\n",
    "class AssembleHeaderException(Exception):\n",
    "    def __init__(self, msg):\n",
    "        self.message = msg\n",
    "\n",
    "\n",
    "class Url:\n",
    "    def __init__(this, host, path, schema):\n",
    "        this.host = host\n",
    "        this.path = path\n",
    "        this.schema = schema\n",
    "        pass\n",
    "\n",
    "\n",
    "class WebsocketDemo:\n",
    "    def __init__(self,APPId,APISecret,APIKey,Text):\n",
    "        self.appid = APPId\n",
    "        self.apisecret = APISecret\n",
    "        self.apikey = APIKey\n",
    "        self.text = Text\n",
    "        self.url = 'https://api.xf-yun.com/v1/private/s9a87e3ec'\n",
    "\n",
    "    # calculate sha256 and encode to base64\n",
    "    def sha256base64(self,data):\n",
    "        sha256 = hashlib.sha256()\n",
    "        sha256.update(data)\n",
    "        digest = base64.b64encode(sha256.digest()).decode(encoding='utf-8')\n",
    "        return digest\n",
    "\n",
    "\n",
    "    def parse_url(self,requset_url):\n",
    "        stidx = requset_url.index(\"://\")\n",
    "        host = requset_url[stidx + 3:]\n",
    "        schema = requset_url[:stidx + 3]\n",
    "        edidx = host.index(\"/\")\n",
    "        if edidx <= 0:\n",
    "            raise AssembleHeaderException(\"invalid request url:\" + requset_url)\n",
    "        path = host[edidx:]\n",
    "        host = host[:edidx]\n",
    "        u = Url(host, path, schema)\n",
    "        return u\n",
    "\n",
    "\n",
    "    # build websocket auth request url\n",
    "    def assemble_ws_auth_url(self,requset_url, method=\"POST\", api_key=\"\", api_secret=\"\"):\n",
    "        u = self.parse_url(requset_url)\n",
    "        host = u.host\n",
    "        path = u.path\n",
    "        now = datetime.now()\n",
    "        date = format_date_time(mktime(now.timetuple()))\n",
    "        #print(date)\n",
    "        # date = \"Thu, 12 Dec 2019 01:57:27 GMT\"\n",
    "        signature_origin = \"host: {}\\ndate: {}\\n{} {} HTTP/1.1\".format(host, date, method, path)\n",
    "        #print(signature_origin)\n",
    "        signature_sha = hmac.new(api_secret.encode('utf-8'), signature_origin.encode('utf-8'),\n",
    "                                 digestmod=hashlib.sha256).digest()\n",
    "        signature_sha = base64.b64encode(signature_sha).decode(encoding='utf-8')\n",
    "        authorization_origin = \"api_key=\\\"%s\\\", algorithm=\\\"%s\\\", headers=\\\"%s\\\", signature=\\\"%s\\\"\" % (\n",
    "            api_key, \"hmac-sha256\", \"host date request-line\", signature_sha)\n",
    "        authorization = base64.b64encode(authorization_origin.encode('utf-8')).decode(encoding='utf-8')\n",
    "        #print(authorization_origin)\n",
    "        values = {\n",
    "            \"host\": host,\n",
    "            \"date\": date,\n",
    "            \"authorization\": authorization\n",
    "        }\n",
    "\n",
    "        return requset_url + \"?\" + urlencode(values)\n",
    "\n",
    "\n",
    "    def get_body(self):\n",
    "        body =  {\n",
    "            \"header\": {\n",
    "                \"app_id\": self.appid,\n",
    "                \"status\": 3,\n",
    "                #\"uid\":\"your_uid\"\n",
    "            },\n",
    "            \"parameter\": {\n",
    "                \"s9a87e3ec\": {\n",
    "                    #\"res_id\":\"your_res_id\",\n",
    "                    \"result\": {\n",
    "                        \"encoding\": \"utf8\",\n",
    "                        \"compress\": \"raw\",\n",
    "                        \"format\": \"json\"\n",
    "                    }\n",
    "                }\n",
    "            },\n",
    "            \"payload\": {\n",
    "                \"input\": {\n",
    "                    \"encoding\": \"utf8\",\n",
    "                    \"compress\": \"raw\",\n",
    "                    \"format\": \"plain\",\n",
    "                    \"status\": 3,\n",
    "                    \"text\": base64.b64encode(self.text.encode(\"utf-8\")).decode('utf-8')\n",
    "                }\n",
    "            }\n",
    "        }\n",
    "        return body\n",
    "\n",
    "    def get_result(self):\n",
    "        request_url = self.assemble_ws_auth_url(self.url, \"POST\", self.apikey, self.apisecret)\n",
    "        headers = {'content-type': \"application/json\", 'host':'api.xf-yun.com', 'app_id':self.appid}\n",
    "        body = self.get_body()\n",
    "        response = requests.post(request_url, data = json.dumps(body), headers = headers)\n",
    "        # print('onMessage：\\n' + response.content.decode())\n",
    "        tempResult = json.loads(response.content.decode())\n",
    "        # print('text字段解析：\\n' + base64.b64decode(tempResult['payload']['result']['text']).decode())\n",
    "        return json.loads(base64.b64decode(tempResult['payload']['result']['text']).decode())\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "#NOTE Put your dataframes here, and modify the column names accordingly\n",
    "tunit_df = pd.DataFrame()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def get_result(result):\n",
    "    if len(result) > 0:\n",
    "        print(str(result))\n",
    "        return str(result)\n",
    "    else:\n",
    "        return pd.NA\n",
    "    \n",
    "def generate_result(df):\n",
    "    APPId = \"\"\n",
    "    APISecret = \"\"\n",
    "    APIKey = \"\"\n",
    "\n",
    "    for i, row in df.loc[:, :].iterrows():\n",
    "        demo = WebsocketDemo(APPId, APISecret, APIKey, row['sentences'])\n",
    "        result = demo.get_result()\n",
    "        df.loc[i, \"政治术语纠错\"] = get_result(result['pol']) \n",
    "        df.loc[i, \"别字纠错\"] = get_result(result['char']) \n",
    "        df.loc[i, \"别词纠错\"] = get_result(result['word'])\n",
    "        df.loc[i, \"语法纠错-冗余\"] = get_result(result['redund']) \n",
    "        df.loc[i, \"语法纠错-缺失\"] = get_result(result['miss']) \n",
    "        df.loc[i, \"语法纠错-乱序\"] = get_result(result['order']) \n",
    "        df.loc[i, \"搭配纠错\"] = get_result(result['dapei']) \n",
    "        df.loc[i, \"标点纠错\"] = get_result(result['punc']) \n",
    "        df.loc[i, \"成语纠错\"] = get_result(result['idm']) \n",
    "        df.loc[i, \"机构名纠错\"] = get_result(result['org']) \n",
    "        df.loc[i, \"领导人职称纠错\"] = get_result(result['leader']) \n",
    "        df.loc[i, \"数字纠错\"] = get_result(result['number']) \n",
    "        df.loc[i, \"地名纠错\"] = get_result(result['addr'])\n",
    "        df.loc[i, \"全文人名纠错\"] = get_result(result['name']) \n",
    "        df.loc[i, \"句式杂糅/语义重复\"] = get_result(result['grammar_pc']) \n",
    "        if i % 100 == 0:\n",
    "            print(i, \"iters done...\")\n",
    "\n",
    "kd_tunit_df = tunit_df.copy()\n",
    "generate_result(kd_tunit_df)"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Result processsing"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {},
   "outputs": [],
   "source": [
    "kd_tunit_result = pd.read_csv('kedaxunfei/tunit_df_result.csv', encoding='utf-8')\n",
    "kd_sen_result = pd.read_csv('kedaxunfei/sen_df_result.csv', encoding='utf-8')\n",
    "\n",
    "# Process the result to combine all types of error message into label\n",
    "def result_processing(df, tru_df):\n",
    "    df['纠错数'] = 15 - df.apply(lambda x: x.isnull().sum(), axis='columns')\n",
    "    for i, row in df.iterrows():\n",
    "        df.loc[i, 'prediction_label'] = False if row['纠错数'] > 0 else True\n",
    "    df['ground_truth_label'] = tru_df['ground_truth_label']\n",
    "\n",
    "# Plug in your ground truth dataframe here as second args\n",
    "result_processing(kd_tunit_result, tunit_df)\n",
    "kd_tunit_result.to_csv('kedaxunfei/tunit_df_result_mod.csv', index=False, encoding='utf-8-sig')\n"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.9.10"
  },
  "orig_nbformat": 4,
  "vscode": {
   "interpreter": {
    "hash": "75273f6ed8af91899ebc591bf6ae2fd0716c5db2515d7097a000123632f4e53a"
   }
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}