growth-jobs/run_category_model_v1.py

#! /usr/bin/env python
# -*- coding: utf-8 -*-
# vim:fenc=utf-8
#
# Copyright © 2024 StrayWarrior <i@straywarrior.com>


import sys
import os
sys.path.append(os.path.join(os.path.dirname(__file__), 'src'))

import time
import json
from datetime import datetime, timedelta
import pandas as pd
from argparse import ArgumentParser
from long_articles.category_models import CategoryRegressionV1
from long_articles.consts import category_feature_v2, category_name_map_v2, reverse_category_name_map_v2
from common.database import MySQLManager
from common import db_operation
from common.logging import LOG
from config.dev import Config

NIGHT_ACCOUNTS = ('gh_12523d39d809','gh_df4a630c04db','gh_f67df16f4670','gh_ca44517edda9','gh_a66c1316fd5e','gh_4242c478bbba','gh_60b0c23fcc7c','gh_33b3470784fc','gh_ec1bcb283daf','gh_234ab9ff490d','gh_7715a626a4c6','gh_1bfe1d257728','gh_9db5e3ac2c93','gh_9d1ae5f9ceac','gh_7208b813f16d','gh_e56ddf195d91','gh_a43aecffe81b','gh_d4a7d2ce54fd','gh_c2b458818b09','gh_349a57ef1c44','gh_89bfe54ad90f','gh_b929ed680b62','gh_f8e8a931ff56','gh_916f4fad5ce0','gh_0d7c5f4c38a9','gh_bceef3f747c2','gh_706456719017','gh_fd51a5e33fc6','gh_5372093f5fb0','gh_957ff8e08e1b','gh_64fc629d3ec2','gh_c8b69797912a','gh_6909b38ad95f','gh_1e69a1b4dc1a','gh_0763523103e4','gh_9b83a9ad7da0','gh_82b416f27698','gh_a60647e98cd9','gh_3ce2fa1956ea','gh_44127c197525','gh_06834aba13a5','gh_c33809af68bc','gh_82cf39ef616e','gh_a342ef23c48e','gh_c9cc1471af7d','gh_291ec369f017','gh_810a439f320a','gh_00f942061a0d','gh_7662653b0e77','gh_d192d757b606','gh_391702d26b3b','gh_3e90f421c974','gh_30d189fe56c7','gh_7ebfbbf675ee','gh_3f84c2b9a1a2','gh_bccbe3681e22','gh_005fc1cb4b73','gh_21d120007b64','gh_3d5f24fd3311','gh_3621aaa6c4a0','gh_aee2dca32701','gh_c25c6040c4b2','gh_641019d44876','gh_95ba63e5cf18','gh_efd90dcf48ac','gh_5e1464b76ff6','gh_5765f834684c','gh_81bec2f4f577','gh_401396006e13','gh_7c33726c5147','gh_bbd8a52ba98b','gh_f74ca3104604'
)

def prepare_raw_data(dt_begin, dt_end):
    data_fields = ['dt', 'gh_id', 'account_name', 'title', 'similarity',
                   'view_count_rate', 'category', 'read_avg',
                   'read_avg_rate', 'first_pub_interval', '`index`']
    fields_str = ','.join(data_fields)
    db_manager = MySQLManager(Config().MYSQL_LONG_ARTICLES)
    night_accounts_condition = str(NIGHT_ACCOUNTS)
    sql = f"""
        SELECT {fields_str} FROM datastat_score WHERE dt BETWEEN {dt_begin} AND {dt_end}
            AND similarity > 0 AND category IS NOT NULL AND read_avg > 500
            AND read_avg_rate BETWEEN 0.3 AND 3 AND view_count_rate > 0
            AND `index` in (1, 2)
            AND (FROM_UNIXTIME(coalesce(publish_timestamp, 0), '%H') < '15'
                OR gh_id in {night_accounts_condition})
            AND dt NOT BETWEEN 20250105 AND 20250215
        """
    rows = db_manager.select(sql)
    df = pd.DataFrame(rows, columns=data_fields)
    df.rename(columns={'`index`': 'index'}, inplace=True)
    df = df.drop_duplicates(['dt', 'gh_id', 'title'])
    return df


def clear_old_version(db_manager, dt):
    update_timestamp = int(time.time())
    sql = f"""
        UPDATE account_category
        SET status = 0, update_timestamp = {update_timestamp}
        WHERE dt < {dt} and status = 1 and version = 2
    """
    rows = db_manager.execute(sql)
    print(f"updated rows for clear: {rows}")

def get_last_version(db_manager, dt):
    sql = f"""
        SELECT gh_id, category_map
        FROM account_category
        WHERE dt = (SELECT max(dt) FROM account_category WHERE dt < {dt} AND
        status = 1)
    """
    data = db_manager.select(sql)
    return data

def compare_version(db_manager, dt_version, new_version, account_id_map):
    last_version = get_last_version(db_manager, dt_version)
    last_version = { entry[0]: json.loads(entry[1]) for entry in last_version }
    new_version = { entry['gh_id']: json.loads(entry['category_map']) for entry in new_version }
    # new record
    all_gh_ids = set(list(new_version.keys()) + list(last_version.keys()))
    for gh_id in all_gh_ids:
        account_name = account_id_map.get(gh_id, None)
        if gh_id not in last_version:
            print(f"new account {account_name}: {new_version[gh_id]}")
        elif gh_id not in new_version:
            print(f"old account {account_name}: {last_version[gh_id]}")
        else:
            new_cates = new_version[gh_id]
            old_cates = last_version[gh_id]
            for cate in new_cates:
                if cate not in old_cates:
                    print(f"account {account_name} new cate: {cate} {new_cates[cate]}")
            for cate in old_cates:
                if cate not in new_cates:
                    print(f"account {account_name} old cate: {cate} {old_cates[cate]}")


def main():
    parser = ArgumentParser()
    parser.add_argument('-n', '--dry-run', action='store_true', help='do not update database')
    parser.add_argument('--run-at', help='dt, also for version')
    parser.add_argument('--print-matrix', action='store_true')
    parser.add_argument('--print-residual', action='store_true')
    args = parser.parse_args()

    run_date = datetime.today()
    if args.run_at:
        run_date = datetime.strptime(args.run_at, "%Y%m%d")
    begin_dt = 20240914
    end_dt = (run_date - timedelta(1)).strftime("%Y%m%d")
    dt_version = end_dt
    LOG.info(f"data range: {begin_dt} - {end_dt}")

    raw_df = prepare_raw_data(begin_dt, end_dt)

    cate_model = CategoryRegressionV1(category_feature_v2, category_name_map_v2)
    df = cate_model.preprocess_data(raw_df)

    if args.dry_run and args.print_matrix:
        cate_model.build_and_print_matrix(df)
        return

    create_timestamp = int(time.time())
    update_timestamp = create_timestamp

    records_to_save = []

    param_to_category_map = reverse_category_name_map_v2
    account_ids = df['gh_id'].unique()
    account_id_map = df[['account_name', 'gh_id']].drop_duplicates() \
        .set_index('gh_id')['account_name'].to_dict()

    account_negative_cates = {k: [] for k in account_ids}
    P_VALUE_THRESHOLD = 0.15
    for account_id in account_ids:
        sub_df = df[df['gh_id'] == account_id]
        account_name = account_id_map[account_id]
        sample_count = len(sub_df)
        if sample_count < 5:
            continue
        params, t_stats, p_values = cate_model.run_ols_linear_regression(
            sub_df, args.print_residual, P_VALUE_THRESHOLD)
        current_record = {}
        current_record['dt'] = dt_version
        current_record['gh_id'] = account_id
        current_record['category_map'] = {}
        param_names = cate_model.get_param_names()
        for name, param, p_value in zip(param_names, params, p_values):
            cate_name = param_to_category_map.get(name, None)
            # 用于排序的品类相关性
            if abs(param) > 0.1 and p_value < P_VALUE_THRESHOLD and cate_name is not None:
                scale_factor = min(0.1 / p_value, 1)
                print(f"{account_id} {account_name} {cate_name} {param:.3f} {p_value:.3f}")
                truncate_param = round(max(min(param, 0.25), -0.25) * scale_factor, 6)
                current_record['category_map'][cate_name] = truncate_param
            # 用于冷启文章分配的负向品类
            if param < -0.1 and cate_name is not None and p_value < P_VALUE_THRESHOLD:
                account_negative_cates[account_id].append(cate_name)
                # print((account_name, cate_name, param, p_value))
        if not current_record['category_map']:
            continue
        current_record['category_map'] = json.dumps(current_record['category_map'], ensure_ascii=False)
        current_record['status'] = 1
        current_record['version'] = 2
        current_record['create_timestamp'] = create_timestamp
        current_record['update_timestamp'] = update_timestamp
        records_to_save.append(current_record)

    db_manager = MySQLManager(Config().MYSQL_LONG_ARTICLES)
    if args.dry_run:
        compare_version(db_manager, dt_version, records_to_save, account_id_map)
        return

    rows = db_manager.batch_insert('account_category', records_to_save, ignore=True)
    if rows != len(records_to_save):
        for record in records_to_save:
            sql = f"""
                UPDATE account_category
                SET category_map = '{record['category_map']}',
                    update_timestamp = {record['update_timestamp']}
                WHERE dt = {record['dt']} AND gh_id = '{record['gh_id']}'
                    AND category_map != '{record['category_map']}'
                    AND version = 2
            """
            update_rows = db_manager.execute(sql)
            print(f"updated rows: {update_rows}, {record['gh_id']}")
    clear_old_version(db_manager, dt_version)

    # 过滤空账号
    for account_id in [*account_negative_cates.keys()]:
        if not account_negative_cates[account_id]:
            account_negative_cates.pop(account_id)

    # print(json.dumps(account_negative_cates, ensure_ascii=False, indent=2))

if __name__ == '__main__':
    pd.set_option('display.max_columns', None)
    pd.set_option('display.max_rows', None)
    main()