下面是编程之家 jb51.cc 通过网络收集整理的代码片段。
编程之家小编现在分享给大家，也给大家做个参考。
#! /usr/bin/env python
# -*- coding: utf-8 -*-
import os
import time
import tornado.httpserver
import tornado.ioloop
import tornado.options
import tornado.web
import torndb
import random
import sys
from tornado.escape import json_encode
from tornado.options import define,options
define("port",default=9000,help="run on the given port",type=int)
define("mysql_host",default="127.0.0.1:3306",help="database host")
define("mysql_database",default="diancan",help="diancanle database name")
define("mysql_user",default="root",help="diancanle database user")
define("mysql_password",default="123456",help="diancanle database password")
class Application(tornado.web.Application):
def __init__(self):
handlers = [
(r"/",ShopListHandler),#外卖店列表页面
(r"/auth/register",AuthRegisterHandler),#注册页面
(r"/auth/login",AuthLoginHandler),#登录叶面
(r"/food_list/(\d+)",FoodListHandler),#选择外卖店后的进入的该外卖店的商品列表页面
(r"/FoodListSubmitHandler",FoodListSubmitHandler),#处理顾客选择某个商品后的提交
(r"/admin",AdminHanlder),#管理员对外卖店和商品管理的页面
(r"/admin/add_shop_info",AdminAddShopHandler),(r"/admin/add_shop_good",AdminAddShopGoodHandler),(r"/admin/start_diancan",AdminStartDiancanInfoHandler),(r"/admin/get_diancaninfo",AdminGetDiancanInfoHandler),(r"/admin/stop_diancan",AdminStopDiancanHandler),(r"/admin/search_order_his",AdminSearchOrderHisHandler),]
settings = dict(
template_path=os.path.join(os.path.dirname(__file__),"template"),static_path=os.path.join(os.path.dirname(__file__),"static"),# xsrf_cookies=True,cookie_secret="thisisdiancanle",login_url="/auth/login",debug=True
)
tornado.web.Application.__init__(self,handlers,**settings)
class BaseHandler(tornado.web.RequestHandler):
def get_current_user(self):
return self.get_secure_cookie("user_id")
class AuthRegisterHandler(BaseHandler):
def get(self):
self.render("register.html",warn_info="")
def post(self):
user_mail = self.get_argument("user_mail")
user_passwd = self.get_argument("user_passwd")
db = torndb.Connection('127.0.0.1','dinning','root','123456')
select_sql = "SELECT * FROM user WHERE USER_NAME = '%s' and user_passwd = '%s'"%(user_mail,user_passwd)
query_resutl = db.get(select_sql)
if query_resutl is not None:
self.render("register.html",warn_info="请填写正确的信息")
return
sql_add_new_user = "insert into user(user_name,user_passwd,type) VALUE('%s','%s',%d)"%(user_mail,1)
db.execute(sql_add_new_user)
self.set_secure_cookie("user_id",user_mail)
self.redirect("/")
class AuthLoginHandler(BaseHandler):
def get(self):
self.render("login.html",warn_info="")
def post(self):
user_mail = self.get_argument("user_mail")
user_passwd = self.get_argument("user_passwd")
conn = torndb.Connection('127.0.0.1','123456')
select_sql = "select * from user where user_name = '%s' and user_passwd = '%s'"%(user_mail,user_passwd)
query_result = conn.get(select_sql)
if query_result is None:
self.render("login.html",warn_info="请输入正确的信息")
return ;
self.set_secure_cookie("user_id",user_mail)
self.redirect('/')
class ShopListHandler(BaseHandler):
@tornado.web.authenticated
def get(self):
recommend_food_cur_list = [u"西红柿炒蛋",u"春笋炒肉",u"红烧狮子头"];
#推荐商店的名称、商店热门商品、总销量
# recommend_shop_list = [
# ("杭州老娘舅","鱼香肉丝","1000","/food_list/1000"),# ("麦当劳","汉堡","900","/food_list/1001"),# ("五号便当","土豆丝","800","/food_list/1002"),# ("德克士","可乐","901","/food_list/1003"),# ("味捷外卖","红烧狮子头","1002","/food_list/1004")
# ]
conn = torndb.Connection('127.0.0.1','123456')
select_sql = "select shop_id,shop_name,shop_image from shop"
select_result = conn.query(select_sql)
recommend_shop_list = [];
for item in select_result:
recommend_shop_list.append((item.shop_name,"/food_list/" + str(item.shop_id),item.shop_image))
# recommend_shop_list = [
# ("杭州老娘舅","/food_list/1004")
# ]
self.render("shop_list.html",# recommend_shop_name_cur="味捷外卖",# recommend_shop_descrition_cur="中国餐饮外卖行业领导者",# recommend_food_cur_list=recommend_food_cur_list,recommend_shop_list=recommend_shop_list
# recommend_shop_name=u"杭州老娘舅",# hot_sell_food=u"杭州老娘舅",# hot_sell_sum=1009
)
class FoodListHandler(BaseHandler):
@tornado.web.authenticated
def get(self,shop_id):
conn = torndb.Connection('127.0.0.1','123456')
select_shop_name_sql = "select distinct shop_name from shop where shop_id = '%s'" %(shop_id);
select_shop_name_result = conn.get(select_shop_name_sql);
try:
shop_name_temp = select_shop_name_result['shop_name']
except:
self.redirect('/')
return
select_sql = "select food_id,food_name,food_img from shop_food where shop_id = '%s'"%(shop_id)
print select_sql
select_result = conn.query(select_sql)
food_list = []
for item in select_result:
food_list.append((item.food_name,item.food_id,item.food_img))
self.render("food_list.html",food_page_title=shop_name_temp,food_list=food_list)
# def post(self):
# self.write("test")
# # food_name = self.get_argument("food_name")
# # self.write(food_name)
class FoodListSubmitHandler(BaseHandler):
@tornado.web.authenticated
def post(self):
status_success = {
"status":True,"value":1
}
status_error = {
"status":False,"value":2
}
status_shop_error = {
"status":False,"value":3,"shop_name_should":""
}
shop_name = self.get_argument('shop_name')
food_name = self.get_argument('food_name')
food_id = self.get_argument('food_id')
user_name = self.get_secure_cookie("user_id")
conn = torndb.Connection('127.0.0.1','123456')
select_user_cur_order="select count(*) as num from order_cur where user_name = '%s' " % user_name
select_user_cur_order_result = conn.get(select_user_cur_order)
print select_user_cur_order_result
select_shop_name = "select shop_name from order_cur where user_name = 'test' "
select_shop_name_result = conn.get(select_shop_name)
try:
shop_name_should = select_shop_name_result["shop_name"]
except:
status_error["value"]=4
self.write(json_encode(status_error))
return
if shop_name_should != shop_name:
status_shop_error["shop_name_should"] = shop_name_should
self.write(json_encode(status_shop_error))
return
if select_user_cur_order_result["num"] == 0:
insert_cur_user_order_cur = "insert into order_cur(order_id,user_name,shop_name) value ('%s','%s')"% \
(str(random.randint(1,1000000)),shop_name);
print insert_cur_user_order_cur
try:
conn.execute(insert_cur_user_order_cur)
except:
self.write(json_encode(status_error))
return
else:
update_cur_user_order_cur = "update order_cur set food_name = '%s' where user_name = '%s'" %(food_name,user_name)
print update_cur_user_order_cur
try:
conn.execute(update_cur_user_order_cur)
except:
self.write(json_encode(status_error))
return
self.write(json_encode(status_success))
class AdminHanlder(BaseHandler):
@tornado.web.authenticated
def get(self):
conn = torndb.Connection('127.0.0.1','123456')
select_shop = "select distinct shop_name FROM shop"
list = conn.query(select_shop)
shop_list = []
shop_list_search = []
for item in list:
shop_list.append(item.shop_name)
shop_list_search.append(item.shop_name)
shop_list_search.append("全部")
user_list_search = []
select_user_sql = "select user_name from user"
select_user_list = conn.query(select_user_sql)
for user_item in select_user_list:
user_list_search.append(user_item.user_name)
user_list_search.append("全部")
self.render("admin.html",shop_list=shop_list,order_list=[],shop_list_search=shop_list_search,user_list_search=user_list_search)
class AdminAddShopHandler(BaseHandler):
def post(self):
status_success = {
"status":True,"value":2
}
shop_name = self.get_argument('shop_name')
shop_addr = self.get_argument('shop_addr')
conn = torndb.Connection('127.0.0.1','123456')
add_shop_sql = "insert into shop(shop_name,shop_addr) VALUE ('%s','%s')"%(shop_name,shop_addr)
try:
conn.execute(add_shop_sql)
except:
self.write(json_encode(status_error))
return
self.write(json_encode(status_success))
class AdminAddShopGoodHandler(BaseHandler):
def post(self):
status_success = {
"status":True,"value":1
}
status_error = {
"status":False,"value":2
}
shop_name = self.get_argument('shop_name')
shop_food = self.get_argument('shop_food')
conn = torndb.Connection('127.0.0.1','123456')
get_shop_addr = conn.get("select shop_id,shop_addr from shop where shop_name = '%s'" %(shop_name))
add_shop_sql = "insert into shop_food(shop_id,shop_addr,food_name) VALUE \
('%d','%s')"%(get_shop_addr["shop_id"],get_shop_addr['shop_addr'],shop_food)
print add_shop_sql
try:
conn.execute(add_shop_sql)
except:
self.write(json_encode(status_error))
return
self.write(json_encode(status_success))
class AdminStartDiancanInfoHandler(BaseHandler):
def post(self):
status_success = {
"status":True,"value":1
}
status_error = {
"status":False,"value":2
}
try:
shop_name = self.get_argument('shop_name')
except:
self.write(json_encode(status_error))
print "here is called"
return
conn = torndb.Connection('127.0.0.1','123456')
add_order_cur_sql = "insert into order_cur(order_id,shop_name) VALUE \
('%s','%s')"%(0,"test","test_food",shop_name)
print add_order_cur_sql
try:
conn.execute(add_order_cur_sql)
except:
self.write(json_encode(status_error))
return
self.write(json_encode(status_success))
class AdminGetDiancanInfoHandler(BaseHandler):
def post(self):
status_success = {
"status":True,"shop_name":"","result_list":[]
}
status_error = {
"status":False,"result_list":[]
}
# {"user_name":"user_name1","food_name":"food_name1"},# {"user_name":"user_name2","food_name":"food_name2"},conn = torndb.Connection('127.0.0.1','123456')
select_order_cur_shop_name_sql = "select distinct shop_name from order_cur where user_name != 'test'"
select_order_cur_shop_name_result = conn.get(select_order_cur_shop_name_sql)
try:
shop_name = select_order_cur_shop_name_result["shop_name"]
except:
print "order cur is empty"
return;
status_success["shop_name"] = shop_name
status_error["shop_name"] = shop_name
select_order_cur = "select user_name,food_name from order_cur where user_name != 'test'"
select_order_list = conn.query(select_order_cur)
for item_order in select_order_list:
status_success["result_list"].append(item_order)
if len(status_success["result_list"]) == 0:
self.write(status_error)
return
self.write(status_success)
class AdminStopDiancanHandler(BaseHandler):
def post(self):
# shop_name = self.get_argument('shop_name')
# shop_addr = self.get_argument('shop_addr')
conn = torndb.Connection('127.0.0.1','123456')
order_cur_select = "select order_id,shop_name from order_cur where user_name != 'test'"
order_cur_list = conn.query(order_cur_select)
for order_cur_item in order_cur_list:
order_his_insert = "insert into order_his(order_id,shop_name) value('%s','%s')"% \
(order_cur_item.order_id,order_cur_item.user_name,order_cur_item.food_name,order_cur_item.shop_name)
print order_his_insert
conn.execute(order_his_insert)
delete_order_cur = "delete from order_cur";
conn.execute(delete_order_cur);
self.redirect('/admin')
class AdminSearchOrderHisHandler(BaseHandler):
def post(self):
pass
# result = {
# "errorid":"0"
# }
# shop_name = self.get_argument("shop_name")
# user_name = self.get_argument("user_name")
# search_date_start = self.get_argument("start_date")
# search_date_end = self.get_argument("end_date")
# print shop_name,search_date_start,search_date_end
#
# search_sql = "SELECT user_name,order_time FROM order_his where user_name = '%s'" \
# " and shop_name = '%s' and order_time > '%s' and order_time < '%s'" %\
# (user_name,search_date_end);
# conn = torndb.Connection('127.0.0.1','123456')
# conn.get(search_sql);
#
# self.write(json_encode(result))
if __name__ == '__main__':
reload(sys)
sys.setdefaultencoding('utf8')
tornado.options.parse_command_line()
http_server = tornado.httpserver.HTTPServer(Application())
http_server.listen(options.port)
tornado.ioloop.IOLoop.instance().start()
以上是编程之家(jb51.cc)为你收集整理的全部代码内容，希望文章能够帮你解决所遇到的程序开发问题。
如果觉得编程之家网站内容还不错，欢迎将编程之家网站推荐给程序员好友。
版权声明：本文内容由互联网用户自发贡献，该文观点与技术仅代表作者本人。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容， 请发送邮件至 dio@foxmail.com 举报，一经查实，本站将立刻删除。