"""
author: CT
description: 从 Hive dim_keepa_asin_info 读取 keepa 数据，过滤 updated_time 超过 3 个月的 asin，
             附带当前月份字段 month（yyyy-MM），导出到 PG us_asin_profit_keepa_add，触发爬虫重新抓取 Keepa
执行示例: spark-submit export_keepa_asin_del.py us
"""
import os
import sys
from datetime import datetime
from dateutil.relativedelta import relativedelta

sys.path.append(os.path.dirname(sys.path[0]))

from pyspark.sql import functions as F

from utils.spark_util import SparkUtil
from utils.db_util import DBUtil

if __name__ == '__main__':
    site_name = sys.argv[1]

    spark = SparkUtil.get_spark_session(f'export_keepa_asin_del_{site_name}')

    # 1. 全量读取 keepa 数据（updated_time 过滤改到 Spark 端）
    sql = f"""
        SELECT asin, updated_time
        FROM dim_keepa_asin_info
        WHERE site_name = '{site_name}'
    """
    print(f"sql=\n{sql}")
    df_all = spark.sql(sqlQuery=sql)

    # 2. Spark 端过滤超过 3 个月的 asin（数据读取后处理，不在 Hive SQL 中算）
    three_months_ago = (datetime.now() - relativedelta(months=3)).strftime('%Y-%m-%d %H:%M:%S')
    current_month    = datetime.now().strftime('%Y-%m')
    print(f"过滤阈值 updated_time < {three_months_ago}；附加 month={current_month}")

    df_need_export = df_all.filter(F.col('updated_time') < F.lit(three_months_ago)) \
        .select(
            F.col('asin'),
            F.lit(current_month).alias('month'),
        ).cache()
    print(f"导出数据量：{df_need_export.count():,}")

    con_info = DBUtil.get_connection_info(db_type='postgresql_cluster', site_name=site_name)
    table_name = 'us_asin_profit_keepa_add'
    df_need_export.write.format("jdbc") \
        .option("url", con_info["url"]) \
        .option("dbtable", table_name) \
        .option("user", con_info["username"]) \
        .option("password", con_info["pwd"]) \
        .mode("append") \
        .save()

    df_need_export.unpersist()
    print("success")
