266 lines
8.9 KiB
Python
266 lines
8.9 KiB
Python
from typing import Union
|
||
from fastapi import FastAPI, Header, BackgroundTasks
|
||
from urllib.parse import unquote
|
||
from flashSMS import sendFlash
|
||
from tgMesaage import sendTg
|
||
import re
|
||
import pickle
|
||
import os
|
||
from fastapi.responses import JSONResponse
|
||
import asyncio
|
||
from pydantic import BaseModel
|
||
import time
|
||
|
||
|
||
app = FastAPI()
|
||
|
||
|
||
class Item(BaseModel):
|
||
query: Union[str, None] = None
|
||
|
||
|
||
# 判断uid是否存在
|
||
def findUid(uiddata, phoneNum):
|
||
isFind = False
|
||
newDict = {}
|
||
if os.path.exists("data.txt"):
|
||
# print('文件存在')
|
||
with open("data.txt", "rb") as f:
|
||
newDict = pickle.load(f)
|
||
if newDict[uiddata] is not None:
|
||
print("newDict[uid]", newDict[uiddata])
|
||
isFind = True
|
||
else:
|
||
newDict[uiddata] = phoneNum
|
||
with open("data.txt", "wb") as f:
|
||
pickle.dump(newDict, f)
|
||
isFind = False
|
||
return isFind
|
||
else:
|
||
# print('文件不存在')
|
||
isFind = False
|
||
newDict[uiddata] = phoneNum
|
||
with open("data.txt", "wb") as f:
|
||
pickle.dump(newDict, f)
|
||
return isFind
|
||
|
||
|
||
# findUid('test','13243889782')
|
||
|
||
|
||
# 正则查找11位手机号
|
||
def findPhoneNum(text):
|
||
pattern = re.compile(r"1[356789]\d{9}")
|
||
# strs = '小明的手机号是13987692110，你明天打给他'
|
||
result = pattern.findall(text)
|
||
print(result)
|
||
# ['13987692110']
|
||
if len(result) > 0:
|
||
return result[0]
|
||
else:
|
||
return None
|
||
|
||
|
||
def resData(inputArguments, uid):
|
||
print("inputArguments", inputArguments)
|
||
try:
|
||
inputArguments = unquote(inputArguments, "utf-8")
|
||
except:
|
||
inputArguments = ""
|
||
# 正则获取11位手机号
|
||
phonenumber = findPhoneNum(inputArguments)
|
||
print("phonenumber", phonenumber)
|
||
# # phonenumber='13243889782'
|
||
if phonenumber is not None and uid is not None:
|
||
warntext = "[异度世界],微信加群验证码:3399"
|
||
# sendTg(f"用户手机号{phonenumber},开始发送验证码")
|
||
# 判断用户uid是都触发过短信,不允许触发第2次
|
||
if findUid(uid, phonenumber) is True:
|
||
print("用户uid已存在,不触发短信")
|
||
# sendTg(f"用户{uid}已存在,不触发短信")
|
||
else:
|
||
# 发送闪信
|
||
sendFlash(phonenumber, warntext)
|
||
else:
|
||
print(f"用户输入:{inputArguments},手机号不正确,或者uid参数异常")
|
||
# sendTg(f"用户输入:{inputArguments},手机号不正确,或者uid参数异常")
|
||
|
||
|
||
def urlSend(num):
|
||
try:
|
||
inputArguments = unquote(num, "utf-8")
|
||
except:
|
||
inputArguments = ""
|
||
# 正则获取11位手机号
|
||
phonenumber = findPhoneNum(inputArguments)
|
||
print("phonenumber", phonenumber)
|
||
# # phonenumber='13243889782'
|
||
if phonenumber is not None:
|
||
warntext = "[异度世界],微信加群验证码:3399"
|
||
sendFlash(phonenumber, warntext)
|
||
else:
|
||
print(f"用户输入:{inputArguments},手机号不正确")
|
||
sendTg(f"用户输入:{inputArguments},手机号不正确,或者uid参数异常")
|
||
|
||
|
||
def returnRes(err_code, outputArguments):
|
||
# 成功的响应 0成功 -1失败
|
||
content = {
|
||
"err_code": err_code,
|
||
"data_list": [{"outputArguments": outputArguments}],
|
||
}
|
||
headers = {"Content-Type": "application/json"}
|
||
return JSONResponse(content=content, headers=headers)
|
||
|
||
|
||
@app.get("/")
|
||
# def read_root():
|
||
# return {"Hello": "World"}
|
||
def read_items(
|
||
appid: Union[str, None] = Header(default=None),
|
||
bid: Union[str, None] = Header(default=None),
|
||
requestid: Union[str, None] = Header(default=None),
|
||
uid: Union[str, None] = Header(default=None),
|
||
):
|
||
print("console :,", "appid", appid, "bid", bid, "requestid", requestid, "uid", uid)
|
||
return {"appid": appid, "bid": bid, "requestid": requestid, "uid": uid}
|
||
|
||
|
||
# 莉莉云机器人
|
||
@app.get("/lili")
|
||
# def read_root():
|
||
# return {"Hello": "World"}
|
||
def lili():
|
||
# 直接返回字符串
|
||
return "lili"
|
||
|
||
|
||
# 通过网址发送闪信
|
||
@app.get("/send")
|
||
# def read_root():
|
||
# return {"Hello": "World"}
|
||
def read_items(
|
||
background_tasks: BackgroundTasks,
|
||
num: Union[str, None] = None,
|
||
):
|
||
print("num", num)
|
||
background_tasks.add_task(urlSend, num)
|
||
content = {"err_code": 0, "data_list": [{"outputArguments": "test"}]}
|
||
headers = {"Content-Type": "application/json"}
|
||
return JSONResponse(content=content, headers=headers)
|
||
|
||
|
||
# 微信对话-用户输入
|
||
@app.get(
|
||
"/input",
|
||
)
|
||
async def getphonenum(
|
||
background_tasks: BackgroundTasks,
|
||
inputArguments: Union[str, None] = None,
|
||
appid: Union[str, None] = Header(default=None),
|
||
bid: Union[str, None] = Header(default=None),
|
||
requestid: Union[str, None] = Header(default=None),
|
||
uid: Union[str, None] = Header(default=None),
|
||
):
|
||
err_code = 0
|
||
outputArguments = "验证码发送成功,请留意短信"
|
||
print("inputArguments", inputArguments)
|
||
try:
|
||
inputArguments = unquote(inputArguments, "utf-8")
|
||
except:
|
||
inputArguments = ""
|
||
# 正则获取11位手机号
|
||
phonenumber = findPhoneNum(inputArguments)
|
||
print("phonenumber", phonenumber)
|
||
# # phonenumber='13243889782'
|
||
if phonenumber is not None and uid is not None:
|
||
warntext = "[异度世界],微信加群验证码:3399"
|
||
# sendTg(f"用户手机号{phonenumber},开始发送验证码")
|
||
# 判断用户uid是都触发过短信,不允许触发第2次
|
||
if findUid(uid, phonenumber) is True:
|
||
err_code = -1
|
||
outputArguments = "验证码已下发1次,不允许重复触发"
|
||
print("用户uid已存在,不触发短信")
|
||
# sendTg(f"用户{uid}已存在,不触发短信")
|
||
else:
|
||
# 发送闪信
|
||
# sendFlash(phonenumber, warntext)
|
||
# 后台发送
|
||
background_tasks.add_task(sendFlash, phonenumber, warntext)
|
||
|
||
# 成功的响应 0成功 -1失败
|
||
content = {
|
||
"err_code": err_code,
|
||
"data_list": [{"outputArguments": outputArguments}],
|
||
}
|
||
headers = {"Content-Type": "application/json"}
|
||
return JSONResponse(content=content, headers=headers)
|
||
|
||
else:
|
||
err_code = -1
|
||
outputArguments = "验证码发送失败,请检查手机号格式,或其他异常"
|
||
print(f"用户输入:{inputArguments},手机号不正确,或者uid参数异常")
|
||
# sendTg(f"用户输入:{inputArguments},手机号不正确,或者uid参数异常")
|
||
|
||
# 成功的响应 0成功 -1失败
|
||
content = {
|
||
"err_code": err_code,
|
||
"data_list": [{"outputArguments": outputArguments}],
|
||
}
|
||
headers = {"Content-Type": "application/json"}
|
||
return JSONResponse(content=content, headers=headers)
|
||
|
||
# # 微信对话平台一定要选择: 随机回复
|
||
|
||
|
||
# 通过网址发送闪信
|
||
@app.post("/callback")
|
||
# def read_root():
|
||
# return {"Hello": "World"}
|
||
async def callback(
|
||
# item: Item,
|
||
inputArguments: Union[str, None] = None,
|
||
appid: Union[str, None] = Header(default=None),
|
||
bid: Union[str, None] = Header(default=None),
|
||
requestid: Union[str, None] = Header(default=None),
|
||
uid: Union[str, None] = Header(default=None),
|
||
):
|
||
# time.sleep(10)
|
||
# print("item", item)
|
||
try:
|
||
inputArguments = unquote(inputArguments, "utf-8")
|
||
except:
|
||
inputArguments = ""
|
||
# 正则获取11位手机号
|
||
phonenumber = findPhoneNum(inputArguments)
|
||
print("phonenumber", phonenumber)
|
||
|
||
content = {
|
||
# "ans_node_name": "小微闲聊",
|
||
# "title": "小薇兄你好",
|
||
"answer": "你好呀123！",
|
||
"answer_type": "text",
|
||
# "bid_stat": {
|
||
# "curr_time": "20190515-18:07:37",
|
||
# "err_msg": "微信通用意图.肯定.branches.s_Any@.arguments's element.slot slot does not exist!",
|
||
# "latest_time": "20190523-16:06:33",
|
||
# "latest_valid": False,
|
||
# "up_ret": -1,
|
||
# },
|
||
# "confidence": 1,
|
||
# "from_user_name": "o9U-85tEZToQxIF8ht6o-KkagxO0",
|
||
# "status": "FAQ",
|
||
# "to_user_name": "xalsjfasf1ljasjdf1",
|
||
# "options": [
|
||
# {
|
||
# "ans_node_id": 12355896,
|
||
# "ans_node_name": "自建应用",
|
||
# "answer": "有详情入口的应用消息一般是通过api进行推送的入口不E_BREAKLINE_BREAK2、企业号或企业微信后台推送的图文消息，在微信PC侧会有“详情”入口，在企业微信手机端、PC端以及微信手机端则不会有“详情”入口。LINE_BREAKLINE_BREAK注意：微信电脑端侧仅会显示摘要和详情，不显示标题。企业微信电脑端、微信微信手机端、企业微信手机端则会显示标题和摘要。若未填写摘要，则不显示摘要。",
|
||
# "confidence": 0.795230507850647,
|
||
# "title": "自建应用消息没有详情入口",
|
||
# }
|
||
# ],
|
||
}
|
||
headers = {"Content-Type": "application/json"}
|
||
return JSONResponse(content=content, headers=headers)
|