"""
@author: luojunhui
"""
import json
import time
import traceback
from typing import List, Dict
from pymysql.cursors import DictCursor
from tqdm import tqdm
from applications import aiditApi
from applications import bot
from applications import log
from applications import WeixinSpider
from applications.const import ArticleCollectorConst
from applications.db import DatabaseConnector
from applications.functions import Functions
from config import denet_config, long_articles_config
empty_dict = {}
const = ArticleCollectorConst()
functions = Functions()
spider = WeixinSpider()
class UpdateArticleInfoFromAIGC(object):
"""
从aigc获取文章信息
"""
def __init__(self):
self.aigc_db_client = DatabaseConnector(db_config=denet_config)
self.long_articles_db_client = DatabaseConnector(db_config=long_articles_config)
self.aigc_db_client.connect()
self.long_articles_db_client.connect()
def get_published_articles(self) -> List[Dict]:
"""
获取当天发布文章的List
"""
sql = f"""
SELECT trace_id, push_type
FROM long_articles_published_trace_id
WHERE create_timestamp > UNIX_TIMESTAMP(DATE_SUB(CURDATE(), INTERVAL 1 DAY)) AND status = {const.INIT_STATUS};
"""
article_list = self.long_articles_db_client.fetch(sql, cursor_type=DictCursor)
return article_list
def get_article_info_from_aigc(self, trace_id: str) -> Dict:
"""
从aigc获取发布结果
"""
sql = f"""
SELECT t2.crawler_channel_content_id, t2.publish_stage_url, t2.publish_timestamp, t1.result_data
from publish_content_miniprogram t1
join publish_content t2 on t1.publish_content_id = t2.id
where t1.trace_id = '{trace_id}' and t2.status = {const.PUBLISHED_STATUS};
"""
article_info = self.aigc_db_client.fetch(sql, cursor_type=DictCursor)
if article_info:
return article_info[0]
else:
return empty_dict
def get_article_info_by_trace_id(self, trace_id: str) -> Dict:
"""
通过trace_id来查询文章信息
"""
select_sql = f"""
SELECT t1.gh_id, t1.account_name, t2.article_title
FROM long_articles_match_videos t1
JOIN long_articles_text t2
ON t1.content_id = t2.content_id
WHERE t1.trace_id = '{trace_id}';
"""
article_info = self.long_articles_db_client.fetch(select_sql, cursor_type=DictCursor)
if article_info:
return article_info[0]
else:
return empty_dict
def update_each_article(self, article: Dict):
"""
更新每个文章的信息
"""
trace_id = article["trace_id"]
push_type = article["push_type"]
article_info = self.get_article_info_from_aigc(trace_id)
if article_info:
channel_content_id = article_info["crawler_channel_content_id"]
published_url = article_info["publish_stage_url"]
publish_timestamp = int(article_info["publish_timestamp"] / 1000)
result_data = json.loads(article_info["result_data"])
root_source_id_list = [
functions.extract_path(item["productionPath"])["root_source_id"] for item in result_data
]
wx_sn = None
if published_url:
response = spider.get_article_text(content_link=published_url)
code = response['code']
match code:
case const.ARTICLE_SUCCESS_CODE:
long_url = response['data']['data']['content_link']
wx_sn = functions.extract_params_from_url(url=long_url, key="sn")
status = const.SUCCESS_STATUS
case const.ARTICLE_DELETE_CODE:
log(
task="update_article_info_from_aigc",
function="update_each_article",
status="fail",
message=trace_id,
data={
"msg": "文章被删文",
"publish_timestamp": publish_timestamp,
"article_delete_timestamp": int(time.time()),
"duration": int(time.time()) - publish_timestamp
}
)
status = const.FAIL_STATUS
case const.ARTICLE_ILLEGAL_CODE:
log(
task="update_article_info_from_aigc",
function="update_each_article",
status="fail",
message=trace_id,
data={
"msg": "文章被判断违规",
"publish_timestamp": publish_timestamp,
"illegal_timestamp": int(time.time()),
"duration": int(time.time()) - publish_timestamp
}
)
article_info = self.get_article_info_by_trace_id(trace_id)
if article_info:
error_detail = response.get("msg")
insert_sql = f"""
INSERT IGNORE INTO illegal_articles
(gh_id, account_name, title, wx_sn, publish_date, illegal_reason)
VALUES
(%s, %s, %s, %s, %s, %s);
"""
affected_rows = self.long_articles_db_client.save(
query=insert_sql,
params=(
article_info['gh_id'],
article_info['account_name'],
article_info['article_title'],
wx_sn,
functions.timestamp_to_str(publish_timestamp),
error_detail
)
)
if affected_rows:
bot(
title="文章违规告警(new task)",
detail={
"account_name": article_info['account_name'],
"gh_id": article_info['gh_id'],
"title": article_info['article_title'],
"wx_sn": wx_sn,
"publish_date": functions.timestamp_to_str(publish_timestamp),
"error_detail": error_detail,
},
mention=False
)
aiditApi.delete_articles(
gh_id=article_info['gh_id'],
title=article_info['article_title']
)
status = const.FAIL_STATUS
case _:
status = const.FAIL_STATUS
else:
if push_type == const.BULK_AUTO_PUSH:
status = const.INIT_STATUS
else:
status = const.SUCCESS_STATUS
update_sql = f"""
UPDATE long_articles_published_trace_id
SET published_url = %s, status = %s, wx_sn = %s, publish_timestamp = %s, crawler_channel_content_id = %s, root_source_id_list = %s
WHERE trace_id = %s;
"""
self.long_articles_db_client.save(
query=update_sql,
params=(published_url, status, wx_sn, publish_timestamp, channel_content_id, json.dumps(root_source_id_list), trace_id)
)
else:
update_sql = f"""
UPDATE long_articles_published_trace_id
SET status = %s
WHERE trace_id = %s;
"""
self.long_articles_db_client.save(
query=update_sql, params=(const.FAIL_STATUS, trace_id)
)
def deal(self):
"""
main function
"""
article_list = self.get_published_articles()
log(
task="update_article_info_from_aigc",
function="deal",
data=article_list,
message="total got {} articles this time".format(len(article_list))
)
for article in tqdm(article_list, desc="更新文章信息"):
try:
self.update_each_article(article)
except Exception as e:
log(
task="update_article_info_from_aigc",
function="update_each_article",
status="fail",
message="update_article_fail",
data={
"trace_id": article["trace_id"],
"error": str(e),
"traceback": traceback.format_exc()
}
)