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)