top_automatic_traceability/common/odps_data.py

import json
import datetime
from odps import ODPS

# ODPS服务配置
ODPS_CONFIG = {
    'ENDPOINT': 'http://service.cn.maxcompute.aliyun.com/api',
    'ACCESSID': 'LTAIWYUujJAm7CbH',
    'ACCESSKEY': 'RfSjdiWwED1sGFlsjXv0DlfTnZTG1P',
    'PROJECT': 'loghubods'
}
class OdpsDataCount:
    @classmethod
    def get_data_count(cls, dt):
        odps = ODPS(
            access_id=ODPS_CONFIG['ACCESSID'],
            secret_access_key=ODPS_CONFIG['ACCESSKEY'],
            project=ODPS_CONFIG['PROJECT'],
            endpoint=ODPS_CONFIG['ENDPOINT']
        )
        data_values = []
        try:
            sql = f'SELECT uid,videoid,return_uv,type,type_owner,channel,channel_owner,title FROM loghubods.all_apptype_top1000_return WHERE dt = "{dt}" and  rank <= 100'
            with odps.execute_sql(sql).open_reader() as reader:
                for row in reader:
                    data_values.append(json.dumps( {"uid": row[0], "videoid": row[1], "return_uv": row[2], "type": row[3], "type_owner": row[4], "channel": row[5], "channel_owner": row[6], "title": row[7], "dt": str(dt)}, ensure_ascii=False ))
        except Exception as e:
            print(f"An error occurred: {e}")
            return data_values
        return data_values

    @classmethod
    def main(cls):
        dt = (datetime.datetime.now() - datetime.timedelta(hours=1)).strftime('%Y%m%d%H')
        data_count = cls.get_data_count(dt= dt)
        print(len(data_count))
        return data_count

if __name__ == '__main__':
    OdpsDataCount.main()