Agent/examples/demand/data_query_tools.py

import hashlib
from zoneinfo import ZoneInfo

from odps import ODPS
from odps.errors import ODPSError
from datetime import date, datetime, timedelta
import json
from pathlib import Path

from examples.demand.mysql import mysql_db


def get_odps_data(sql):
    # 配置信息
    access_id = 'LTAI9EBa0bd5PrDa'
    access_key = 'vAalxds7YxhfOA2yVv8GziCg3Y87v5'
    project = 'loghubods'
    endpoint = 'http://service.odps.aliyun.com/api'

    # 1. 初始化 ODPS 入口
    o = ODPS(access_id, access_key, project, endpoint=endpoint)

    try:
        # 2. 执行 SQL 并获取结果
        # execute_sql 会等待任务完成，使用 open_reader 读取数据
        with o.execute_sql(sql).open_reader() as reader:
            # reader 类似于 Java 中的 List<Record>
            # 我们可以直接将其转换为 Python 的 list
            records = [record for record in reader]
            return records

    except ODPSError as e:
        print(f"ODPS 错误: {e}")
        return None


def execute_odps_sql(sql) -> bool:
    # 配置信息
    access_id = 'LTAI9EBa0bd5PrDa'
    access_key = 'vAalxds7YxhfOA2yVv8GziCg3Y87v5'
    project = 'loghubods'
    endpoint = 'http://service.odps.aliyun.com/api'

    o = ODPS(access_id, access_key, project, endpoint=endpoint)
    try:
        instance = o.execute_sql(sql)
        instance.wait_for_success()
        return True
    except ODPSError as e:
        print(f"ODPS 错误: {e}")
        return False


_STRATEGY_GAP = "当下供需gap"
_STRATEGY_GAP_FENCI = "当下供需gap-分词"
_STRATEGY_GAP_ZAOAN = "当下供需gap-早安"
_ZAOAN_BARE_DEMAND_NAMES = {"早安", "晨安", "早上好", "清晨", "晨起"}
_HIVE_TABLE = "loghubods.dwd_multi_demand_pool_di"
_HIVE_DT_FMT = "%Y%m%d"  # 分区格式：yyyymmdd，如 20260519
_CHINA_TZ = ZoneInfo("Asia/Shanghai")


def _hive_partition_dt() -> str:
    """中国时区（Asia/Shanghai）当天日期，格式 yyyymmdd。"""
    return datetime.now(_CHINA_TZ).date().strftime(_HIVE_DT_FMT)


def _hive_yesterday_partition_dt() -> str:
    """中国时区（Asia/Shanghai）昨天日期，格式 yyyymmdd。"""
    return (datetime.now(_CHINA_TZ).date() - timedelta(days=1)).strftime(_HIVE_DT_FMT)


def _fenci_demand_filter_key(name: str, merge_leve2: str, type_str: str) -> tuple[str, str, str]:
    """当下供需gap-分词策略的去重键（不含 dt，用于跨天过滤）。"""
    return (name.strip(), merge_leve2.strip(), type_str.strip())


def _get_demand_filter_words(bizdate: str | None = None) -> list[str]:
    """
    从 ODPS demand_filter_word 读取当天过滤词。
    dt 为中国时区当天（yyyymmdd）；查询失败时返回空列表（不过滤）。
    """
    dt = bizdate or _hive_partition_dt()
    sql = f"""
SELECT  filter_word
FROM    demand_filter_word
WHERE   dt = '{dt}'
"""
    records = get_odps_data(sql)
    if not records:
        return []
    words: list[str] = []
    seen: set[str] = set()
    for record in records:
        word = str(record[0] or "").strip()
        if not word or word in seen:
            continue
        seen.add(word)
        words.append(word)
    return words


def _text_contains_filter_word(text: str, filter_words: list[str]) -> bool:
    """若 text 包含任一过滤词（等价 LIKE '%word%'）返回 True。"""
    if not text or not filter_words:
        return False
    return any(word in text for word in filter_words if word)


def _get_yesterday_fenci_demand_keys() -> set[tuple[str, str, str]]:
    """
    从 MySQL demand_content 读取昨天产生的全部需求（含未写入 ODPS 的），
    用于「当下供需gap-分词」策略的跨天去重。
    """
    yesterday = _hive_yesterday_partition_dt()
    rows = mysql_db.select(
        "demand_content",
        columns="merge_leve2, name, ext_data",
        where="dt = %s",
        where_params=(yesterday,),
    )
    keys: set[tuple[str, str, str]] = set()
    for row in rows:
        name = str(row.get("name") or "").strip()
        merge_leve2 = str(row.get("merge_leve2") or "").strip()
        if not name or not merge_leve2:
            continue
        ext_data = _parse_ext_data(row.get("ext_data"))
        type_str = str(ext_data.get("type") or "").strip()
        keys.add(_fenci_demand_filter_key(name, merge_leve2, type_str))
    return keys


def _escape_odps_string(value: object) -> str:
    return str(value).replace("'", "''")


def _format_odps_string_array(values: list) -> str:
    if not values:
        return "ARRAY()"
    parts = [f"'{_escape_odps_string(v)}'" for v in values]
    return f"ARRAY({','.join(parts)})"


def _parse_ext_data(ext_data_raw: object) -> dict:
    if isinstance(ext_data_raw, dict):
        return ext_data_raw
    if isinstance(ext_data_raw, str) and ext_data_raw.strip():
        try:
            return json.loads(ext_data_raw)
        except json.JSONDecodeError:
            return {}
    return {}


def _build_hive_select_part(
        strategy: str,
        demand_id: str,
        demand_name: str,
        weight: float,
        type_str: str,
        video_count: int,
        video_ids: list[str],
        extend_json: str,
) -> str:
    return (
        "SELECT "
        f"'{_escape_odps_string(strategy)}' AS strategy, "
        f"'{_escape_odps_string(demand_id)}' AS demand_id, "
        f"'{_escape_odps_string(demand_name)}' AS demand_name, "
        f"{weight} AS weight, "
        f"'{_escape_odps_string(type_str)}' AS `type`, "
        f"{video_count} AS video_count, "
        f"{_format_odps_string_array(video_ids)} AS video_list, "
        f"'{_escape_odps_string(extend_json)}' AS extend"
    )


def _insert_hive_select_parts(select_parts: list[str], partition_dt: str) -> bool:
    if not select_parts:
        return True
    union_sql = "\nUNION ALL\n".join(select_parts)
    insert_sql = f"""
INSERT INTO TABLE {_HIVE_TABLE}
PARTITION (dt='{partition_dt}')
(strategy, demand_id, demand_name, weight, `type`, video_count, video_list, extend)
{union_sql}
"""
    return execute_odps_sql(insert_sql)


def write_dwd_multi_demand_pool_di_to_hive(rows: list[dict]) -> int:
    """
    将行数据映射并写入 loghubods.dwd_multi_demand_pool_di（尽力插入，不校验结果）。

    分区与 demand_id 的日期均为中国时区当天（yyyymmdd），不使用行内 dt 字段。
    写入前从 demand_filter_word（dt=当天）拉取过滤词，demand_name 包含任一过滤词则跳过。
    执行两次 INSERT（同表、同分区），策略不同：
    1) 当下供需gap: demand_name=merge_leve2+' '+name, demand_id=md5(strategy+demand_name+type+dt)
    2) 当下供需gap-分词: demand_name=name, demand_id=md5(strategy+name+品类+type+dt)
       - 写入前过滤掉昨天 MySQL demand_content 中已产生的全部需求（按 name+品类+type 去重）
    """
    if not rows:
        return 0

    china_today = _hive_partition_dt()
    filter_words = _get_demand_filter_words(china_today)
    yesterday_keys = _get_yesterday_fenci_demand_keys()
    gap_parts: list[str] = []
    fenci_parts: list[str] = []
    fenci_skipped = 0
    filter_word_skipped = 0

    print(
        f"[hive] 过滤词加载: dt={china_today}, count={len(filter_words)}"
    )

    for row in rows:
        merge_leve2 = str(row.get("merge_leve2") or "").strip()
        name = str(row.get("name") or "").strip()
        if not merge_leve2 or not name:
            continue

        weight = round(float(row.get("score") or 0.0), 6)

        ext_data = _parse_ext_data(row.get("ext_data"))
        type_str = str(ext_data.get("type") or "").strip()
        video_ids = ext_data.get("video_ids") or []
        if not isinstance(video_ids, list):
            video_ids = []
        video_ids = [str(v).strip() for v in video_ids if v is not None and str(v).strip()]
        video_count = len(video_ids)
        extend_json = json.dumps({"品类": merge_leve2}, ensure_ascii=False)

        demand_name_gap = f"{merge_leve2} {name}"
        # 等价 SQL LIKE '%filter_word%'：demand_name 包含任一过滤词则不写入 ODPS
        if _text_contains_filter_word(demand_name_gap, filter_words):
            filter_word_skipped += 1
            continue

        demand_id_gap = hashlib.md5(
            f"{_STRATEGY_GAP}{demand_name_gap}{type_str}{china_today}".encode("utf-8")
        ).hexdigest()
        gap_parts.append(
            _build_hive_select_part(
                _STRATEGY_GAP, demand_id_gap, demand_name_gap,
                weight, type_str, video_count, video_ids, extend_json,
            )
        )

        filter_key = _fenci_demand_filter_key(name, merge_leve2, type_str)
        if filter_key in yesterday_keys:
            fenci_skipped += 1
            continue

        demand_id_fenci = hashlib.md5(
            f"{_STRATEGY_GAP_FENCI}{name}{merge_leve2}{type_str}{china_today}".encode("utf-8")
        ).hexdigest()
        fenci_parts.append(
            _build_hive_select_part(
                _STRATEGY_GAP_FENCI, demand_id_fenci, name,
                weight, type_str, video_count, video_ids, extend_json,
            )
        )

    print(
        f"[hive] 过滤词跳过: skipped={filter_word_skipped}, "
        f"gap_to_write={len(gap_parts)}, fenci_to_write={len(fenci_parts)}"
    )

    if not gap_parts:
        return 0

    print(
        f"[hive] {_STRATEGY_GAP_FENCI} 过滤昨天需求: "
        f"yesterday_dt={_hive_yesterday_partition_dt()}, "
        f"yesterday_total={len(yesterday_keys)}, skipped={fenci_skipped}, "
        f"to_write={len(fenci_parts)}"
    )

    _insert_hive_select_parts(gap_parts, china_today)
    if fenci_parts:
        _insert_hive_select_parts(fenci_parts, china_today)
    return len(gap_parts) + len(fenci_parts)


def write_dwd_zaoan_demand_pool_to_hive(rows: list[dict]) -> int:
    """
    写入策略「当下供需gap-早安」到 loghubods.dwd_multi_demand_pool_di。
    固定早安词（早安/晨安/早上好/清晨/晨起）demand_name 只用词本身，不拼品类。
    其余 demand_name = merge_leve2 + ' ' + name。
    demand_id = md5(strategy + demand_name + type + dt)
    该策略已由上游筛过，不再套用过滤词 / 昨天分词去重。
    """
    if not rows:
        return 0

    china_today = _hive_partition_dt()
    select_parts: list[str] = []

    for row in rows:
        merge_leve2 = str(row.get("merge_leve2") or "").strip()
        name = str(row.get("name") or "").strip()
        if not merge_leve2 or not name:
            continue

        weight = round(float(row.get("score") or 0.0), 6)
        ext_data = _parse_ext_data(row.get("ext_data"))
        type_str = str(ext_data.get("type") or "").strip()
        video_ids = ext_data.get("video_ids") or []
        if not isinstance(video_ids, list):
            video_ids = []
        video_ids = [str(v).strip() for v in video_ids if v is not None and str(v).strip()]
        video_count = len(video_ids)
        extend_json = json.dumps({"品类": merge_leve2}, ensure_ascii=False)

        demand_name = name if name in _ZAOAN_BARE_DEMAND_NAMES else f"{merge_leve2} {name}"
        demand_id = hashlib.md5(
            f"{_STRATEGY_GAP_ZAOAN}{demand_name}{type_str}{china_today}".encode("utf-8")
        ).hexdigest()
        select_parts.append(
            _build_hive_select_part(
                _STRATEGY_GAP_ZAOAN, demand_id, demand_name,
                weight, type_str, video_count, video_ids, extend_json,
            )
        )

    if not select_parts:
        print(f"[hive] {_STRATEGY_GAP_ZAOAN} 无有效行，跳过")
        return 0

    print(f"[hive] {_STRATEGY_GAP_ZAOAN} 写入: dt={china_today}, rows={len(select_parts)}")
    ok = _insert_hive_select_parts(select_parts, china_today)
    if not ok:
        raise RuntimeError(f"写入 Hive 失败: strategy={_STRATEGY_GAP_ZAOAN}, dt={china_today}")
    return len(select_parts)


def write_feature_point_data_to_hive(names: list[str]) -> int:
    """
    将需求名称写入 Hive 表 feature_point_data（按北京时间当天分区）。
    仅写入以下字段：
    - 特征点
    - 总分发曝光pv（固定 5000）
    - 质bn_rovn（固定 0.1）
    """
    normalized_names = [str(name).strip() for name in names if name is not None and str(name).strip()]
    if not normalized_names:
        return 0

    dt = datetime.now(ZoneInfo("Asia/Shanghai")).strftime("%Y%m%d")
    select_parts = []
    for name in normalized_names:
        safe_name = name.replace("'", "''")
        select_parts.append(
            "SELECT "
            f"'{safe_name}' AS `特征点`, "
            "5000 AS `总分发曝光pv`, "
            "0.1 AS `质bn_rovn`"
        )

    union_sql = "\nUNION ALL\n".join(select_parts)
    insert_sql = f"""
INSERT INTO TABLE feature_point_data
PARTITION (dt='{dt}')
(`特征点`, `总分发曝光pv`, `质bn_rovn`)
{union_sql}
"""
    ok = execute_odps_sql(insert_sql)
    if not ok:
        return 0
    return len(normalized_names)


def get_demand_merge_level2_names():
    date_time = datetime.now(ZoneInfo("Asia/Shanghai")).date() - timedelta(days=1)
    day = date_time.strftime("%Y%m%d")
    count = 50
    sql_query = f'''
select *
from (
select
dt,
merge二级品类,
sum(当日分发曝光pv) as 分发曝光pv,
sum(累计分享回流uv) AS bn_总回流,
sum(当日分发回流uv)/(sum(当日分发曝光pv)+100) as 质bn_rovn,

case when sum(当日分发曝光pv)>=10000 then
    case when sum(当日分发回流uv)/(sum(当日分发曝光pv)+100)<0.035
        then -1*(count(DISTINCT 视频id)/avg(总日分发视频数))/((sum(累计分享回流uv)/avg(总日回流uv)))
        else 10*(sum(累计分享回流uv)/avg(总日回流uv)*sum(当日分发回流uv)/(sum(当日分发曝光pv)+100))/(count(DISTINCT 视频id)/avg(总日分发视频数))
    end
else 0 end AS 总供需分,

case when sum(当日分发曝光pv)>=10000 then
    case when sum(当日分发回流uv)/(sum(当日分发曝光pv)+100)<0.035
        then -1*(COUNT(DISTINCT CASE WHEN 推荐天数间隔<3 THEN 视频id END ) /avg(总日分发视频数))/(sum(累计分享回流uv)/avg(总日回流uv))
        else 10*(sum(累计分享回流uv)/avg(总日回流uv)*sum(当日分发回流uv)/(sum(当日分发曝光pv)+1000))/(COUNT(DISTINCT CASE WHEN 推荐天数间隔<3 THEN 视频id END ) /avg(总日分发视频数))
    end
else 0 end AS 新供需分,

count(DISTINCT 视频id) as 分发视频量,
count(DISTINCT if(推荐天数间隔<3,视频id,null)) as 3日新推荐视频量,

case when sum(当日分发曝光pv)>=10000 and sum(当日分发回流uv)/(sum(当日分发曝光pv)+100)>0.035
then (avg(总日分发视频数)*(10*(sum(当日分发回流uv)/(sum(当日分发曝光pv)+100))*(sum(累计分享回流uv)/avg(总日回流uv) ))/0.5-count(DISTINCT 视频id))/3
end as 缺量,

case when sum(当日分发曝光pv)>=10000 and sum(当日分发回流uv)/(sum(当日分发曝光pv)+100)<=0.035
then (avg(总日分发视频数)*(10*(sum(当日分发回流uv)/(sum(当日分发曝光pv)+100))*(sum(累计分享回流uv)/avg(总日回流uv) ))/(2)-count(DISTINCT 视频id))/3
end as 控量,

avg(总日回流uv) AS 总日回流uv,
avg(总日分发视频数) AS 总日分发视频数,
avg(总日推荐视频数) AS 总日推荐视频数,

COUNT(DISTINCT CASE WHEN 总回流uv>0 THEN 视频id END )/avg(总日分发视频数) AS 回流视频个数占比,
sum(当日分发回流uv) AS bn_当日分发回流,
sum(当日分发回流uv)/avg(总日回流uv) AS 分发拉回回流uv占比,
sum(累计分享回流uv)/avg(总日回流uv) AS 回流uv占比,
count(DISTINCT 视频id)/avg(总日分发视频数) AS 分发视频量占比,
COUNT(DISTINCT CASE WHEN 是否当日新推荐=1 THEN 视频id END ) /avg(总日分发视频数) AS 新推荐视频量占比

from loghubods.video_dimension_detail_add_column
where dt = '{day}'
group by dt, merge二级品类
) t1 
where t1.缺量>= {count}

'''

    data = get_odps_data(sql_query)
    result_list = []
    if data:
        for r in data:
            lack_count = r[9]
            if lack_count > 1000:
                count = 70
            elif 500 < lack_count <= 1000:
                count = 60
            elif 100 < lack_count <= 500:
                count = 40
            elif 50 < lack_count <= 100:
                count = 20
            else:
                count = 10
            if count == 0:
                continue
            result_list.append({
                "cluster_name": r[1],
                "platform_type": "piaoquan",
                "count": count,
            })
    return result_list


def get_rov_by_merge_leve2_and_video_ids(merge_leve2, video_ids):
    merge_level_in_clause = f"'{merge_leve2}'"
    video_ids_in_clause = ", ".join([f"'{video_id}'" for video_id in video_ids])
    end_date = (date.today() - timedelta(days=1)).strftime("%Y%m%d")
    start_date = (date.today() - timedelta(days=14)).strftime("%Y%m%d")
#     sql_query = f'''
# SELECT
#     v.videoid,
#     CASE
#         WHEN COALESCE(SUM(COALESCE(t3.`当日分发曝光pv`, 0)), 0) < 1000 THEN 0
#         ELSE COALESCE(AVG(NULLIF(t3.rov_t0, 0)), 0)
#     END AS avg_rov_t0
# FROM
# (
#     SELECT
#         t2.videoid,
#         t2.merge_leve2
#     FROM videoods.content_profile t1
#     JOIN loghubods.video_merge_tag t2
#         ON t1.content_id = t2.videoid
#     WHERE
#         t1.status = 3
#         AND t1.is_deleted = 0
#         AND t2.merge_leve2 IN ({merge_level_in_clause})
# ) v
# LEFT JOIN loghubods.video_dimension_detail_add_column t3
#     ON v.videoid = t3.视频id
#     AND t3.dt >= '{start_date}'
#     AND t3.dt <= '{end_date}'
# WHERE v.videoid in ({video_ids_in_clause})
# GROUP BY
#     v.videoid
# ;
#     '''
    sql_query = f'''
    SELECT
        CAST(t3.视频id AS STRING) AS 视频id_str,
        CASE
            WHEN COALESCE(SUM(COALESCE(t3.`当日分发曝光pv`, 0)), 0) < 1000 THEN 0
            ELSE COALESCE(AVG(NULLIF(t3.rov_t0, 0)), 0)
        END AS avg_rov_t0
    FROM
    loghubods.video_dimension_detail_add_column t3

    WHERE t3.视频id in ({video_ids_in_clause})
    AND t3.dt >= '{start_date}'
    AND t3.dt <= '{end_date}'
    GROUP BY
        t3.视频id
    ;
        '''
    data = get_odps_data(sql_query)
    result_dict = {}
    if data:
        result_dict = {r[0]: r[1] for r in data}
    return result_dict


def get_rov_by_tree_and_video_ids(video_ids):
    video_ids_in_clause = ", ".join([f"'{video_id}'" for video_id in video_ids])
    last_year_today = date.today() - timedelta(days=365)
    start_date = last_year_today.strftime("%Y%m%d")
    end_date = (last_year_today + timedelta(days=7)).strftime("%Y%m%d")
    sql_query = f'''
SELECT
    CAST(t3.视频id AS STRING) AS 视频id_str,
    CASE
        WHEN COALESCE(SUM(COALESCE(t3.`当日分发曝光pv`, 0)), 0) < 1000 THEN 0
        ELSE COALESCE(AVG(NULLIF(t3.rov_t0, 0)), 0)
    END AS avg_rov_t0
FROM
loghubods.video_dimension_detail_add_column t3

WHERE t3.视频id in ({video_ids_in_clause})
AND t3.dt >= '{start_date}'
AND t3.dt <= '{end_date}'
GROUP BY
    t3.视频id
;
    '''
    data = get_odps_data(sql_query)
    result_dict = {}
    if data:
        result_dict = {r[0]: r[1] for r in data}
    return result_dict


def get_changwen_weight(account_name):
    bizdatemax_date = date.today() - timedelta(days=1)
    bizdatemin_date = bizdatemax_date - timedelta(days=30)
    bizdatemax = bizdatemax_date.strftime("%Y%m%d")
    bizdatemin = bizdatemin_date.strftime("%Y%m%d")

    sql_query = f'''
SELECT 
        公众号名
        ,videoid
        ,一级品类
        ,二级品类
        ,头部曝光
        ,头部曝光uv
        ,头部realplay
        ,头部realplay_uv
        ,头部分享
        ,头部分享uv
        ,头部回流人数 AS 头部回流数
        ,推荐曝光数
        ,当日分发曝光uv
        ,推荐realplay
        ,分发realplay_uv
        ,推荐分享数
        ,当日分发分享uv
        ,推荐回流数
        ,当日回流进入分发曝光次数 AS vov分子
FROM    (
            SELECT  DISTINCT a.公众号名
                    ,a.videoid
                    ,e.merge_leve1 AS 一级品类
                    ,e.merge_leve2 AS 二级品类
                    ,a.title
                    ,a.进入分发人数
                    ,头部曝光pv AS 头部曝光
                    ,头部realplay_pv AS 头部realplay
                    ,头部分享pv AS 头部分享
                    ,a.当日分发曝光pv AS 推荐曝光数
                    ,a.当日分发播放pv
                    ,分发realplay_pv AS 推荐realplay
                    ,分发realplay_pv / a.当日分发播放pv AS 真实播放率pv
                    ,当日分发播放uv
                    ,c.realplay_uv AS 分发真实播uv
                    ,c.realplay_uv / a.当日分发播放uv AS 真实播放率uv
                    ,a.当日分发分享pv AS 推荐分享数
                    ,a.当日分发分享pv / a.当日分发曝光pv AS str
                    ,NVL(b.当日分发回流人数,0) AS 推荐回流数
                    ,NVL(b.当日回流进入分发人数,0) AS 当日回流进入分发人数
                    ,NVL(b.当日回流进入分发曝光次数,0) AS 当日回流进入分发曝光次数
                    ,NVL(b.当日回流进入分发曝光次数,0) / a.当日分发曝光pv AS vov分子
                    ,d.头部回流人数
                    ,当日分发曝光uv
                    ,头部曝光uv
                    ,当日分发分享uv
                    ,头部分享uv
                    ,分发realplay_uv
                    ,头部realplay_uv
            FROM    (
                        SELECT  account_name AS 公众号名
                                ,videoid
                                ,title
                                ,COUNT(DISTINCT mid) AS 进入分发人数
                                ,COUNT(
                                      CASE    WHEN pagesource REGEXP 'category$|recommend$|-pages/user-videos-detail$' AND businesstype = 'videoView' THEN mid END
                                ) AS 当日分发曝光pv
                                ,COUNT(DISTINCT 
                                      CASE    WHEN pagesource REGEXP 'category$|recommend$|-pages/user-videos-detail$' AND businesstype = 'videoView' THEN mid END
                                ) AS 当日分发曝光uv
                                ,COUNT(
                                      CASE    WHEN pagesource REGEXP 'pages/user-videos-share$' AND businesstype = 'videoView' THEN mid END
                                ) AS 头部曝光pv
                                ,COUNT(DISTINCT 
                                      CASE    WHEN pagesource REGEXP 'pages/user-videos-share$' AND businesstype = 'videoView' THEN mid END
                                ) AS 头部曝光uv
                                ,COUNT(
                                      CASE    WHEN pagesource REGEXP 'category$|recommend$|-pages/user-videos-detail$' AND businesstype = 'videoPlay' THEN mid END
                                ) AS 当日分发播放pv
                                ,COUNT(DISTINCT 
                                      CASE    WHEN pagesource REGEXP 'category$|recommend$|-pages/user-videos-detail$' AND businesstype = 'videoPlay' THEN mid END
                                ) AS 当日分发播放uv
                                ,COUNT(
                                      CASE    WHEN pagesource REGEXP 'category$|recommend$|-pages/user-videos-detail$' AND businesstype = 'videoShareFriend' THEN mid END
                                ) AS 当日分发分享pv
                                 ,COUNT(DISTINCT 
                                      CASE    WHEN pagesource REGEXP 'category$|recommend$|-pages/user-videos-detail$' AND businesstype = 'videoShareFriend' THEN mid END
                                ) AS 当日分发分享uv
                                ,COUNT(
                                      CASE    WHEN pagesource REGEXP 'pages/user-videos-share$' AND businesstype = 'videoShareFriend' THEN mid END
                                ) AS 头部分享pv
                                 ,COUNT(DISTINCT 
                                      CASE    WHEN pagesource REGEXP 'pages/user-videos-share$' AND businesstype = 'videoShareFriend' THEN mid END
                                ) AS 头部分享uv
                        FROM    (
                                    SELECT  DISTINCT a.mid
                                            ,a.videoid
                                            ,a.businesstype
                                            ,a.pagesource
                                            ,a.subsessionid
                                            ,account_name
                                            ,e.title
                                    FROM    loghubods.video_action_log_rp a
                                    LEFT JOIN loghubods.user_wechat_identity_info_ha b
                                    ON      a.mid = CONCAT('weixin_openid_',b.open_id)
                                    AND     b.dt = MAX_PT("loghubods.user_wechat_identity_info_ha")
                                    LEFT JOIN loghubods.gzh_fans_info d
                                    ON      b.union_id = d.union_id
                                    AND     d.dt = MAX_PT("loghubods.gzh_fans_info")
                                    LEFT JOIN videoods.wx_video e
                                    ON      a.videoid = e.id
                                    WHERE   a.dt >= '{bizdatemin}'
                                    AND     a.dt <= '{bizdatemax}'
                                    AND     businesstype IN ('videoView','videoPlay','videoShareFriend')
                                    AND     d.user_create_time IS NOT NULL
                                    AND     account_name = '{account_name}'
                                    AND a.videoid IN (
                                        SELECT 
                                        DISTINCT content_id AS videoid
                                        FROM
                                        videoods.content_profile
                                        WHERE status=3
                                        AND is_deleted = 0
                                    
                                    )
                                ) t
                        GROUP BY 公众号名
                                 ,videoid
                                 ,title
                    ) a
            LEFT JOIN   (
                            SELECT  t.account_name AS 公众号名
                                    ,t.videoid
                                    ,COUNT(DISTINCT s.machinecode) AS 当日分发回流人数
                                    ,COUNT(DISTINCT v.mid) AS 当日回流进入分发人数
                                    ,COUNT(v.mid) AS 当日回流进入分发曝光次数
                            FROM    (
                                        SELECT  DISTINCT a.subsessionid
                                                ,a.videoid
                                                ,a.mid
                                                ,d.account_name
                                                ,GET_JSON_OBJECT(extparams,'$.recomTraceId') AS recomtraceid
                                        FROM    loghubods.video_action_log_rp a
                                        LEFT JOIN loghubods.user_wechat_identity_info_ha b
                                        ON      a.mid = CONCAT('weixin_openid_',b.open_id)
                                        AND     b.dt = MAX_PT("loghubods.user_wechat_identity_info_ha")
                                        LEFT JOIN loghubods.gzh_fans_info d
                                        ON      b.union_id = d.union_id
                                        AND     d.dt = MAX_PT("loghubods.gzh_fans_info")
                                        WHERE   a.dt >= '{bizdatemin}'
                                        AND     a.dt <= '{bizdatemax}'
                                        AND     a.businesstype = 'videoShareFriend'
                                        AND     a.pagesource REGEXP 'category$|recommend$|-pages/user-videos-detail$'
                                        AND     d.user_create_time IS NOT NULL
                                        AND     d.account_name = '{account_name}'
                                    ) t
                            LEFT JOIN   (
                                            SELECT  DISTINCT subsessionid
                                                    ,machinecode
                                                    ,recomtraceid
                                                    ,clickobjectid
                                            FROM    loghubods.user_share_log
                                            WHERE   dt >= '{bizdatemin}'
                                            AND     dt <= '{bizdatemax}'
                                            AND     topic = 'click'
                                        ) s
                            ON      t.recomtraceid = s.recomtraceid
                            AND     t.videoid = s.clickobjectid
                            LEFT JOIN   (
                                            SELECT  subsessionid
                                                    ,mid
                                                    ,videoid
                                            FROM    loghubods.video_action_log_rp
                                            WHERE   dt >= '{bizdatemin}'
                                            AND     dt <= '{bizdatemax}'
                                            AND     pagesource REGEXP 'category$|recommend$|-pages/user-videos-detail$'
                                            AND     businesstype = 'videoView'
                                        ) v
                            ON      s.subsessionid = v.subsessionid
                            AND     s.machinecode = v.mid
                            GROUP BY account_name
                                     ,t.videoid
                        ) b
            ON      a.公众号名 = b.公众号名
            AND     a.videoid = b.videoid
            LEFT JOIN   (
                            SELECT  d.account_name AS 公众号名
                                    ,a.videoid
                                    ,COUNT(DISTINCT a.mid) AS realplay_uv
                                    ,COUNT(
                                          CASE    WHEN a.pagesource REGEXP 'category$|recommend$|-pages/user-videos-detail$' THEN a.mid END
                                    ) AS 分发realplay_pv
                                    ,COUNT(CASE    WHEN a.pagesource REGEXP 'pages/user-videos-share$' THEN a.mid END) AS 头部realplay_pv
                                     ,COUNT(DISTINCT 
                                          CASE    WHEN a.pagesource REGEXP 'category$|recommend$|-pages/user-videos-detail$' THEN a.mid END
                                    ) AS 分发realplay_uv
                                    ,COUNT(DISTINCT CASE    WHEN a.pagesource REGEXP 'pages/user-videos-share$' THEN a.mid END) AS 头部realplay_uv
                            FROM    loghubods.ods_video_play_log_day a
                            LEFT JOIN   (
                                            SELECT  DISTINCT open_id
                                                    ,union_id
                                            FROM    loghubods.user_wechat_identity_info_ha
                                            WHERE   dt = MAX_PT("loghubods.user_wechat_identity_info_ha")
                                        ) b
                            ON      a.mid = CONCAT('weixin_openid_',b.open_id)
                            LEFT JOIN loghubods.gzh_fans_info d
                            ON      b.union_id = d.union_id
                            AND     d.dt = MAX_PT("loghubods.gzh_fans_info")
                            WHERE   a.dt >= '{bizdatemin}'
                            AND     a.dt <= '{bizdatemax}'
                            AND     a.businesstype = 'videoRealPlay'
                            AND     d.user_create_time IS NOT NULL
                            AND     d.account_name = '{account_name}'
                            GROUP BY d.account_name
                                     ,a.videoid
                            ORDER BY 分发realplay_pv DESC
                        ) c
            ON      a.公众号名 = c.公众号名
            AND     a.videoid = c.videoid
            LEFT JOIN   (
                            SELECT  t.account_name AS 公众号名
                                    ,t.videoid
                                    ,COUNT(DISTINCT s.machinecode) AS 头部回流人数
                            FROM    (
                                        SELECT  DISTINCT a.shareobjectid AS videoid
                                                ,a.shareid
                                                ,a.machinecode
                                                ,d.account_name
                                        FROM    loghubods.user_share_log a
                                        LEFT JOIN loghubods.user_wechat_identity_info_ha b
                                        ON      a.machinecode = CONCAT('weixin_openid_',b.open_id)
                                        AND     b.dt = MAX_PT("loghubods.user_wechat_identity_info_ha")
                                        LEFT JOIN loghubods.gzh_fans_info d
                                        ON      b.union_id = d.union_id
                                        AND     d.dt = MAX_PT("loghubods.gzh_fans_info")
                                        WHERE   a.dt >= '{bizdatemin}'
                                        AND     a.dt <= '{bizdatemax}'
                                        AND     a.topic = 'share'
                                        AND     a.pagesource REGEXP 'pages/user-videos-share$'
                                        AND     d.user_create_time IS NOT NULL
                                        AND     d.account_name = '{account_name}'
                                    ) t
                            LEFT JOIN   (
                                            SELECT  DISTINCT shareid
                                                    ,machinecode
                                                    ,clickobjectid
                                            FROM    loghubods.user_share_log
                                            WHERE   dt >= '{bizdatemin}'
                                            AND     dt <= '{bizdatemax}'
                                            AND     topic = 'click'
                                        ) s
                            ON      t.shareid = s.shareid
                            GROUP BY account_name
                                     ,t.videoid
                        ) d
            ON      a.公众号名 = d.公众号名
            AND     a.videoid = d.videoid
            LEFT JOIN loghubods.video_merge_tag e
            ON      a.videoid = e.videoid
        ) 
ORDER BY 推荐曝光数 DESC
    '''
    result_list = []
    data = get_odps_data(sql_query)
    if data:
        for r in data:
            result_list.append(
                {
                    "account_name": r[0],
                    "videoid": r[1],
                    "一级品类": r[2],
                    "二级品类": r[3],
                    "ext_data": {
                        "头部曝光": r[4],
                        "头部曝光uv": r[5],
                        "头部realplay": r[6],
                        "头部realplay_uv": r[7],
                        "头部分享": r[8],
                        "头部分享uv": r[9],
                        "头部回流数": r[10],
                        "推荐曝光数": r[11],
                        "当日分发曝光uv": r[12],
                        "推荐realplay": r[13],
                        "分发realplay_uv": r[14],
                        "推荐分享数": r[15],
                        "当日分发分享uv": r[16],
                        "推荐回流数": r[17],
                        "vov分子": r[18],
                    },
                }
            )

    # 输出到 examples/demand/data/changwen_data/
    output_dir = Path(__file__).parent / "data" / "changwen_data"
    output_dir.mkdir(parents=True, exist_ok=True)
    output_file = output_dir / f"{account_name}.json"
    with output_file.open("w", encoding="utf-8") as f:
        json.dump(result_list, f, ensure_ascii=False, indent=2)

    return result_list


def get_zengzhang_weight(account_name):
    bizdatemax_date = date.today() - timedelta(days=1)
    bizdatemin_date = bizdatemax_date - timedelta(days=30)
    bizdatemax = bizdatemax_date.strftime("%Y%m%d")
    bizdatemin = bizdatemin_date.strftime("%Y%m%d")

    sql_query = f'''
SELECT  合作方名
        ,合作方简称
        ,videoid
        ,一级品类
        ,二级品类
        ,SUM(头部曝光) as 头部曝光
        ,SUM(头部曝光uv) as 头部曝光uv
        ,SUM(头部realplay) as  头部realplay
        ,SUM(头部realplay_uv) as 头部realplay_uv
        ,SUM(头部分享) as 头部分享
        ,SUM(头部分享uv) as 头部分享uv
        ,SUM(头部回流数)  as 头部回流数
        ,SUM(推荐曝光数) as 推荐曝光数
        ,SUM(当日分发曝光uv) as 当日分发曝光uv
        ,SUM(推荐realplay) as 推荐realplay
        ,SUM(分发realplay_uv) as 分发realplay_uv
        ,SUM(推荐分享数) as 推荐分享数
        ,SUM(当日分发分享uv) as 当日分发分享uv
        ,SUM(推荐回流数) as 推荐回流数
        ,SUM(vov分子) as vov分子
FROM    loghubods.dws_growth_partner_vid_data
WHERE   dt BETWEEN '{bizdatemin}' AND '{bizdatemax}'
AND     合作方名 = '{account_name}'
GROUP BY 合作方名
         ,合作方简称
         ,videoid
         ,一级品类
         ,二级品类
ORDER BY SUM(推荐曝光数)
;
    '''
    result_list = []
    data = get_odps_data(sql_query)
    if data:
        for r in data:
            result_list.append(
                {
                    "account_name": r[0],
                    "合作方简称": r[1],
                    "videoid": r[2],
                    "一级品类": r[3],
                    "二级品类": r[4],
                    "ext_data": {
                        "头部曝光": r[5],
                        "头部曝光uv": r[6],
                        "头部realplay": r[7],
                        "头部realplay_uv": r[8],
                        "头部分享": r[9],
                        "头部分享uv": r[10],
                        "头部回流数": r[11],
                        "推荐曝光数": r[12],
                        "当日分发曝光uv": r[13],
                        "推荐realplay": r[14],
                        "分发realplay_uv": r[15],
                        "推荐分享数": r[16],
                        "当日分发分享uv": r[17],
                        "推荐回流数": r[18],
                        "vov分子": r[19],
                    },
                }
            )

    # 输出到 examples/demand/data/zengzhang_data/
    output_dir = Path(__file__).parent / "data" / "zengzhang_data"
    output_dir.mkdir(parents=True, exist_ok=True)
    output_file = output_dir / f"{account_name}.json"
    with output_file.open("w", encoding="utf-8") as f:
        json.dump(result_list, f, ensure_ascii=False, indent=2)

    return result_list


def get_merge_leve2_by_video_ids(video_ids, batch_size=2000):
    result = {}
    if not video_ids:
        return result

    normalized_ids = [str(video_id) for video_id in video_ids if video_id is not None]
    for i in range(0, len(normalized_ids), batch_size):
        batch_ids = normalized_ids[i:i + batch_size]
        escaped_ids = [video_id.replace("'", "''") for video_id in batch_ids]
        video_ids_in_clause = ", ".join([f"'{video_id}'" for video_id in escaped_ids])
        sql_query = f'''
            SELECT videoid, merge_leve2
            FROM loghubods.video_merge_tag
            WHERE videoid IN ({video_ids_in_clause})
        '''
        data = get_odps_data(sql_query)
        if not data:
            continue

        for row in data:
            result[str(row[0])] = row[1]

    return result


def get_all_decode_task_result_rows():
    return mysql_db.select(
        "workflow_decode_task_result",
        columns="id, channel_content_id, merge_leve2",
    )


def update_decode_task_result_merge_leve2(channel_content_id, merge_leve2):
    return mysql_db.update(
        "workflow_decode_task_result",
        {"merge_leve2": str(merge_leve2)},
        "channel_content_id = %s",
        (str(channel_content_id),),
    )


def backfill_merge_leve2_for_decode_task_result():
    rows = get_all_decode_task_result_rows()
    updated_count = 0
    skipped_count = 0
    valid_content_ids = []

    for row in rows:
        channel_content_id = row.get("channel_content_id")
        if channel_content_id is None:
            skipped_count += 1
            continue

        channel_content_id = str(channel_content_id)
        if len(channel_content_id) > 8:
            skipped_count += 1
            continue

        valid_content_ids.append(channel_content_id)

    merge_leve2_map = get_merge_leve2_by_video_ids(valid_content_ids, batch_size=2000)

    for channel_content_id in valid_content_ids:
        merge_leve2 = merge_leve2_map.get(channel_content_id)
        if not merge_leve2:
            continue
        affected = update_decode_task_result_merge_leve2(channel_content_id, merge_leve2)
        if affected > 0:
            updated_count += affected

    return {
        "total": len(rows),
        "updated": updated_count,
        "skipped": skipped_count,
    }

#
# if __name__ == '__main__':
#     backfill_merge_leve2_for_decode_task_result()