log.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
# @Version     : Python 3.11.4
# @Software    : Sublime Text 4
# @Author      : StudentCWZ
# @Email       : StudentCWZ@outlook.com
# @Date        : 2023/11/19 14:59
# @File        : log.py
# @Description :
"""


import os
from datetime import datetime
from sqlalchemy import text

from application.extensions.init_sqlalchemy import db


class Log(db.Model):
    """Basic user model"""
    __tablename__ = os.environ.get('TableName', 'logs')

    id = db.Column(db.Integer, primary_key=True, autoincrement=True)
    date_time = db.Column(db.DateTime, nullable=False)
    uuid = db.Column(db.String(80))
    mid = db.Column(db.String(80))
    mid_type = db.Column(db.String(80))
    mac_wifi = db.Column(db.String(80))
    mac_voice = db.Column(db.String(80))
    code = db.Column(db.SmallInteger)
    query = db.Column(db.String(80))
    terminal_domain = db.Column(db.String(80))
    terminal_intent = db.Column(db.String(80))
    distribution_gree_domain = db.Column(db.String(80))
    distribution_gree_intent = db.Column(db.String(80))
    response_text = db.Column(db.Text)
    emotion_class = db.Column(db.String(80))
    skill_id = db.Column(db.String(80))
    voice_portal = db.Column(db.SmallInteger)
    service_nlu = db.Column(db.String(80))
    service_type = db.Column(db.SmallInteger)
    slots = db.Column(db.Text)
    yzs_request_id = db.Column(db.String(80))
    yzs_remote_ip = db.Column(db.String(80))
    yzs_app_key = db.Column(db.String(80))
    yzs_ud_id = db.Column(db.String(80))
    yzs_user_id = db.Column(db.String(80))
    yzs_intent = db.Column(db.Text)
    yzs_general = db.Column(db.Text)
    yzs_nlu_time = db.Column(db.String(80))
    get_body_time = db.Column(db.String(80))
    gree_nlu_time = db.Column(db.String(80))
    get_homeid_time = db.Column(db.String(80))
    tencent_nlu_time = db.Column(db.String(80))
    cost_time = db.Column(db.String(80))

    def __repr__(self) -> str:
        return f'<Log {self.uuid}>'

    def to_dict(self) -> dict:
        """object to dict"""
        _dic = {}
        for key, value in self.__dict__.items():
            if str(key).startswith("_"):
                # 前缀带下划线不要
                continue
            _dic[key] = value
        return _dic

    @classmethod
    def batch_save(cls, objects):
        with db.session.begin_nested():
            for obj in objects:
                log = Log(**obj)
                # 将字符串转换为 datetime 对象
                date_time_obj = datetime.strptime(log.date_time, '%Y-%m-%d %H:%M:%S')
                partition_date = date_time_obj.strftime('%Y_%m')
                partition_name = f'{log.__tablename__}_{partition_date}'

                db.session.execute(text(f"""
                CREATE TABLE IF NOT EXISTS {partition_name} PARTITION OF {log.__tablename__}
                FOR VALUES FROM ('{date_time_obj.strftime('%Y-%m-01')}') TO 
                ('{date_time_obj.strftime('%Y-%m-01')}'::date + interval '1 month');
                """))
                db.session.add(log)

        db.session.commit()