123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171 |
- import uvicorn
- import fastapi
- from fastapi.middleware.cors import CORSMiddleware
- 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
- 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")
- # db connect
- db = dataset.connect('mysql://choozmo:pAssw0rd@db.ptt.cx:3306/arkcard?charset=utf8mb4')
- table = db['users']
- # 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'
- # nft collection api
- @app.get("/collection/{userid}")
- def collection(userid):
- 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):
- # generate qr code from user id
- results = table.find_one(userid=userid)
- address = results['useraddress']
- qr = qrcode.QRCode(
- version=1,
- box_size=10,
- border=5)
- qr.add_data(address)
- qr.make(fit=True)
- img_qr = qr.make_image(fill='black', back_color='white')
- filename = "qrcode/" + userid + '.png'
- img_save = img_qr.save(filename)
- # open file
- im = Image.open(filename)
- # convert
- imgByteArr = io.BytesIO()
- im.save(imgByteArr, format='PNG')
- # decode to ascii
- x = imgByteArr.getvalue()
- base64.b64encode(x).decode('ascii')
- encode = base64.b64encode(x).decode('ascii')
- # add to db
- data = dict(base64=encode)
- result = data['base64']
- # print(data['base64'])
- # table.insert(data)
- table.update(dict(userid=userid, base=result), ['userid'])
- # follow event
- @handler.add(FollowEvent)
- def handle_follow(event):
- # get user id when follow
- real_user_id = event.source.user_id
- rand_num = '_test_000_000_002'
- user_id = event.source.user_id + rand_num
- # 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']
- # write in sql
- data = dict(userid=real_user_id, useraddress=user_address)
- table.insert(data)
- print(" Follow event: " + event.source.user_id)
- 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給對方!',
- actions=[
- URITemplateAction(
- label='打開發送頁面',
- uri='http://q.ptt.cx/a1/index-line.html?userid=' + 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/receive/' + 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/collection/' + event.source.user_id), ])
- line_bot_api.reply_message(
- event.reply_token,
- TemplateSendMessage(
- alt_text="Receive",
- template=button_template_message))
- else:
- line_bot_api.reply_message(
- event.reply_token,
- TextSendMessage(text="更多的服務內容,歡迎請上我們的官網!(http://ark.cards)"))
- if __name__ == '__main__':
- uvicorn.run("main:app", host="0.0.0.0", port=8228, reload=True)
|