176 lines
7.9 KiB
Python
176 lines
7.9 KiB
Python
|
import argparse
|
|||
|
|
from datetime import datetime
|
|||
|
|
import pymysql
|
|||
|
|
from elasticsearch import Elasticsearch, helpers
|
|||
|
|
import json
|
|||
|
|
# 复用现有配置
|
|||
|
|
from config import ES_HOST, ES_PORT, ES_USER, ES_PASSWORD, MYSQL_HOST_APP, MYSQL_USER_APP, MYSQL_PASSWORD_APP, MYSQL_DB_APP
|
|||
|
|
from tqdm import tqdm
|
|||
|
|
import requests
|
|||
|
|
from config import jx_adr
|
|||
|
|
|
|||
|
|
class MySQLToESSync:
|
|||
|
|
def __init__(self):
|
|||
|
|
self.es = Elasticsearch(
|
|||
|
|
[f"http://{ES_HOST}:{ES_PORT}"],
|
|||
|
|
basic_auth=(ES_USER, ES_PASSWORD)
|
|||
|
|
)
|
|||
|
|
self.mysql_conn = pymysql.connect(
|
|||
|
|
host=MYSQL_HOST_APP,
|
|||
|
|
user=MYSQL_USER_APP,
|
|||
|
|
password=MYSQL_PASSWORD_APP,
|
|||
|
|
db=MYSQL_DB_APP,
|
|||
|
|
charset='utf8mb4',
|
|||
|
|
cursorclass=pymysql.cursors.DictCursor
|
|||
|
|
)
|
|||
|
|
|
|||
|
|
# 更新全量表
|
|||
|
|
def sync_all(self, start_time=None, end_time=None):
|
|||
|
|
# 更新全量表到 es
|
|||
|
|
with self.mysql_conn.cursor(pymysql.cursors.DictCursor) as cursor:
|
|||
|
|
query = self._build_query("news_info",start_time, end_time)
|
|||
|
|
cursor.execute(query)
|
|||
|
|
# 支持两种同步模式
|
|||
|
|
res = cursor.fetchall()
|
|||
|
|
# 判断es中是否有该news_id
|
|||
|
|
for data in tqdm(res, desc="同步全量表", unit="条"):
|
|||
|
|
news_id = data['news_id']
|
|||
|
|
# 从es中查询
|
|||
|
|
try:
|
|||
|
|
exists = self.es.get(index="news_info", id=news_id)
|
|||
|
|
except:
|
|||
|
|
exists = None
|
|||
|
|
if exists:
|
|||
|
|
pass
|
|||
|
|
else:
|
|||
|
|
# es中不存在，直接新增
|
|||
|
|
self.es.index(
|
|||
|
|
index="news_info",
|
|||
|
|
id=news_id,
|
|||
|
|
document={
|
|||
|
|
"news_id": data.get('news_id', ""),
|
|||
|
|
"input_date": data.get('input_date', ""),
|
|||
|
|
"words": data.get('words', ""),
|
|||
|
|
"title_txt": data.get('title_txt', ""),
|
|||
|
|
"key_word": data.get('key_word', ""),
|
|||
|
|
"CN_content": data.get('CN_content', ""),
|
|||
|
|
"EN_content": data.get('EN_content',""),
|
|||
|
|
"URL": data.get('URL', ""),
|
|||
|
|
"abstract": data.get('abstract', ""),
|
|||
|
|
"title_EN": data.get('title_EN', ""),
|
|||
|
|
"category": data.get('category', ""),
|
|||
|
|
"sourcename": data.get('sourcename', ""),
|
|||
|
|
"lang": data.get('lang', ""),
|
|||
|
|
"deleted": data.get('deleted', ""),
|
|||
|
|
"create_time": data.get('create_time', ""),
|
|||
|
|
"update_time": data.get('update_time', ""),
|
|||
|
|
"file_date": data.get('file_date', ""),
|
|||
|
|
"file_name": data.get('file_name', ""),
|
|||
|
|
}
|
|||
|
|
)
|
|||
|
|
print(f"新闻 {news_id} 更新到 ES 成功")
|
|||
|
|
|
|||
|
|
|
|||
|
|
# 更新标签表
|
|||
|
|
def sync_tags(self, start_time=None, end_time=None):
|
|||
|
|
|
|||
|
|
with self.mysql_conn.cursor(pymysql.cursors.DictCursor) as cursor:
|
|||
|
|
query = self._build_query("news_tags",start_time, end_time)
|
|||
|
|
cursor.execute(query)
|
|||
|
|
|
|||
|
|
res = cursor.fetchall()
|
|||
|
|
# 判断es中是否有该news_id
|
|||
|
|
for data in tqdm(res, desc="同步新闻标签进度", unit="条"):
|
|||
|
|
news_id = data['news_id']
|
|||
|
|
# 从es中查询
|
|||
|
|
try:
|
|||
|
|
# 更新全量表到 es
|
|||
|
|
es_doc = self.es.get(index="news_info", id=news_id)
|
|||
|
|
news_tags = None
|
|||
|
|
if es_doc and es_doc.get('found'):
|
|||
|
|
news_tags = es_doc['_source'].get('news_tags', None)
|
|||
|
|
if not news_tags:
|
|||
|
|
# 将mysql的数据写入es
|
|||
|
|
self.es.update(
|
|||
|
|
index="news_info",
|
|||
|
|
id=news_id,
|
|||
|
|
doc={
|
|||
|
|
"news_tags": {
|
|||
|
|
"id": news_id,
|
|||
|
|
"abstract": data.get('abstract', ""),
|
|||
|
|
"title": data.get('title', ""),
|
|||
|
|
"rewrite_content": data.get('rewrite_content', ""),
|
|||
|
|
"industry_label": json.loads(data.get('industry_label', [])),
|
|||
|
|
"industry_confidence": json.loads(data.get('industry_confidence', [])),
|
|||
|
|
"industry_score": json.loads(data.get('industry_score', [])),
|
|||
|
|
"concept_label": json.loads(data.get('concept_label', [])),
|
|||
|
|
"concept_confidence": json.loads(data.get('concept_confidence', [])),
|
|||
|
|
"concept_score": json.loads(data.get('concept_score', [])),
|
|||
|
|
"public_opinion_score": data.get('public_opinion_score', 10),
|
|||
|
|
"China_factor": data.get('China_factor', 0.1),
|
|||
|
|
"source": data.get('source', "其他"),
|
|||
|
|
"source_impact": data.get('source_impact', 5),
|
|||
|
|
"news_score": data.get('news_score', 0.0),
|
|||
|
|
"news_id": news_id,
|
|||
|
|
"deleted": '0',
|
|||
|
|
"create_time": data.get('create_time', ""),
|
|||
|
|
"update_time": data.get('update_time', "")
|
|||
|
|
}
|
|||
|
|
}
|
|||
|
|
)
|
|||
|
|
print(f"新闻 {news_id} 标签更新到 ES 成功")
|
|||
|
|
|
|||
|
|
|
|||
|
|
except Exception as e:
|
|||
|
|
print(e)
|
|||
|
|
|
|||
|
|
# 更新精选表
|
|||
|
|
def sync_chooses(self, start_time=None, end_time=None):
|
|||
|
|
with self.mysql_conn.cursor(pymysql.cursors.DictCursor) as cursor:
|
|||
|
|
query = "select * from news_tags where news_score >= 80"
|
|||
|
|
cursor.execute(query)
|
|||
|
|
res = cursor.fetchall()
|
|||
|
|
# 判断es中是否有该news_id
|
|||
|
|
for data in tqdm(res, desc="同步精选表数据", unit="条"):
|
|||
|
|
# 从es中查询
|
|||
|
|
try:
|
|||
|
|
news_id = data['news_id']
|
|||
|
|
news_score = data['news_score']
|
|||
|
|
# 获取返回数据里面的 新闻id
|
|||
|
|
adr = jx_adr.replace("news_id", str(news_id))
|
|||
|
|
response = requests.get(adr)
|
|||
|
|
if response.status_code != 200:
|
|||
|
|
print(f"新闻id:{news_id} 得分:{news_score}, 调用精选接口失败, 错误码:{response.status_code}")
|
|||
|
|
print(f"新闻id:{news_id} 得分:{news_score}, 调用精选接口成功")
|
|||
|
|
except Exception as e:
|
|||
|
|
print(e)
|
|||
|
|
|
|||
|
|
|
|||
|
|
|
|||
|
|
|
|||
|
|
|
|||
|
|
def _build_query(self, table_name, start_time, end_time):
|
|||
|
|
base_query = f"""
|
|||
|
|
SELECT *
|
|||
|
|
FROM {table_name}
|
|||
|
|
"""
|
|||
|
|
if start_time and end_time:
|
|||
|
|
return f"{base_query} WHERE create_time BETWEEN '{start_time}' AND '{end_time}'"
|
|||
|
|
return base_query
|
|||
|
|
|
|||
|
|
if __name__ == "__main__":
|
|||
|
|
parser = argparse.ArgumentParser(description='MySQL to ES Sync Tool')
|
|||
|
|
parser.add_argument('--full', action='store_true', help='Full synchronization')
|
|||
|
|
parser.add_argument('--start', type=str, help='Start time (YYYY-MM-DD HH:MM:SS)')
|
|||
|
|
parser.add_argument('--end', type=str, help='End time (YYYY-MM-DD HH:MM:SS)')
|
|||
|
|
|
|||
|
|
args = parser.parse_args()
|
|||
|
|
|
|||
|
|
if not args.full and not (args.start and args.end):
|
|||
|
|
raise ValueError("需要指定--full全量同步或--start/--end时间范围")
|
|||
|
|
|
|||
|
|
sync = MySQLToESSync()
|
|||
|
|
sync.sync_all(args.start, args.end)
|
|||
|
|
sync.sync_tags(args.start, args.end)
|
|||
|
|
sync.sync_chooses(args.start, args.end)
|
|||
|
|
|