python Flask与微信小程序统计管理-Toy模板网

这篇具有很好参考价值的文章主要介绍了python Flask与微信小程序统计管理。希望对大家有所帮助。如果存在错误或未考虑完全的地方，请大家不吝赐教，您也可以点击"举报违法"按钮提交疑问。

common/models/stat/StatDailyMember.py


DROP TABLE IF EXISTS `stat_daily_member`;

CREATE TABLE `stat_daily_member` (
  `id` int(11) unsigned NOT NULL AUTO_INCREMENT,
  `date` date NOT NULL COMMENT '日期',
  `member_id` int(11) NOT NULL DEFAULT '0' COMMENT '会员id',
  `total_shared_count` int(11) NOT NULL DEFAULT '0' COMMENT '当日分享总次数',
  `total_pay_money` decimal(10,2) NOT NULL DEFAULT '0.00' COMMENT '当日付款总金额',
  `updated_time` timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP COMMENT '最后一次更新时间',
  `created_time` timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP COMMENT '插入时间',
  PRIMARY KEY (`id`),
  KEY `idx_date_member_id` (`date`,`member_id`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='会员日统计';

flask-sqlacodegen 'mysql://root:root@127.0.0.1/food_db' --tables stat_daily_member --outfile "common/models/stat/StatDailyMember.py"  --flask

common/models/stat/StatDailySite.py

DROP TABLE IF EXISTS `stat_daily_site`;

CREATE TABLE `stat_daily_site` (
  `id` int(11) unsigned NOT NULL AUTO_INCREMENT,
  `date` date NOT NULL COMMENT '日期',
  `total_pay_money` decimal(10,2) NOT NULL DEFAULT '0.00' COMMENT '当日应收总金额',
  `total_member_count` int(11) NOT NULL COMMENT '会员总数',
  `total_new_member_count` int(11) NOT NULL COMMENT '当日新增会员数',
  `total_order_count` int(11) NOT NULL COMMENT '当日订单数',
  `total_shared_count` int(11) NOT NULL,
  `updated_time` timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP COMMENT '最后一次更新时间',
  `created_time` timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP COMMENT '插入时间',
  PRIMARY KEY (`id`),
  KEY `idx_date` (`date`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='全站日统计';

flask-sqlacodegen 'mysql://root:root@127.0.0.1/food_db' --tables stat_daily_site --outfile "common/models/stat/StatDailySite.py"  --flask

common/models/stat/StatDailyFood.py

DROP TABLE IF EXISTS `stat_daily_food`;

CREATE TABLE `stat_daily_food` (
  `id` int(11) unsigned NOT NULL AUTO_INCREMENT,
  `date` date NOT NULL,
  `food_id` int(11) NOT NULL DEFAULT '0' COMMENT '菜品id',
  `total_count` int(11) NOT NULL DEFAULT '0' COMMENT '售卖总数量',
  `total_pay_money` decimal(10,2) NOT NULL DEFAULT '0.00' COMMENT '总售卖金额',
  `updated_time` timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP COMMENT '最后一次更新时间',
  `created_time` timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP COMMENT '插入时间',
  PRIMARY KEY (`id`),
  KEY `date_food_id` (`date`,`food_id`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='书籍售卖日统计';

flask-sqlacodegen 'mysql://root:root@127.0.0.1/food_db' --tables stat_daily_food --outfile "common/models/stat/StatDailyFood.py"  --flask

jobs/tasks/stat/daily.py

# -*- coding: utf-8 -*-

from application import app,db
from common.libs.Helper import getFormatDate,getCurrentDate
from common.models.member.Member import Member
from common.models.pay.PayOrder import PayOrder
from common.models.stat.StatDailyFood import StatDailyFood
from common.models.stat.StatDailySite import StatDailySite
from common.models.stat.StatDailyMember import StatDailyMember
from common.models.food.WxShareHistory import WxShareHistory
from common.models.food.FoodSaleChangeLog import FoodSaleChangeLog
from sqlalchemy import func
import random
'''
python manager.py runjob -m stat/daily -a member|food|site -p 2018-07-01
'''
class JobTask():
    def __init__(self):
        pass

    def run(self, params):
        act = params['act'] if 'act' in params else ''
        date = params['param'][0] if params['param'] and len(params['param']) else getFormatDate(format="%Y-%m-%d")
        if not act:
            return

        date_from = date + " 00:00:00"
        date_to = date + " 23:59:59"
        func_params = {
            'act': act,
            'date':date,
            'date_from':date_from,
            'date_to':date_to
        }
        if act == "member":
            self.statMember( func_params )
        elif act == "food":
            self.statFood( func_params )
        elif act == "site":
            self.statSite( func_params)
        elif act == "test":
            self.test()

        app.logger.info("it's over~~")
        return

    '''
    会员统计
    '''
    def statMember(self,params):
        act = params['act']
        date = params['date']
        date_from = params['date_from']
        date_to = params['date_to']
        app.logger.info( "act:{0},from:{1},to:{2}".format( act,date_from,date_to ) )

        member_list = Member.query.all()
        if not member_list:
            app.logger.info( "no member list" )
            return

        for member_info in member_list:
            tmp_stat_member = StatDailyMember.query.filter_by( date = date,member_id = member_info.id ).first()
            if tmp_stat_member:
                tmp_model_stat_member = tmp_stat_member
            else:
                tmp_model_stat_member = StatDailyMember()
                tmp_model_stat_member.date = date
                tmp_model_stat_member.member_id = member_info.id
                tmp_model_stat_member.created_time = getCurrentDate()

            tmp_stat_pay =  db.session.query( func.sum(PayOrder.total_price).label("total_pay_money")) \
                    .filter( PayOrder.member_id  == member_info.id ,PayOrder.status == 1 )\
                    .filter( PayOrder.created_time >= date_from,PayOrder.created_time <= date_to ).first()
            tmp_stat_share_count = WxShareHistory.query.filter( PayOrder.member_id  == member_info.id  )\
                    .filter( PayOrder.created_time >= date_from,PayOrder.created_time <= date_to ).count()

            tmp_model_stat_member.total_shared_count = tmp_stat_share_count
            tmp_model_stat_member.total_pay_money = tmp_stat_pay[ 0 ] if tmp_stat_pay[ 0 ] else 0.00
            '''
            为了测试效果模拟数据
            '''
            tmp_model_stat_member.total_shared_count = random.randint(50,100)
            tmp_model_stat_member.total_pay_money = random.randint(1000,1010)
            tmp_model_stat_member.updated_time = getCurrentDate()
            db.session.add( tmp_model_stat_member )
            db.session.commit()

        return

    '''
    Food统计
    '''
    def statFood(self,params):
        act = params['act']
        date = params['date']
        date_from = params['date_from']
        date_to = params['date_to']
        app.logger.info( "act:{0},from:{1},to:{2}".format( act,date_from,date_to ) )

        stat_food_list = db.session.query(FoodSaleChangeLog.food_id, func.sum(FoodSaleChangeLog.quantity).label("total_count"),
                         func.sum(FoodSaleChangeLog.price).label("total_pay_money")) \
            .filter(FoodSaleChangeLog.created_time >= date_from, FoodSaleChangeLog.created_time <= date_to)\
            .group_by( FoodSaleChangeLog.food_id ).all()

        if not stat_food_list:
            app.logger.info("no data")
            return

        for item in stat_food_list:
            tmp_food_id = item[ 0 ]
            tmp_stat_food = StatDailyFood.query.filter_by(date=date, food_id = tmp_food_id ).first()
            if tmp_stat_food:
                tmp_model_stat_food = tmp_stat_food
            else:
                tmp_model_stat_food = StatDailyFood()
                tmp_model_stat_food.date = date
                tmp_model_stat_food.food_id = tmp_food_id
                tmp_model_stat_food.created_time = getCurrentDate()

            tmp_model_stat_food.total_count = item[1]
            tmp_model_stat_food.total_pay_money = item[2]
            tmp_model_stat_food.updated_time = getCurrentDate()

            '''
            为了测试效果模拟数据
            '''
            tmp_model_stat_food.total_count = random.randint(50, 100)
            tmp_model_stat_food.total_pay_money = random.randint(1000, 1010)

            db.session.add( tmp_model_stat_food )
            db.session.commit()

        return
    '''
    site统计
    '''
    def statSite(self,params):
        act = params['act']
        date = params['date']
        date_from = params['date_from']
        date_to = params['date_to']
        app.logger.info( "act:{0},from:{1},to:{2}".format( act,date_from,date_to ) )

        stat_pay = db.session.query(func.sum(PayOrder.total_price).label("total_pay_money")) \
            .filter(PayOrder.status == 1) \
            .filter(PayOrder.created_time >= date_from, PayOrder.created_time <= date_to).first()

        stat_member_count = Member.query.count()
        stat_new_member_count = Member.query.filter(Member.created_time >= date_from,
                            Member.created_time <= date_to).count()

        stat_order_count = PayOrder.query.filter_by( status = 1 )\
            .filter(PayOrder.created_time >= date_from, PayOrder.created_time <= date_to)\
            .count()

        stat_share_count = WxShareHistory.query.filter(WxShareHistory.created_time >= date_from
            , WxShareHistory.created_time <= date_to).count()

        tmp_stat_site = StatDailySite.query.filter_by(date=date).first()
        if tmp_stat_site:
            tmp_model_stat_site = tmp_stat_site
        else:
            tmp_model_stat_site = StatDailySite()
            tmp_model_stat_site.date = date
            tmp_model_stat_site.created_time = getCurrentDate()

        tmp_model_stat_site.total_pay_money = stat_pay[ 0 ] if stat_pay[ 0 ] else 0.00
        tmp_model_stat_site.total_new_member_count = stat_new_member_count
        tmp_model_stat_site.total_member_count = stat_member_count
        tmp_model_stat_site.total_order_count = stat_order_count
        tmp_model_stat_site.total_shared_count = stat_share_count
        tmp_model_stat_site.updated_time = getCurrentDate()
        '''
        为了测试效果模拟数据
        '''
        tmp_model_stat_site.total_pay_money = random.randint(1000, 1010)
        tmp_model_stat_site.total_new_member_count = random.randint(50, 100)
        tmp_model_stat_site.total_member_count += tmp_model_stat_site.total_new_member_count
        tmp_model_stat_site.total_order_count = random.randint(900, 1000)
        tmp_model_stat_site.total_shared_count = random.randint(1000, 2000)
        db.session.add(tmp_model_stat_site)
        db.session.commit()

    def test(self):
        import datetime
        from common.libs.Helper import getFormatDate
        now = datetime.datetime.now()
        for i in reversed( range( 1,30 ) ):
            date_before = now + datetime.timedelta( days = -i )
            date = getFormatDate( date = date_before,format = "%Y-%m-%d" )
            tmp_params = {
                'act': 'test',
                'date': date,
                'date_from': date + " 00:00:00",
                'date_to':  date + " 23:59:59"
            }
            self.testFood( date )
            self.statFood( tmp_params )
            self.statMember( tmp_params )
            self.statSite( tmp_params )

    def testFood(self,date):
        from common.models.food.Food import Food
        list = Food.query.all()
        if list:
            for item in list:
                model =  FoodSaleChangeLog()
                model.food_id =  item.id
                model.quantity =  random.randint( 1,10 )
                model.price = model.quantity * item.price
                model.member_id = 1
                model.created_time = date + " " + getFormatDate( format = "%H:%M:%S")
                db.session.add( model )
                db.session.commit()

statMember(self, params)

这段代码是一个函数statMember的实现，它接受一个参数params，其中包含了act、date、date_from和date_to等信息。函数的主要功能是统计会员信息。

首先，函数会根据传入的参数打印出act、date_from和date_to的值。然后，它会查询数据库中的所有会员信息，并进行遍历。

在遍历过程中，函数会根据日期和会员ID查询对应的每日会员统计信息StatDailyMember。如果找到了对应的记录，则将其赋值给tmp_model_stat_member；否则，创建一个新的StatDailyMember对象，并设置其日期、会员ID和创建时间。

接下来，函数会使用数据库查询语句计算该会员的总支付金额，并将结果保存在total_pay_money字段中。

最后，函数会返回统计结果。

statSite(self,params)

这段代码是一个函数statSite的定义，它接受一个参数params。函数内部首先从params中获取act、date、date_from和date_to的值，并将它们打印出来。

接下来，函数使用SQLAlchemy进行数据库查询，统计了以下几个指标：

stat_pay：计算了满足条件的支付订单的总金额。
stat_member_count：统计了会员的总数量。
stat_new_member_count：统计了在指定日期范围内新注册的会员数量。
stat_order_count：统计了在指定日期范围内已支付的订单数量。
stat_share_count：统计了在指定日期范围内的分享历史数量。

请注意，这段代码中使用了一些数据库模型（如PayOrder、Member和WxShareHistory），以及一些SQLAlchemy的查询方法（如filter()、count()和first()）来实现数据统计功能。

.count()

count()是一个用于统计字符串、列表、元组等可迭代对象中某个元素出现的次数的方法。它可以用于字符串、列表、元组等数据类型。

在字符串中，.count()方法可以统计指定字符或子字符串在字符串中出现的次数。例如：

python

string = "Hello, World!"
count = string.count("o")
print(count)  # 输出结果为2

在列表和元组中，.count()方法可以统计指定元素在列表或元组中出现的次数。例如：

python

my_list = [1, 2, 3, 4, 2, 2]
count = my_list.count(2)
print(count)  # 输出结果为3

.count()方法返回的是指定元素在可迭代对象中出现的次数。

test(self)

(imooc) [root@localhost order]# python manager.py runjob -m stat/daily -a test

加载30天的数据到数据库

这段代码是一个名为test的函数，它包含了一些操作。让我逐步解释一下：

首先，代码导入了datetime模块和getFormatDate函数。
然后，创建了一个当前时间的变量now，使用datetime.datetime.now()获取当前时间。
接下来，使用一个反向的循环，从1到29，创建一个变量date_before，表示当前时间减去i天的日期。
使用getFormatDate函数将date_before格式化为"%Y-%m-%d"的日期格式，并赋值给变量date。
创建一个临时参数字典tmp_params，包含了一些键值对。
调用了self.testFood(date)函数，传入了date作为参数。
调用了self.statFood(tmp_params)函数，传入了tmp_params作为参数。
调用了self.statMember(tmp_params)函数，传入了tmp_params作为参数。
调用了self.statSite(tmp_params)函数，传入了tmp_params作为参数。

这段代码的作用是进行一系列的测试、统计操作，根据不同的日期进行相应的处理。文章来源地址https://www.toymoban.com/news/detail-826456.html

到了这里，关于python Flask与微信小程序统计管理的文章就介绍完了。如果您还想了解更多内容，请在右上角搜索TOY模板网以前的文章或继续浏览下面的相关文章，希望大家以后多多支持TOY模板网！

python Flask与微信小程序统计管理

common/models/stat/StatDailyMember.py

common/models/stat/StatDailySite.py

common/models/stat/StatDailyFood.py

jobs/tasks/stat/daily.py

statMember(self, params)

statSite(self,params)

.count()

test(self)

觉得文章有用就打赏一下文章作者

支付宝扫一扫打赏

微信扫一扫打赏

支付宝扫一扫领取红包，优惠每天领

二维码1

二维码2

python Flask与微信小程序 统计管理

common/models/stat/StatDailyMember.py

common/models/stat/StatDailySite.py

common/models/stat/StatDailyFood.py

jobs/tasks/stat/daily.py

statMember(self, params)

statSite(self,params)

.count()

test(self)

相关文章

觉得文章有用就打赏一下文章作者

支付宝扫一扫打赏

微信扫一扫打赏

支付宝扫一扫领取红包，优惠每天领

二维码1

二维码2

python Flask与微信小程序统计管理