Initial commit

Ex.py

+from extract.ex_ly_sku import ExLySku
+from utils.msg_handler import MsgHandler
+import traceback
+import sys
+
+
+if __name__ == '__main__':
+
+    """
+    ex_sku: 采集联营SKU
+    del_sku: 删除MongoDB里的原始数据
+    """
+    if len(sys.argv) > 1:
+        if sys.argv[1] == 'ex_sku':
+            try:
+                ExLySku.ex_sku()
+            except:
+                MsgHandler.send_dd_msg("【联营SKU价格采集任务】 失败\n" +
+                                       "失败信息：" + str(traceback.print_exc()))
+        elif sys.argv[1] == 'del_sku':
+            try:
+                # ExLySku.del_sku()
+                pass
+            except:
+                MsgHandler.send_dd_msg("【Mongo数据删除任务】 失败\n" +
+                                       "失败信息：" + str(traceback.print_exc()))

Test.py

+from pipeline.pi_order import PiOrder
+from pipeline.pi_user import PiUser
+
+
+def main():
+
+    PiOrder.pipeline_order_items()
+
+    # pi_user = PiUser('用户流')
+    # pi_user.pipeline_user_without_login()
+    pass
+
+
+if __name__ == '__main__':
+    main()
+

config/conn_list.py

+import pymysql
+from config.db import *
+import pymongo as pm
+
+
+class ConnList:
+
+    @staticmethod
+    def Order():
+        conf = order_server
+        host = str(conf['host'])
+        user = conf['user']
+        password = conf['password']
+        db_name = conf['db_name']
+
+        return pymysql.connect(host, user, password, db_name, charset='utf8')
+
+    @staticmethod
+    def IcData():
+        conf = ic_data_server
+        host = str(conf['host'])
+        user = conf['user']
+        password = conf['password']
+        db_name = conf['db_name']
+
+        return pymysql.connect(host, user, password, db_name, charset='utf8')
+
+    @staticmethod
+    def MongoDB():
+        conf = mongodb_server
+        host = str(conf['host'])
+
+        return pm.MongoClient(host)

config/connection_list.py

+from utils.db_connection import DbConnection
+from config.db import *
+
+
+def order():
+
+    conf = order_server
+    host = str(conf['host'])
+    user = conf['user']
+    password = conf['password']
+    db_name = conf['db_name']
+    conn = DbConnection(host, user, password, db_name)
+
+    return conn.connection()
+
+
+def ic_data():
+
+    conf = ic_data_server
+    host = str(conf['host'])
+    user = conf['user']
+    password = conf['password']
+    db_name = conf['db_name']
+    conn = DbConnection(host, user, password, db_name)
+
+    return conn.connection()
+

config/db.py

+#!/usr/bin/env python
+# -*- coding:utf-8 -*-
+
+order_server = {
+    'host': '172.18.137.22',
+    'user': 'huntdbslave',
+    'password': 'mLssy2@@!!@$#yy',
+    'db_name': 'hunt2016'
+}
+
+ic_data_server = {
+    'host': '172.18.137.21',
+    'user': 'dtuserRead',
+    'password': 'xL#Rym2902m2lLX2y322',
+    'db_name': 'icdata'
+}
+
+mongodb_server = {
+    'host': 'mongodb://ichunt:huntmon66499@172.18.137.23/ichunt'
+}

config/order_sql.py

+#!/usr/bin/env python
+# -*- coding:utf-8 -*-
+
+
+class Order:
+
+    def user_no_order(self, where):
+
+        sql = "SELECT count(user_id) \
+            FROM lie_user_main \
+            WHERE user_id not in (SELECT user_id FROM lie_order) \
+            AND is_test = 0 \
+            AND create_time between 1262275200 and 1535731200"
\ No newline at end of file

extract/base.py

+
+
+class Base:
+
+    def __init__(self, name):
+        print('抽取数据：', name)
+
+    # 条件组装
+    @staticmethod
+    def condition_to_str(condition):
+
+        # 组合条件
+        con_str = ''
+        index = 0
+        if len(condition) > 0:
+            con_str += ' AND '
+            for row in condition:
+                index += 1
+                if index < len(condition):
+                    con_str += row + ' AND '
+                else:
+                    con_str += row
+        return con_str
+
+    # SELECT条件组装
+    @staticmethod
+    def col_to_str(col):
+
+        col_str = ''
+        index = 0
+        for row in col:
+            # 去除前面的XX.XX
+            if row.find('.'):
+                print(row)
+                row = row.split('.')[1]
+            index += 1
+            if index < len(col):
+                col_str += row + ' , '
+            else:
+                col_str += row + ' '
+        return col_str
+
+    # 结果格式转换
+    @staticmethod
+    def result_to_dict(col, result):
+
+        r_list = []
+        for r in result:
+            index = 0
+            r_dict = {}
+            for c in col:
+                r_dict[c] = r[index]
+                index += 1
+            r_list.append(r_dict)
+        return r_list
+
+

extract/ex_base.py

+
+
+class Base:
+
+    def __init__(self, name):
+        print('抽取数据：', name)
+
+    # WHERE条件组装
+    @staticmethod
+    def condition_to_str(condition):
+
+        # 组合条件
+        con_str = ''
+        index = 0
+        if len(condition) > 0:
+            con_str += ' AND '
+            for row in condition:
+                index += 1
+                if index < len(condition):
+                    con_str += row + ' AND '
+                else:
+                    con_str += row
+        return con_str
+
+    # SELECT条件组装
+    @staticmethod
+    def col_to_str(col):
+
+        col_str = ''
+        index = 0
+        for row in col:
+            index += 1
+            if index < len(col):
+                col_str += row + ' , '
+            else:
+                col_str += row + ' '
+        return col_str
+
+    # 结果格式转换
+    @staticmethod
+    def result_to_dict(col, result):
+
+        r_list = []
+        for r in result:
+            index = 0
+            r_dict = {}
+            for c in col:
+                # 去除前面的XX.XX
+                if '.' in c:
+                    c = c.split('.')[1]
+                r_dict[c] = r[index]
+                index += 1
+            r_list.append(r_dict)
+        return r_list
+
+

extract/ex_ly_sku.py

+from extract.ex_base import Base
+from config.conn_list import ConnList
+from utils.date_handler import DateHandler
+from utils.msg_handler import MsgHandler
+
+
+class ExLySku(Base):
+
+    @staticmethod
+    def ex_sku():
+
+        # 发送任务开始消息
+        MsgHandler.send_dd_msg("【联营SKU价格采集任务】 启动")
+        start = DateHandler.now_datetime()
+
+        # 获取mongo客户端
+        client = ConnList.MongoDB()
+        db = client['ichunt']
+        table = db['sku_price_log']
+
+        # 文件写入
+        file_name = r'/data3/hdfs_data/ly_sku_price/' + 'sku_' + DateHandler.now_date(days=0, d_type=2) + '.txt'
+        fo = open(file_name, "w")
+
+        # 偏转值和获取值
+        skip = 0
+        limit = 10000
+        index = 0
+
+        # 筛选条件
+        query = {"time": {"$gte": DateHandler.date_time(0), "$lte": DateHandler.date_time(-1)}}
+        print(query)
+
+        while True:
+            if index > 2000:
+                break
+            # 获取数据
+            result = table.find(query).skip(skip).limit(limit)
+            # 数据读取完毕，跳出循环
+            if index == result.count():
+                break
+            for row in result:
+                fo.write(
+                    str(row['sku_id']) + '_' + DateHandler.now_date(days=0, d_type=2) + '|' +
+                    str(row['tiered']) + '|' +
+                    str(row['time']) + "\n")
+                index += 1
+            # 更新偏移量
+            skip += limit
+            print("当前数据量：", index)
+
+        # 发送任务结束消息
+        end = DateHandler.now_datetime()
+        dur = DateHandler.cal_duration(start, end, t_type=2)
+        MsgHandler.send_dd_msg("【联营SKU价格采集任务】 结束\n" +
+                               "任务耗时：" + str(dur) + "分钟\n" +
+                               "采集数量：" + str(index))
+        # 关闭文件流
+        fo.close()
+
+    @staticmethod
+    def del_sku():
+
+        # 发送任务开始消息
+        MsgHandler.send_dd_msg("【Mongo数据删除任务】 启动")
+        start = DateHandler.now_datetime()
+
+        # 获取mongo客户端
+        client = ConnList.MongoDB()
+        db = client['ichunt']
+        table = db['sku_price_log']
+
+        # 筛选条件
+        query = {"time": {"$gte": DateHandler.date_time(1), "$lte": DateHandler.date_time(0)}}
+
+        # 删除文档
+        del_doc = table.delete_many(query)
+
+        # 发送任务结束消息
+        end = DateHandler.now_datetime()
+        dur = DateHandler.cal_duration(start, end, t_type=2)
+        MsgHandler.send_dd_msg("【Mongo数据删除任务】 结束\n" +
+                               "任务耗时：" + str(dur) + "分钟\n" +
+                               "删除数量：" + str(del_doc.deleted_count))

extract/ex_order.py

+from extract.ex_base import Base
+from config.conn_list import ConnList
+from utils.db_handler import DBHandler
+
+
+class ExOrder(Base):
+
+    """
+    获取全部订单 v1.0 zzl
+
+    1.去掉测试账号 is_test = 0
+    2.去掉尽调单 is_test = 0
+    """
+    def all_order(self, condition):
+
+        # 筛选字段
+        col = ['user_id', 'order_id', 'order_source', 'order_type', 'status', 'order_amount',
+               'order_goods_type', 'currency']
+        col_str = super().col_to_str(col)
+
+        # 订单数据库
+        db = ConnList.Order()
+        start_time = condition['start_time']
+        end_time = condition['end_time']
+        con_str = super().condition_to_str(condition['condition'])
+        sql = "SELECT %s \
+                FROM \
+                lie_order \
+                WHERE \
+                create_time BETWEEN  %d AND  %d \
+                AND is_type = 0 \
+                AND order_type = 1 \
+                AND user_id NOT IN ( SELECT user_id FROM lie_user_main WHERE is_test = 1 ) %s" \
+              % (col_str, start_time, end_time, con_str)
+        results = DBHandler.read(db=db, sql=sql)
+
+        # 结果格式转换为字典
+        final_result = super().result_to_dict(col, results)
+        print('final_result', len(final_result))
+        print(final_result[0])
+
+        return final_result
+
+    """
+    获取用户订单汇总 v1.0 zzl
+    
+    1. 关联会员表获取手机、邮箱
+    2. 关联税务表获得公司名。客户姓名
+    """
+    def person_order_detail(self, condition):
+
+        # 筛选字段
+        col = ['o.user_id', 'o.order_id', 'o.order_source', 'o.order_type', 'o.status', 'o.order_amount',
+               'o.order_goods_type', 'o.currency', 'o.pay_time', 'o.sale_id', 'i.tax_title', 'i.nike_name',
+               'u.mobile', 'u.email']
+        col_str = super().col_to_str(col)
+
+        # 订单数据库
+        db = ConnList.Order()
+        start_time = condition['start_time']
+        end_time = condition['end_time']
+        con_str = super().condition_to_str(condition['condition'])
+        sql = "SELECT %s FROM \
+                lie_order o\
+                LEFT JOIN lie_order_invoice i \
+                ON i.order_id = o.order_id \
+                LEFT JOIN lie_user_main u \
+                ON o.user_id = u.user_id \
+                WHERE \
+                o.pay_time BETWEEN  %d AND  %d \
+                AND o.is_type = 0 \
+                AND o.order_type = 1 \
+                AND u.is_test = 0 %s" \
+              % (col_str, start_time, end_time, con_str)
+        results = DBHandler.read(db=db, sql=sql)
+
+        # 结果格式转换为字典
+        final_result = super().result_to_dict(col, results)
+
+        return final_result
+
+    """
+    获取订单明细 V1.0 zzl
+    
+    1. 关联订单表获取
+    """
+    def order_items(self, condition):
+
+        # 筛选字段
+        col = ['i.order_id', 'u.user_id', 'i.goods_id', 'o.order_sn', 'o.status']
+        col_str = super().col_to_str(col)
+
+        # 订单数据库
+        db = ConnList.Order()
+        start_time = condition['start_time']
+        end_time = condition['end_time']
+        con_str = super().condition_to_str(condition['condition'])
+        sql = "SELECT %s FROM \
+                        lie_order_items i\
+                        LEFT JOIN lie_order o \
+                        ON i.order_id = o.order_id \
+                        LEFT JOIN lie_user_main u \
+                        ON u.user_id = o.user_id \
+                        WHERE \
+                        o.create_time BETWEEN  %d AND  %d \
+                        AND o.is_type = 0 \
+                        AND o.order_type = 1 \
+                        AND o.order_goods_type = 1 \
+                        AND u.is_test = 0 %s" \
+              % (col_str, start_time, end_time, con_str)
+        results = DBHandler.read(db=db, sql=sql)
+
+        # 结果格式转换为字典
+        final_result = super().result_to_dict(col, results)
+
+        return final_result
+

extract/ex_user.py

+from extract.ex_base import Base
+from config.conn_list import ConnList
+from utils.db_handler import DBHandler
+
+
+class ExUser(Base):
+
+    """
+    获取注册用户
+    """
+    def reg_user(self, condition):
+
+        # 筛选字段
+        col = ['user_id', 'create_time', 'mobile', 'email']
+        col_str = super().col_to_str(col)
+
+        # 用户数据库
+        db = ConnList.Order()
+        start_time = condition['start_time']
+        end_time = condition['end_time']
+        con_str = super().condition_to_str(condition['condition'])
+
+        sql = "SELECT %s \
+                FROM \
+                lie_user_main \
+                WHERE \
+                create_time BETWEEN  %d AND  %d \
+                AND is_test = 0  \
+                AND is_type = 0 %s" \
+              % (col_str, start_time, end_time, con_str)
+        results = DBHandler.read(db=db, sql=sql)
+
+        # 结果格式转换为字典
+        final_result = super().result_to_dict(col, results)
+        print('final_result', len(final_result))
+
+        return final_result

extract/extract.py

+import abc
+
+
+class ExtractBase:
+
+    def __init__(self, where):
+        self.where = where
+
+    @abc.abstractclassmethod
+    def extract(self):
+        pass
+
+

extract/extract_base.py

+import abc
+
+
+class ExtractBase:
+
+    def __init__(self, where):
+        self.where = where
+
+    @abc.abstractclassmethod
+    def extract(self):
+        pass
+

extract/extract_coupon.py

+import traceback
+from extract.extract import ExtractBase
+
+
+class ExtractCoupon(ExtractBase):
+
+    def extract(self):
+
+        results = {}
+        start_time = self.where['start_time']
+        end_time = self.where['end_time']
+        condition = self.where['condition']
+        # 组合条件
+        con_str = ''
+        index = 0
+        if len(condition) > 0:
+            con_str += ' AND '
+            for row in condition:
+                index += 1
+                if index < len(condition):
+                    con_str += row + ' AND '
+                else:
+                    con_str += row
+
+        db = ''
+        cursor = db.cursor()
+        sql = "SELECT \
+                user_id \
+                FROM \
+                lie_user_coupon \
+                WHERE \
+                create_time BETWEEN  %d AND  %d \
+                %s " \
+              % (start_time, end_time, con_str)
+
+        try:
+            cursor.execute(sql)
+            results = cursor.fetchall()
+        except:
+            db.rollback()
+            traceback.print_exc()
+        finally:
+            db.close()
+
+        return results
+
+    # 新用户
+    def newUserExtract(self):
+
+        str = 'user_id in (SELECT user_id FROM lie_user_main WHERE create_time BETWEEN %d AND %d)' \
+              % (self.where['start_time'], self.where['end_time'])
+        self.where['condition'].append(str)
+        return self.extract()

extract/extract_coupon_new_user.py

+from extract.extract_coupon import ExtractCoupon
+
+
+class ExtractCouponNewUser(ExtractCoupon):
+
+    def __init__(self, where):
+        super().__init__(where)
+
+    def extract(self):
+
+        print(self.where)

extract/extract_order.py

+import traceback
+from extract.extract import ExtractBase
+
+
+class ExtractOrder(ExtractBase):
+
+    def extract(self):
+
+        results = {}
+        start_time = self.where['start_time']
+        end_time = self.where['end_time']
+        condition = self.where['condition']
+        # 组合条件
+        con_str = ''
+        index = 0
+        if len(condition) > 0:
+            con_str += ' AND '
+            for row in condition:
+                index += 1
+                if index < len(condition):
+                    con_str += row + ' AND '
+                else:
+                    con_str += row
+        db = ''
+        cursor = db.cursor()
+        sql = "SELECT \
+                user_id, \
+	            order_id, \
+	            order_source, \
+	            order_type, \
+	            STATUS, \
+	            order_amount, \
+	            order_goods_type, \
+	            currency \
+                FROM \
+	            lie_order \
+                WHERE \
+	            create_time BETWEEN  %d AND  %d \
+	            AND is_type = 0 \
+                AND order_type = 1 \
+	            AND user_id NOT IN ( SELECT user_id FROM lie_user_main WHERE is_test = 1 ) %s" \
+              % (start_time, end_time, con_str)
+
+        try:
+            cursor.execute(sql)
+            results = cursor.fetchall()
+        except:
+            db.rollback()
+            traceback.print_exc()
+        finally:
+            db.close()
+
+        return results
+
+    def newUserOrder(self):
+
+        results = {}
+        start_time = self.where['start_time']
+        end_time = self.where['end_time']
+        condition = self.where['condition']
+        # 组合条件
+        con_str = ''
+        index = 0
+        if len(condition) > 0:
+            con_str += ' AND '
+            for row in condition:
+                index += 1
+                if index < len(condition):
+                    con_str += row + ' AND '
+                else:
+                    con_str += row
+        db = ''
+        cursor = db.cursor()
+        sql = "SELECT \
+                u.user_id, \
+                o.order_id, \
+                o.order_source, \
+                o.order_type, \
+                o.STATUS, \
+                o.order_amount, \
+                o.order_goods_type, \
+                o.currency \
+                FROM \
+                lie_order o \
+                LEFT JOIN lie_user_main u \
+                ON u.user_id = o.user_id \
+                WHERE \
+                (u.create_time BETWEEN  %d AND  %d) AND \
+                (o.create_time BETWEEN %d AND %d) \
+                AND o.is_type = 0 \
+                AND o.order_type = 1 \
+                AND u.is_test = 0 %s" \
+              % (start_time, end_time, start_time, end_time, con_str)
+        try:
+            cursor.execute(sql)
+            results = cursor.fetchall()
+        except:
+            db.rollback()
+            traceback.print_exc()
+        finally:
+            db.close()
+
+        return results
+
+    # 订单加入发票信息
+    def activityOrder(self):
+
+        results = {}
+        start_time = self.where['start_time']
+        end_time = self.where['end_time']
+        condition = self.where['condition']
+        # 组合条件
+        con_str = ''
+        index = 0
+        if len(condition) > 0:
+            con_str += ' AND '
+            for row in condition:
+                index += 1
+                if index < len(condition):
+                    con_str += row + ' AND '
+                else:
+                    con_str += row
+        db = ''
+        cursor = db.cursor()
+        sql = "SELECT \
+                o.user_id, \
+                o.order_id, \
+                o.order_source, \
+                o.order_type, \
+                o.STATUS, \
+                o.order_amount, \
+                o.order_goods_type, \
+                o.currency, \
+                o.pay_time, \
+                i.tax_title, \
+                i.nike_name, \
+                u.mobile, \
+                u.email, \
+                o.sale_id \
+                FROM \
+                lie_order o\
+                LEFT JOIN lie_order_invoice i \
+                ON i.order_id = o.order_id \
+                LEFT JOIN lie_user_main u \
+                ON o.user_id = u.user_id \
+                WHERE \
+                o.pay_time BETWEEN  %d AND  %d \
+                AND o.is_type = 0 \
+                AND o.order_type = 1 \
+                AND u.is_test = 0 %s" \
+              % (start_time, end_time, con_str)
+        try:
+            cursor.execute(sql)
+            results = cursor.fetchall()
+        except:
+            db.rollback()
+            traceback.print_exc()
+        finally:
+            db.close()
+
+        return results
+
+

extract/extract_order_price.py

+import traceback
+from extract.extract import ExtractBase
+
+
+class ExtractOrderPrice(ExtractBase):
+
+    def extract(self):
+
+        results = {}
+        start_time = self.where['start_time']
+        end_time = self.where['end_time']
+        condition = self.where['condition']
+        # 组合条件
+        con_str = ''
+        index = 0
+        if len(condition) > 0:
+            con_str += ' AND '
+            for row in condition:
+                index += 1
+                if index < len(condition):
+                    con_str += row + ' AND '
+                else:
+                    con_str += row
+        db = ''
+        cursor = db.cursor()
+        sql = "SELECT \
+                u.user_id, \
+                o.order_id, \
+                p.currency, \
+                p.price, \
+                p.price_type, \
+                o.status \
+                FROM \
+                lie_order_price p \
+                LEFT JOIN lie_order o \
+                ON o.order_id = p.order_id \
+                LEFT JOIN lie_user_main u \
+                ON u.user_id = o.user_id \
+                WHERE \
+                p.create_time BETWEEN  %d AND  %d \
+                AND o.is_type = 0 \
+                AND o.order_type = 1 \
+                AND u.is_test = 0 \
+                % s" \
+              % (start_time, end_time, con_str)
+
+
+        try:
+            cursor.execute(sql)
+            results = cursor.fetchall()
+        except:
+            db.rollback()
+            traceback.print_exc()
+        finally:
+            db.close()
+
+        return results
+

extract/extract_user.py

+import traceback
+from extract.extract import ExtractBase
+
+
+class ExtractUser(ExtractBase):
+
+    def extract(self):
+
+        results = {}
+        start_time = self.where['start_time']
+        end_time = self.where['end_time']
+        condition = self.where['condition']
+        # 组合条件
+        con_str = ''
+        index = 0
+        if len(condition) > 0:
+            con_str += ' AND '
+            for row in condition:
+                index += 1
+                if index < len(condition):
+                    con_str += row + ' AND '
+                else:
+                    con_str += row
+
+        db = ''
+        cursor = db.cursor()
+        sql = "SELECT \
+                user_id \
+                FROM \
+                lie_user_main \
+                WHERE \
+                create_time BETWEEN  %d AND  %d \
+                AND is_test = 0 %s " \
+              % (start_time, end_time, con_str)
+
+        try:
+            cursor.execute(sql)
+            results = cursor.fetchall()
+        except:
+            db.rollback()
+            traceback.print_exc()
+        finally:
+            db.close()
+
+        return results

main/Test.py

+from translate.trans_order import TransOrder
+
+
+def main():
+
+    trans = TransOrder()
+    trans.translate()
+
+if __name__ == '__main__':
+    main()

manager/manager.py

+import abc
+
+
+class ManagerBase:
+
+    def __init__(self, name):
+        self.name = name
+
+    @abc.abstractclassmethod
+    def manager(self):
+        pass

manager/manager_coupon.py

+from manager.manager import ManagerBase
+from translate.trans_coupon import TransCoupon
+from translate.trans_user import TransUser
+from translate.trans_order_price import TransOrderPrice
+from extract.extract_coupon import ExtractCoupon
+from extract.extract_order_price import ExtractOrderPrice
+
+
+class ManagerCoupon(ManagerBase):
+
+    def manager(self):
+        print(self.name)
+
+    def couponView(self, where):
+
+        con = {'start_time': where['start_time'], 'end_time': where['end_time'], 'condition': where['condition']}
+        extract = ExtractCoupon(con)
+        data = extract.newUserExtract()
+        trans = TransCoupon(data)
+        trans.transCouponUser()
+        trans.transCouponCount()
+
+    def couponDetail(self, where):
+        con = {'start_time': where['start_time'], 'end_time': where['end_time'], 'condition': where['condition']}
+        extract = ExtractOrderPrice(con)
+        data = extract.extract()
+        trans = TransUser(data)
+        trans.transUser()
+        trans_order = TransOrderPrice(data)
+        trans_order.transOrder()
+        trans_order.transOrderAmount(6.95)
+
+
+

manager/manager_order.py

+from manager.manager import ManagerBase
+from translate.trans_order import TransOrder
+from translate.trans_user import TransUser
+from extract.extract_order import ExtractOrder
+from extract.extract_user import ExtractUser
+
+
+class ManagerOrder(ManagerBase):
+
+    def manager(self):
+        print(self.name)
+
+    def orderView(self, where):
+
+        con = {'start_time': where['start_time'], 'end_time': where['end_time'], 'condition': where['condition']}
+        data = ExtractOrder(con).extract()
+        trans = TransOrder(data)
+        trans.translateOrderPeople()
+        trans.translateOrderPaidPeople()
+        trans.translateOrderAmount(6.95)
+        trans.translateOrderPaidAmount(6.95)
+
+    def userView(self, where):
+
+        con = {'start_time': where['start_time'], 'end_time': where['end_time'], 'condition': where['condition']}
+        data = ExtractUser(con).extract()
+        trans = TransUser(data)
+        trans.transUser()
+
+    def orderNewUserView(self, where):
+
+        con = {'start_time': where['start_time'], 'end_time': where['end_time'], 'condition': where['condition']}
+        data = ExtractOrder(con).newUserOrder()
+        trans = TransOrder(data)
+        trans.translateOrderPeople()
+        trans.translateOrderPaidPeople()
+        trans.translateOrderAmount(6.95)
+        trans.translateOrderPaidAmount(6.95)
+
+    def orderAc(self, where):
+        con = {'start_time': where['start_time'], 'end_time': where['end_time'], 'condition': where['condition']}
+        data = ExtractOrder(con).activityOrder()
+        trans = TransOrder(data)
+        trans.translateOrderDetail(6.95)
+
+

pipeline/Base.py

+
+
+class Base:
+
+    def __init__(self, name):
+        self.name = name
+        print(name)
+

pipeline/ex_base.py

+
+
+class Base:
+
+    def __init__(self, name):
+        self.name = name
+        print(name)
+

pipeline/pi_order.py

+from extract.ex_order import ExOrder
+from translate.ts_order import TsOrder
+
+
+class PiOrder:
+
+    @staticmethod
+    def pipeline_order_people():
+
+        where = {'start_time': 1539273600, 'end_time': 1541001600, 'condition': ['o.order_goods_type = 1',
+                                                                                 'o.order_pay_type != 3',
+                                                                                 'o.order_source like \'%pf=-1%\'',
+                                                                                 'o.status in (7 ,8, 10)']}
+        # E
+        ex_order = ExOrder('订单详情')
+        data = ex_order.person_order_detail(where)
+        # T
+        ts_order = TsOrder('会员订单', data=data)
+        ts_order.trans_order_people(6.95)
+
+    @staticmethod
+    def pipeline_order_items():
+
+        where = {'start_time': 1538150400, 'end_time': 1541952000, 'condition': ['i.supplier_id = 17']}
+
+        # E
+        ex_order = ExOrder('order detail')
+        data = ex_order.order_items(where)
+        # T
+        ts_order = TsOrder('', data=data)
+        ts_order.trans_order_items_canal("L0000089")
+
+
+
+

pipeline/pi_user.py

+from pipeline.ex_base import Base
+from extract.ex_user import ExUser
+from translate.ts_user import TsUser
+
+
+class PiUser(Base):
+
+    def pipeline_user_without_login(self):
+
+        where = {'start_time': 1527782400, 'end_time': 1535644800, 'condition': []}
+
+        # E
+        ex_user = ExUser('新用户')
+        data = ex_user.reg_user(where)
+
+        # T
+        ts_user = TsUser('用户转化', data=data)
+        ts_user.trans_user_without_login(90)

translate/base.py

+
+
+class Base:
+
+    def __init__(self, name):
+        print('数据转换：', name)
+

translate/ex_base.py

+
+
+class Base:
+
+    def __init__(self, name):
+        print('数据转换：', name)
+

translate/trans.py

+import abc
+
+
+class Base:
+
+    def __init__(self, data):
+        self.data = data
+
+    @abc.abstractclassmethod
+    def translate(self):
+        pass
+
+
+
+
+
+

translate/trans_base.py

+import abc
+
+
+class TranslateBase:
+
+    def __init__(self, data):
+        self.data = data
+
+    @abc.abstractclassmethod
+    def translate(self):
+        pass

translate/trans_coupon.py

+
+
+class TransCoupon:
+
+    def translate(self):
+
+        print(len(self.data))
+
+    def transCouponUser(self):
+
+        # user_id列表
+        user_list = []
+
+        for row in self.data:
+            user_id = row[0]
+            if user_id not in user_list:
+                user_list.append(user_id)
+        print('领券人数', len(user_list))
+
+        return len(user_list)
+
+    def transCouponCount(self):
+
+        print('领券数量', len(self.data))
+        return len(self.data)
+

translate/trans_order.py

+import xlwt
+import time
+import traceback
+
+
+class TransOrder:
+
+    def translate(self):
+
+        print('订单数量' + len(self.data))
+        return len(self.data)
+
+    # 订单人数
+    def translateOrderPeople(self):
+
+        # user_id列表
+        user_list = []
+
+        for row in self.data:
+            user_id = row[0]
+            if user_id not in user_list:
+                user_list.append(user_id)
+
+        print('下单人数', len(user_list))
+        return len(user_list)
+
+    # 付款人数
+    def translateOrderPaidPeople(self):
+
+        # user_id列表
+        user_list = []
+
+        for row in self.data:
+            user_id = row[0]
+            status = row[4]
+            if status > 2:
+                if user_id not in user_list:
+                    user_list.append(user_id)
+
+        print('付款人数', len(user_list))
+        return len(user_list)
+
+    # 订单金额
+    def translateOrderAmount(self, exchange):
+
+        # 汇总金额
+        order_amount = 0
+
+        for row in self.data:
+            amount = float(row[5])
+            currency = row[7]
+            # 美金换算
+            if currency == 2:
+                amount *= exchange
+            order_amount += amount
+
+        print('订单金额', order_amount)
+        return order_amount
+
+    # 付款金额
+    def translateOrderPaidAmount(self, exchange):
+
+        # 汇总金额
+        order_amount = 0
+
+        for row in self.data:
+            amount = float(row[5])
+            currency = row[7]
+            status = row[4]
+            if status > 2:
+                # 美金换算
+                if currency == 2:
+                    amount *= exchange
+                order_amount += amount
+
+        print('付款金额', order_amount)
+        return order_amount
+
+    # 订单详情
+    def translateOrderDetail(self, exchange):
+
+        print(len(self.data))
+
+        # 数据集合
+        data = {}
+        # excel
+        wb = xlwt.Workbook()
+        sheet = wb.add_sheet("结果")
+        sheet.write(0, 0, '姓名&公司')
+        sheet.write(0, 1, '手机号')
+        sheet.write(0, 2, '付款时间')
+        sheet.write(0, 3, '实付金额')
+        sheet.write(0, 4, '业务员')
+
+        for row in self.data:
+            user_id = row[0]
+            amount = float(row[5])
+            currency = row[7]
+            pay_time = row[8]
+            tax_title = row[9]
+            tax_name = row[10]
+            mobile = row[11]
+            email = row[12]
+            sale_id = row[13]
+            # 币种转换
+            if currency == 2:
+                amount *= exchange
+            # 汇总用户
+            if user_id not in data:
+                data[user_id] = {}
+                data[user_id]['amount'] = amount
+                data[user_id]['pay_time'] = pay_time
+                data[user_id]['tax_title'] = tax_title
+                data[user_id]['tax_name'] = tax_name
+                data[user_id]['mobile'] = mobile
+                data[user_id]['email'] = email
+                data[user_id]['pay_time'] = time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(pay_time))
+                data[user_id]['sale_man'] = sale_id
+                db = ic_data()
+                cursor = db.cursor()
+                sql = 'select name from user_info where userId = %d' % (sale_id)
+                try:
+                    cursor.execute(sql)
+                    results = cursor.fetchall()
+                    if len(results) > 0:
+                        data[user_id]['sale_man'] = results[0][0]
+                except:
+                    db.rollback()
+                    traceback.print_exc()
+                finally:
+                    db.close()
+
+            else:
+                data[user_id]['amount'] += amount
+
+        # 生成EXCEL
+        print('data', len(data))
+        index = 1
+        for user_id in data:
+            sheet.write(index, 0, data[user_id]['tax_name'] + ' ' + data[user_id]['tax_title'])
+            sheet.write(index, 1, data[user_id]['mobile'] if data[user_id]['mobile'] != '' else data[user_id]['email'])
+            sheet.write(index, 2, data[user_id]['pay_time'])
+            sheet.write(index, 3, data[user_id]['amount'])
+            sheet.write(index, 4, data[user_id]['sale_man'])
+            index += 1
+
+        wb.save('result.xls')

translate/trans_order_price.py

+
+
+class TransOrderPrice:
+
+    def translate(self):
+
+        print('商品条数', len(self.data))
+        return len(self.data)
+
+    # 获取订单数
+    def transOrder(self):
+
+        # order_id列表
+        order_list = []
+        for row in self.data:
+            order_id = row[1]
+            if order_id not in order_list:
+                order_list.append(order_id)
+        print('订单数', len(order_list))
+
+        return len(order_list)
+
+    # 订单付款金额
+    def transOrderAmount(self, exchange):
+
+        order_amount = 0
+        for row in self.data:
+            currency = row[2]
+            price = float(abs(row[3]))
+            type = row[4]
+            status = row[5]
+            if -3 <= type <= -1 and status > 2:
+                if currency == 2:
+                    price *= exchange
+                order_amount += price
+
+        print('订单金额', order_amount)
+        return order_amount
+
+
+

translate/trans_user.py

+
+
+class TransUser:
+
+    def translate(self):
+
+        print('数量', len(self.data))
+        return len(self.data)
+
+    def transUser(self):
+
+        # user_id列表
+        user_list = []
+
+        for row in self.data:
+            user_id = row[0]
+            if user_id not in user_list:
+                user_list.append(user_id)
+        print('人数', len(user_list))
+
+        return len(user_list)

translate/ts_base.py

+
+
+class Base:
+
+    def __init__(self, name, data):
+        self.data = data
+        print('数据转换：', name)
+
+    # 人数去重
+    def no_repeat_people(self, data):
+
+        user_list = []
+        for row in data:
+            user_id = row['user_id']
+            if user_id not in user_list:
+                user_list.append(user_id)
+
+        print(len(user_list))
+

translate/ts_order.py

+from translate.ts_base import Base
+from utils.excel_handler import ExcelHandler
+from config.conn_list import ConnList
+from utils.db_handler import DBHandler
+import time
+import requests
+
+
+class TsOrder(Base):
+
+    """
+    获取用户订单明细 zzl v1.0
+
+    需生成Excel
+    """
+    def trans_order_people(self, exchange):
+
+        # Excel标题、内容
+        title = ['姓名&公司', '手机号', '付款时间', '实付金额', '业务员']
+        content = ['tax_title', 'mobile', 'pay_time', 'order_amount', 'sale_name']
+        # user_id列表
+        ul = []
+        # 结果字典
+        rd = {}
+
+        for row in self.data:
+            user_id = row['user_id']
+            order_amount = float(row['order_amount'])
+            currency = row['currency']
+            pay_time = row['pay_time']
+            tax_title = row['tax_title']
+            tax_name = row['nike_name']
+            mobile = row['mobile']
+            email = row['email']
+            sale_id = row['sale_id']
+            # 币种转换
+            if currency == 2:
+                order_amount *= exchange
+            # 用户去重
+            if user_id not in ul:
+                ul.append(user_id)
+                rd[user_id] = {}
+                rd[user_id]['order_amount'] = order_amount
+                rd[user_id]['pay_time'] = time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(pay_time))
+                rd[user_id]['tax_title'] = tax_name + ' ' + tax_title
+                rd[user_id]['mobile'] = mobile if mobile != '' else email
+                # 跨库获取交易员
+                db = ConnList.IcData()
+                sql = 'select name from user_info where userId = %d' % (sale_id)
+                results = DBHandler.read(db=db, sql=sql)
+                sale_man = results[0][0] if len(results) > 0 else ''
+                rd[user_id]['sale_name'] = sale_man
+            else:
+                rd[user_id]['order_amount'] += order_amount
+        ExcelHandler.write_to_excel(title, content, rd, "result", result_type=1)
+
+    """
+    获取指定渠道的订单 zzl v1.0
+    
+    通过基石API获取渠道
+    """
+    def trans_order_items_canal(self, canal):
+
+        # 订单集合
+        order_list = []
+        # 下单用户
+        user_list = []
+        # 付款用户
+        user_paid_list = []
+
+        for row in self.data:
+            order_id = row['order_id']
+            user_id = row['user_id']
+            goods_id = row['goods_id']
+            order_sn = row['order_sn']
+            order_status = row['status']
+
+            # request请求
+            url = "http://footstone.ichunt.net/webapi/sku_list"
+            body = {"sku_id": goods_id}
+            r = requests.post(url, data=body)
+            data = r.json()['data']['data']
+            canal_str = data[str(goods_id)]['canal']
+            # 渠道筛选
+            if canal_str == canal:
+                print(order_id, canal_str)
+                if order_status > 2:
+                    user_paid_list.append(user_id)
+                else:
+                    user_list.append(user_id)
+
+        print(user_list)
+        print(user_paid_list)
+
+
+
+

translate/ts_user.py

+from translate.ts_base import Base
+from utils.excel_handler import ExcelHandler
+from config.conn_list import ConnList
+from utils.db_handler import DBHandler
+import time
+
+
+class TsUser(Base):
+
+    """
+    获取一定时间内未登录用户
+    """
+    def trans_user_without_login(self, day):
+
+        # Excel标题、内容
+        title = ['账号', '注册时间']
+        content = ['mobile', 'create_time']
+        # 结果字典
+        rd = []
+        # 遍历结果集
+        for row in self.data:
+            mobile = row['mobile']
+            email = row['email']
+            create_time = row['create_time']
+            user_id = row['user_id']
+            # 获取登录时间
+            db = ConnList.Order()
+            sql = 'select last_login_time from lie_user_login_log where user_id = %d' % user_id
+            print(user_id)
+            results = DBHandler.read(db=db, sql=sql)
+            # 标记位
+            index = 0
+            # 遍历登录日志结果
+            for r in results:
+                last_login_time = r[0]
+                # 时间间隔
+                dur_time = last_login_time - create_time
+                # 下次登录时间大于一天且小于X天
+                if 86400 < dur_time < 60 * 60 * 24 * day:
+                    break
+                # 标记位自增
+                index += 1
+            if index == len(results):
+                rd.append({'mobile': mobile if mobile != '' else email,
+                           'create_time': time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(create_time))})
+        print(len(rd))
+        # 生成Excel
+        ExcelHandler.write_to_excel(title, content, rd, "result", result_type=2)
+
+

utils/date_handler.py

+import time
+import datetime
+
+
+class DateHandler:
+
+    """
+    获取字符串格式时间
+    d_type 1:Y-M-D 2:YMD
+    """
+    @staticmethod
+    def now_date(days, d_type):
+
+        today = datetime.date.today()
+        day = today - datetime.timedelta(days=days)
+        if d_type == 1:
+            return str(day)
+        elif d_type == 2:
+            return str(day.year) + str(day.month) + str(day.day)
+
+    """
+    获取当前时间的时间戳
+    """
+    @staticmethod
+    def now_datetime():
+
+        return int(time.time())
+
+    """
+    获取UNIX格式时间    
+    """
+    @staticmethod
+    def date_time(days):
+
+        today = datetime.date.today()
+        day = today - datetime.timedelta(days=days)
+        return int(time.mktime(time.strptime(str(day), '%Y-%m-%d')))
+
+    """
+    根据时间戳计算持续时间
+    t_type 1:秒 2:分 3:小时
+    """
+    @staticmethod
+    def cal_duration(start, end, t_type):
+
+        if t_type == 1:
+            pass
+        elif t_type == 2:
+            minute = (end - start) / 60
+            return round(minute, 2)
+

utils/db_connection.py

+import pymysql
+
+
+class DbConnection:
+
+    def __init__(self, host, user, password, db_name):
+        self.host = host
+        self.user = user
+        self.password = password
+        self.db_name = db_name
+
+    def connection(self):
+
+        return pymysql.connect(self.host, self.user, self.password, self.db_name, charset='utf8')
+
+

utils/db_handler.py

+import traceback
+
+
+class DBHandler:
+
+    @staticmethod
+    def read(db, sql):
+
+        results = {}
+        cursor = db.cursor()
+
+        try:
+            cursor.execute(sql)
+            results = cursor.fetchall()
+        except:
+            db.rollback()
+            traceback.print_exc()
+        finally:
+            db.close()
+
+        return results

utils/excel.py

+# workbook相关
+from openpyxl.workbook import Workbook
+# ExcelWriter，封装了很强大的excel写的功能
+from openpyxl.writer.excel import ExcelWriter
+# 一个eggache的数字转为列字母的方法
+from openpyxl.utils import get_column_letter
+from openpyxl.reader.excel import load_workbook
+
+
+class ExcelHandel:
+
+    def __init__(self, name):
+        self.name = name
+
+    def write_to_excel(self, title, content, file_name):
+
+        wb = Workbook()
+        ew = ExcelWriter(workbook=wb)
+        dest_filename = file_name.decode('utf-8')
+        ws = wb.worksheets[0]
+        ws.title = "name"
+        # 标题行
+        for h_x in range(1, len(title) + 1):
+            h_col = get_column_letter(h_x)
+            # print h_col
+            ws.cell('%s%s' % (h_col, 1)).value = '%s' % (title[h_x - 1])
+        # 写第二行及其以后的那些行
+        i = 2
+        for record in content:
+            record_list = str(record).strip().split("\t")
+            for x in range(1, len(record_list) + 1):
+                col = get_column_letter(x)
+                ws.cell('%s%s' % (col, i)).value = '%s' % (record_list[x - 1].decode('utf-8'))
+            i += 1
+        # 写文件
+        ew.save(filename=dest_filename)

utils/excel_handler.py

+import xlwt
+
+
+class ExcelHandler:
+
+    """
+    result_type 1:嵌套字典 2.列表字典
+    """
+    @staticmethod
+    def write_to_excel(title, content, result, file_name, result_type):
+
+        wb = xlwt.Workbook()
+        sheet = wb.add_sheet("结果")
+        col_index = 0
+        row_index = 0
+        # 标题填充
+        for t in title:
+            sheet.write(row_index, col_index, t)
+            col_index += 1
+        # 内容填充
+        col_index = 0
+        row_index = 1
+        # 结果遍历
+        if result_type == 1:
+            for r in result:
+                for c in content:
+                    sheet.write(row_index, col_index, result[r][c])
+                    col_index += 1
+                col_index = 0
+                row_index += 1
+        elif result_type == 2:
+            for r in result:
+                for c in content:
+                    sheet.write(row_index, col_index, r[c])
+                    col_index += 1
+                col_index = 0
+                row_index += 1
+        # 结果保存
+        wb.save(file_name + '.xls')
+

utils/msg_handler.py

+import requests
+
+
+robot_api = 'https://oapi.dingtalk.com/robot/send?access_token='
+access_token = '030f5d480c8f3a7c199fdcfc91e0e5d3e54149c12d51c4a5a3c0bc07a8be308a'
+
+
+class MsgHandler:
+
+    @staticmethod
+    def send_dd_msg(msg, robot=access_token):
+        data = {
+            "msgtype": "text",
+            "text": {
+                "content": msg
+            }
+        }
+        requests.post(robot_api + robot, json=data)