第一次提交

config/config.ini

@@ -0,0 +1,8 @@
+[996]
+access_key_id=1
+access_key_secret=1
+endpoint=dataworks.cn-hangzhou.aliyuncs.com
+[xianxia]
+access_key_id=LTAI5t8Uv8MUS2Kn5Yn5scUK
+access_key_secret=RB17dFF6ITVAXt5OqG1WFmeZNj1xgj
+endpoint=dataworks.cn-hangzhou.aliyuncs.com

docker-compose.yaml

@@ -0,0 +1,13 @@
+version: '3.4'
+services:
+    sql-runner:
+        build:
+            context: .
+            dockerfile: Dockerfile
+        restart: always
+        container_name: aliapi
+        image: aliapi:latest
+
+
+
+# docker-compose up --build

dockerfile

@@ -0,0 +1,20 @@
+# 使用阿里云的 Python 3.11 镜像
+FROM registry.cn-hangzhou.aliyuncs.com/yinzhou_docker_hub/python:3.11-alpine
+
+# 设置工作目录
+WORKDIR /opt/aliapi
+
+# 设置时区为 Asia/Shanghai
+ENV TZ=Asia/Shanghai
+
+# 将 requirements.txt 文件复制到容器中
+COPY requirements.txt .
+
+# 安装依赖
+RUN pip install --no-cache-dir -r requirements.txt -i https://pypi.tuna.tsinghua.edu.cn/simple
+
+# 将其他文件复制到容器中
+COPY . .
+
+# 运行应用程序
+ENTRYPOINT ["python3", "run.py"]

run.py

@@ -0,0 +1,21 @@
+from utils.ali_client import create_client,get_listtask,get_task_detail
+import json
+from utils.log import Log
+
+
+if __name__ == '__main__':
+    # 创建一个新的Log实例，确保每天创建一个新的日志文件
+    log = Log().getlog()
+
+    aliclient=create_client('xianxia');
+
+    # response_dict=get_listtask(aliclient,39023);
+    # response = json.loads(response_dict)
+    # print(response_dict)
+    # print('1111111111111111111111')
+    # print(response['body'])
+    # print(response['body']['PagingInfo']['TotalCount'])
+
+    # response_dict = get_task_detail(aliclient, 1023194720,'Dev');
+    response_dict = get_task_detail(aliclient, 1023194720,'Prod');
+    log.info(response_dict)

utils/ali_client.py

@@ -0,0 +1,62 @@
+from alibabacloud_tea_openapi import models as open_api_models
+from alibabacloud_dataworks_public20240518.client import Client as dataworks_public20240518Client
+from utils.loadconfig import get_ali_config
+from alibabacloud_dataworks_public20240518 import models as dataworks_public_20240518_models
+from alibabacloud_tea_util import models as util_models
+import json
+from alibabacloud_tea_util.client import Client as UtilClient
+from utils.log import Log
+
+
+def create_client(env):
+    # 创建一个新的Log实例，确保每天创建一个新的日志文件
+    log = Log().getlog()
+    db_config = get_ali_config(env)
+    log.info(db_config)
+    config = open_api_models.Config(
+        # 必填，请确保代码运行环境设置了环境变量 ALIBABA_CLOUD_ACCESS_KEY_ID。,
+        access_key_id=db_config.get('access_key_id'),
+        access_key_secret=db_config.get('access_key_secret')
+        # 必填，请确保代码运行环境设置了环境变量 ALIBABA_CLOUD_ACCESS_KEY_SECRET。,
+    )
+    # Endpoint 请参考 https://api.aliyun.com/product/dataworks-public
+    config.endpoint =db_config.get('endpoint')
+    return dataworks_public20240518Client(config)
+
+
+# 获取任务列表
+def get_listtask(client,project_id):
+    list_tasks_request = dataworks_public_20240518_models.ListTasksRequest(
+        project_id=project_id
+    )
+    runtime = util_models.RuntimeOptions()
+    try:
+        # 复制代码运行请自行打印 API 的返回值
+        response = client.list_tasks_with_options(list_tasks_request, runtime)
+        # 打印 API 的返回值
+        return json.dumps(response.to_map(), indent=2, ensure_ascii=False)
+    except Exception as error:
+        # 此处仅做打印展示，请谨慎对待异常处理，在工程项目中切勿直接忽略异常。
+        # 错误 message
+        print(error.message)
+        # 诊断地址
+        print(error.data.get("Recommend"))
+        UtilClient.assert_as_string(error.message)
+
+def get_task_detail(client,task_id,project_env):
+    get_task_request = dataworks_public_20240518_models.GetTaskRequest(
+        id=1023194720,
+        project_env=project_env
+    )
+    runtime = util_models.RuntimeOptions()
+    try:
+        # 调用 API 并获取返回值
+        response = client.get_task_with_options(get_task_request, runtime)
+        # 打印 API 的返回值
+        return json.dumps(response.to_map(), indent=2, ensure_ascii=False)
+    except Exception as error:
+        print(f"Error occurred: {error.message}")
+        if hasattr(error, 'data') and error.data:
+            print(f"Recommend: {error.data.get('Recommend')}")
+
+

utils/date_time.py

@@ -0,0 +1,53 @@
+#!/usr/bin/env python
+# -*- coding: utf-8 -*-
+
+import datetime
+import time
+
+
+def str_time():
+    return time.strftime('%Y-%m-%d %H:%M:%S', time.localtime())
+
+
+def bizdate(day):
+    # 获取当前日期
+    current_date = datetime.date.today()
+    # 计算前一天的日期
+    previous_date = current_date - datetime.timedelta(days=day)
+    # 格式化日期输出
+    return previous_date.strftime('%Y%m%d')
+
+
+def t_month(day):
+    # 获取当前日期
+    current_date = datetime.date.today()
+    # 计算前一天的日期
+    previous_date = current_date - datetime.timedelta(days=day)
+    # 格式化日期输出
+    return previous_date.strftime('%Y%m00')
+
+
+def get_ltv_day():
+    day=(1, 2, 3, 4, 5, 6, 7, 14, 30, 60, 90, 150, 300)
+    return ','.join(["'"+bizdate(i)+"'" for i in day])
+
+def get_parameter():
+    return {
+        '${bizdate}': bizdate(1),
+        '${t_month}': t_month(1),
+        '${intra_day}': bizdate(0),
+        '${30_days_later}': bizdate(30),
+        '${300_days_later}': bizdate(301),
+        '${ltv_day}': get_ltv_day(),
+    }
+
+
+if __name__ == "__main__":
+    # hhh = {
+    #     '${bizdate}': bizdate(1),
+    #     '${t_month}': t_month(1),
+    #     '${intra_day}': bizdate(0),
+    #     '${30_days_later}': bizdate(30),
+    #     '${300_days_later}': bizdate(301)
+    # }
+    print(get_ltv_day())

utils/loadconfig.py

@@ -0,0 +1,36 @@
+from configparser import ConfigParser
+import os
+
+
+ENVIRONMENT_VARIABLE = os.getenv("ENVIRONMENT_VARIABLE", 996)
+def get_path():
+    current_directory = os.path.dirname(os.path.abspath(__file__))
+    root_path = os.path.abspath(os.path.dirname(current_directory) + os.path.sep + ".")
+    return root_path
+
+
+def get_pg_config():
+    conf = ConfigParser()
+    conf.read(get_path() + '/config/config.ini')
+    return {
+        "database": conf[ENVIRONMENT_VARIABLE]['database'],
+        "user": conf[ENVIRONMENT_VARIABLE]['user'],
+        "password": conf[ENVIRONMENT_VARIABLE]['password'],
+        "host": conf[ENVIRONMENT_VARIABLE]['host'],
+        "port": conf[ENVIRONMENT_VARIABLE]['port'],
+        # 设置默认超时时间statement_timeout=五分钟
+        "options":"-c statement_timeout=500000 -c idle_in_transaction_session_timeout=600000"
+    }
+
+def get_ali_config(ENVIRONMENT_VARIABLE):
+
+    conf = ConfigParser()
+    conf.read(get_path() + '/config/config.ini')
+    return {
+        "access_key_id": conf[ENVIRONMENT_VARIABLE]['access_key_id'],
+        "access_key_secret": conf[ENVIRONMENT_VARIABLE]['access_key_secret'],
+        "endpoint": conf[ENVIRONMENT_VARIABLE]['endpoint']
+    }
+
+if __name__ == '__main__':
+    print(get_ali_config())

utils/log.py

@@ -0,0 +1,58 @@
+import logging
+import os
+from datetime import datetime
+
+# 定义全局变量 log_path
+cur_path = os.path.dirname(os.path.realpath(__file__))
+log_path = os.path.join(os.path.dirname(cur_path), 'logs')
+
+
+class Log():
+    def __init__(self, logger_name='my_logger'):
+        self.logger = logging.getLogger(logger_name)
+        if self.logger.hasHandlers():
+            self.logger.handlers.clear()
+        self.logger.setLevel(logging.INFO)
+
+        if not os.path.exists(log_path):
+            os.makedirs(log_path)
+
+        self.update_log_file()
+
+    def update_log_file(self):
+        current_date = datetime.now().strftime("%Y_%m_%d")
+        self.log_name = os.path.join(log_path, f'{current_date}.log')
+
+        for handler in self.logger.handlers[:]:
+            self.logger.removeHandler(handler)
+
+        fh = logging.FileHandler(self.log_name, 'a', encoding='utf-8')
+        fh.setLevel(logging.INFO)
+
+        ch = logging.StreamHandler()
+        ch.setLevel(logging.INFO)
+
+        formatter = logging.Formatter('[%(asctime)s] %(filename)s line:%(lineno)d [%(levelname)s]%(message)s')
+        fh.setFormatter(formatter)
+        ch.setFormatter(formatter)
+
+        self.logger.addHandler(fh)
+        self.logger.addHandler(ch)
+
+    def getlog(self):
+        today = datetime.now().strftime("%Y_%m_%d")
+        log_date = os.path.basename(self.log_name).split('.')[0]
+        if today != log_date:
+            self.update_log_file()
+        return self.logger
+
+    def info(self, msg, *args, **kwargs):
+        logger = self.getlog()
+        logger.info(msg, *args, **kwargs)
+
+
+if __name__ == "__main__":
+    log = Log().getlog()
+    log.info("---测试开始----")
+    log.error("操作步骤1,2,3")
+    log.warning("----测试结束----")