基于Python MySQL实现校园疫情监控平台系统【优质毕业设计、课程设计项目】.zip

from app.service import service_select from app.core.mysql import MysqlPool import json import csv import ast import os mysqlPool = MysqlPool() # 帮助方法,合并对象 def obj_update(*config): config_temp = {} for o in config: config_temp.update(o) return config_temp # 权限集合 dict_auth = {} # 控制器父类 class Controller: def __init__(self, config): """ 构造函数 @param {Dictionary} config 配置参数 """ # 配置参数 self.config = config or {} # 添加服务 self.service = service_select(self.config["service"]) cg = { # 选择的模板那路径模板 "tpl": "./index/", # 选择的服务 "service": "user", # 注册get请求路由 "get": ["list", "view", "table"], # 注册post请求路由 "post": [], # 注册get api路由 "get_api": [ "del", "get_list", "get_obj", "count", "count_group", "sum", "sum_group", "avg", "avg_group", "list_group", "bar_group" ], # 注册post api路由 "post_api": ["add", "del", "set", "import_db", "export_db", "upload"], "interact": [], "unique": [] } if config: if "interact" in config: config["interact"].extend(cg["interact"]) else: config["interact"] = cg["interact"] if "get" in config: config["get"].extend(cg["get"]) else: config["get"] = cg["get"] if "post" in config: config["post"].extend(cg["post"]) else: config["post"] = cg["post"] if "get_api" in config: config["get_api"].extend(cg["get_api"]) else: config["get_api"] = cg["get_api"] if "post_api" in config: config["post_api"].extend(cg["post_api"]) else: config["post_api"] = cg["post_api"] if "unique" in config: config["unique"].extend(cg["unique"]) else: config["unique"] = cg["unique"] # 公共模型,用于在render（）为传递模板数据补充 def model(self, ctx, model): m = {} m.update(model) # # model_temp.user = ctx.session.user # 获取导航 service = service_select("nav") m["nav_top"] = service.Get_list({"location": "top"}) m["nav_side"] = service.Get_list({"location": "side"}) m["nav_foot"] = service.Get_list({"location": "foot"}) # 获取轮播图 service = service_select("slides") m["list_slides"] = service.Get_list({}) # 获取公告 service = service_select("notice") m["list_notice"] = service.Get_list({}, {"orderby": "`update_time` desc"}) # 交互模型接口 if ("interact" in self.config) and self.config["interact"]: m = self.model_interact(ctx, m) m["query"] = ctx.query m["body"] = ctx.body m["auth"] = ctx.auth return m # 交互对象 def interact_obj(self, ctx, o): interact = self.config["interact"] if interact: source_table = service_select( self.config["service"]).config["table"] source_field = source_table + "_id" # 评论 if "comment" in interact: service = service_select("comment") source_id = o[source_field] o["comment_list"] = service.Get_list( { "source_table": source_table, "source_field": source_field, "source_id": source_id }, { "page": 1, "size": 10, }) o["comment_len"] = service.Count({ "source_table": source_table, "source_field": source_field, "source_id": source_id, }) # 评分 if "score" in interact: service = service_select("score") source_id = o[source_field] o["score_list"] = service.Get_list( { "source_table": source_table, "source_field": source_field, "source_id": source_id }, { "page": 1, "size": 10, }) o["score_len"] = service.Avg( { "source_table": source_table, "source_field": source_field, "source_id": source_id, }, {"field": "score_num"}, ) # 收藏 if "collect" in interact: service = service_select("collect") source_id = o[source_field] o["collect_list"] = service.Get_list( { "source_table": source_table, "source_field": source_field, "source_id": source_id }, { "page": 1, "size": 10, }) o["collect_len"] = service.Count({ "source_table": source_table, "source_field": source_field, "source_id": source_id, }) # 点赞 if "praise" in interact: service = service_select("praise") source_id = o[source_field] o["praise_list"] = service.Get_list( { "source_table": source_table, "source_field": source_field, "source_id": source_id, }, { "page": 1, "size": 10, }) o["praise_len"] = service.Count({ "source_table": source_table, "source_field": source_field, "source_id": source_id, }) return o # 交互列表 def interact_list(self, ctx, list_1): interact = self.config["interact"] if interact: source_table = service_select( self.config["service"]).config["table"] source_field = source_table + "_id" # 评论数 if "comment" in interact: service = service_select("comment") for o in list_1: source_id = o[source_field] o["comment_len"] = service.Count({ "source_table": source_table, "source_field": source_field, "source_id": source_id, }) # 平均分 if "score" in interact: service = service_select("score") for o in list_1: source_id = o[source_field] o["score_len"] = service.Avg( { "source_table": source_table, "source_field": source_field, "source_id": source_id, }, {"field": "score_num"}, ) # 收藏人数 if "collect" in interact: service = service_select("collect") for o in list_1: source_id = o[source_field] o["collect_len"] = service.Count({ "source_table": source_table, "source_field": source_field, "source_id": source_id, }) # 点赞人数 if "praise" in interact: service = service_select("praise") for o in list_1: source_id = o[source_field] o["praise_len"] = service.Count({ "source_table": source_table, "source_field": source_field, "source_id": source_id, }) # 交互模型 def model_interact(self, ctx, m): if ("list" in m) and m["list"]: self.interact_list(ctx, m["list"]) elif ("obj" in m) and m["obj"]: self.interact_obj(ctx, m["obj"]) return m """ 公共参数校验 """ def Check_param(self, ctx): return True # 首页 def Index(self, ctx): """首页 @param {Object} ctx http请求上下文 @return {Object} 返回html页面 """ query = dict(ctx.query) config_plus = {} if "field" in query: field = query.pop("field") config_plus["field"] = field if "page" in query: page = query.pop("page") config_plus["page"] = page if "size" in query: size = query.pop("size") config_plus["size"] = size result_list = self.service.Get_list( query, obj_update(self.config, config_plus)) result_dict = {"list": result_list} model = self.model(ctx, result_dict) return ctx.render(self.config["tpl"] + "index" + ".html", model) def Api(self, ctx): return {"demo": "hello world!"} # 列表页面 def List(self, ctx): """ 列表页面 @param {Object} ctx http请求上下文 @return {Object} 返回html页面 """ query = dict(ctx.query) config_plus = {} if "field" in query: field = query.pop("field") config_plus["field"] = field if "page" in query: page = query.pop("page") config_plus["page"] = page if "size" in query: size = query.pop("size") config_plus["size"] = size result_list = self.service.Get_list( query, obj_update(