AutoScraperX/core/utils/log/aliyun_log.py

"""
公共方法，包含：生成log
"""
import json
import time

from aliyun.log import LogClient, PutLogsRequest, LogItem

from config import settings

proxies = {"http": None, "https": None}
from core.utils.trace_utils import get_current_trace_id  # 导入工具函数


class AliyunLogger(object):
    """
    阿里云日志方法
    """

    def __init__(self, platform, mode, env="prod"):
        self.platform = platform
        self.mode = mode
        self.env = env

    # 写入阿里云日志
    def logging(
            self, code, message, data=None, trace_id=None, account=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
        """
        # 设置阿里云日志服务的访问信息
        # if message is None:
        #     message = LOG_CODES.get(code, "未知错误")
        # if data is None:
        #     data = {}
        # 优先使用传入的 trace_id，否则从上下文获取
        current_trace_id = get_current_trace_id() or ""
        accessKeyId = settings.ALIYUN_ACCESS_KEY_ID
        accessKey = settings.ALIYUN_ACCESS_KEY_SECRET
        if self.env == "dev":
            project = "crawler-log-dev"
            logstore = "crawler-log-dev"
            endpoint = "cn-hangzhou.log.aliyuncs.com"
        else:
            project = "crawler-log-prod"
            logstore = "crawler-fetch"
            endpoint = "cn-hangzhou.log.aliyuncs.com"

        # 创建 LogClient 实例
        client = LogClient(endpoint, accessKeyId, accessKey)
        log_group = []
        log_item = LogItem()

        """
        生成日志消息体格式，例如
        crawler:xigua
        message:不满足抓取规则 
        mode:search
        timestamp:1686656143
        """
        message = message.replace("\r", " ").replace("\n", " ")
        contents = [
            (f"TraceId", str(current_trace_id)),
            (f"code", str(code)),
            (f"platform", str(self.platform)),
            (f"mode", str(self.mode)),
            (f"message", str(message)),
            (f"data", json.dumps(data, ensure_ascii=False) if data else ""),
            (f"account", str(account)),
            ("timestamp", str(int(time.time()))),
        ]

        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)