252 lines
9.4 KiB
Python
252 lines
9.4 KiB
Python
|
import subprocess
|
|||
|
|
from bs4 import BeautifulSoup
|
|||
|
|
import datetime
|
|||
|
|
import os
|
|||
|
|
from urllib.parse import urlparse, urljoin
|
|||
|
|
import time
|
|||
|
|
import pymysql
|
|||
|
|
from pathlib import Path
|
|||
|
|
import logging
|
|||
|
|
from apscheduler.schedulers.blocking import BlockingScheduler
|
|||
|
|
from apscheduler.events import EVENT_JOB_EXECUTED, EVENT_JOB_ERROR
|
|||
|
|
import json
|
|||
|
|
import re
|
|||
|
|
from config import * # 导入配置
|
|||
|
|
from mqsend import send_message # 导入消息发送函数
|
|||
|
|
from elasticsearch import Elasticsearch
|
|||
|
|
|
|||
|
|
|
|||
|
|
def clean_control_characters(text):
|
|||
|
|
return re.sub(r'[\x00-\x1F\x7F-\x9F]', '', text)
|
|||
|
|
|
|||
|
|
def do_work(target):
|
|||
|
|
# 创建data目录
|
|||
|
|
os.makedirs('data', exist_ok=True)
|
|||
|
|
|
|||
|
|
today = datetime.datetime.today().strftime('%Y-%m-%d')
|
|||
|
|
new_target = f"{target}/{today}/"
|
|||
|
|
try:
|
|||
|
|
sub_result = subprocess.run(['curl', new_target], capture_output=True, text=True)
|
|||
|
|
|
|||
|
|
if sub_result.returncode == 0:
|
|||
|
|
soup = BeautifulSoup(sub_result.stdout, 'html.parser')
|
|||
|
|
links = soup.find_all('a', href=True)
|
|||
|
|
|
|||
|
|
for link in links:
|
|||
|
|
href = link['href']
|
|||
|
|
if href.startswith('/'):
|
|||
|
|
|
|||
|
|
file_url = f"{target}/{urljoin(new_target, href)}"
|
|||
|
|
# 生成本地保存路径
|
|||
|
|
local_path = os.path.join(file_path+'data', href.lstrip('/')) # 添加日期目录层级
|
|||
|
|
# 创建目录结构
|
|||
|
|
os.makedirs(os.path.dirname(local_path), exist_ok=True)
|
|||
|
|
# 下载文件
|
|||
|
|
if not os.path.exists(local_path):
|
|||
|
|
logger.info(f"文件不存在,准备下载: {file_url}")
|
|||
|
|
download_cmd = ['curl', '-s', '-o', local_path, file_url]
|
|||
|
|
dl_result = subprocess.run(download_cmd)
|
|||
|
|
if dl_result.returncode == 0:
|
|||
|
|
logger.info(f"下载成功: {local_path}")
|
|||
|
|
# 同步到MySQL
|
|||
|
|
sync_to_database(local_path)
|
|||
|
|
else:
|
|||
|
|
logger.error(f"下载失败: {file_url}")
|
|||
|
|
|
|||
|
|
else:
|
|||
|
|
logger.warning(f"忽略非相对路径: {href}")
|
|||
|
|
else:
|
|||
|
|
logger.error(f"获取失败: {sub_result.stderr}")
|
|||
|
|
except Exception as e:
|
|||
|
|
logger.error(f"处理 {new_target} 时发生未知错误: {e}")
|
|||
|
|
|
|||
|
|
|
|||
|
|
def write_to_mysql(data_list, local_path):
|
|||
|
|
conn = pymysql.connect(
|
|||
|
|
host=MYSQL_HOST_APP,
|
|||
|
|
port=MYSQL_PORT_APP,
|
|||
|
|
user=MYSQL_USER_APP,
|
|||
|
|
password=MYSQL_PASSWORD_APP,
|
|||
|
|
db=MYSQL_DB_APP,
|
|||
|
|
charset='utf8mb4'
|
|||
|
|
)
|
|||
|
|
|
|||
|
|
try:
|
|||
|
|
with conn.cursor() as cursor:
|
|||
|
|
for data in data_list:
|
|||
|
|
# 新增JSON结构解析逻辑
|
|||
|
|
category_data = data.get('c', [{}])
|
|||
|
|
category = lang = sourcename = ""
|
|||
|
|
if category_data:
|
|||
|
|
category = category_data[0].get('category', '')
|
|||
|
|
b_data = category_data[0].get('b', [{}])
|
|||
|
|
if b_data:
|
|||
|
|
lang = b_data[0].get('lang', '')
|
|||
|
|
d_data = b_data[0].get('d', [{}])
|
|||
|
|
if d_data:
|
|||
|
|
sourcename = d_data[0].get('sourcename', '')
|
|||
|
|
|
|||
|
|
# 处理特殊字符转义
|
|||
|
|
en_content = data.get('EN_content',0).replace('quot;', '"')
|
|||
|
|
|
|||
|
|
# 修改后的SQL语句
|
|||
|
|
sql = """INSERT INTO news_info
|
|||
|
|
(news_id, input_date, words, title_txt, key_word,
|
|||
|
|
CN_content, EN_content, URL, abstract, title_EN,
|
|||
|
|
category, sourcename, lang, deleted, create_time,
|
|||
|
|
update_time, file_date, file_name)
|
|||
|
|
VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s)
|
|||
|
|
ON DUPLICATE KEY UPDATE
|
|||
|
|
category=VALUES(category),
|
|||
|
|
sourcename=VALUES(sourcename),
|
|||
|
|
lang=VALUES(lang)"""
|
|||
|
|
|
|||
|
|
values = (
|
|||
|
|
data['id'],
|
|||
|
|
data.get('input_date', ""),
|
|||
|
|
data.get('words', ""),
|
|||
|
|
data.get('title_txt', ""),
|
|||
|
|
data.get('key_word', ""),
|
|||
|
|
data.get('CN_content', ""),
|
|||
|
|
en_content, # 使用处理后的英文内容
|
|||
|
|
data.get('URL', ""),
|
|||
|
|
data.get('abstract', ""),
|
|||
|
|
data.get('title_EN', ""),
|
|||
|
|
category,
|
|||
|
|
sourcename,
|
|||
|
|
lang,
|
|||
|
|
0,
|
|||
|
|
datetime.datetime.now(),
|
|||
|
|
datetime.datetime.now(),
|
|||
|
|
datetime.datetime.strptime(Path(local_path).parent.name, '%Y-%m-%d').date(),
|
|||
|
|
Path(local_path).name
|
|||
|
|
)
|
|||
|
|
cursor.execute(sql, values)
|
|||
|
|
logger.info(f"成功写入mysql: {data['id']}")
|
|||
|
|
|
|||
|
|
#
|
|||
|
|
|
|||
|
|
# 发送消息到MQ
|
|||
|
|
send_message(json.dumps(data))
|
|||
|
|
conn.commit()
|
|||
|
|
except Exception as e:
|
|||
|
|
logger.error(f"写入mysql失败: {str(e)}")
|
|||
|
|
finally:
|
|||
|
|
conn.close()
|
|||
|
|
return True
|
|||
|
|
|
|||
|
|
# 写入ES
|
|||
|
|
def write_to_es(data_list, local_path):
|
|||
|
|
# 初始化ES连接（添加在文件顶部）
|
|||
|
|
es = Elasticsearch(
|
|||
|
|
[f"http://{ES_HOST}:{ES_PORT}"], # 将协议直接包含在hosts中
|
|||
|
|
basic_auth=(ES_USER, ES_PASSWORD)
|
|||
|
|
)
|
|||
|
|
try:
|
|||
|
|
for data in data_list:
|
|||
|
|
# 插入ES核心逻辑
|
|||
|
|
category_data = data.get('c', [{}])
|
|||
|
|
category = lang = sourcename = ""
|
|||
|
|
if category_data:
|
|||
|
|
category = category_data[0].get('category', '')
|
|||
|
|
b_data = category_data[0].get('b', [{}])
|
|||
|
|
if b_data:
|
|||
|
|
lang = b_data[0].get('lang', '')
|
|||
|
|
d_data = b_data[0].get('d', [{}])
|
|||
|
|
if d_data:
|
|||
|
|
sourcename = d_data[0].get('sourcename', '')
|
|||
|
|
|
|||
|
|
es.index(
|
|||
|
|
index="news_info",
|
|||
|
|
id=data['id'],
|
|||
|
|
document={
|
|||
|
|
"news_id": data.get('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',0).replace('quot;', '"'),
|
|||
|
|
"URL": data.get('URL', ""),
|
|||
|
|
"abstract": data.get('abstract', ""),
|
|||
|
|
"title_EN": data.get('title_EN', ""),
|
|||
|
|
"category": category,
|
|||
|
|
"sourcename": sourcename,
|
|||
|
|
"lang": lang,
|
|||
|
|
"deleted": "0",
|
|||
|
|
"create_time":datetime.datetime.now(),
|
|||
|
|
"update_time": datetime.datetime.now(),
|
|||
|
|
"file_date": datetime.datetime.strptime(Path(local_path).parent.name, '%Y-%m-%d').date(),
|
|||
|
|
"file_name": Path(local_path).name
|
|||
|
|
}
|
|||
|
|
)
|
|||
|
|
logger.info(f"成功写入ES文档ID: {data['id']}")
|
|||
|
|
|
|||
|
|
except Exception as e:
|
|||
|
|
logger.error(f"写入ES失败: {str(e)}")
|
|||
|
|
|
|||
|
|
def sync_to_database(local_path):
|
|||
|
|
try:
|
|||
|
|
with open(local_path, 'r', encoding='utf-16') as f: # 修改编码方式
|
|||
|
|
raw_data = f.read()
|
|||
|
|
cleaned_data = clean_control_characters(raw_data)
|
|||
|
|
data_list = json.loads(cleaned_data)
|
|||
|
|
write_to_mysql(data_list, local_path) # 传递本地路径
|
|||
|
|
write_to_es(data_list, local_path) # 传递本地路径
|
|||
|
|
|
|||
|
|
return True
|
|||
|
|
|
|||
|
|
except Exception as e:
|
|||
|
|
logger.error(f"处理文件 {local_path} 时发生未知错误: {e}")
|
|||
|
|
return False
|
|||
|
|
|
|||
|
|
|
|||
|
|
|
|||
|
|
|
|||
|
|
# 初始化日志配置（添加在文件顶部）
|
|||
|
|
logger = logging.getLogger()
|
|||
|
|
logger.setLevel(logging.INFO)
|
|||
|
|
formatter = logging.Formatter('%(asctime)s - %(levelname)s - %(message)s')
|
|||
|
|
|
|||
|
|
# 文件日志
|
|||
|
|
file_handler = logging.FileHandler('zzck_nohup.log')
|
|||
|
|
file_handler.setFormatter(formatter)
|
|||
|
|
logger.addHandler(file_handler)
|
|||
|
|
|
|||
|
|
# 控制台日志
|
|||
|
|
console_handler = logging.StreamHandler()
|
|||
|
|
console_handler.setFormatter(formatter)
|
|||
|
|
logger.addHandler(console_handler)
|
|||
|
|
|
|||
|
|
# 调度任务监听器（添加在job函数后）
|
|||
|
|
def scheduler_listener(event):
|
|||
|
|
if event.exception:
|
|||
|
|
logger.error("任务执行失败！", exc_info=event.exception)
|
|||
|
|
else:
|
|||
|
|
logger.info("任务执行完成")
|
|||
|
|
|
|||
|
|
if __name__ == "__main__":
|
|||
|
|
scheduler = BlockingScheduler()
|
|||
|
|
target = "zzbtw.ckxx.net" # 确保target变量在此处定义
|
|||
|
|
|
|||
|
|
# 修改调度任务添加方式
|
|||
|
|
scheduler.add_job(
|
|||
|
|
do_work,
|
|||
|
|
'interval',
|
|||
|
|
minutes=1,
|
|||
|
|
id='zzck_sync_job',
|
|||
|
|
args=[target], # 添加参数传递
|
|||
|
|
max_instances=1
|
|||
|
|
)
|
|||
|
|
|
|||
|
|
# 添加事件监听
|
|||
|
|
scheduler.add_listener(scheduler_listener, EVENT_JOB_EXECUTED | EVENT_JOB_ERROR)
|
|||
|
|
|
|||
|
|
try:
|
|||
|
|
logger.info("程序启动，调度器开始运行...")
|
|||
|
|
scheduler.start()
|
|||
|
|
except (KeyboardInterrupt, SystemExit):
|
|||
|
|
logger.info("收到停止信号，正在关闭调度器...")
|
|||
|
|
scheduler.shutdown()
|
|||
|
|
logger.info("调度器已关闭")
|