LongArticleTaskServer/applications/api/aliyun_log_api.py

"""
@author: luojunhui
"""

import datetime
import json
import time

from aliyun.log import LogClient, PutLogsRequest, LogItem


def log(task, function, status="success", message=None, data=None):
    """
    @:param task 任务
    @:param
    :return:
    """
    if data is None:
        data = {}
    accessKeyId = "LTAIP6x1l3DXfSxm"
    accessKey = "KbTaM9ars4OX3PMS6Xm7rtxGr1FLon"
    project = "changwen-alg"
    log_store = "long_articles_job"
    endpoint = "cn-hangzhou.log.aliyuncs.com"
    # 创建 LogClient 实例
    client = LogClient(endpoint, accessKeyId, accessKey)
    log_group = []
    log_item = LogItem()
    contents = [
        (f"task", str(task)),
        (f"function", str(function)),
        (f"message", str(message)),
        (f"status", str(status)),
        (f"data", json.dumps(data, ensure_ascii=False) if data else ""),
        ("dateTime", datetime.datetime.now().__str__()),
        ("timestamp", str(int(time.time()))),
    ]

    log_item.set_contents(contents)
    log_group.append(log_item)
    # 写入日志
    request = PutLogsRequest(
        project=project,
        logstore=log_store,
        topic="",
        source="",
        logitems=log_group,
        compress=False,
    )
    try:
        client.put_logs(request)
    except Exception as e:
        print("日志失败")
        print(e)