Agent/examples/demand/test.py

from odps import ODPS
from odps.errors import ODPSError


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_data():
    sql = '''
    SELECT  type
                    ,channel
                    ,vid
                    ,cate1
                    ,cate2
                    ,title
                    ,url1
                    ,url2
                    ,SUBSTR(decode_result,1,20) AS decode_result
                    ,is_new
                    ,extend
                    ,dt
            FROM    loghubods.dwd_topic_decode_result_di
            WHERE   dt = '20260415'
            limit 1;
    '''

    data = get_odps_data(sql)
    if data:
        for r in data:
            print(r[0],r[1],r[2],r[3],r[4],r[5],r[6],r[7],r[8],r[9],r[10],r[11])

if __name__ == '__main__':
    get_data()