import uvicorn import fastapi from fastapi.middleware.cors import CORSMiddleware from fastapi.responses import HTMLResponse from linebot import LineBotApi, WebhookHandler from linebot.models import ( MessageEvent, TextMessage, TextSendMessage, FollowEvent, TemplateSendMessage, ButtonsTemplate, URITemplateAction, ) import dataset import requests import json import qrcode from PIL import Image import base64, io from random import randrange import models app = fastapi.FastAPI() app.add_middleware( CORSMiddleware, allow_origins=['*'], allow_credentials=True, allow_methods=["*"], allow_headers=["*"], ) # bot config line_bot_api = LineBotApi("SJT7VPT4RMQFLcS27jQBy3FcC24gtDrkcwJWZ5Xzqesr5T78LOKudHEJzt0k3b2S7n4KPwf27J7DVz2c8NQ4plSaaQylEeB1cYrfejaE/RPG/lCIQBYe4iBTzo26s4i2PcmT89837per/lTyvhVIKAdB04t89/1O/w1cDnyilFU=") handler = WebhookHandler("411ae3ef7e766739ed2c2c27b249d010") # callback event @app.post("/callback") async def callback(request: fastapi.Request): signature = request.headers['X-Line-Signature'] body = await request.body() handler.handle(body.decode('utf-8'), signature) return 'OK' # follow event @handler.add(FollowEvent) def handle_follow(event): # get user id when follow real_user_id = event.source.user_id rand_num = str(randrange(99999)) # db connect db = dataset.connect('mysql://choozmo:pAssw0rd@db.ptt.cx:3306/arkcard?charset=utf8mb4') table = db['users'] print(rand_num) user_id = event.source.user_id + rand_num result = table.find_one(userid=real_user_id) if result: line_bot_api.reply_message( event.reply_token, TextSendMessage(text='很高興再見到您！')) else: # create user account url = 'https://nft-api-staging.joyso.io/api/v1/accounts' headers = {'Authorization': 'Basic bmZ0OmMxOTEzOWMzYjM3YjdjZWU3ZmY3OTFiZGU3NzdjZWNl'} # setup for temp use (unique id) data = 'uid=' + user_id r = requests.post(url=url, headers=headers, data=data) # extract the account address dict_str = json.loads(r.text) user_account = dict_str['account'] user_address = user_account['address'] # generate qr code from user id qr = qrcode.QRCode( version=1, box_size=10, border=5) qr.add_data(user_address) qr.make(fit=True) img_qr = qr.make_image(fill='black', back_color='white') filename = "/var/www/ArkCard-Linebot/ArkCard-web/qrcode/" + real_user_id + '.png' img_save = img_qr.save(filename) # add to db result = "/qrcode/"+str(real_user_id)+'.png' data = dict(userid=real_user_id, useraddress=user_address, base=result) table.insert(data) line_bot_api.reply_message( event.reply_token, TextSendMessage(text='歡迎加入好友')) # message handler @handler.add(MessageEvent, message=TextMessage) def message(event): if '我要發送' in event.message.text: button_template_message = ButtonsTemplate( title=' ', text='點擊並打開收藏的NFT，可以選擇想要發送的NFT給對方！', actions=[ URITemplateAction( label='打開發送頁面', uri='http://ark.cards/collect.html?' + event.source.user_id),]) line_bot_api.reply_message( event.reply_token, TemplateSendMessage( alt_text="Receive", template=button_template_message)) elif '我要接收' in event.message.text: button_template_message = ButtonsTemplate( title=' ', text='點擊並打開接收頁面，即可分享接收地址給對方！', actions=[ URITemplateAction( label='打開接收頁面', uri='http://ark.cards/qr-code.html?' + event.source.user_id),]) line_bot_api.reply_message( event.reply_token, TemplateSendMessage( alt_text="Receive", template=button_template_message)) elif 'NFT商店' in event.message.text: button_template_message = ButtonsTemplate( title=' ', text='點擊並打開NFT商品頁，就可以購買您所想要的NFT商品哦！', actions=[ URITemplateAction( label='打開NFT商品頁面', uri='http://ark.cards/shop.html?' + event.source.user_id),]) line_bot_api.reply_message( event.reply_token, TemplateSendMessage( alt_text="Receive", template=button_template_message)) else: button_template_message = ButtonsTemplate( title=' ', text='更多的服務內容，歡迎請上我們的官網！', actions=[ URITemplateAction( label='ArkCard的官網', uri='http://ark.cards'),]) line_bot_api.reply_message( event.reply_token, TemplateSendMessage( alt_text="Receive", template=button_template_message)) # nft collection api @app.get("/collection/{userid}") def collection(userid): # db connect db = dataset.connect('mysql://choozmo:pAssw0rd@db.ptt.cx:3306/arkcard?charset=utf8mb4') table1 = db['users'] table2 = db['nft'] if table1.find_one(userid=userid) and not table2.find_one(userid=userid): return "您尚未擁有NFT商品，歡迎前往NFT商店購買！" elif not table2.find_one(userid=userid): return "使用者不存在" else: xname = userid sql = 'SELECT a.* FROM nft a INNER JOIN (SELECT userid FROM users WHERE userid="' + xname + '") b ON b.userid = a.userid' result = db.query(sql) rows = {} i = 0 for row in result: rows[i] = row i += 1 return rows # receive handler @app.get("/receive/{userid}") def receive(userid): # db connect db = dataset.connect('mysql://choozmo:pAssw0rd@db.ptt.cx:3306/arkcard?charset=utf8mb4') table = db['users'] table.find_one(userid=userid) if not table.find_one(userid=userid): return "ERROR: User Not Found" else: result = table.find_one(userid=userid) return {"userid": result['userid'], "base": result['base'], "useraddress": result['useraddress']} # send handler @app.post("/send") async def receive(userModel : models.TransactionNft): # db connect db = dataset.connect('mysql://choozmo:pAssw0rd@db.ptt.cx:3306/arkcard?charset=utf8mb4') table2 = db['nft'] # input nftid = userModel.nftid address = userModel.address if not table2.find_one(id=nftid): print("error: nft not found") return {'msg': 'nft not found'} else: statement = 'SELECT userid FROM users WHERE useraddress ="'+address+'"' for row in db.query(statement): result = row['userid'] user_obj = table2.find_one(id=nftid) user_obj['userid'] = result table2.update(dict(user_obj), ['id']) return {'msg': 'OK'} # shop handler @app.get("/shop/{userid}") def buy(userid): # db connect db = dataset.connect('mysql://choozmo:pAssw0rd@db.ptt.cx:3306/arkcard?charset=utf8mb4') sql = 'SELECT * FROM arkcard.nft WHERE userid IS NULL LIMIT 5' result = db.query(sql) rows = {} i = 0 for row in result: rows[i] = row i += 1 return rows @app.get("/buy/{userid}/{nftid}") async def buy(userid,nftid): # db connect db = dataset.connect('mysql://choozmo:pAssw0rd@db.ptt.cx:3306/arkcard?charset=utf8mb4') table2 = db['nft'] if not table2.find_one(id=nftid): print("error: nft not found") return "該NFT商品不存在！如果有疑問，請洽網站的服務信箱！" else: user_obj = table2.find_one(id=nftid) user_obj['userid'] = userid table2.update(dict(user_obj), ['id']) return "您已購買成功！" if __name__ == '__main__': uvicorn.run("main:app", host="0.0.0.0", port=8228, reload=True)