LongArticleTaskServer/applications/tasks/monitor_tasks/gzh_article_monitor.py

import time
import datetime
from os import WCONTINUED
from typing import Optional, List

from tqdm import tqdm

from applications.api import feishu_robot
from applications.api import delete_illegal_gzh_articles
from applications.crawler.wechat import get_article_detail
from applications.crawler.wechat import get_article_list_from_account
from applications.utils import str_to_md5


class MonitorConst:
    # 文章违规状态
    ILLEGAL_STATUS = 1
    INIT_STATUS = 0

    # 监测周期
    MONITOR_CYCLE = 3 * 24 * 3600

    # article code
    ARTICLE_ILLEGAL_CODE = 25012
    ARTICLE_DELETE_CODE = 25005
    ARTICLE_SUCCESS_CODE = 0
    ARTICLE_UNKNOWN_CODE = 10000

    # Task status
    TASK_SUCCESS_CODE = 2
    TASK_FAIL_CODE = 99


class OutsideGzhArticlesManager(MonitorConst):

    def __init__(self, pool):
        self.pool = pool

    async def update_article_illegal_status(
        self, article_id: int, illegal_reason: str
    ) -> None:
        query = f"""
            update outside_gzh_account_monitor
            set illegal_status = %s, illegal_reason = %s
            where id = %s and illegal_status = %s
        """
        await self.pool.async_save(
            query=query,
            params=(self.ILLEGAL_STATUS, illegal_reason, article_id, self.INIT_STATUS),
        )

    async def whether_published_in_a_week(self, gh_id: str) -> bool:
        """
        判断该账号一周内是否有发文，如有，则说无需抓
        """
        query = f"""
            select id, publish_timestamp from outside_gzh_account_monitor
            where gh_id = %s
            order by publish_timestamp desc
            limit %s;
        """
        response, error = await self.pool.async_fetch(query=query, params=(gh_id, 1))
        if response:
            publish_timestamp = response[0]["publish_timestamp"]
            if publish_timestamp is None:
                return False
            else:
                return int(time.time()) - publish_timestamp <= self.MONITOR_CYCLE
        else:
            return False


class OutsideGzhArticlesCollector(OutsideGzhArticlesManager):

    async def fetch_outside_account_list(self):
        query = f"""
            select 
                t2.group_source_name as account_source, 
                t3.name as account_name,
                t3.gh_id as gh_id,
                t3.status as status
            from wx_statistics_group_source t1
                join wx_statistics_group_source_account t2 on t2.group_source_name = t1.account_source_name
                join publish_account t3 on t3.id = t2.account_id
            where
                t1.mode_type = '代运营服务号';
        """
        response, error = await self.pool.async_fetch(query=query, db_name="aigc")
        return response

    async def fetch_each_account(self, account: dict):
        gh_id = account["gh_id"]
        # 判断该账号本周是否已经发布过
        if await self.whether_published_in_a_week(gh_id):
            return

        fetch_response = get_article_list_from_account(gh_id)
        try:
            msg_list = fetch_response.get("data", {}).get("data", [])
            if msg_list:
                for msg in tqdm(
                    msg_list, desc=f"insert account {account['account_name']}"
                ):
                    await self.save_each_msg_to_db(msg, account)

            else:
                print(f"crawler failed: {account['account_name']}")
        except Exception as e:
            print(
                f"crawler failed: account_name: {account['account_name']}\n"
                f"error: {e}\n"
            )

    async def save_each_msg_to_db(self, msg: dict, account: dict):
        base_info = msg["AppMsg"]["BaseInfo"]
        detail_info = msg["AppMsg"]["DetailInfo"]
        app_msg_id = base_info["AppMsgId"]
        create_timestamp = base_info["CreateTime"]
        publish_type = base_info["Type"]

        # insert each article
        for article in detail_info:
            link = article["ContentUrl"]
            article_detail = get_article_detail(link)
            response_code = article_detail["code"]
            if response_code == self.ARTICLE_ILLEGAL_CODE:
                illegal_reason = article_detail.get("msg")
                # bot and return
                await feishu_robot.bot(
                    title="文章违规告警",
                    detail={
                        "账号名称": article["account_name"],
                        "标题": article["title"],
                        "违规理由": illegal_reason,
                        "发布日期": datetime.datetime.fromtimestamp(
                            create_timestamp
                        ).strftime("%Y-%m-%d %H:%M:%S"),
                        "账号合作商": article["account_source"],
                    },
                    env="outside_gzh_monitor",
                    mention=False,
                )

            elif response_code == self.ARTICLE_SUCCESS_CODE:
                insert_query = f"""
                    insert ignore into outside_gzh_account_monitor
                    (account_name, gh_id, account_source, account_type, app_msg_id, publish_type, position, title, link, 
                    channel_content_id, crawler_timestamp, publish_timestamp)
                    values
                    (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s);
                """
                await self.pool.async_save(
                    query=insert_query,
                    params=(
                        account["account_name"],
                        account["gh_id"],
                        account["account_source"],
                        "服务号",
                        app_msg_id,
                        publish_type,
                        article["ItemIndex"],
                        article["Title"],
                        link,
                        article_detail["data"]["data"]["channel_content_id"],
                        int(time.time()),
                        int(article_detail["data"]["data"]["publish_timestamp"] / 1000),
                    ),
                )
            else:
                continue

    async def deal(self):
        account_list = await self.fetch_outside_account_list()
        for account in tqdm(account_list):
            try:
                await self.fetch_each_account(account)
            except Exception as e:
                print(f"crawler failed: {account['account_name']}, error: {e}")


class OutsideGzhArticlesMonitor(OutsideGzhArticlesManager):

    async def fetch_article_list_to_check(self):
        publish_timestamp_threshold = int(time.time()) - self.MONITOR_CYCLE
        fetch_query = f"""
            select id, account_name, gh_id, account_source, account_type, 
                title, link, from_unixtime(publish_timestamp) as publish_date
            from outside_gzh_account_monitor
            where illegal_status = {self.INIT_STATUS} and publish_timestamp > {publish_timestamp_threshold};
        """
        response, error = await self.pool.async_fetch(query=fetch_query)
        return response

    async def check_each_article(self, article: dict):
        """
        check each article
        """
        link = article["link"]
        article_detail = get_article_detail(link)
        response_code = article_detail["code"]
        if response_code == self.ARTICLE_ILLEGAL_CODE:
            illegal_reason = article_detail.get("msg")
            # illegal_reason = '测试报警功能'
            feishu_robot.bot(
                title="文章违规告警",
                detail={
                    "账号名称": article["account_name"],
                    "标题": article["title"],
                    "违规理由": illegal_reason,
                    "发布日期": str(article["publish_date"]),
                    "账号合作商": article["account_source"],
                },
                env="outside_gzh_monitor",
                mention=False,
            )
            article_id = article["id"]
            await self.update_article_illegal_status(article_id, illegal_reason)
        else:
            return

    async def deal(self):
        article_list = await self.fetch_article_list_to_check()
        for article in tqdm(article_list, desc="外部服务号监控"):
            try:
                await self.check_each_article(article)

            except Exception as e:
                print(
                    f"crawler failed: account_name: {article['account_name']}\n"
                    f"link: {article['link']}\n"
                    f"title: {article['title']}\n"
                    f"error: {e}\n"
                )
        return self.TASK_SUCCESS_CODE


class InnerGzhArticlesMonitor(MonitorConst):
    def __init__(self, pool):
        self.pool = pool

    async def whether_title_unsafe(self, title: str) -> bool:
        """
        :param title: gzh article title
        :return: bool
        """
        title_md5 = str_to_md5(title)
        query = f"""
            select title_md5 from article_unsafe_title where title_md5 = '{title_md5}';
        """
        response, error = await self.pool.async_fetch(query=query)
        return True if response else False


    async def fetch_article_list_to_check(self, run_date: str = None) -> Optional[List]:
        """
        :param run_date: 执行日期，格式为“%Y-%m-%d”, default None
        """
        if not run_date:
            run_date = datetime.datetime.today().strftime("%Y-%m-%d")

        run_timestamp = int(
            datetime.datetime.strptime(run_date, "%Y-%m-%d").timestamp()
        )
        start_timestamp = run_timestamp - self.MONITOR_CYCLE
        query = f"""
            select ghId, accountName, title, ContentUrl, wx_sn, from_unixtime(publish_timestamp) as publish_timestamp
            from official_articles_v2
            where publish_timestamp >= {start_timestamp}
            order by publish_timestamp desc;
        """
        response, error = await self.pool.async_fetch(
            query=query, db_name="piaoquan_crawler"
        )
        if error:
            await feishu_robot.bot(
                title="站内微信公众号发文监测任务异常",
                detail={"error": error, "message": "查询数据库异常"},
            )
            return None
        else:
            return response

    async def check_each_article(self, article: dict):
        gh_id, account_name, title, url, wx_sn, publish_date = article
        try:
            response = get_article_detail(url, is_cache=False)
            response_code = response["code"]
            if response_code == self.ARTICLE_ILLEGAL_CODE:
                error_detail = article.get("msg")
                query = 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_row = await self.pool.async_save(
                    query=query,
                    params=(
                        gh_id,
                        account_name,
                        title,
                        wx_sn,
                        publish_date,
                        error_detail,
                    ),
                )
                if affected_row:
                    if await self.whether_title_unsafe(title):
                        return

                    await feishu_robot.bot(
                        title="文章违规告警",
                        detail={
                            "account_name": account_name,
                            "gh_id": gh_id,
                            "title": title,
                            "wx_sn": wx_sn.decode("utf-8"),
                            "publish_date": str(publish_date),
                            "error_detail": error_detail,
                        },
                        mention=False,
                        env="prod"
                    )
                    await delete_illegal_gzh_articles(gh_id, title)

        except Exception as e:
            print(f"crawler failed: {article['account_name']}, error: {e}")

    async def deal(self):
        article_list = await self.fetch_article_list_to_check()
        for article in tqdm(article_list, desc="站内文章监测任务"):
            await self.check_each_article(article)

        return self.TASK_SUCCESS_CODE