(working_dir)
| 101 | # print(extract_results) |
| 102 | return extract_results |
| 103 | def create_db_table_mysql(working_dir): |
| 104 | con = pymysql.connect(host='localhost',port=4321, user='root', |
| 105 | passwd='123', charset='utf8mb4') |
| 106 | cur=con.cursor() |
| 107 | dbname=os.path.basename(working_dir) |
| 108 | |
| 109 | cur.execute(f"drop database if exists {dbname};") |
| 110 | cur.execute(f"create database {dbname} character set utf8mb4;") |
| 111 | |
| 112 | # 使用库 |
| 113 | cur.execute(f"use {dbname};") |
| 114 | cur.execute("drop table if exists entities;") |
| 115 | # 建表 |
| 116 | cur.execute("create table entities\ |
| 117 | (entity_name varchar(500), description varchar(10000),source_id varchar(1000),\ |
| 118 | degree int,parent varchar(1000),level int ,INDEX en(entity_name))character set utf8mb4 COLLATE utf8mb4_unicode_ci;") |
| 119 | |
| 120 | cur.execute("drop table if exists relations;") |
| 121 | cur.execute("create table relations\ |
| 122 | (src_tgt varchar(190),tgt_src varchar(190), description varchar(10000),\ |
| 123 | weight int,level int ,INDEX link(src_tgt,tgt_src))character set utf8mb4 COLLATE utf8mb4_unicode_ci;") |
| 124 | |
| 125 | |
| 126 | cur.execute("drop table if exists communities;") |
| 127 | cur.execute("create table communities\ |
| 128 | (entity_name varchar(500), entity_description varchar(10000),findings text,INDEX en(entity_name)\ |
| 129 | )character set utf8mb4 COLLATE utf8mb4_unicode_ci ;") |
| 130 | cur.close() |
| 131 | con.close() |
| 132 | |
| 133 | def insert_data_to_mysql(working_dir): |
| 134 | dbname=os.path.basename(working_dir) |
no outgoing calls
no test coverage detected