当前位置：首页 > news >正文

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

news 2026/2/9 17:10:18

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):passdef 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:returndate_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" )returnfor 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_memberelse:tmp_model_stat_member = StatDailyMember()tmp_model_stat_member.date = datetmp_model_stat_member.member_id = member_info.idtmp_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_counttmp_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")returnfor 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_foodelse:tmp_model_stat_food = StatDailyFood()tmp_model_stat_food.date = datetmp_model_stat_food.food_id = tmp_food_idtmp_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_siteelse:tmp_model_stat_site = StatDailySite()tmp_model_stat_site.date = datetmp_model_stat_site.created_time = getCurrentDate()tmp_model_stat_site.total_pay_money = stat_pay[ 0 ] if stat_pay[ 0 ] else 0.00tmp_model_stat_site.total_new_member_count = stat_new_member_counttmp_model_stat_site.total_member_count = stat_member_counttmp_model_stat_site.total_order_count = stat_order_counttmp_model_stat_site.total_shared_count = stat_share_counttmp_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_counttmp_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 datetimefrom common.libs.Helper import getFormatDatenow = 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 Foodlist = Food.query.all()if list:for item in list:model =  FoodSaleChangeLog()model.food_id =  item.idmodel.quantity =  random.randint( 1,10 )model.price = model.quantity * item.pricemodel.member_id = 1model.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作为参数。

这段代码的作用是进行一系列的测试、统计操作，根据不同的日期进行相应的处理。