manyi/venv/Lib/site-packages/aip/kg.py

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

"""
知识图谱
"""

import re
import sys
import math
import time
from .base import AipBase
from .base import base64
from .base import json
from .base import urlencode
from .base import quote

class AipKg(AipBase):

    """
    知识图谱
    """

    __createTaskUrl = 'https://aip.baidubce.com/rest/2.0/kg/v1/pie/task_create'

    __updateTaskUrl = 'https://aip.baidubce.com/rest/2.0/kg/v1/pie/task_update'

    __taskInfoUrl = 'https://aip.baidubce.com/rest/2.0/kg/v1/pie/task_info'

    __taskQueryUrl = 'https://aip.baidubce.com/rest/2.0/kg/v1/pie/task_query'

    __taskStartUrl = 'https://aip.baidubce.com/rest/2.0/kg/v1/pie/task_start'

    __taskStatusUrl = 'https://aip.baidubce.com/rest/2.0/kg/v1/pie/task_status'

    
    def createTask(self, name, template_content, input_mapping_file, output_file, url_pattern, options=None):
        """
            创建任务
        """
        options = options or {}

        data = {}
        data['name'] = name
        data['template_content'] = template_content
        data['input_mapping_file'] = input_mapping_file
        data['output_file'] = output_file
        data['url_pattern'] = url_pattern

        data.update(options)

        return self._request(self.__createTaskUrl, data)
    
    def updateTask(self, id, options=None):
        """
            更新任务
        """
        options = options or {}

        data = {}
        data['id'] = id

        data.update(options)

        return self._request(self.__updateTaskUrl, data)
    
    def getTaskInfo(self, id, options=None):
        """
            获取任务详情
        """
        options = options or {}

        data = {}
        data['id'] = id

        data.update(options)

        return self._request(self.__taskInfoUrl, data)
    
    def getUserTasks(self, options=None):
        """
            以分页的方式查询当前用户所有的任务信息
        """
        options = options or {}

        data = {}

        data.update(options)

        return self._request(self.__taskQueryUrl, data)
    
    def startTask(self, id, options=None):
        """
            启动任务
        """
        options = options or {}

        data = {}
        data['id'] = id

        data.update(options)

        return self._request(self.__taskStartUrl, data)
    
    def getTaskStatus(self, id, options=None):
        """
            查询任务状态
        """
        options = options or {}

        data = {}
        data['id'] = id

        data.update(options)

        return self._request(self.__taskStatusUrl, data)