(entity1,entity2,working_dir,level=0)
| 294 | return None |
| 295 | |
| 296 | def search_nodes_link(entity1,entity2,working_dir,level=0): |
| 297 | # cursor = db.cursor() |
| 298 | # db_name=os.path.basename(working_dir) |
| 299 | # sql=f"select * from {db_name}.relations where src_tgt=%s and tgt_src=%s and level=%s" |
| 300 | # cursor.execute(sql,(entity1,entity2,level)) |
| 301 | # ret=cursor.fetchall() |
| 302 | # if len(ret)==0: |
| 303 | # sql=f"select * from {db_name}.relations where src_tgt=%s and tgt_src=%s and level=%s" |
| 304 | # cursor.execute(sql,(entity2,entity1,level)) |
| 305 | # ret=cursor.fetchall() |
| 306 | # if len(ret)==0: |
| 307 | # return None |
| 308 | # else: |
| 309 | # return ret[0] |
| 310 | db = pymysql.connect(host='localhost',port=4321, user='root', |
| 311 | passwd='123', charset='utf8mb4') |
| 312 | cursor = db.cursor() |
| 313 | db_name=os.path.basename(working_dir) |
| 314 | sql=f"select * from {db_name}.relations where src_tgt=%s and tgt_src=%s " |
| 315 | cursor.execute(sql,(entity1,entity2)) |
| 316 | ret=cursor.fetchall() |
| 317 | if len(ret)==0: |
| 318 | sql=f"select * from {db_name}.relations where src_tgt=%s and tgt_src=%s " |
| 319 | cursor.execute(sql,(entity2,entity1)) |
| 320 | ret=cursor.fetchall() |
| 321 | if len(ret)==0: |
| 322 | return None |
| 323 | else: |
| 324 | return ret[0] |
| 325 | def search_chunks(working_dir,entity_set): |
| 326 | db = pymysql.connect(host='localhost',port=4321, user='root', |
| 327 | passwd='123', charset='utf8mb4') |
no outgoing calls
no test coverage detected