"""
author: CT
description: 从 Hive dim_keepa_asin_info 读取 keepa 数据，过滤 updated_time 超过 3 个月的 asin，
             附带当前月份字段 month（yyyy-MM），导出到 PG us_asin_profit_keepa_add，触发爬虫重新抓取 Keepa
             导出前 LEFT ANTI 剔除 PG 表中已存在的 asin，避免重复触发
执行示例: 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).cache()
    print(f"全量读取 keepa 数据:{df_all.count()}")

    # 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}")

    con_info = DBUtil.get_connection_info(db_type='postgresql_cluster', site_name=site_name)
    table_name = 'us_asin_profit_keepa_add'

    # 3. 读取 PG 已导出 asin 集合，用于 LEFT ANTI 剔除（避免重复触发爬虫）
    #    PG 端只取 asin 列全量，Spark 端 dropDuplicates 去重
    df_pg_existing = spark.read.format("jdbc") \
        .option("url", con_info["url"]) \
        .option("dbtable", table_name) \
        .option("user", con_info["username"]) \
        .option("password", con_info["pwd"]) \
        .load() \
        .select('asin') \
        .dropDuplicates(['asin']).cache()
    print(f"PG 已导出 asin 数量：{df_pg_existing.count():,}")

    # 4. 过滤 3 个月以前 + LEFT ANTI 剔除已导出
    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'),
        ) \
        .join(df_pg_existing, on='asin', how='left_anti') \
        .cache()
    print(f"导出数据量：{df_need_export.count():,}")

    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")
