-- Databricks notebook source
-- CREATE OR REPLACE TABLE DM.DM_TD_EXT_CHC_PACK2MARKET (
--   MARKET STRING,
--   PACK_CODE STRING,
--   PACK_DESC STRING,
--   STGH_DESC STRING,
--   PACK_LCH STRING,
--   PROD_CODE STRING,
--   CMPS_CODE STRING,
--   CMPS_DESC STRING,
--   ATC1_CODE STRING,
--   ATC2_CODE STRING,
--   ATC3_CODE STRING,
--   ATC4_CODE STRING,
--   APP1_CODE STRING,
--   APP2_CODE STRING,
--   APP3_CODE STRING,
--   BIO_DESC STRING,
--   GENE_ORIG_DESC STRING,
--   ETH_OTC_DESC STRING,
--   NRDL_DESC STRING,
--   NRDL_ENTRY_DATE STRING,
--   EDL_DESC STRING,
--   TCM_DESC STRING,
--   PAED_DESC STRING,
--   GQCE_DESC STRING,
--   VBP_DESC STRING,
--   MANU_CODE STRING,
--   MANU_DESC STRING,
--   MNFL_CODE STRING,
--   MNFL_DESC STRING,
--   CORP_CODE STRING,
--   CORP_DESC STRING,
--   BRANDTYPE STRING,
--   BU STRING,
--   STARTTIME STRING,
--   ENDTIME STRING,
--   MARKET_RATIO STRING,
--   ETL_INSERT_DT TIMESTAMP,
--   ETL_UPDATE_DT TIMESTAMP
-- )
-- USING delta
-- -- LOCATION 'abfss://master@azcdatalakeprd.dfs.core.chinacloudapi.cn/DM/dm_td_ext_chc_pack2market';
-- -- 上面是生产环境location，下面是测试环境location
-- LOCATION 'abfss://master@retaildlstoragetest.dfs.core.chinacloudapi.cn/DM/dm_td_ext_chc_pack2market';

-- COMMAND ----------

-- CREATE OR REPLACE TABLE DM.DM_TD_EXT_chc_PACK2MARKET_KEYCOMPETITOR (
--   KEY_COMPETITOR STRING,
--   MARKET STRING,
--   PACK_CODE STRING)
-- USING delta
-- -- LOCATION 'abfss://master@azcdatalakeprd.dfs.core.chinacloudapi.cn/DM/dm_td_ext_chc_pack2market_keycompetitor';
-- -- 上面是生产环境location，下面是测试环境location
-- LOCATION 'abfss://master@retaildlstoragetest.dfs.core.chinacloudapi.cn/DM/dm_td_ext_chc_pack2market_keycompetitor';

-- COMMAND ----------

-- CREATE or REPLACE TABLE DM.DM_TD_EXT_chc_PACK2MARKET_CLASS (
--   MARKET STRING,
--   PACK_CODE STRING,
--   CLASS STRING)
-- USING delta
-- -- LOCATION 'abfss://master@azcdatalakeprd.dfs.core.chinacloudapi.cn/DM/dm_td_ext_chc_pack2market_class';
-- -- 上面是生产环境location，下面是测试环境location
-- LOCATION 'abfss://master@retaildlstoragetest.dfs.core.chinacloudapi.cn/DM/dm_td_ext_chc_pack2market_class';

-- COMMAND ----------

-- 使用 EXCEPT（集合差操作）简化逻辑
INSERT OVERWRITE TABLE DM.DM_TD_EXT_CHC_PACK2MARKET
-- 包含规则
SELECT DISTINCT
  T2.MARKET,
  T1.PACK_CODE,
  T1.PACK_DESC,
  T1.STGH_DESC,
  T1.PACK_LCH,
  T1.PROD_CODE,
  T1.CMPS_CODE,
  T1.CMPS_DESC,
  T1.ATC1_CODE,
  T1.ATC2_CODE,
  T1.ATC3_CODE,
  T1.ATC4_CODE,
  T1.APP1_CODE,
  T1.APP2_CODE,
  T1.APP3_CODE,
  T1.BIO_DESC,
  T1.GENE_ORIG_DESC AS GENE_ORIG,
  T1.ETH_OTC_DESC AS ETH_OTC_D,
  T1.NRDL_DESC,
  T1.NRDL_ENTRY_DATE AS NRDL_ENTR,
  T1.EDL_DESC,
  T1.TCM_DESC,
  T1.PAED_DESC,
  T1.GQCE_DESC,
  T1.VBP_DESC,
  T1.MANU_CODE,
  T1.MANU_DESC,
  T1.MNFL_CODE,
  T1.MNFL_DESC,
  T1.CORP_CODE,
  T1.CORP_DESC,
  T1.BRANDTYPE,
  T2.BU,
  T2.STARTTIME,
  T2.ENDTIME,
  COALESCE(T2.EXTEND_MARKET_RATIO, '1') AS MARKET_RATIO,
  FROM_UTC_TIMESTAMP(CURRENT_TIMESTAMP(),'UTC+8') AS ETL_INSERT_DT,
  FROM_UTC_TIMESTAMP(CURRENT_TIMESTAMP(),'UTC+8') AS ETL_UPDATE_DT
FROM DM.DM_TD_EXT_CHC_PACK_PROPERTY AS T1
LEFT JOIN (
  SELECT 
    MARKET,
    BU,
    STARTTIME,
    ENDTIME,
    EXTEND_MARKET_RATIO,
    ATC1_CODE,
    ATC2_CODE,
    ATC3_CODE,
    ATC4_CODE,
    NFC1_CODE,
    NFC2_CODE,
    NFC3_CODE,
    CORPORATION_CODE,
    MANUFACTURER_CODE,
    PRODUCT_CODE,
    PACK_CODE,
    STRENGTH,
    MOLECULE_CODE
  FROM DWD.DWD_GND_IMS_TBLMARKET_BYMONTH
  WHERE EXTEND_MARKET IS NULL AND (NOT_IN_FLAG IS NULL OR NOT_IN_FLAG = '1')
) AS T2
  ON (T2.ATC1_CODE IS NULL OR T1.ATC1_CODE = T2.ATC1_CODE)
  AND (T2.ATC2_CODE IS NULL OR T1.ATC2_CODE = T2.ATC2_CODE)
  AND (T2.ATC3_CODE IS NULL OR T1.ATC3_CODE = T2.ATC3_CODE)
  AND (T2.ATC4_CODE IS NULL OR T1.ATC4_CODE = T2.ATC4_CODE)
  AND (T2.NFC1_CODE IS NULL OR T1.APP1_CODE = T2.NFC1_CODE)
  AND (T2.NFC2_CODE IS NULL OR T1.APP2_CODE = T2.NFC2_CODE)
  AND (T2.NFC3_CODE IS NULL OR T1.APP3_CODE = T2.NFC3_CODE)
  AND (T2.CORPORATION_CODE IS NULL OR T1.CORP_CODE = T2.CORPORATION_CODE)
  AND (T2.MANUFACTURER_CODE IS NULL OR T1.MANU_CODE = T2.MANUFACTURER_CODE)
  AND (T2.PRODUCT_CODE IS NULL OR T1.PROD_CODE = T2.PRODUCT_CODE)
  AND (T2.PACK_CODE IS NULL OR T1.PACK_CODE = T2.PACK_CODE)
  AND (T2.STRENGTH IS NULL OR T1.STGH_DESC = T2.STRENGTH)
  AND (T2.MOLECULE_CODE IS NULL OR T1.CMPS_CODE = T2.MOLECULE_CODE)
WHERE T2.MARKET IS NOT NULL

EXCEPT

-- 排除规则
SELECT DISTINCT
  T2.MARKET,
  T1.PACK_CODE,
  T1.PACK_DESC,
  T1.STGH_DESC,
  T1.PACK_LCH,
  T1.PROD_CODE,
  T1.CMPS_CODE,
  T1.CMPS_DESC,
  T1.ATC1_CODE,
  T1.ATC2_CODE,
  T1.ATC3_CODE,
  T1.ATC4_CODE,
  T1.APP1_CODE,
  T1.APP2_CODE,
  T1.APP3_CODE,
  T1.BIO_DESC,
  T1.GENE_ORIG_DESC AS GENE_ORIG,
  T1.ETH_OTC_DESC AS ETH_OTC_D,
  T1.NRDL_DESC,
  T1.NRDL_ENTRY_DATE AS NRDL_ENTR,
  T1.EDL_DESC,
  T1.TCM_DESC,
  T1.PAED_DESC,
  T1.GQCE_DESC,
  T1.VBP_DESC,
  T1.MANU_CODE,
  T1.MANU_DESC,
  T1.MNFL_CODE,
  T1.MNFL_DESC,
  T1.CORP_CODE,
  T1.CORP_DESC,
  T1.BRANDTYPE,
  T2.BU,
  T2.STARTTIME,
  T2.ENDTIME,
  COALESCE(T2.EXTEND_MARKET_RATIO, '1') AS MARKET_RATIO,
  FROM_UTC_TIMESTAMP(CURRENT_TIMESTAMP(),'UTC+8') AS ETL_INSERT_DT,
  FROM_UTC_TIMESTAMP(CURRENT_TIMESTAMP(),'UTC+8') AS ETL_UPDATE_DT
FROM DM.DM_TD_EXT_CHC_PACK_PROPERTY AS T1
LEFT JOIN (
  SELECT 
    MARKET,
    BU,
    STARTTIME,
    ENDTIME,
    EXTEND_MARKET_RATIO,
    ATC1_CODE,
    ATC2_CODE,
    ATC3_CODE,
    ATC4_CODE,
    NFC1_CODE,
    NFC2_CODE,
    NFC3_CODE,
    CORPORATION_CODE,
    MANUFACTURER_CODE,
    PRODUCT_CODE,
    PACK_CODE,
    STRENGTH,
    MOLECULE_CODE  
  FROM DWD.DWD_GND_IMS_TBLMARKET_BYMONTH
  WHERE EXTEND_MARKET IS NULL AND NOT_IN_FLAG = '0'
) AS T2
  ON (T2.ATC1_CODE IS NULL OR T1.ATC1_CODE = T2.ATC1_CODE)
  AND (T2.ATC2_CODE IS NULL OR T1.ATC2_CODE = T2.ATC2_CODE)
  AND (T2.ATC3_CODE IS NULL OR T1.ATC3_CODE = T2.ATC3_CODE)
  AND (T2.ATC4_CODE IS NULL OR T1.ATC4_CODE = T2.ATC4_CODE)
  AND (T2.NFC1_CODE IS NULL OR T1.APP1_CODE = T2.NFC1_CODE)
  AND (T2.NFC2_CODE IS NULL OR T1.APP2_CODE = T2.NFC2_CODE)
  AND (T2.NFC3_CODE IS NULL OR T1.APP3_CODE = T2.NFC3_CODE)
  AND (T2.CORPORATION_CODE IS NULL OR T1.CORP_CODE = T2.CORPORATION_CODE)
  AND (T2.MANUFACTURER_CODE IS NULL OR T1.MANU_CODE = T2.MANUFACTURER_CODE)
  AND (T2.PRODUCT_CODE IS NULL OR T1.PROD_CODE = T2.PRODUCT_CODE)
  AND (T2.PACK_CODE IS NULL OR T1.PACK_CODE = T2.PACK_CODE)
  AND (T2.STRENGTH IS NULL OR T1.STGH_DESC = T2.STRENGTH)
  AND (T2.MOLECULE_CODE IS NULL OR T1.CMPS_CODE = T2.MOLECULE_CODE)
WHERE T2.MARKET IS NOT NULL;

-- COMMAND ----------

INSERT OVERWRITE TABLE DM.DM_TD_EXT_CHC_PACK2MARKET_KEYCOMPETITOR
WITH kc AS (
  SELECT
    keycompetitor, no,
    ROW_NUMBER() OVER (ORDER BY
      CASE
        WHEN ATC1_Code     IS NOT NULL THEN 1
        WHEN ATC2_Code     IS NOT NULL THEN 2
        WHEN ATC3_Code     IS NOT NULL THEN 3
        WHEN ATC4_Code     IS NOT NULL THEN 4
        WHEN Molecule_Code IS NOT NULL THEN 5
        WHEN Product_Code  IS NOT NULL THEN 6
        WHEN Pack_Code     IS NOT NULL THEN 7
        ELSE 999
      END,
      CASE WHEN NFC1_CODE IS NULL THEN 0 ELSE 1 END,
      CASE WHEN NFC2_CODE IS NULL THEN 0 ELSE 1 END,
      CASE WHEN NFC3_CODE IS NULL THEN 0 ELSE 1 END
    ) AS no1,
    ATC1_Code, ATC2_Code, ATC3_Code, ATC4_Code,
    NFC1_Code, NFC2_Code, NFC3_Code,
    corporation_code, Manufacturer_Code,
    Product_Code, Pack_Code, Strength, Molecule_Code, market
  FROM dwd.dwd_gnd_ims_tblkeycompetitor
),
joined AS (
  SELECT
    /*+ BROADCAST(kc) */
    COALESCE(kc.keycompetitor, 'OTHERS') AS Key_Competitor,  -- ✅ UPDATE 逻辑内联
    t1.market, t1.PACK_CODE
    /*bug 调试区
    ,
    kc.no,
    kc.no1,
    t1.PROD_COD, t1.CMPS_COD, t1.CMPS_DES,
    t1.ATC1_COD, t1.ATC2_COD, t1.ATC3_COD, t1.ATC4_COD,
    t1.APP1_COD, t1.APP2_COD, t1.APP3_COD,
    t1.BIO_DESC, t1.GENE_ORIG_DESC, t1.ETH_OTC_DESC,
    t1.NRDL_DESC, t1.NRDL_Entry_Date, t1.EDL_DESC, t1.TCM_DESC,
    t1.PAED_DESC, t1.GQCE_DESC, t1.VBP_DESC,
    t1.MANU_COD, t1.MANU_DES, t1.MNFL_COD, t1.MNFL_DES,
    t1.CORP_COD, t1.CORP_DES, t1.BrandType, t1.bu, t1.Market_Ratio
    */
  ,ROW_NUMBER() OVER (
      PARTITION BY t1.market, t1.PACK_CODE, t1.PROD_CODE
      ORDER BY kc.no1 DESC, kc.no DESC
    ) AS _rn
  FROM DM.DM_TD_EXT_CHC_PACK2MARKET t1
  LEFT JOIN kc
    ON  (kc.ATC1_Code         IS NULL OR t1.ATC1_CODE = kc.ATC1_Code)
    AND (kc.ATC2_Code         IS NULL OR t1.ATC2_CODE = kc.ATC2_Code)
    AND (kc.ATC3_Code         IS NULL OR t1.ATC3_CODE = kc.ATC3_Code)
    AND (kc.ATC4_Code         IS NULL OR t1.ATC4_CODE = kc.ATC4_Code)
    AND (kc.NFC1_Code         IS NULL OR t1.APP1_CODE = kc.NFC1_Code)
    AND (kc.NFC2_Code         IS NULL OR t1.APP2_CODE = kc.NFC2_Code)
    AND (kc.NFC3_Code         IS NULL OR t1.APP3_CODE = kc.NFC3_Code)
    AND (kc.corporation_code  IS NULL OR t1.CORP_CODE = kc.corporation_code)
    AND (kc.Manufacturer_Code IS NULL OR t1.MANU_CODE = kc.Manufacturer_Code)
    AND (kc.Product_Code      IS NULL OR t1.PROD_CODE = kc.Product_Code)
    AND (kc.Pack_Code         IS NULL OR t1.PACK_CODE = kc.Pack_Code)
    AND (kc.Strength          IS NULL OR t1.STGH_DESC = kc.Strength)
    AND (kc.Molecule_Code     IS NULL OR t1.CMPS_CODE = kc.Molecule_Code)
    AND (kc.market            IS NULL OR t1.market   = kc.market)
)
SELECT distinct Key_Competitor,market,pack_code 
FROM joined
WHERE _rn = 1;

-- COMMAND ----------

INSERT OVERWRITE TABLE DM.DM_TD_EXT_CHC_PACK2MARKET_CLASS
SELECT 
  /*+ BROADCAST(T2) */
  DISTINCT
  T1.MARKET,
  T1.PACK_CODE,
  COALESCE( T2.CLASS, 'OTHERS') AS CLASS
FROM DM.DM_TD_EXT_CHC_PACK2MARKET T1
LEFT JOIN DWD.DWD_GND_TBLCLASS T2
  ON  (T2.MARKET            IS NULL OR T1.MARKET    = T2.MARKET)
  AND (T2.PACK_CODE         IS NULL OR T1.PACK_CODE = T2.PACK_CODE)
  AND (T2.PRODUCT_CODE      IS NULL OR T1.PROD_CODE = T2.PRODUCT_CODE)
  AND (T2.MOLECULE_CODE     IS NULL OR T1.CMPS_CODE = T2.MOLECULE_CODE)
  AND (T2.CORPORATION_CODE  IS NULL OR T1.CORP_CODE = T2.CORPORATION_CODE)
  AND (T2.MANUFACTURER_CODE IS NULL OR T1.MANU_CODE = T2.MANUFACTURER_CODE)
  AND (T2.ATC1_CODE         IS NULL OR T1.ATC1_CODE = T2.ATC1_CODE)
  AND (T2.ATC2_CODE         IS NULL OR T1.ATC2_CODE = T2.ATC2_CODE)
  AND (T2.ATC3_CODE         IS NULL OR T1.ATC3_CODE = T2.ATC3_CODE)
  AND (T2.ATC4_CODE         IS NULL OR T1.ATC4_CODE = T2.ATC4_CODE)
  AND (T2.NFC1_CODE         IS NULL OR T1.APP1_CODE = T2.NFC1_CODE)
  AND (T2.NFC2_CODE         IS NULL OR T1.APP2_CODE = T2.NFC2_CODE)
  AND (T2.NFC3_CODE         IS NULL OR T1.APP3_CODE = T2.NFC3_CODE)
  AND (T2.STRENGTH          IS NULL OR T1.STGH_DESC = T2.STRENGTH)