[Python] A story about making a LINE Bot with a practical manned function on its own without using Salesforce [Messaging API]
at first
I will fold the explanation a lot. sorry. Maybe the code is also dirty. sorry. -
If you use LINE Bot for business purposes, which is probably used by almost everyone in Japan (?), You need a manned function, right? (Perhaps) Therefore, we will build a reservation system that can be manned as easily as possible! Do! !! !!
goal
Conclusion: This is the product.
LINEBOT: Reservation system
I wondered if it would be easier to understand what kind of code was actually made by looking at the actual code! The operator is connected to me (laughs) Manned functions are possible by linking the database and PUSH Message! !! With the free version, you can only do up to 1000 cases a month, so if you actually use it for business, you should use it with a paid light plan.
Required installation, etc.
Typical
pip install Flask
pip install line-bot-sdk
pip install flask_sqlalchemy
pip install Pillow
・
・
Others
Directory structure
It is like this. There may be some breaks, but I am eager to divide the actions into minimum parts and manage them in an easy-to-understand manner.
├── projects
├── static
└── contents
├── FollowAction.py
├── ImageAction.py
├── PostBackAction.py
└── TextAction.py
└── server.py
└── Settings.py
└── config.ini
『server.py』 Here, I play the role of setting up a server, but I also design the database here. It's better to divide it, This time, it's okay.
server.py
# coding:utf-8
from flask import Flask, request, abort
from flask_sqlalchemy import SQLAlchemy
from linebot import (
LineBotApi, WebhookHandler
)
from linebot.exceptions import (
InvalidSignatureError
)
from linebot.models import (
MessageEvent, TextMessage, TextSendMessage, PostbackEvent, ButtonsTemplate, PostbackTemplateAction, TemplateSendMessage, FollowEvent
, LocationSendMessage, LocationMessage, ImagemapSendMessage, MessageTemplateAction, ConfirmTemplate, DatetimePickerAction, DatetimePickerTemplateAction
, ImageMessage, ImageSendMessage, FlexSendMessage, BoxComponent, TextComponent, BubbleContainer, ButtonComponent,MessageAction
, CarouselContainer, URIAction, ImageComponent, PostbackAction, CameraAction, CameraRollAction, QuickReplyButton, QuickReply
)
import re
from PIL import Image
from io import BytesIO
import configparser
#Each action module
from contents import FollowAction, ImageAction, TextAction, PostBackAction
app = Flask(__name__)
""" ------------------------Read configuration file---------------------------"""
#Read configuration file
cfg = configparser.ConfigParser()
cfg.read('config.ini')
#SQLALCHEMY
url = cfg["SQL"]["URL"]
app.config['SQLALCHEMY_DATABASE_URI'] = url
db = SQLAlchemy(app)
#LINE settings
chansec = cfg["LINE"]["SECRET"]
acctoken = cfg["LINE"]["TOKEN"]
line_bot_api = LineBotApi(acctoken)
handler = WebhookHandler(chansec)
superuser_securekey = cfg["SUPERUSER"]["SECUREKEY"]
superuser_securepwd = cfg["SUPERUSER"]["SECUREPASS"]
#1 is valid whether to enable parrot return
Parrot_return = False
#Domain server URL
base_url = cfg["SERVER"]["URL"]
"""--------------------------------------------------------------------------------------------"""
"""--------------------------------Database definition----------------------------"""
class lineuser(db.Model):#Database for general users
__tablename__ = "lineuser"
__table_args__ = {'mysql_collate': 'utf8_general_ci'}
user_id = db.Column(db.String(80), primary_key=True)
username = db.Column(db.String(255))
tel = db.Column(db.String(255))
plan = db.Column(db.String(255))
usermessage = db.Column(db.String(255))
step = db.Column(db.Integer)
status = db.Column(db.String(255))
retention = db.Column(db.String(255))
requested_at = db.Column(db.String(255))
def __init__(self, user_id, username, tel, plan, usermessage, step, status, retention, requested_at):
self.user_id = user_id
self.username = username
self.tel = tel
self.plan = plan
self.usermessage = usermessage
self.step = step
self.retention = retention
self.status = status
self.requested_at = requested_at
def __repr__(self):
return '<lineuser %r>' % self.user_id
class administrator(db.Model):#Administrator database
__tablename__ = "administrator"
__table_args__ = {'mysql_collate': 'utf8_general_ci'}
admin_id = db.Column(db.String(80), primary_key=True)
admin_name = db.Column(db.String(255))
password = db.Column(db.String(255))
adminstatus = db.Column(db.String(255))
to_user = db.Column(db.String(255))
page_options = db.Column(db.Integer)
def __init__(self, admin_id, admin_name, password, adminstatus, to_user, page_options):
self.admin_id = admin_id
self.admin_name = admin_name
self.password = password
self.adminstatus = adminstatus
self.to_user = to_user
self.page_options = page_options
def __repr__(self):
return '<administrator %r>' % self.admin_id
"""---------------------------------------------------------------------------------------------------"""
"""--------------------------------Messaging API processing----------------------------"""
#Where to receive requests from LINE
@app.route("/callback", methods=['POST'])
def callback():
signature = request.headers['X-Line-Signature']
body = request.get_data(as_text=True)
app.logger.info("Request body: " + body)
try:
handler.handle(body, signature)
except InvalidSignatureError:
abort(400)
return 'OK'
#If you are followed
@handler.add(FollowEvent)
def on_follow(event):
FollowAction.Start(line_bot_api, event, db, lineuser, administrator)
#If a message comes
@handler.add(MessageEvent, message=TextMessage)
def on_message(event):
TextAction.Start(line_bot_api, event, db, lineuser, administrator, Parrot_return)
#If you press a button
@handler.add(PostbackEvent)
def on_postback(event):
PostBackAction.Start(line_bot_api, event, db, lineuser, administrator)
#When the image is sent
@handler.add(MessageEvent, message=ImageMessage)
def Image_message(event):
ImageAction.Start(line_bot_api, event, db, lineuser, administrator)
"""----------------------------------------------------------------------------------------------------"""
if __name__ == "__main__":
app.run(port=????)#Specify your favorite port number
Configuration files "Settings.py" and config.ini
import configparser
cfg = configparser.ConfigParser()
cfg['DEFAULT'] = {
'debug': True
}
cfg['LINE'] = {
'SECRET': "Here is the LINE secret key"
, 'TOKEN': "Access token here"
}
cfg["SQL"] = {
"URL": "Describe the URI of the DB used in SQLALCHEMY here"
}
cfg["SUPERUSER"] = {
"SECUREKEY": "KEY for administrator use"
, "SECUREPASS": "PASS for administrator use"
}
cfg["SERVER"] = {
"URL": "Server URL"
}
with open('config.ini', 'w') as config_file:
cfg.write(config_file)
Make a DB table
make
>>python Settings.py
>>python
from server import db
db.create_all()
exit()
『TextAction.py』 This is the process when the following code, TEXT input is received from LINE.
TextAction.py
from linebot import (
LineBotApi, WebhookHandler
)
from linebot.exceptions import (
InvalidSignatureError
)
from linebot.models import (
MessageEvent, TextMessage, TextSendMessage, PostbackEvent, ButtonsTemplate, PostbackTemplateAction, TemplateSendMessage, FollowEvent
, LocationSendMessage, LocationMessage, ImagemapSendMessage, MessageTemplateAction, ConfirmTemplate, DatetimePickerAction, DatetimePickerTemplateAction
, ImageMessage, ImageSendMessage, FlexSendMessage, BoxComponent, TextComponent, BubbleContainer, ButtonComponent,MessageAction
, CarouselContainer, URIAction, ImageComponent, PostbackAction, CameraAction, CameraRollAction, QuickReplyButton, QuickReply
)
import re
import configparser
""" ------------------------Read configuration file---------------------------"""
#Read configuration file
cfg = configparser.ConfigParser()
cfg.read('config.ini')
superuser_securekey = cfg["SUPERUSER"]["SECUREKEY"]
superuser_securepwd = cfg["SUPERUSER"]["SECUREPASS"]
"""-----------------------------------------------------------------------------------------"""
def Start(line_bot_api, event, db, lineuser, administrator, Parrot_return):
txt = event.message.text
user_id = event.source.user_id
reply_token = event.reply_token
if Parrot_return == 1:
line_bot_api.reply_message(reply_token, TextSendMessage(text=txt))
else:
if db.session.query(administrator).filter(administrator.admin_id == user_id).count() == True and txt == "Administrator use":
user = db.session.query(lineuser).filter_by(user_id=user_id).first()
if user.status == "end":
line_bot_api.push_message(to=user_id, messages=TextSendMessage(text="Start using it as an administrator."))
user.status = "admin"
db.session.add(user)
db.session.commit()
else:
line_bot_api.reply_message(reply_token, TextSendMessage(text="Please wait until you complete normal use!"))
if not db.session.query(lineuser).filter(lineuser.user_id == user_id).count():
reg = lineuser(user_id, None, None, None, None, 0, None, None, None)
db.session.add(reg)
db.session.commit()
user = db.session.query(lineuser).filter_by(user_id=user_id).first()
if txt == superuser_securekey and user.status == "end":
if not db.session.query(administrator).filter(administrator.admin_id == user_id).count():
user.status = "preadmin"
db.session.add(user)
db.session.commit()
line_bot_api.push_message(to=user_id,
messages=TextSendMessage(text="One more step! Please tell me your password.")) #One step before administrator registration
else:
line_bot_api.push_message(to=user_id,
messages=TextSendMessage(text="You already have administrator privileges.")) #I'm already registered as an administrator
#############Status check and correction confirmation##########
status = user.status
if status == None:
user.status = "name"
db.session.add(user)
db.session.commit()
line_bot_api.reply_message(reply_token, TextSendMessage(text="Please tell me your name!"))
elif status == "name":
yestxt = "Yes"
notxt = "No"
bubble = BubbleContainer(
body=BoxComponent(
layout='horizontal', margin="xs",
contents=[
TextComponent("『%Is it okay with "s"?" % txt, size='md', wrap=True, type="text", margin="xs", )]),
footer=BoxComponent(
layout="horizontal", spacing="sm",
contents=[ButtonComponent(style="link", height="sm",
action=PostbackAction(label=yestxt,
data="name@@@0@@@" + txt,
display_text=yestxt)),
ButtonComponent(style="link", height="sm",
action=PostbackAction(label=notxt, data="no@@@0",
display_text=notxt))], flex=0))
image_confirm_to_message = FlexSendMessage(alt_text="Confirm message", contents=bubble)
line_bot_api.reply_message(reply_token, image_confirm_to_message)
elif status == "tel":
yestxt = "Yes"
notxt = "No"
bubble = BubbleContainer(
body=BoxComponent(
layout='horizontal', margin="xs",
contents=[
TextComponent("The phone number is "%s ”is it all right?" % txt, size='md', wrap=True, type="text", margin="xs", )]),
footer=BoxComponent(
layout="horizontal", spacing="sm",
contents=[ButtonComponent(style="link", height="sm",
action=PostbackAction(label=yestxt,
data="tel@@@1@@@" + txt,
display_text=yestxt)),
ButtonComponent(style="link", height="sm",
action=PostbackAction(label=notxt, data="no@@@1",
display_text=notxt))], flex=0))
image_confirm_to_message = FlexSendMessage(alt_text="Confirm message", contents=bubble)
line_bot_api.reply_message(reply_token, image_confirm_to_message)
elif status == "plan":
plan_ls = ["A plan", "B plan", "C plan", "GOLD plan"]
if txt in plan_ls:
yestxt = "Yes"
notxt = "No"
bubble = BubbleContainer(
body=BoxComponent(
layout='horizontal', margin="xs",
contents=[
TextComponent("The plan you want is "%s ”is it all right?" % txt, size='md', wrap=True, type="text", margin="xs", )]),
footer=BoxComponent(
layout="horizontal", spacing="sm",
contents=[ButtonComponent(style="link", height="sm",
action=PostbackAction(label=yestxt,
data="plan@@@2@@@" + txt,
display_text=yestxt)),
ButtonComponent(style="link", height="sm",
action=PostbackAction(label=notxt, data="no@@@2",
display_text=notxt))], flex=0))
image_confirm_to_message = FlexSendMessage(alt_text="Confirm message", contents=bubble)
line_bot_api.reply_message(reply_token, image_confirm_to_message)
elif status == "message":
yestxt = "Yes"
notxt = "No"
bubble = BubbleContainer(
body=BoxComponent(
layout='horizontal', margin="xs",
contents=[
TextComponent("The final message is "%s ”is it all right?" % txt, size='md', wrap=True, type="text", margin="xs", )]),
footer=BoxComponent(
layout="horizontal", spacing="sm",
contents=[ButtonComponent(style="link", height="sm",
action=PostbackAction(label=yestxt,
data="message@@@3@@@" + txt,
display_text=yestxt)),
ButtonComponent(style="link", height="sm",
action=PostbackAction(label=notxt, data="no@@@3",
display_text=notxt))], flex=0))
image_confirm_to_message = FlexSendMessage(alt_text="Confirm message", contents=bubble)
line_bot_api.reply_message(reply_token, image_confirm_to_message)
elif status == "end":
if txt == "Connect":
yestxt = "Yes"
notxt = "No"
bubble = BubbleContainer(
body=BoxComponent(
layout='horizontal', margin="xs",
contents=[
TextComponent("Do you want to apply for connection to the operator?", size='md', wrap=True, type="text", margin="xs", )]),
footer=BoxComponent(
layout="horizontal", spacing="sm",
contents=[ButtonComponent(style="link", height="sm",
action=PostbackAction(label=yestxt,
data="end@@@4@@@" + txt,
display_text=yestxt)),
ButtonComponent(style="link", height="sm",
action=PostbackAction(label=notxt, data="no@@@1",
display_text=notxt))], flex=0))
image_confirm_to_message = FlexSendMessage(alt_text="Verification", contents=bubble)
line_bot_api.reply_message(reply_token, image_confirm_to_message)
else:
line_bot_api.reply_message(reply_token, TextSendMessage(text="To apply for a connection to the operator, enter "Connect"."))
elif status == "preadmin" and txt != superuser_securekey:
if txt == cfg["SUPERUSER"]["SECUREPASS"]:
add_admin = administrator(user_id, None, 0, None, None, None)
db.session.add(add_admin)
db.session.commit()
user.status = "end"
db.session.add(user)
db.session.commit()
line_bot_api.reply_message(reply_token, TextSendMessage(text="I have granted you as an administrator. Please enter "Use as an administrator"."))
else:
user.status = "end"
db.session.add(user)
db.session.commit()
line_bot_api.reply_message(reply_token, TextSendMessage(text="Please enter again from the beginning."))
elif status == "response":
text = txt + " --from %s"%user.username
line_bot_api.push_message(to=user.retention, messages=TextSendMessage(text=text))
elif status == "admin":
admin = db.session.query(administrator).filter_by(admin_id=user_id).first()
#Enter the administrator name
if admin.admin_name == None and admin.adminstatus != "admin_name":
admin.adminstatus = "admin_name"
db.session.add(user)
db.session.commit()
line_bot_api.push_message(to=user_id, messages=TextSendMessage(text="Please enter the administrator's name."))
#Confirmation of administrator name input
elif admin.adminstatus == "admin_name":
qtxt = "Administrator: "%Are you sure you like "s"?" % txt
yestxt = "Yes"
notxt = "No"
bubble = BubbleContainer(
body=BoxComponent(
layout='horizontal',
margin="xs",
contents=[TextComponent(text=qtxt, size='md', wrap=True, type="text", margin="xs", )]),
footer=BoxComponent(
layout="horizontal",
spacing="sm",
contents=[ButtonComponent(style="link", height="sm",
action=PostbackAction(label=yestxt,
data="admin_name@@@1@@@" + txt,
display_text=yestxt)),
ButtonComponent(style="link", height="sm",
action=PostbackAction(label=notxt, data="no@@@1@@@no",
display_text=notxt))], flex=0))
message = FlexSendMessage(alt_text="Please enter the administrator name.", contents=bubble)
line_bot_api.push_message(to=user_id, messages=message)
elif txt == "Disconnect":
if admin.to_user != None:
line_bot_api.push_message(to=admin.to_user, messages=TextSendMessage(text="The connection with the operator has ended."))
line_bot_api.reply_message(reply_token, TextSendMessage(text="The manned connection has been disconnected."))
to_user = db.session.query(lineuser).filter_by(user_id=admin.to_user).first()
to_user.retention = None
to_user.status = "end"
db.session.add(to_user)
db.session.commit()
admin.to_user = None
db.session.add(user)
db.session.commit()
#Receive input for "normal use"
elif txt == "Normal use":
if admin.to_user != None:
line_bot_api.reply_message(reply_token,
TextSendMessage(text="Manned support is in progress. Be sure to switch to "normal use" after turning off manned support with "disconnect"."))
else:
line_bot_api.reply_message(reply_token, TextSendMessage(text="I switched to normal use. For manned support, please enter "administrator use"."))
user.status = "end"
db.session.add(user)
db.session.commit()
to_user = db.session.query(lineuser).filter_by(user_id=user_id).first()
to_user.retention = None
to_user.status = "end"
db.session.add(to_user)
db.session.commit()
admin.to_user = None
db.session.add(user)
db.session.commit()
else:
if admin.to_user != None:
line_bot_api.push_message(to=admin.to_user, messages=TextSendMessage(text=txt))
line_bot_api.reply_message(reply_token, TextSendMessage(text="%I sent a message with s."%txt))
else:
line_bot_api.reply_message(reply_token, TextSendMessage(text="Not connected to anyone."))
Sorry for the very crude article. I will go on and on.
『PostBackAction.py』 When you receive a button input or such a special input It is a mechanism to process.
PostBackAction.py
Learn more or give us feedback
from linebot import (
LineBotApi, WebhookHandler
)
from linebot.exceptions import (
InvalidSignatureError
)
from linebot.models import (
MessageEvent, TextMessage, TextSendMessage, PostbackEvent, ButtonsTemplate, PostbackTemplateAction, TemplateSendMessage, FollowEvent
, LocationSendMessage, LocationMessage, ImagemapSendMessage, MessageTemplateAction, ConfirmTemplate, DatetimePickerAction, DatetimePickerTemplateAction
, ImageMessage, ImageSendMessage, FlexSendMessage, BoxComponent, TextComponent, BubbleContainer, ButtonComponent,MessageAction
, CarouselContainer, URIAction, ImageComponent, PostbackAction, CameraAction, CameraRollAction, QuickReplyButton, QuickReply, FillerComponent,IconComponent
)
#from .ExtraCreateMessage import admin_sight_client_list
import configparser
""" ------------------------Read configuration file---------------------------"""
#Read configuration file
cfg = configparser.ConfigParser()
cfg.read('config.ini')
base_url = cfg["SERVER"]["URL"]
def create_image_container(title_name, url):
container = BubbleContainer(
body=BoxComponent(layout="vertical", spacing="sm", margin="sm", contents=[
BoxComponent(layout="vertical", contents=[
ImageComponent(url=url, size="full", aspect_mode="cover", aspect_ratio="1:1", gravity="top"),
BoxComponent(layout="vertical", contents=[
BoxComponent(layout="vertical", contents=[
TextComponent(text=title_name, size="xl", color="#ffffff", weight="bold")
]),
BoxComponent(layout="vertical", contents=[
FillerComponent(),
BoxComponent(layout="baseline", spacing="sm",contents=[
FillerComponent(),
TextComponent(text="Make this plan", offset_top="-2px", color="#ffffff",flex=0, action=MessageAction(text=title_name)),
FillerComponent(),
]),
FillerComponent(),
], border_width="1px", border_color="#ffffff", corner_radius="4px",
height="40px", spacing="sm", margin="xxl"),
], position="absolute", offset_bottom="0px", offset_start="0px", offset_end="0px",
background_color="#03303Acc", padding_all="20px", padding_top="18px"),
], padding_all="0px"),
],padding_all="0px")
)
return container
def Start(line_bot_api, event, db, lineuser, administrator):
user_id = event.source.user_id
reply_token = event.reply_token
postback_msg = event.postback.data
msg = postback_msg.split("@@@") # @@@Change to
user = db.session.query(lineuser).filter_by(user_id=user_id).first()
step = user.step
if len(msg) == 2:
post_msg = msg[0] #YES or NO
post_num = int(msg[1]) #Question number
elif len(msg) == 3:
post_msg = msg[0]
post_num = int(msg[1])
data = msg[2]
if post_msg == "admin_name":
if post_num == 1:
ad_name = data
admin = db.session.query(administrator).filter_by(admin_id=user_id).first()
admin.admin_name = data
admin.adminstatus = None
db.session.add(admin)
db.session.commit()
line_bot_api.reply_message(reply_token, TextSendMessage(text="%Thank you for registering as an administrator, Mr. s!"%ad_name))
elif step == post_num:
if post_msg == "name":
user.username = data
user.status = "tel"
user.step = step + 1
db.session.add(user)
db.session.commit()
line_bot_api.reply_message(reply_token, TextSendMessage(text="%Thank you, Mr. s."
"Then enter your phone number!"
"* This is for demonstration purposes, so please do not enter your real phone number! !!" % data))
elif post_msg == "tel":
user.tel = data
user.step = step + 1
user.status = "plan"
db.session.add(user)
db.session.commit()
line_bot_api.reply_message(reply_token, TextSendMessage(
text="%Thank you, Mr. s. Please select the plan you want to use next." % user.username))
plan_ls = ["A plan", "B plan", "C plan", "GOLD plan"]
url_ls = ["A.png ", "B.png ", "C.png ", "GOLD.png "]
bubble_container = []
for (p, url) in zip(plan_ls, url_ls):
url_ = base_url + "static/" + url
container = create_image_container(p, url_)
bubble_container.append(container)
bubble2 = CarouselContainer(contents=bubble_container)
select_messa = FlexSendMessage(alt_text="Please select a plan.", contents=bubble2)
line_bot_api.push_message(to=user_id, messages=select_messa)
elif post_msg == "plan":
user.plan = data
user.step = step + 1
user.status = "message"
db.session.add(user)
db.session.commit()
line_bot_api.reply_message(reply_token, TextSendMessage(
text="%Thank you, Mr. s. If you have any message at the end, please enter it! If none, enter "None"." % user.username))
elif post_msg == "message":
user.usermessage = data
user.step = step + 1
user.status = "end"
db.session.add(user)
db.session.commit()
line_bot_api.reply_message(reply_token, TextSendMessage(
text="%Thank you, Mr. s. This completes the application! * If you want to connect to the operator as it is, enter "Connect"." % user.username))
### --------Manned processing---------- ####
elif post_msg == "end":
line_bot_api.reply_message(reply_token, TextSendMessage(
text="operator(Producer)I am applying for connection to. Please wait for a while."))
yestxt = "Connect"
id_ls = db.session.query(administrator.admin_id).all()
for id_ in id_ls:
try:
txt = "%An operator connection application has been received from Mr. s. Please click the "Connect" button to take action." %user.username
bubble = BubbleContainer(
body=BoxComponent(
layout='horizontal',
margin="xs",
contents=[TextComponent(txt, size='md', wrap=True, type="text", margin="xs", )]),
footer=BoxComponent(
layout="horizontal",
spacing="sm",
contents=[ButtonComponent(style="link", height="sm",
action=PostbackTemplateAction(label=yestxt,
data="demand@@@4@@@" + user_id,
display_text=yestxt))], flex=0))
admin_access_message = FlexSendMessage(alt_text="A correspondence application has come.", contents=bubble)
line_bot_api.push_message(to=id_[0], messages=admin_access_message)
except:
pass
elif post_msg == "demand":
to_user_id = data
to_user = db.session.query(lineuser).filter_by(user_id=to_user_id).first()
to_user.retention = user_id
to_user.status = "response"
db.session.add(to_user)
db.session.commit()
admin = db.session.query(administrator).filter_by(admin_id=user_id).first()
admin.to_user = to_user_id
db.session.add(admin)
db.session.commit()
adminuser = db.session.query(lineuser).filter_by(user_id=user_id).first()
adminuser.status = "admin"
db.session.add(adminuser)
db.session.commit()
adminname = admin.admin_name
line_bot_api.push_message(to=to_user_id, messages=TextSendMessage(text="Of the operator%Connected with s"%adminname))
line_bot_api.reply_message(reply_token, TextSendMessage(text="%Connected with s."%to_user.username))
###----------------------------------------------------------####
else:
line_bot_api.reply_message(reply_token, TextSendMessage(text="Once again thank you."))
else:
if user.status == "end":
line_bot_api.reply_message(reply_token, TextSendMessage(text="I canceled it."))
else:
line_bot_api.reply_message(reply_token, TextSendMessage(text="It's not up to date!"))
『ImageAction.py』 A code that reacts when an image is sent.
ImageAction.py
from linebot import (
LineBotApi, WebhookHandler
)
from linebot.exceptions import (
InvalidSignatureError
)
from linebot.models import (
MessageEvent, TextMessage, TextSendMessage, PostbackEvent, ButtonsTemplate, PostbackTemplateAction, TemplateSendMessage, FollowEvent
, LocationSendMessage, LocationMessage, ImagemapSendMessage, MessageTemplateAction, ConfirmTemplate, DatetimePickerAction, DatetimePickerTemplateAction
, ImageMessage, ImageSendMessage, FlexSendMessage, BoxComponent, TextComponent, BubbleContainer, ButtonComponent,MessageAction
, CarouselContainer, URIAction, ImageComponent, PostbackAction, CameraAction, CameraRollAction, QuickReplyButton, QuickReply, FillerComponent,IconComponent
)
#from .ExtraCreateMessage import admin_sight_client_list
import configparser
""" ------------------------Read configuration file---------------------------"""
#Read configuration file
cfg = configparser.ConfigParser()
cfg.read('config.ini')
base_url = cfg["SERVER"]["URL"]
def create_image_container(title_name, url):
container = BubbleContainer(
body=BoxComponent(layout="vertical", spacing="sm", margin="sm", contents=[
BoxComponent(layout="vertical", contents=[
ImageComponent(url=url, size="full", aspect_mode="cover", aspect_ratio="1:1", gravity="top"),
BoxComponent(layout="vertical", contents=[
BoxComponent(layout="vertical", contents=[
TextComponent(text=title_name, size="xl", color="#ffffff", weight="bold")
]),
BoxComponent(layout="vertical", contents=[
FillerComponent(),
BoxComponent(layout="baseline", spacing="sm",contents=[
FillerComponent(),
TextComponent(text="Make this plan", offset_top="-2px", color="#ffffff",flex=0, action=MessageAction(text=title_name)),
FillerComponent(),
]),
FillerComponent(),
], border_width="1px", border_color="#ffffff", corner_radius="4px",
height="40px", spacing="sm", margin="xxl"),
], position="absolute", offset_bottom="0px", offset_start="0px", offset_end="0px",
background_color="#03303Acc", padding_all="20px", padding_top="18px"),
], padding_all="0px"),
],padding_all="0px")
)
return container
def Start(line_bot_api, event, db, lineuser, administrator):
user_id = event.source.user_id
reply_token = event.reply_token
postback_msg = event.postback.data
msg = postback_msg.split("@@@") # @@@Change to
user = db.session.query(lineuser).filter_by(user_id=user_id).first()
step = user.step
if len(msg) == 2:
post_msg = msg[0] #YES or NO
post_num = int(msg[1]) #Question number
elif len(msg) == 3:
post_msg = msg[0]
post_num = int(msg[1])
data = msg[2]
if post_msg == "admin_client_list_next":
admin = db.session.query(administrator).filter_by(admin_id=user_id).first()
page_options = admin.page_options
admin.page_options = page_options + 1
db.session.add(admin)
db.session.commit()
#admin_sight_client_list(user_id)
elif post_msg == "admin_client_list_prev":
admin = db.session.query(administrator).filter_by(admin_id=user_id).first()
page_options = admin.page_options
if page_options > 0:
admin.page_options = page_options - 1
db.session.add(admin)
db.session.commit()
#admin_sight_client_list(user_id)
else:
pass
#admin_sight_client_list(user_id)
elif post_msg == "admin_name":
if post_num == 1:
ad_name = data
admin = db.session.query(administrator).filter_by(admin_id=user_id).first()
admin.admin_name = data
admin.adminstatus = None
db.session.add(admin)
db.session.commit()
line_bot_api.reply_message(reply_token, TextSendMessage(text="%Thank you for registering as an administrator, Mr. s!"%ad_name))
elif step == post_num:
if post_msg == "name":
user.username = data
user.status = "tel"
user.step = step + 1
db.session.add(user)
db.session.commit()
line_bot_api.reply_message(reply_token, TextSendMessage(text="%Thank you, Mr. s."
"Then enter your phone number!"
"* This is for demonstration purposes, so please do not enter your real phone number! !!" % data))
elif post_msg == "tel":
user.tel = data
user.step = step + 1
user.status = "plan"
db.session.add(user)
db.session.commit()
line_bot_api.reply_message(reply_token, TextSendMessage(
text="%Thank you, Mr. s. Please select the plan you want to use next." % user.username))
plan_ls = ["A plan", "B plan", "C plan", "GOLD plan"]
url_ls = ["A.png ", "B.png ", "C.png ", "GOLD.png "]
bubble_container = []
for (p, url) in zip(plan_ls, url_ls):
url_ = base_url + "static/" + url
container = create_image_container(p, url_)
bubble_container.append(container)
bubble2 = CarouselContainer(contents=bubble_container)
select_messa = FlexSendMessage(alt_text="Please select a plan.", contents=bubble2)
line_bot_api.push_message(to=user_id, messages=select_messa)
elif post_msg == "plan":
user.plan = data
user.step = step + 1
user.status = "message"
db.session.add(user)
db.session.commit()
line_bot_api.reply_message(reply_token, TextSendMessage(
text="%Thank you, Mr. s. If you have any message at the end, please enter it! If none, enter "None"." % user.username))
elif post_msg == "message":
user.usermessage = data
user.step = step + 1
user.status = "end"
db.session.add(user)
db.session.commit()
line_bot_api.reply_message(reply_token, TextSendMessage(
text="%Thank you, Mr. s. This completes the application! * If you want to connect to the operator as it is, enter "Connect"." % user.username))
### --------Manned processing---------- ####
elif post_msg == "end":
line_bot_api.reply_message(reply_token, TextSendMessage(
text="operator(Producer)I am applying for connection to. Please wait for a while."))
yestxt = "Connect"
id_ls = db.session.query(administrator.admin_id).all()
for id_ in id_ls:
try:
txt = "%An operator connection application has been received from Mr. s. Please click the "Connect" button to take action." %user.username
bubble = BubbleContainer(
body=BoxComponent(
layout='horizontal',
margin="xs",
contents=[TextComponent(txt, size='md', wrap=True, type="text", margin="xs", )]),
footer=BoxComponent(
layout="horizontal",
spacing="sm",
contents=[ButtonComponent(style="link", height="sm",
action=PostbackTemplateAction(label=yestxt,
data="demand@@@4@@@" + user_id,
display_text=yestxt))], flex=0))
admin_access_message = FlexSendMessage(alt_text="A correspondence application has come.", contents=bubble)
line_bot_api.push_message(to=id_[0], messages=admin_access_message)
except:
pass
elif post_msg == "demand":
to_user_id = data
to_user = db.session.query(lineuser).filter_by(user_id=to_user_id).first()
to_user.retention = user_id
to_user.status = "response"
db.session.add(to_user)
db.session.commit()
admin = db.session.query(administrator).filter_by(admin_id=user_id).first()
admin.to_user = to_user_id
db.session.add(admin)
db.session.commit()
adminuser = db.session.query(lineuser).filter_by(user_id=user_id).first()
adminuser.status = "admin"
db.session.add(adminuser)
db.session.commit()
adminname = admin.admin_name
line_bot_api.push_message(to=to_user_id, messages=TextSendMessage(text="Of the operator%Connected with s"%adminname))
line_bot_api.reply_message(reply_token, TextSendMessage(text="%Connected with s."%to_user.username))
###----------------------------------------------------------####
else:
line_bot_api.reply_message(reply_token, TextSendMessage(text="Once again thank you."))
else:
if user.status == "end":
line_bot_api.reply_message(reply_token, TextSendMessage(text="I canceled it."))
else:
line_bot_api.reply_message(reply_token, TextSendMessage(text="It's not up to date!"))
long! !! !! But next and last!
『FollowAction.py』 Why didn't you draw the follow action first ... Well, it's code that runs only the first time, and it doesn't matter if it's there or not.
FollowAction.py
from linebot import (
LineBotApi, WebhookHandler
)
from linebot.exceptions import (
InvalidSignatureError
)
from linebot.models import (
MessageEvent, TextMessage, TextSendMessage, PostbackEvent, ButtonsTemplate, PostbackTemplateAction, TemplateSendMessage, FollowEvent
, LocationSendMessage, LocationMessage, ImagemapSendMessage, MessageTemplateAction, ConfirmTemplate, DatetimePickerAction, DatetimePickerTemplateAction
, ImageMessage, ImageSendMessage, FlexSendMessage, BoxComponent, TextComponent, BubbleContainer, ButtonComponent,MessageAction
, CarouselContainer, URIAction, ImageComponent, PostbackAction, CameraAction, CameraRollAction, QuickReplyButton, QuickReply
)
def Start(line_bot_api, event, db, lineuser, adminstrator):
reply_token = event.reply_token
line_bot_api.reply_message(reply_token, TextSendMessage(text="Thank you for following me! It is a reservation system with a manned function!"))
user_id = event.source.user_id
#Register user information.
if not db.session.query(lineuser).filter(lineuser.user_id == user_id).count():
reg = lineuser(user_id, None, None, None, None, 0, None, None, None)
db.session.add(reg)
db.session.commit()
Thank you for your hard work
Thank you for your hard work. I also have a strong sense of writing my own notes, so I thought it would be helpful for someone.
I think there are some inefficient processes, but it's a test, so there's no problem! I thought, I made it at explosive speed. Python is the best! !! !!
Recommended Posts