Agent/examples/piaoquan_demand/data_query_tools.py

from odps import ODPS
from odps.errors import ODPSError
from datetime import date, timedelta


from agent import tool


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 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
;
    '''
    data = get_odps_data(sql_query)
    result_dict = {}
    if data:
        result_dict = {r[0]: r[1] for r in data}
    return result_dict


if __name__ == '__main__':
    videos = ["64429933"]
    print(get_rov_by_merge_leve2_and_video_ids('历史名人', videos))