qc-dgp/dgp/tests/util/db_order_util.py

#!/usr/bin/env python
# -*- coding: utf-8 -*-

"""
__title__ = '订单工具类，用于查询，修改订单等'

@Time    : 2020/9/30 12:38
@Author  : Kenny-PC
@Software: PyCharm

# code is far away from bugs with the god animal protecting
    I love animals. They taste delicious.
              ┏┓      ┏┓
            ┏┛┻━━━┛┻┓
            ┃      ☃      ┃
            ┃  ┳┛  ┗┳  ┃
            ┃      ┻      ┃
            ┗━┓      ┏━┛
                ┃      ┗━━━┓
                ┃  神兽保佑    ┣┓
                ┃　永无BUG！   ┏┛
                ┗┓┓┏━┳┓┏┛
                  ┃┫┫  ┃┫┫
                  ┗┻┛  ┗┻┛
"""

from util.MySQLConnection import MySQLConnection
import pymysql

# 数据导入表采用replace替换主键orderid的方法
def batch_save_order(data):
    if data is None or len(data) == 0:
        print('数据为空，不执行数据库操作！')
    else:
        sql = 'replace INTO quchen_text.`order` (amount,channel,channel_id,date,from_novel,order_id,order_time,platform,reg_time,stage,user_id) VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s);'
        connect = MySQLConnection()
        try:
            num = connect.batch(sql, data)
            # 提交
            connect.commit()
            print('订单数据最终入库【{num}】条'.format(num=num))
        except Exception as e:
            print('订单数据入库失败：', e)
        finally:
            connect.close()


# 获取各平台的订单数量
def get_platform_order_count(date):
    sql = 'SELECT platform, COUNT(1) AS num FROM quchen_text.`order` WHERE date = %s GROUP BY platform'
    connect = MySQLConnection()
    platform_order_count = []
    try:
        platform_order_count = connect.query(sql, date)
    except Exception as e:
        print('各平台的订单数据查询失败：', e)
    finally:
        connect.close()
    return platform_order_count

def get_account_info_list(platform):
    sql=f"select text from order_account_text where platform='{platform}'"
    db = pymysql.connect('rm-bp1c9cj79872tx3aaro.mysql.rds.aliyuncs.com', 'superc', 'Cc719199895', 'quchen_text')
    cur=db.cursor()
    try:
        cur.execute(sql)
        platform= cur.fetchall()
    except Exception as e:
        print(e)
    li=[]
    for i in platform:
        li.append(i[0].split(","))
    return li


if __name__ == '__main__':
    get_account_info_list("掌中云")