carry_efficiency_tool/utils/aliyun_log.py

# -*- coding: utf-8 -*-
"""
公共方法，包含：生成log / 删除log
"""
import json
from typing import Optional
from aliyun.log import PutLogsRequest, LogClient, LogItem
from loguru import logger

proxies = {"http": None, "https": None}


class AliyunLogger:

    # 写入阿里云日志
    @staticmethod
    def logging(channel: str,
            principal: str,
            channel_user: str,
            channel_video_id: str,
            message: str,
            code: str,
            data: Optional[str] = None,
            pq_video_id: Optional[str] = None):
        """
        写入阿里云日志
        测试库: https://sls.console.aliyun.com/lognext/project/crawler-log-dev/logsearch/crawler-log-dev
        正式库: https://sls.console.aliyun.com/lognext/project/crawler-log-prod/logsearch/crawler-log-prod
        """
        accessKeyId = "LTAIWYUujJAm7CbH"
        accessKey = "RfSjdiWwED1sGFlsjXv0DlfTnZTG1P"

        project = "crawler-log-prod"
        logstore = "rewriting-log"
        endpoint = "cn-hangzhou.log.aliyuncs.com"
        try:
            contents = [
                ("principal", principal),
                ("channel", channel),
                ("channel_user", str(channel_user) if channel_user is not None else ""),
                ("channel_video_id", str(channel_video_id) if channel_video_id is not None else ""),
                ("message", str(message) if message is not None else ""),
                ("code", str(code) if code is not None else ""),
                ("data", json.dumps(data, ensure_ascii=False) if data else ""),
                ("pq_video_id", pq_video_id if pq_video_id else "")
            ]
            # 创建 LogClient 实例
            client = LogClient(endpoint, accessKeyId, accessKey)
            log_group = []
            log_item = LogItem()
            log_item.set_contents(contents)
            log_group.append(log_item)
            # 写入日志
            request = PutLogsRequest(
                project=project,
                logstore=logstore,
                topic="",
                source="",
                logitems=log_group,
                compress=False,
            )

            client.put_logs(request)
        except Exception as e:
            logger.error(f"[+] 阿里云日志写入失败{e}")