- author : Shuxin_Wang
- email : 213202122@seu.edu.cn
- time : 2022/6/2
- version: 0.4.7
目录
- 简单查询，只有
SearchManage一个查询类，单线程查询，简单解析；
- 新增
Entities,EntitiesSearch基本类，Tools.py提供功能函数接口; - 支持多线程查询;
- 支持N维列表数据输入；
- 新增解析工具函数，解析类;
- （key='ids'时）支持正则表达目标解析，支持自定义解析函数的接口；
- 支持wikidata数据模型中7种值类型数据的解析；
- 重新规划文件位置，类
SearchManage作为类EntitiesSearch的子类 - 修复线程数过大导致的错误；
- 新增
Wikipedia类，可实现wikipedia.suggest(query)的多线程调用； - 解析claims下的具体值时，每个值的基本格式为元组，元组基本结构形如 (
值类型，主值1, ···)； - 修复大量bug，提高稳定性；
- 新增类
SparqlQuery可进行自定义多线程SPARQL查询； - 新增类
BingQuery多线程爬取模糊搜索结果 - 工具函数作为工具类下的静态方法；
- 类
Entities继承类JsonAnalysis，类Wikipedia继承类EntitiesSearch； - 引入**线程池
<concurrent.futures.ThreadPoolExecutor>代替基本线程事务**，进一步提高性能；
- 新增
SpellCheck进行多线程爬取拼写纠错结果； - 完成底层接口的注释；
- 重新整理文件目录和模块结构；
- 完成全英注释；
- 新增类
DbpediaLookUp针对DBpedia知识图谱的多线程查询； - 修改部分bug；
- 可通过爬取ask(
https://www.ask.com/web)搜索进行SpellCheck，并提供相应解析函数AnalysisTools.ask_analysis；
classDiagram
Entities ..> AnalysisTools:依赖
Entities ..> Tools:依赖
EntitiesSearch ..> Tools:依赖
Entities --|>RequestAnalysis : 继承
EntitiesSearch ..o Entities: 集合
SearchManage --|> EntitiesSearch :继承
Wikipedia --|> EntitiesSearch:继承
SparqlQuery --|> EntitiesSearch:继承
BingQuery --|> EntitiesSearch:继承
SpellCheck --|> EntitiesSearch:继承
DbpediaLookUp --|> EntitiesSearch:继承
class Tools{
@ staticmethod
}
class AnalysisTools{
@ staticmethod
}
class RequestAnalysis{
- request
- analysis
- ready_
}
class Entities{
- index
- entities
- params
}
class EntitiesSearch{
+ entities_num
+ key
+ keys
+ m_num
+ Queue[Entities]: search_queue
+ Queue[Entites]: re_queue
+ List[Entities]: re_list
+ queryParams
- index_
}
class SearchManage{
- url
}
class Wikipedia{
}
class SparqlQuery{
- url_
- returnFormat
}
class BingQuery{
- url_
}
class SpellCheck{
- url_
}
class DbpediaLookUp{
}
from searchmanage import SearchManage
search_m1 = SearchManage(key='search', m_num=20)
search_m2 = SearchManage(key='ids', m_num=10)
"""
url_api(str): 查询网址，默认为"https://www.wikidata.org/w/api.php";
key(str): 查询action,'search' 或者 'ids';
m_num(int): 指定线程数目;
""" def search_run(self, points: list, keys: Union[str, List[str]] = None,
timeout: float = 30.0, time_stop: float = 30.0, block_num: int = 10,
function_=None, args: tuple = None, **kwargs) -> dict- 输入N维文本或ID列表，多线程查询并根据解析关键词返回解析结果
- 参数说明
points(list)：N维文本或ID列表；keys(Union[str, List[str]])：解析关键词或其列表；timeout(float)：每个线程查询的最大限制时间，到超过这个时间，会视为查询失败，默认为30.0；time_stop(float)：当某个线程查询失败时，会进行重新查询，该参数表示进行重新查询的间隔时间，默认为30.0；block_num(int)：查询失败进行重新查询的最大次数，当达到最大次数时，返回RunTimeError异常，默认10；function_(function)：自定义解析函数；函数要求：第一个参数为待解析json数据；当值为None时，使用默认解析方法，默认为None；args(tuple)：自定义解析函数的除第一个参数的其他参数元组，默认为None；**kwargs：可以用来修改查询参数字典中的值，比如我们要修改返回值数量('limit')为20，可以直接将limit = 20作为函数参数，修改其他查询参数同理；
- 返回
(dict)- 解析关键词为键值，对应值为对应位置映射的解析结果；
-
labels,descriptions,aliases
以
labels为例子labels,labels/,labels//→labels下字典labels/xx→xx语言下的值
-
claims
claims,claims/,claims//→claims下字典claims/P,claims/P/,claims/P//→属性ID列表claims/P/value→所有属性下的值，值的格式为元组，(值类型，主值1, ···)claims/Pxx,claims/Pxx/,claims/Pxx//→Pxx下的字典claims/Pxx/value→Pxx的具体值，值的格式为元组，(值类型，主值1, ···)claims/Pxx/qualifiers-order→Pxx限定词顺序列表claims/Pxx/qualifiers,claims/Pxx/qualifiers/,claims/Pxx/qualifiers//→Pxx限定词下字典claims/Pxx/references,claims/Pxx/references/,claims/Pxx/references//→Pxx引用下字典
-
sitelinks
sitelinks,sitelinks/,sitelinks//→外部链接下的字典
# key = 'search'
{
'search': # 文本,
'action': 'wbsearchentities',
'format': 'json',
'language': 'en',
'type': 'item',
'limit': 10,
'strictlanguage': None,
'continue': None,
'props': None
}
# key = 'ids'
{
'ids': # 实体ID,
'action': 'wbgetentities',
'format': 'json',
'languages': 'en',
'redirects': None,
'sites': None,
'title': None,
'props': None,
'languagefallback': None,
'normalize': None,
'sitefilter': None
}- 需要安装库
wikipedia-->pip install wikipedia
from searchmanage import Wikipedia
w1 = Wikipedia(m_num = 10)
"""
m_num(int):指定运行线程数量
"""- 运行前请检查一下能否打开网址https://en.wikipedia.org/w/api.php，如果不能打开，请先检查网络。
def search_run(self, points: list, time_stop: float = 30.0, block_num: int = 10,
function_=None, args: tuple = None, **kwargs) -> list- 输入N维字符列表多线程调用
wikipedia中相关函数，默认为函数wikipedia.suggest() - 参数说明：
points(list)：待查询的N维字符列表；time_stop(float)：当某个线程查询失败时，会进行重新查询，该参数表示进行重新查询的间隔时间，默认30.0；block_num(int)：查询失败进行重新查询的最大次数，当达到最大次数时，返回RunTimeError异常，默认10；function(function)：多线程调用wikipedia中的**相关函数**，默认为函数wikipedia.suggest，默认值为None；args(tuple)：调用相关函数的除第一个参数的其他参数元组，默认为None；**kwargs：用于功能拓展，暂无实际功能；
- 返回
(list)：- 对应位置映射的结果列表
- 需要安装库
SPARQLWrapper--->pip install sparqlwrapper
from searchmanage import SparqlQuery
sql = SparqlQuery(m_num = 10, format_ = 'json')
"""
m_num(int):指定运行线程数量;
format_(str):查询返回格式，默认为'json';
url_(str):查询域名节点，默认为<SparqlQuery.URL_ = "https://query.wikidata.org/sparql">;
sparql_(str): 查询sparql格式语句，默认为<SparqlQuery.SPARQL_>;
"""def search_run(self, points: list, timeout: float = 30.0, time_stop: float = 30.0,
block_num: int = 10, function_=None, args: tuple = None, **kwargs) -> dict-
输入N维ID列表，返回结果字典；
-
参数说明
points(list)：N维ID列表；timeout(float)：每个线程查询的最大限制时间，到超过这个时间，会视为查询失败，默认为30.0；time_stop(float)：当某个线程查询失败时，会进行重新查询，该参数表示进行重新查询的间隔时间，默认为30.0；block_num(int)：查询失败进行重新查询的最大次数，当达到最大次数时，返回RunTimeError异常，默认10；function_(function)：自定义解析函数；函数要求：第一个参数为待解析json数据；当值为None时，使用默认解析方法，默认为None；args(tuple)：自定义解析函数的除第一个参数的其他参数元组，默认为None；**kwargs：用于功能拓展，暂时无实际功能；
-
返回
(dict)- 解析关键词为键值，对应值为对应位置映射的解析结果；
- 符合Sparql语句基本格式；
- 带查询变量位置使用
python字符串格式化%s标识； - 目前**只支持*一个***查询变量；
# 默认sparql_
SparqlQuery.SPARQL_ = """
SELECT ?item ?itemLabel
WHERE
{
?item wdt:P31 wd:%s.
SERVICE wikibase:label { bd:serviceParam wikibase:language "[AUTO_LANGUAGE],en". }
}from searchmanage import BingQuery
b = BingQuery(m_num = 10)
"""
url_(str): 模糊搜索域名节点，默认为<BingQuery.URL_ = "https://www.bing.com/search">;
m_num(int):指定运行线程数量;
"""def search_run(self, points: list, timeout: float = 30.0, time_stop: float = 30.0,
block_num: int = 10, function_=None, args: tuple = None, **kwargs) -> list- 输入N维字符列表，输出模糊结果的
url网址 - 参数说明
points(list)：N维字符列表；timeout(float)：每个线程查询的最大限制时间，到超过这个时间，会视为查询失败，默认为30.0；time_stop(float)：当某个线程查询失败时，会进行重新查询，该参数表示进行重新查询的间隔时间，默认为30.0；block_num(int)：查询失败进行重新查询的最大次数，当达到最大次数时，返回RunTimeError异常，默认10；function_(function)：自定义正则解析函数；函数要求：第一个参数为待正则解析*字符串*；当值为None时，使用默认正则解析方法<REG_ = r"https://www.wikidata.org/wiki/Q\d+"，默认值为None；args(tuple)：自定义解析函数的除第一个参数的其他参数元组，默认为None；**kwargs：用于功能拓展，暂时无实际功能；
- 返回
(list)- 模糊搜索结果的
url网址映射列表
- 模糊搜索结果的
from searchmanage import SpellCheck
sc = SpellCheck(m_num = 10)
"""
url_(str): 爬取拼写结果的域名节点，默认为<BingQuery.URL_ = "https://www.bing.com/search">;
m_num(int):指定运行线程数量;
"""def search_run(self, points: list, timeout: float = 30.0, time_stop: float = 30.0,
block_num: int = 10, function_=None, args: tuple = None, **kwargs) -> list- 输入N维字符列表，输出纠错后结果
- 参数说明
points(list)：N维字符列表；timeout(float)：每个线程查询的最大限制时间，到超过这个时间，会视为查询失败，默认为30.0；time_stop(float)：当某个线程查询失败时，会进行重新查询，该参数表示进行重新查询的间隔时间，默认为30.0；block_num(int)：查询失败进行重新查询的最大次数，当达到最大次数时，返回RunTimeError异常，默认10；function_(function)：自定义解析函数；函数要求：第一个参数为requests对象的text属性；当值为None时，获取网页搜索推荐纠错结果，默认值为None；args(tuple)：自定义解析函数的除第一个参数的其他参数元组，默认为None；**kwargs：用于功能拓展，暂时无实际功能；
- 返回
(list)- 单词纠错结果映射列表
from searchmanage import DbpediaLookUp
db = DbpediaLookUp(m_num=10)
"""
key(str):
针对实体某部分进行搜索，可选"query","label", "comment"
or "category", 默认为全局搜索"query";
m_num(int):指定运行线程数量;
"""def search_run(self, points: list, patten: str = "search", is_all: bool = False, timeout: float = 30.0,
time_stop: float = 30.0, block_num: int = 10, function_=None, args: tuple = (), **kwargs) -> dict- 输入N维文本列表，多线程查询并根据解析关键词返回解析结果
- 参数说明
points(list)：N维文本列表；patten(str)：查询模式，可选'search'或'prefix'，其中'prefix'为前缀搜索，对应URL为"https://lookup.dbpedia.org/api/prefix"，默认为普通搜索为'search'，对应URL为"https://lookup.dbpedia.org/api/search"；is_all(bool)：是否返回所有解析结果，解析键值见查询参数字典与解析键值，默认为False，返回解析键值的前四个；timeout(float)：每个线程查询的最大限制时间，到超过这个时间，会视为查询失败，默认为30.0；time_stop(float)：当某个线程查询失败时，会进行重新查询，该参数表示进行重新查询的间隔时间，默认为30.0；block_num(int)：查询失败进行重新查询的最大次数，当达到最大次数时，返回RunTimeError异常，默认10；function_(function)：自定义解析函数；函数要求：第一个参数为待解析json数据；当值为None时，使用默认解析方法，默认为None；args(tuple)：自定义解析函数的除第一个参数的其他参数元组，默认为None；**kwargs：可以用来修改查询参数字典中的值，比如我们要修改返回值数量('maxResults')为20，可以直接将maxResults = 20作为函数参数，修改其他查询参数同理，查询参数见查询参数字典与解析键值；
- 返回
(dict)- 解析键值为键值，对应值为对应位置映射的解析结果；
# 查询字典格式
PARAM_DBPEDIA_QUERY = {
"query": None,
"label": None,
"comment": None,
"category": None,
"typeName": None,
"maxResults": 10,
"format": "json",
"minRelevance": None
}
"""Parameters using in Dbpedia look up."""
# 解析键值列表，元组第一位为1表明结果为单数，为2说明结果为复数，以列表形式存在
DBPEDIA_KEYS = [(1, 'label'), (1, 'resource'), (2, 'typeName'), (2, 'type'), (1, 'score'),
(1, 'refCount'), (1, 'comment'), (2, 'redirectlabel'), (2, 'category')]
"""Analysis keys using in Dbpedia look up json data."""- 列首的整数表示该值类型解析几个主值
# Wikidata Value-type
value_type = {
'wikibase-entityid': [1, 'id', 'entity-type', 'numeric-id'],
'globecoordinate': [2, 'latitude', 'longitude', 'precision', 'globe'],
'time': [1, 'time', 'precision', 'before', 'after', 'timezone'],
'string': None,
'monolingualtext': [2, 'text', 'language'],
'quantity': [1, 'amount', 'lowerBound', 'upperBound']
}# Wikidata Data-type
data_type = {
'commonsMedia': 'string',
'globe-coordinate': 'globecoordinate',
'wikibase-item': 'wikibase-entityid',
'wikibase-property': 'wikibase-entityid',
'string': 'string',
'monolingualtext': 'monolingualtext',
'external-id': 'string',
'quantity': 'quantity',
'time': 'time',
'url': 'string',
'math': 'string',
'geo-shape': 'string',
'musical-notation': 'string',
'tabular-data': 'string',
'wikibase-lexeme': 'wikibase-entityid',
'wikibase-form': 'wikibase-entityid',
'wikibase-sense': 'wikibase-entityid'
}from searchmanage import SearchManage, Wikipedia, SparqlQuery, BingQuery, Tools, SpellCheck, DbpediaLookUp
if __name__ == "__main__":
# Read data from csv
d, d_t = Tools.read_csv("data\\1C9LFOKN.csv", is_header=True, out_data_t=True, is_print=True)
print(len(d_t))
# Example data
p1 = [['SEU', 'Chain', 'English'], ['computer', 'games', 'computer game'], ['graph', 'wikipedia'],
['SEU', 'Chain', 'English'], ['computer', 'games', 'computer game'], ['graph', 'wikipedia'],
['SEU', 'Chain', 'English'], ['computer', 'games', 'computer game'], ['graph', 'wikipedia']]
p2 = ["Q3918", "Q355304", "Q106589826", "Q3918", "Q355304", "Q106589826", "Q3918", "Q355304",
"Q106589826", "Q3918", "Q355304", "Q106589826"]
# SearchManage: key = 'search'
s1 = SearchManage(key='search', m_num=10)
r1 = s1.search_run(p1, keys='all', limit=20)
# print(r1)
# SearchManage: key = 'ids'
s2 = SearchManage(key='ids', m_num=20)
r2 = s2.search_run(r1['id'], keys=['labels/en', 'claims/P//', 'claims/P/value'])
# print(r2)
# Wikipedia: <wikipedia.suggest()>
w1 = Wikipedia(m_num=10)
r3 = w1.search_run(p1)
# print(r3)
# SparqlQuery: <sparql_ = SparqlQuery.SPARQL_>
sql1 = SparqlQuery(m_num=12, format_='json')
r4 = sql1.search_run(p2[0:3], timeout=60)
# print(r4)
# BingQuery: <url_ = BingQuery.URL_>
b1 = BingQuery(m_num=24)
r5 = b1.search_run(p1)
print(r5)
p3 = [["elgant palm trre garden", "elgant palm trre", "the sotheast univrsity"],
["elgant", "elgant trre", "the sotheast univrssity"],
["elgat palm trre garden", "elgat palm trre", "the sothaast univrsity"]]
# SpellCheck <url_ = "https://www.bing.com/search">
sc = SpellCheck(m_num=12)
r6 = sc.search_run(p3)
print(r6)
# DbpediaLookUp->"resource"
db = DbpediaLookUp(m_num=10)
r7 = db.search_run(p1, patten='search', is_all=False, maxResults=20)
print(r7['resource'])
# Dbpedia SPARQL
end_point = "https://dbpedia.org/sparql"
sparql_ = """
SELECT?Type?Rtype
WHERE{
<%s> dbp:type ?Type;
rdf:type ?Rtype.}
"""
sql2 = SparqlQuery(m_num=200, format_='json', url_=end_point, sparql_=sparql_)
r8 = sql2.search_run(r7['resource'], timeout=10000)
print(r8['Type'])