data-verification/Statistical/manage_data.py

#-*- coding: utf-8 -*-
#__author__ = "dukun"

import json
import time
import pymysql
import logging
import numpy as np
import pandas as pd
import configparser
from celery_tasks import cel
#from DBUtils.PooledDB import PooledDB
from dbutils.pooled_db import PooledDB
from datetime import datetime,timedelta,date


#配置输出日志格式
LOG_FORMAT = "%(asctime)s %(filename)s[line:%(lineno)d] %(name)s %(levelname)s %(pathname)s %(message)s "

#配置输出时间的格式，注意月份和天数不要搞乱了
DATE_FORMAT = '%Y-%m-%d  %H:%M:%S %a ' 
logging.basicConfig(level=logging.INFO,
                    format=LOG_FORMAT,
                    datefmt = DATE_FORMAT ,
                    filename=r"./logs/manage.log" #有了filename参数就不会直接输出显示到控制台，而是直接写入文件
                    )

cf = configparser.RawConfigParser()
cf.read(r"./config.ini") 

CHANNEL_HOST = cf.get("database1","host")
CHANNEL_PORT = int(cf.get("database1","port"))
CHANNEL_USER = cf.get("database1","user")
CHANNEL_PASSWORD = cf.get("database1","password")
CHANNEL_DB = cf.get("database1","db")

STAT_HOST = cf.get("database2","host")
STAT_PORT = int(cf.get("database2","port"))
STAT_USER = cf.get("database2","user")
STAT_PASSWORD = cf.get("database2","password")
STAT_DB = cf.get("database2","db")




#通道数据库
channel_db_config = {
    'host' : CHANNEL_HOST,
    'port' : CHANNEL_PORT,
    'user' : CHANNEL_USER,
    'password' : CHANNEL_PASSWORD,
    'db' : CHANNEL_DB,
    'charset' : 'utf8',
    'autocommit' : 1
}

#监控数据库
stat_db_config = {
    'host' : STAT_HOST,
    'port' : STAT_PORT,
    'user' : STAT_USER,
    'password' : STAT_PASSWORD,
    'db' : STAT_DB,
    'charset' : 'utf8',
    'autocommit' : 1
}

#创建数据库连接池
def createPool(db_config):
   spool = PooledDB(pymysql, 5, **db_config) 
   return spool

@cel.task
def save_yesterday_data():
    today = datetime.today()
    yesterday = today - timedelta(days=1)
    yesterday = yesterday.strftime("%Y-%m-%d")

    ins_manage_sql = "INSERT INTO manage_data VALUES(0,'{}',%s,%s,%s,%s,%s,%s,%s,%s,%s,%s)".format(yesterday)
    ins_use_time_sql = "INSERT INTO yesterday_use_time VALUES(0,'{}',%s,%s)".format(yesterday)
    #print(ins_manage_sql)

    conn = createPool(channel_db_config).connection()
    stat_conn = createPool(stat_db_config).connection()
    
    month_df = get_month_data(conn) 
    yesterday_df = get_yesterday_data(conn)
    use_time_df = get_yesterday_use_time(conn)
    
    df = pd.merge(month_df,yesterday_df,how = 'outer',on=['enterprise_id','customer_name'])
    #df = df.fillna(0)
    #df['yes_rate'] = 0
    df['yes_rate'] = df['profit'] / df['income']
    df['month_rate'] = df['month_profit'] / df['income']
    df = df.fillna(0)
    df['today_income'] = -1
    df['today_profit'] = -1
    df = df.reindex(columns=['enterprise_id','customer_name','income','profit','yes_rate','month_income','month_profit',
                            'month_rate','today_income','today_profit'])
    
    yes_profit_sum = float(df['profit'].sum())
    yes_income_sum = float(df['income'].sum())
    month_profit_sum = float(df['month_profit'].sum())
    month_income_sum = float(df['month_income'].sum())
    yes_profit_rate = yes_profit_sum /yes_income_sum
    month_profit_rate = month_profit_sum /month_income_sum

    ins_manage_list = df.values.tolist()
    ins_manage_list.append(['0001','',yes_income_sum,yes_profit_sum,yes_profit_rate,month_income_sum,
                            month_profit_sum,month_profit_rate,-1,-1] )
    #print(ins_manage_list)

    ins_use_time_list = []
    ins_use_time_list.append(('小于10分钟',int(use_time_df['A'].iloc[0])))
    ins_use_time_list.append(('10-20分钟',int(use_time_df['B'].iloc[0])))
    ins_use_time_list.append(('20-30分钟',int(use_time_df['C'].iloc[0])))
    ins_use_time_list.append(('30-40分钟',int(use_time_df['D'].iloc[0])))
    ins_use_time_list.append(('40-50分钟',int(use_time_df['E'].iloc[0])))
    ins_use_time_list.append(('50-60分钟',int(use_time_df['F'].iloc[0])))
    ins_use_time_list.append(('大于60分钟',int(use_time_df['G'].iloc[0])))
    ins_use_time_list.append(('平均时长',float(use_time_df['avg_time'].iloc[0]/60)))
    #print(use_time_df)
    #print(ins_use_time_list)

    cursor = stat_conn.cursor()
    try:
        cursor.executemany(ins_manage_sql,ins_manage_list)
        cursor.executemany(ins_use_time_sql,ins_use_time_list)
        
    except Exception as ex :
        stat_conn.rollback()
        print(ex)


    #print(month_df)
    #print(yesterday_df)
    #print(df)
    #print(use_time_df)
    
    conn.close()
    stat_conn.close()
    print(44444444444)

@cel.task
def save_now_data():
    conn = createPool(channel_db_config).connection()
    stat_conn = createPool(stat_db_config).connection()
    
    income,profit = get_now_data(conn)
    #print(income,profit)

    #now = today.strftime("%Y-%m-%d %H:%M:%S")
    now_time = time.strftime("%Y-%m-%d %H:%M:%S",time.localtime())
    #print(now_time)

    que_sql = " SELECT * FROM manage_data WHERE customer_id = '0000'  "
    ins_sql = """ INSERT INTO manage_data (time,customer_id,today_income,today_profit) VALUES('{}','0000',
                                        {},{}) """.format(now_time,income,profit)
    upd_sql = """ UPDATE manage_data SET time = '{}',today_income= {},today_profit={} 
                  WHERE customer_id = '0000' """.format(now_time,income,profit)
    #print(upd_sql)

    df = pd.read_sql(que_sql,stat_conn)
    
    cursor = stat_conn.cursor()
    try:
        if df.empty is True:
            cursor.execute(ins_sql)
        else:
            cursor.execute(upd_sql)
    except Exception as ex:
        print(ex)
        stat_conn.rollback()
    
    cursor.close()
    conn.close()
    stat_conn.close()

    print(555555555555)

def get_now_data(conn):
    today = datetime.today()

    time_start = today.strftime("%Y-%m-%d") + ' 00:00:00'
    sql = """SELECT SUM(price) income,SUM(price - operator_balance_price - partner_balance_price) profit
            FROM
                flow_order_info
            WHERE 
                check_time BETWEEN '{}'  AND now()  AND status = 6 
                AND enterprise_id NOT IN (57,61,63,64,72)
                AND channel_id NOT IN ( 101,102,103,110,111,112) """.format(time_start)
    #print(sql)            
    df = pd.read_sql(sql,conn)
    
    return df['income'].iloc[0],df['profit'].iloc[0]
    

#获取本月总利润
def get_month_data(conn):

    DATE = None
    STEP = None

    today = datetime.today()
    year = today.year
    month = today.month
    
    yesterday = today - timedelta(days=1)
    yesterday = yesterday.strftime("%Y-%m-%d")
    yesterdayStart = yesterday + ' 00:00:00'
    yesterdayEnd = yesterday + ' 23:59:59'

    date2 = datetime(year, month, today.day)
    if date2.day == 1:
       if month == 1:
          month = 12
          year = year -1
       else:
          month = month - 1

    date1 = datetime(year, month, 1)
    date1_week = date1.isocalendar()[1]
    date2_week = date2.isocalendar()[1]

    if DATE is None:
       DATE = date1
    if STEP is None:
       STEP = (date2 - date1).days

    DATE_STR = DATE.strftime("%Y-%m")


    time_start = DATE.strftime("%Y-%m-%d %H:%M:%S")
    
    #print(time_start)
    time_end = (DATE + timedelta(days=STEP)).strftime("%Y-%m-%d %H:%M:%S")
    #print(time_start)
    #print(time_end)


            
    profit_temp_sql = """SELECT  enterprise_id,customer_name,income,profit
                         FROM
                            (SELECT enterprise_id,SUM(price) income,SUM(price - operator_balance_price - partner_balance_price) profit
                            FROM
                                flow_order_info_{}
                            WHERE 
                                check_time BETWEEN '{}'  AND '{}'  AND status = 6 
                                AND enterprise_id NOT IN (57,61,63,64,72)
                                AND channel_id NOT IN ( 101,102,103,110,111,112)
                                GROUP BY enterprise_id) t1 LEFT JOIN customer_info
                                ON customer_info.customer_id = t1.enterprise_id"""
                    
    profit_sql_temp_1 = """SELECT SUM(price - operator_balance_price - partner_balance_price - 
                            price * cast(0.06 / 1.06 AS decimal(12,4))) profits
                    FROM
                        flow_order_info_{}
                    WHERE
                        apply_date BETWEEN '{}'  AND '{}'  AND status = 6 AND enterprise_id  IN (72) 
                        AND channel_id NOT IN ( 101,102,103,110,111,112)"""


        
    profit_end_sql = """ SELECT  enterprise_id,customer_name,income,profit
                        FROM
                            (SELECT enterprise_id,SUM(price) income,SUM(price - operator_balance_price - partner_balance_price) profit
                            FROM
                                flow_order_info
                            WHERE 
                                check_time BETWEEN '{}'  AND '{}'  AND status = 6 
                                AND enterprise_id NOT IN (57,61,63,64,72)
                                AND channel_id NOT IN ( 101,102,103,110,111,112)
                                GROUP BY enterprise_id) t1 LEFT JOIN customer_info
                                ON customer_info.customer_id = t1.enterprise_id"""
    
    profit_end_sql_1 = """SELECT  SUM(price - operator_balance_price  - partner_balance_price 
                                            - price * cast(0.06 / 1.06 AS decimal(12,4)))  profit
                    FROM
                        flow_order_info
                    WHERE
                        apply_date BETWEEN '{}'  AND '{}'  AND status = 6 AND enterprise_id IN (72)
                        AND channel_id NOT IN ( 101,102,103,110,111,112)"""

    
    sql = ''
    for i in range(date1_week,date2_week+1):
        str_table_suffix = "{0}{1:02d}".format(year,i)
        if i == date2_week:
            if (today - date2) < timedelta(days=7):
                sql += profit_end_sql.format(time_start,yesterdayEnd)
            else:    
                sql += prifit_temp_sql.format(str_table_suffix,time_start,yesterdayEnd)
        else :
            sql += profit_temp_sql.format(str_table_suffix,time_start,yesterdayEnd) + ' UNION '
            
    last_sql = ''' SELECT enterprise_id,customer_name,SUM(income) month_income,SUM(profit) month_profit
                  FROM ({})t6 GROUP BY enterprise_id
                  
               '''.format(sql)

    #print(last_sql)
    df = pd.read_sql(last_sql,conn)
 
    return df


def get_yesterday_data(conn):
    today = datetime.today()
    yesterday = today - timedelta(days=1)
    yesterday = yesterday.strftime("%Y-%m-%d")
    yesterdayStart = yesterday + ' 00:00:00'
    yesterdayEnd = yesterday + ' 23:59:59'

    
    yes_profit_sql =  """SELECT  enterprise_id,customer_name,income,profit
                        FROM
                            (SELECT enterprise_id,SUM(price) income,SUM(price - operator_balance_price - partner_balance_price) profit
                            FROM
                                flow_order_info
                            WHERE 
                                check_time BETWEEN '{}'  AND '{}'  AND status = 6 
                                AND enterprise_id NOT IN (57,61,63,64,72)
                                AND channel_id NOT IN ( 101,102,103,110,111,112)
                                GROUP BY enterprise_id) t1 LEFT JOIN customer_info
                                ON customer_info.customer_id = t1.enterprise_id""".format(yesterdayStart,yesterdayEnd)

    yes_profit_sql_1 =  """SELECT SUM(price) sum_income,
                                    SUM(operator_balance_price + partner_balance_price 
                                                + price * cast(0.06 / 1.06 AS decimal(12,4))) sum_cost
                        FROM
                            flow_order_info
                        WHERE
                            apply_date BETWEEN '{}'  AND '{}'  AND status = 6 AND enterprise_id  IN (72)
                            AND channel_id NOT IN ( 101,102,103,110,111,112)""".format(yesterdayStart,yesterdayEnd)
    #print(yes_profit_sql)

    df = pd.read_sql(yes_profit_sql,conn)
  
    return  df

def get_yesterday_use_time(conn):
    today = datetime.today()
    yesterday = today - timedelta(days=1)
    yesterday = yesterday.strftime("%Y-%m-%d")
    yesterdayStart = yesterday + ' 00:00:00'
    yesterdayEnd = yesterday + ' 23:59:59'


    sql = """ SELECT
             SUM(CASE WHEN diff_time  BETWEEN 0 AND 600 THEN 1 ELSE 0 END) AS 'A',
             SUM(CASE WHEN diff_time  BETWEEN 601 AND 1200 THEN 1 ELSE 0 END) AS 'B',
             SUM(CASE WHEN diff_time  BETWEEN 1201 AND 1800 THEN 1 ELSE 0 END) AS 'C',
             SUM(CASE WHEN diff_time  BETWEEN 1801 AND 2400 THEN 1 ELSE 0 END) AS 'D',
             SUM(CASE WHEN diff_time  BETWEEN 2401 AND 3000 THEN 1 ELSE 0 END) AS 'E',
             SUM(CASE WHEN diff_time  BETWEEN 3001 AND 3600 THEN 1 ELSE 0 END) AS 'F',
             SUM(CASE WHEN diff_time   > 3600 THEN 1 ELSE 0 END) AS 'G',
             (SUM(diff_time) /COUNT(*))  AS 'avg_time'
                FROM 
                    (  SELECT 
                (UNIX_TIMESTAMP(check_time) - UNIX_TIMESTAMP(apply_date)) AS diff_time
                FROM 
                flow_order_info
                WHERE 
                      apply_date BETWEEN '{}'  AND '{}' AND
                      status = 6 AND
                      channel_id NOT IN ( 101,102,103,110,111,112  ) ) AS t1 """.format(yesterdayStart,yesterdayEnd)
                      
    df = pd.read_sql(sql,conn)
    
    return df

save_yesterday_data()
#save_now_data()