LongArticlesJob/tasks/data_tasks/fwh_data_recycle.py

import json
import time
import urllib.parse
from datetime import datetime
from typing import Optional

from tqdm import tqdm

from pymysql.cursors import DictCursor

from applications.api import FeishuBotApi
from applications.db import DatabaseConnector
from applications.utils import str_to_md5
from cold_start.crawler.wechat import get_article_detail
from config import denet_config, long_articles_config, piaoquan_crawler_config


class FwhDataRecycle:
    RECYCLE_INIT_STATUS = 0
    RECYCLE_PROCESSING_STATUS = 1
    RECYCLE_SUCCESS_STATUS = 2
    RECYCLE_FAILED_STATUS = 99

    PUBLISH_SUCCESS_STATUS = 2

    STAT_PERIOD = 2 * 24 * 3600

    def __init__(self):
        self.denet_client = DatabaseConnector(denet_config)
        self.denet_client.connect()

        self.long_articles_client = DatabaseConnector(long_articles_config)
        self.long_articles_client.connect()

        self.piaoquan_client = DatabaseConnector(piaoquan_crawler_config)
        self.piaoquan_client.connect()

        self.feishu_robot = FeishuBotApi()

    def get_group_server_accounts(self):
        fetch_query = f"""
            select gzh_id from article_gzh_developer;
        """
        fetch_response = self.piaoquan_client.fetch(fetch_query, cursor_type=DictCursor)
        gh_id_list = [i["gzh_id"] for i in fetch_response]
        # gh_id_list = ['gh_5e543853d8f0']
        return gh_id_list

    def get_server_account_name(self, gh_id: str) -> Optional[str]:
        fetch_query = f"""
            select account_name from long_articles_group_send_result where gh_id = %s limit 1;
        """
        fetch_response = self.long_articles_client.fetch(
            fetch_query, cursor_type=DictCursor, params=(gh_id,)
        )
        account_name = fetch_response[0]["account_name"] if fetch_response else None
        return account_name


class FwhGroupPublishRecordManager(FwhDataRecycle):

    def get_published_articles(self):
        fetch_query = f"""
            select id, publish_content_id, gh_id, user_group_id
            from long_articles_group_send_result
            where status = %s and recycle_status = %s;
        """
        fetch_response = self.long_articles_client.fetch(
            query=fetch_query,
            cursor_type=DictCursor,
            params=(self.PUBLISH_SUCCESS_STATUS, self.RECYCLE_INIT_STATUS),
        )
        return fetch_response

    def get_article_url_from_aigc_system(self, publish_content_id, user_group_id):
        sql = f"""
            select t1.publish_stage_url
            from publish_content_stage_url t1
            left join publish_content t2 on t1.publish_content_id = t2.id
            where t1.publish_content_id = %s and t1.user_group_id = %s;
        """
        article_info = self.denet_client.fetch(
            sql,
            cursor_type=DictCursor,
            params=(publish_content_id, user_group_id),
        )
        if article_info:
            return article_info[0]
        else:
            return None

    def update_recycle_status(self, record_id, ori_status, new_status):
        update_query = f"""
            update long_articles_group_send_result
            set recycle_status = %s
            where id = %s and recycle_status = %s;
        """
        return self.long_articles_client.save(
            update_query, (new_status, record_id, ori_status)
        )

    def set_article_url(self, record_id, article_url):
        update_query = f"""
            update long_articles_group_send_result
            set url = %s, recycle_status = %s
            where id = %s and recycle_status = %s;
        """
        return self.long_articles_client.save(
            query=update_query,
            params=(
                article_url,
                self.RECYCLE_SUCCESS_STATUS,
                record_id,
                self.RECYCLE_PROCESSING_STATUS,
            ),
        )

    def deal(self):
        publish_records = self.get_published_articles()
        for publish_record in tqdm(publish_records):
            publish_content_id = publish_record["publish_content_id"]
            record_id = publish_record["id"]
            group_id = publish_record["user_group_id"]
            # lock
            self.update_recycle_status(
                record_id, self.RECYCLE_INIT_STATUS, self.RECYCLE_PROCESSING_STATUS
            )

            publish_call_back_info = self.get_article_url_from_aigc_system(
                publish_content_id, group_id
            )
            if publish_call_back_info:
                article_url = publish_call_back_info["publish_stage_url"]
                if article_url:
                    # set record and unlock
                    self.set_article_url(record_id, article_url)
                else:
                    # unlock
                    self.update_recycle_status(
                        record_id,
                        self.RECYCLE_PROCESSING_STATUS,
                        self.RECYCLE_INIT_STATUS,
                    )
            else:
                # unlock
                self.update_recycle_status(
                    record_id, self.RECYCLE_PROCESSING_STATUS, self.RECYCLE_INIT_STATUS
                )

    def monitor(self, date_string: str = datetime.today().strftime("%Y-%m-%d")):
        """
        monitor the publish record
        """
        now = datetime.now()
        if now.hour < 12:
            account_list = self.get_group_server_accounts()
            do_not_publish_account = []
            sql = f"""
                select account_name as '账号名称', gh_id, count(distinct user_group_id) as '发文组数'
                from long_articles_group_send_result
                where publish_date = %s
                group by account_name, gh_id; 
            """
            publish_records = self.long_articles_client.fetch(
                query=sql, cursor_type=DictCursor, params=(date_string,)
            )
            self.feishu_robot.bot(
                title=f"{date_string}服务号发文记录",
                mention=False,
                detail=publish_records,
                env="server_account_publish_monitor",
            )

            publish_account_id_set = set([i["gh_id"] for i in publish_records])
            for account_id in account_list:
                if account_id not in publish_account_id_set:
                    account_name = self.get_server_account_name(account_id)
                    do_not_publish_account.append(
                        {
                            "account_name": account_name,
                            "gh_id": account_id,
                        }
                    )

            if do_not_publish_account:
                self.feishu_robot.bot(
                    title=f"{date_string}发现服务号存在未发文情况",
                    detail=do_not_publish_account,
                    env="server_account_publish_monitor",
                )


class SaveFwhDataToDatabase(FwhDataRecycle):

    def update_article_read_cnt(self, wx_sn, new_read_cnt):
        """
        update article read cnt
        """
        if new_read_cnt <= 0:
            return 0

        update_query = f"""
            update official_articles_v2
            set show_view_count = %s
            where wx_sn = %s;
        """
        return self.piaoquan_client.save(update_query, (new_read_cnt, wx_sn))

    def save_data_to_database(self, article):
        """
        save data to db
        """
        insert_query = f"""
            insert into official_articles_v2
            (ghId, accountName, appMsgId, title, Type, createTime, updateTime, ItemIndex, ContentUrl, show_view_count, 
             wx_sn, title_md5, article_group, channel_content_id, root_source_id_list, publish_timestamp) 
            values (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s);      
        """
        return self.piaoquan_client.save(insert_query, article)

    def get_stat_published_articles(self, gh_id):
        earliest_timestamp = int(time.time()) - self.STAT_PERIOD
        fetch_query = f"""
            select publish_date, account_name, gh_id, user_group_id, url, publish_timestamp
            from long_articles_group_send_result
            where gh_id = %s and recycle_status = %s and create_time > %s;
        """
        return self.long_articles_client.fetch(
            fetch_query,
            DictCursor,
            (gh_id, self.RECYCLE_SUCCESS_STATUS, earliest_timestamp),
        )

    def process_each_account_data(self, account_published_article_list):
        if not account_published_article_list:
            return

        for article in account_published_article_list:
            account_name = article["account_name"]
            gh_id = article["gh_id"]
            user_group_id = article["user_group_id"]
            url = article["url"]
            # get article detail info with spider

            try:
                article_detail_info = get_article_detail(
                    url, is_count=True, is_cache=False
                )
                time.sleep(1)
                content_url = article_detail_info["data"]["data"]["content_link"]
                app_msg_id = content_url.split("mid=")[-1].split("&")[0]
                wx_sn = content_url.split("sn=")[-1]
                publish_timestamp = int(
                    article_detail_info["data"]["data"]["publish_timestamp"] / 1000
                )
                create_time = publish_timestamp
                update_time = publish_timestamp
                item_index = article_detail_info["data"]["data"]["item_index"]
                show_view_count = article_detail_info["data"]["data"]["view_count"]
                title = article_detail_info["data"]["data"]["title"]
                title_md5 = str_to_md5(title)
                channel_content_id = article_detail_info["data"]["data"][
                    "channel_content_id"
                ]
                mini_program_info = article_detail_info["data"]["data"]["mini_program"]
                root_source_id_list = [
                    urllib.parse.parse_qs(urllib.parse.unquote(i["path"]))[
                        "rootSourceId"
                    ][0]
                    for i in mini_program_info
                ]
                root_source_id_list = json.dumps(root_source_id_list)
                try:
                    self.save_data_to_database(
                        article=(
                            gh_id,
                            account_name,
                            app_msg_id,
                            title,
                            "9",
                            create_time,
                            update_time,
                            item_index,
                            content_url,
                            show_view_count,
                            wx_sn,
                            title_md5,
                            user_group_id,
                            channel_content_id,
                            root_source_id_list,
                            publish_timestamp,
                        )
                    )
                except Exception as e:
                    self.update_article_read_cnt(wx_sn, show_view_count)
            except Exception as e:
                print(f"article {url} is not available, skip it")
                print(e)

    def deal(self):
        account_id_list = self.get_group_server_accounts()
        for account_id in account_id_list:
            publish_articles = tqdm(
                self.get_stat_published_articles(account_id),
                desc=f"<crawling> {account_id}",
            )
            self.process_each_account_data(publish_articles)


class FwhDataExportTemp(FwhDataRecycle):

    def get_publish_articles(self, gh_id):
        sql = f"""
            -- select accountName, title, article_group, ItemIndex, show_view_count, from_unixtime(createTime, '%Y-%m-%d'), root_source_id_list
            select accountName, ContentUrl, title, ItemIndex, from_unixtime(createTime, '%Y-%m-%d'), sum(show_view_count), group_concat(article_group) as 'group', root_source_id_list
            from official_articles_v2
            where from_unixtime(publish_timestamp) between '2025-06-09' and '2025-06-13'
            and ghId = '{gh_id}' and article_group is not null
            group by accountName, title, ItemIndex;
        """
        return self.piaoquan_client.fetch(query=sql, cursor_type=DictCursor)

    def get_fission_info(self, root_source_id_list):
        """
        获取裂变信息
        """
        root_source_id_tuple = tuple(json.loads(root_source_id_list))
        query = f"""
            select sum(uv) as 'uv', sum(first_uv) as 'first_uv', sum(split_uv) as 'split_uv', sum(split0) as 'T+0_fission'
            from changwen_data_rootsourceid 
            where root_source_id
            in %s;
        """
        return self.long_articles_client.fetch(
            query=query, cursor_type=DictCursor, params=(root_source_id_tuple,)
        )

    def get_fans_num(self, gh_id, group_id_tuple):
        sql = f"""
            select count(1) as 'fans_count'
            from article_user_group
            where gzh_id = %s and user_group_id in %s
            and is_delete = 0;
        """
        return self.piaoquan_client.fetch(
            query=sql, cursor_type=DictCursor, params=(gh_id, group_id_tuple)
        )

    def deal(self):
        import pandas as pd

        gh_id_list = self.get_group_server_accounts()
        L = []
        for gh_id in gh_id_list:
            publish_articles = self.get_publish_articles(gh_id)
            for article in publish_articles:
                try:
                    group_id_tuple = tuple(article["group"].split(","))
                    fans_count = self.get_fans_num(gh_id, group_id_tuple)[0][
                        "fans_count"
                    ]
                    root_source_id_list = article["root_source_id_list"]
                    fission_info = self.get_fission_info(root_source_id_list)
                    article["uv"] = fission_info[0]["uv"]
                    article["first_uv"] = fission_info[0]["first_uv"]
                    article["split_uv"] = fission_info[0]["split_uv"]
                    article["T+0_fission"] = fission_info[0]["T+0_fission"]
                    article["fans_count"] = fans_count
                    L.append(article)
                except Exception as e:
                    print(f"article {article['ContentUrl']} is not available, skip it")
        df = pd.DataFrame(L)
        df.to_csv("temp2.csv", index=False)