MatchRate/test.py

"""
@author: luojunhui
CREATE TABLE `publish_content_miniprogram` (
  `publish_content_id` varchar(64) NOT NULL COMMENT '发布内容ID',
  `root_share_id` varchar(64) DEFAULT NULL,
  `source` varchar(64) DEFAULT NULL COMMENT '来源',
  `program_id` varchar(64) DEFAULT NULL COMMENT '小程序ID',
  `program_name` varchar(128) DEFAULT NULL COMMENT '小程序名称',
  `program_avatar` varchar(512) DEFAULT NULL COMMENT '小程序头像',
  `production_cover` varchar(1024) DEFAULT NULL COMMENT '卡片封面',
  `production_name` varchar(255) DEFAULT NULL COMMENT '卡片标题',
  `production_path` varchar(1024) DEFAULT NULL COMMENT '卡片路径',
  `video_url` varchar(1024) DEFAULT NULL COMMENT '视频播放地址',
  `hide_flag` int(11) DEFAULT NULL COMMENT '是否隐藏（0-否，1-是）',
  `status` int(11) NOT NULL DEFAULT '2' COMMENT '状态（0-待处理，1-处理中，2-成功，3-失败）',
  `trace_id` varchar(128) DEFAULT NULL COMMENT '请求ID',
  `error_msg` varchar(2048) DEFAULT NULL COMMENT '错误信息',
  `create_timestamp` bigint(20) DEFAULT NULL COMMENT '创建时间戳',
  `update_timestamp` bigint(20) DEFAULT NULL COMMENT '更新时间戳',
  PRIMARY KEY (`publish_content_id`),
  KEY `idx_rootShareId` (`root_share_id`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='发布内容-插入小程序'
"""
import pymysql
from functions import RateDetail


def table_structure():
    """
    sensitive words
    :return:
    """
    connection = pymysql.connect(
        host="rm-t4na9qj85v7790tf84o.mysql.singapore.rds.aliyuncs.com",  # 数据库IP地址，内网地址
        port=3306,  # 端口号
        user="crawler_readonly",  # mysql用户名
        passwd="cyber#crawler_2023",  # mysql用户登录密码
        db="aigc-admin-prod",  # 数据库名
        charset="utf8mb4"  # 如果数据库里面的文本是utf8编码的，charset指定是utf8
    )
    sql = "select status, trace_id, error_msg  from publish_content_miniprogram where create_timestamp > 1716739200000;"
    cursor = connection.cursor()
    cursor.execute(sql)
    data = cursor.fetchall()
    result = [list(line) for line in data]
    return result


result = table_structure()
success, s_r, fail, f_r, processing, p_rate, error_list = RateDetail().each_rate(result)
for error in error_list:
    print(error)