from fastapi import Depends, FastAPI, HTTPException from sqlalchemy.orm import Session # from . import crud, models, schemas import crud, models, schemas from database import SessionLocal, engine # from .database import SessionLocal, engine import json import os import threading import time from pydantic import BaseModel from flask import request, Flask try: import xml.etree.cElementTree as ET except ImportError: import xml.etree.ElementTree as ET import WXBotService from planet import planetXml from tengxunyun import tengxunyunXml from lxml import etree import xml.dom.minidom from pydantic import BaseModel import utils import requests import config from urllib.parse import quote, unquote import re models.Base.metadata.create_all(bind=engine) app = FastAPI() # Dependency def get_db(): db = SessionLocal() try: yield db finally: db.close() # 默认首页 @app.get("/") async def root(): return {"message": "Hello World"} # 微信登录 @app.get("/login") def read_users(code, wxid, db: Session = Depends(get_db)): print('code--', code) new_user = {} # 异度星球-小程序 payload = { 'appid': config.appid, 'secret': config.secret, 'js_code': code, 'grant_type': 'authorization_code' } res = requests.get(config.tokenUrl, params=payload) r = res.json() # 返回参数 session_key/unionid/errmsg/openid/errcode print('r----', r) openId = r['openid'] print('openId----', openId) db_user = crud.get_user_by_openId(db, openId=openId) print('db_user----', db_user) if db_user: # 返回用户完整数据 new_user = db_user else: if wxid: # 机器人id account_wxid = config.account_wxid # 新增用户 new_user = utils.saveWxidInfo(account_wxid, wxid, openId) # "errmsg":r['errmsg'],,"errcode":r['errcode'] # return {"session_key":r['session_key'],"unionid":r['unionid'],"openid":r['openid']} return new_user # H5保存用户h5openid与wxid @app.get("/addH5openid") def addH5openid(h5openid, wxid, db: Session = Depends(get_db)): db_user = crud.get_user_by_h5openid(db, h5openid=h5openid) print('db_user----', db_user) if db_user: # 返回用户完整数据 new_user = db_user else: if wxid: # 机器人id account_wxid = config.account_wxid # 新增用户 new_user = utils.saveWxidInfo(account_wxid=config.account_wxid,wxid=wxid, h5openid=h5openid) # "errmsg":r['errmsg'],,"errcode":r['errcode'] # return {"session_key":r['session_key'],"unionid":r['unionid'],"openid":r['openid']} return new_user # 加入微信群 @app.get("/joingroup") # 群-异度星球 49187487591@chatroom chatroom def read_users(wxid, db: Session = Depends(get_db)): # debug # chatroomId = '49187487591@chatroom' # 判断是否在群中 inGroup = False groupList = WXBotService.get_group_members_list( robot_wxid=config.account_wxid, from_wxid=config.chatroomId) print('groupList', groupList) for groupItem in groupList['data']: if groupItem['wxid'] == wxid: inGroup = True if inGroup: return {"code": 200, "success": "fail", "message": "已经在群聊中"} else: WXBotService.add_group_members_40_more( robot_wxid=config.account_wxid, from_wxid=config.chatroomId, #异度星球群 new_wxid=wxid) print('wxid--', wxid) return {"code": 200, "success": "ok", "message": "群聊邀请已发送"} # 加入微信群 @app.get("/h5group") # 群-异度星球 49187487591@chatroom chatroom def h5group(nickname, db: Session = Depends(get_db)): newtext = unquote(nickname, 'utf-8') pattern = r'收到(\w+)的在看' print(newtext) matches = re.findall(pattern, newtext) if matches: print("匹配到的结果：", matches) # 判断是否在群中 inGroup = False groupList = WXBotService.get_group_members_list( robot_wxid=config.account_wxid, from_wxid=config.chatroomId) # print('groupList', groupList) for groupItem in groupList['data']: if groupItem['nickname'] == matches: inGroup = True if inGroup: return {"code": 200, "success": "fail", "message": "已经在群聊中"} else: new_wxid = None allFriend = WXBotService.get_address_list(config.account_wxid) for friendItem in allFriend['data']["contactList"]: if friendItem['nickname'] == matches: new_wxid = friendItem['wxid'] else: # print('通过nickname查不到wxid') pass if new_wxid: WXBotService.add_group_members_40_more( robot_wxid=config.account_wxid, from_wxid=config.chatroomId, #异度星球群 new_wxid=new_wxid) print('new_wxid--', new_wxid) return {"code": 200, "success": "ok", "message": "群聊邀请已发送"} else: return {"code": 200, "success": "ok", "message": "查不到nickname"} else: print("未找到匹配项,或者有特殊字符") # 创建用户 @app.post("/users") # schemas.UserCreate决定了swagger文档 def create_user(user: schemas.User, db: Session = Depends(get_db)): print('user', user) print('typeuser', type(user)) # 检测是否存在wxid db_user = crud.get_user_by_wxid(db, user.wxid) print('db_user', db_user) if db_user: raise HTTPException(status_code=400, detail="wxid already registered") return crud.create_user(db=db, user=user) # 分页查询-用户 @app.get("/users") def read_users(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)): users = crud.get_users(db, skip=skip, limit=limit) return users # 根据openId查询用户 @app.post("/openId") def read_user(user: schemas.User, db: Session = Depends(get_db)): db_user = crud.get_user_by_openId(db, openId=user.openId) if db_user is None: raise HTTPException(status_code=404, detail="openId not found") return db_user # 根据wxid查询用户 @app.get("/users") def read_user(wxid, db: Session = Depends(get_db)): db_user = crud.get_user_by_wxid(db, wxid=wxid) if db_user is None: raise HTTPException(status_code=404, detail="User-wxid not found") return db_user # 监听机器人http-get消息 @app.get("/wxbot") def helloget(q: dict): data = q t = threading.Thread(target=kp, args=(data, )) # 通过当前线程开启新的线程去执行 t.start() return json.dumps({"Code": 0}) # 监听机器人post日志消息 @app.post("/wxbot") # def hellopost(requestData:schemas.User): def hellopost(requestData: dict): print('requestData---', requestData) data = requestData t = threading.Thread(target=kp, args=(data, )) # 通过当前线程开启新的线程去执行 t.start() return json.dumps({"Code": 0}) def kp(event_data): # unicode [\\uxxxxx] 转 emoji 表情 event_data = json.loads( WXBotService.unicode_to_emoji( json.dumps(event_data, ensure_ascii=False), 'json')) kp_log = '' # 定义鲲鹏日志消息 event_type = event_data.get('event_type') #事件类型 account_wxid = event_data.get('account_wxid') #机器人id data = event_data.get('data') msg_type = data.get('msgtype') #消息类型 msg = data.get('content') to_wxid = data.get('to_wxid') #发送给 to_name = data.get('to_name') filepath = data.get("filepath") subtype = data.get('subtype') #类型 thumbpath = data.get('thumbpath') #图片,存在电脑绝对路径 wxid = data.get('final_from_wxid') #消息来源用户 wx_name = data.get('final_from_name') group = None if data.get('from_wxid') == wxid else data.get('from_wxid') group_name = None if data.get('from_name') == wx_name else data.get( 'from_name') '''event类型判断''' if event_type == 1001: # 1001 创建WeChat进程成功，通讯连接成功 kp_log = f'创建WeChat进程成功，通讯连接成功' elif event_type == 1002: # 1002 登录成功 kp_log = f'登录成功 {data.get("nickname")}（{data.get("wxid")}）' elif event_type == 1003: # 1003 账号注销登录 kp_log = f'账号注销登录 {data.get("wxid")} {data.get("reason")}' elif event_type == 1004: # 1004 切换聊天对象事件 kp_log = f'切换聊天对象事件 {data.get("nickname")}（{data.get("wxid")}）' elif event_type == 1005: # 1005 切换联系人事件 kp_log = f'切换联系人事件 {data.get("nickname")}（{data.get("wxid")}）' elif event_type == 1007: # 1005 二维码刷新事件 kp_log = f'二维码刷新事件 {data.get("qrcode")}' elif event_type == 1008: # 1008 连接断开 kp_log = f'连接断开' elif event_type == 1009: # 1009 群员退群通知 member_list = [] for member in data.get('memberlist'): member_list.append( f'{member.get("nickname")}（{member.get("wxid")}）') kp_log = f'群员退群通知 {"，".join(member_list)} 退出了群聊 {data.get("roomname")}（{data.get("roomid")}） ' \ f'人数：{data.get("membercount")}' elif event_type == 1010: # 1010 群员进群通知 member_list = [] for member in data.get('memberlist'): member_list.append( f'{member.get("nickName")}（{member.get("userName")}）') kp_log = f'群员进群通知 {"，".join(member_list)} 加入了群聊 {data.get("roomname")}（{data.get("roomid")}）' elif event_type == 1011: # 1011 收款消息 pay_msg = ET.fromstring(msg).find('paymsg') pay_msg_type = pay_msg.findtext('PayMsgType') if pay_msg_type == '9': # 二维码收款 status = pay_msg.findtext('status') username = pay_msg.findtext('username') display_name = pay_msg.findtext('displayname') fee = pay_msg.findtext('fee') if status == '0': kp_log = f'收款消息 {display_name}（{username}） 扫描收款二维码' elif status == '1': kp_log = f'收款消息 {display_name}（{username}） 付款成功 {int(fee) / 100} 元' elif status == '2': kp_log = f'收款消息 {display_name}（{username}） 取消付款' elif pay_msg_type == '45': # 貌似有未收款转账或转账红包过期会触发 42 和 45 类型 pass # kp_log = f'收款消息 {pay_msg.findtext("fromusername")} {pay_msg.findtext("content")}' elif event_type == 1040: # 1040 撤回消息事件 if not group: kp_log = f'撤回消息事件 {wx_name}（{wxid}） 撤回了 {msg}' else: kp_log = f'撤回消息事件 {wx_name}（{wxid}） 从群 {group_name}（{group}） 撤回了 {msg}' elif event_type == 1100: # 1100 同步消息 if group: type_str = f'同步消息 群聊消息 群聊：{group_name}（{group}） 发送者：{wx_name}（{wxid}）' elif 'gh_' in wxid: type_str = f'同步消息 公众号消息 {wx_name}（{wxid}）' elif '@openim' in wxid: type_str = f'同步消息 企业微信消息 {wx_name}（{wxid}）' else: type_str = f'同步消息 私聊消息 {wx_name}（{wxid}）' if data.get('isSender') == 1: # 手机发送还是电脑发送这里判断不是很准确 if to_wxid == account_wxid: type_str = f'机器人手机端 发送的 ' + type_str else: type_str = f'机器人电脑端 发送的 ' + type_str if msg_type == 1: # 1 文本消息/自带表情 kp_log = f'{type_str} 文本消息 内容：{msg}' if msg in ['加群']: # 将好友信息=>保存到数据库 newFriendDict = WXBotService.get_friend_info( account_wxid, wxid) # print('newFriendDict[data]---',newFriendDict['data']) userClass = models.User(**(newFriendDict['data'])) # print('userClass====',type(userClass)) # crud.create_user(db=SessionLocal(),user=userClass) # 保存好友信息 # utils.saveWxidInfo(account_wxid, wxid) # 直接邀请加群>40 # WXBotService.add_group_members_40_more(robot_wxid='wxid_sfvsru738v3122', # from_wxid='49187487591@chatroom', # new_wxid='wxid_4413224132412') # 发送小程序 # utils.sendminiProgram(data, account_wxid, wxid, planetXml) # H5链接 WXBotService.send_link_message( account_wxid, wxid, title="欢迎加入微信群", desc="请查看规则", url=f'https://link.hackrobot.cn?wxid={wxid}', pic= "https://www.hackrobot.cn/wp-content/uploads/2023/05/cropped-cropped-Screenshot-2023-05-22-at-17.26.38.png" ) # if msg in ['腾讯云']: # # 发送小程序 # utils.sendminiProgram(data, account_wxid, wxid, tengxunyunXml) elif msg_type == 3: # 3 图片消息 kp_log = f'{type_str} 图片消息 内容：{filepath}' # 本地dat图片解密 decrypt_path = WXBotService.local_decrypt_image( filepath, os.path.join(os.getcwd(), '解密图片')) print(f'{type_str} 图片消息 解密路径：{decrypt_path}') elif msg_type == 34: # 34 语言消息（需要在设置中打开语言自动转文字） xml_msg = ET.fromstring(msg) kp_log = f'{type_str} 语音消息 内容：{xml_msg.find("voicetrans").get("transtext")}' # elif msg_type == 37: # 37 好友请求 # xml_msg = ET.fromstring(msg) # kp_log = f'同步消息 好友请求 {xml_msg.get("fromnickname")}（{xml_msg.get("fromusername")}） ' \ # f'验证内容：{xml_msg.get("content")}' # # 同意好友请求 # WXBotService.agree_friend_request(account_wxid, # xml_msg.get('alias'), # xml_msg.get('encryptusername'), # xml_msg.get('ticket'), # xml_msg.get('scene')) # newFriendWxid = xml_msg.get('alias') # # 将好友信息=>保存到数据库 # # newFriendDict=WXBotService.get_friend_info(account_wxid, newFriendWxid) # # userClass=models.User(**(newFriendDict['data'])) # # crud.create_user(db= SessionLocal(),user=userClass) # newMsg = f'''你好,本消息是bot回复,本人是一名开发者,副业跨境电商,爱折腾homelab,参与交流请回复: 加群''' # WXBotService.send_text_message(robot_wxid=account_wxid, # from_wxid=newFriendWxid, # msg=newMsg) elif msg_type == 42: # 42 名片消息 xml_msg = ET.fromstring(msg) kp_log = f'{type_str} 企业微信名片消息 {xml_msg.get("nickname")}（{xml_msg.get("username")}）' elif msg_type == 43: # 43 视频消息 kp_log = f'{type_str} 视频消息 内容：{filepath}' elif msg_type == 47: # 47 表情包 kp_log = f'{type_str} 表情包消息 网址：{ET.fromstring(msg).find("emoji").get("cdnurl".replace("amp;", ""))}' # 本地下载图片 down_path = WXBotService.local_download_image( msg, os.path.join(os.getcwd(), '下载图片')) print(f'{type_str} 表情包消息 下载路径：{down_path}') elif msg_type == 48: # 48 地理位置 kp_log = f'{type_str} 地理位置 内容：{msg}' elif msg_type == 49: # 49 xml消息 # 获取根节点 root = ET.fromstring(msg) xml_appmsg = root.find('appmsg') xml_title = xml_appmsg.findtext('title') xml_type = xml_appmsg.findtext('type') xml_des = xml_appmsg.findtext('des') xml_url = xml_appmsg.findtext('url') if xml_type == '1': # 1 文本 kp_log = f'收到 {wxid} xml文本消息' elif xml_type == '3': # 3 歌曲 kp_log = f'{type_str} 音乐消息 歌曲：{xml_title} 歌手：{xml_des} 网址：{xml_url}' elif xml_type == '5': # 5 链接 kp_log = f'{type_str} 链接消息 标题：{xml_title} 内容：{xml_des} 网址：{xml_url}' print("kp_log--------------------------------", msg) # 本身也是系统消息-link,会无线循环 # 判断是不是自己发的消息 # if data.get('isSender') != 1: # WXBotService.send_link_message( # account_wxid, "wxid_w2imh7ki5s4q12", "title", "dec", # "https://www.baidu.com", # "https://p1-q.mafengwo.net/s17/M00/E3/9C/CoUBXl_cawCAQWMcAAGGhEoBcRU77.jpeg") elif xml_type == '6': # 6 文件 kp_log = f'{type_str} 文件消息 文件名：{xml_title} 文件大小：{xml_des} 文件路径：{filepath}' elif xml_type == '19': # 19 合并转发聊天记录 kp_log = f'{type_str} 合并转发聊天记录 文件名：{xml_title} 文件大小：{xml_des} 文件路径：{filepath}' elif xml_type == '33': # 33 小程序 kp_log = f'{type_str} 小程序消息 小程序名：{xml_appmsg.findtext("sourcedisplayname")} 标题：{xml_title} ' \ f'appid：{xml_appmsg.findtext("weappinfo/appid")} ' \ f'pagepath：{xml_appmsg.findtext("weappinfo/pagepath")}' print("---------------------------kp_log", msg) elif xml_type == '57': # 引用消息 kp_log = f'{type_str} 引用消息 内容：{msg}' elif xml_type == '2000': # 2000 转账消息 money = xml_appmsg.findtext('wcpayinfo/feedesc').replace( '￥', '') pay_subtype = xml_appmsg.findtext('wcpayinfo/paysubtype') transfer_id = xml_appmsg.findtext('wcpayinfo/transferid') if pay_subtype == '1': kp_log = f'{type_str} 转账消息 {money} 元' # 自动接收转账 WXBotService.transfer_receipt(account_wxid, wxid, transfer_id) elif pay_subtype == '3': kp_log = f'{type_str} 已收款 {money} 元 转账' elif msg_type == 66: # 66 企业微信名片消息 xml_msg = ET.fromstring(msg) kp_log = f'企业微信名片消息 {xml_msg.get("nickname")}（{xml_msg.get("username")}）' elif msg_type == 10000: # 10000 系统消息 kp_log = f'系统消息 {wx_name}（{wxid}） {msg}' #获取通讯录 # allFriends=WXBotService.get_address_list(account_wxid) # print('allFriends',allFriends) # 发送H5链接 if not kp_log: print( '====================================================================================' ) print(event_data) print( '====================================================================================' ) else: print(kp_log)