import io from typing import List, Literal import pydash import requests from pydantic import BaseModel from wechat_token.token工具包 import sixToken from wechat_数据库.企业微信数据库.企微用户表 import wecom企微用户 from wechat_数据库.用户数据库.小程序登陆用户 import weiuser from wechat_项目工具包.数据库的工具包.更新用户积分 import miniUser用户角色 from 第三方工具目录.腾讯云 import oss上传腾讯云图片 class wecomData客户列表响应(BaseModel): errcode: int errmsg: str external_userid: List[str] from pydantic import BaseModel from typing import List, Optional class wecomData外部联系人基本信息(BaseModel): """外部联系人基本信息""" external_userid: str name: str type: int avatar: str gender: int unionid: Optional[str] = None class wecomData跟进人标签(BaseModel): """跟进人标签""" group_name: str tag_name: str type: int tag_id: Optional[str] = None class wecomData跟进成员信息(BaseModel): """跟进成员信息""" userid: str remark: Optional[str] = None description: Optional[str] = None createtime: int state: Optional[str] = None tags: List[wecomData跟进人标签] = [] remark_mobiles: List[str] = [] remark_corp_name: Optional[str] = None add_way: int oper_userid: str class wecomData获取客户详情响应(BaseModel): """获取客户详情接口返回""" errcode: int errmsg: str external_contact: Optional[wecomData外部联系人基本信息] = None follow_user: List[wecomData跟进成员信息] = [] class wecom客户好友管理(BaseModel): # demo_必填: int = Field(..., description="描述") # 必传属发生 # demo_可选: Optional[int] = None # 可选值 # demo_默认值: int = Field(default=100001, description="描述") # 有默认值 # demo_联合类型: Literal["xiao", "zizi"] = Field(..., description="描述") def 企微原用户处理接口(self, data): """""" state上级用户 = pydash.get(obj=data, path="state", default="六果") # 上级分享者的uniond值 这里是企微二维码绑定的 uninoid external_userid = pydash.get(obj=data, path="external_userid") unionid下级用户 = pydash.get(obj=data, path="unionid") print(f"state上级用户===>{state上级用户}", f"unionid下级用户===>{unionid下级用户}") 上级用户, 上级用户状态 = weiuser.objects.get_or_create(unionid=state上级用户) # 如果state的状态不为None:则创建这个用户 下级用户, 下级用户状态 = weiuser.objects.get_or_create(unionid=unionid下级用户) six六果对象, sixBool = weiuser.objects.get_or_create(unionid="六果") if 上级用户.parent is None: # 当这里企微号的state绑定的小程序用户的时候 的 上级用户为None的时候 上级用户.parent = six六果对象 上级用户.external_userid = external_userid 上级用户.user_用户当前的等级 = six六果对象.user_用户当前的等级 + 1 上级用户.codeQrCode = six六果对象.codeQrCode 上级用户.codeRemark = six六果对象.codeRemark 上级用户.codeConfigId = six六果对象.codeConfigId 上级用户.user_绑定企微UNIONID = six六果对象.user_绑定企微UNIONID 上级用户.save() # # 第二步就只需要查询，如果下级用户已经有上级了就直接忽略 if 上级用户.parent is not None: print("这里是上上级用户不为None的时候", ) # 还需要判断一下,下级用户的 父元素为six公司对象的时候,还得更新一下当前的数据,因为这个用户原来就是属于 企微员工的绑定者 if 下级用户.parent == six六果对象 and 下级用户.user_企微角色权限 != miniUser用户角色().企微员工: print(下级用户, "🍎🍎🍎🍎🍎🍎🍎🍎🍎🍎🍎🍎🍎🍎🍎🍎🍎🍎🍎判断成功了") 下级用户.parent = 上级用户 下级用户.user_用户当前的等级 = 上级用户.user_用户当前的等级 + 1 下级用户.user_绑定企微UNIONID = 上级用户.user_绑定企微UNIONID 下级用户.codeConfigId = 上级用户.codeConfigId 下级用户.codeQrCode = 上级用户.codeQrCode 下级用户.codeRemark = 上级用户.codeRemark 下级用户.user_绑定企微usrID账号 = 上级用户.user_绑定企微usrID账号 下级用户.save() if 下级用户.parent is not None: # 这里判断出下级用户有上级，直接原样处理，不做任何的更改状态 print("下级用户的父不为不为不为元素为None的时候", 上级用户) 下级用户.external_userid = external_userid 下级用户.save() if 下级用户.parent is None: print("下级用户的父元素为None的时候", 上级用户) 下级用户.parent = 上级用户 下级用户.user_用户当前的等级 = 上级用户.user_用户当前的等级 + 1 下级用户.external_userid = external_userid 下级用户.codeQrCode = 上级用户.codeQrCode 下级用户.codeRemark = 上级用户.codeRemark 下级用户.codeConfigId = 上级用户.codeConfigId 下级用户.user_绑定企微UNIONID = 上级用户.user_绑定企微UNIONID 下级用户.save() def 获取客户列表(self, userid: Literal["YuanBao1975"]) -> wecomData客户列表响应: """根据企微员工的ID,一次获取当前企微员所有的好友列表""" wecom_token = sixToken.get_wecom_token() uri = f"https://qyapi.weixin.qq.com/cgi-bin/externalcontact/list" json_data = requests.get(url=uri, params={"access_token": wecom_token, "userid": userid}) # print("获取客户列表获取客户列表获取客户列表获取客户列表获取客户列表获取客户列表获取客户列表", json_data.json()) get_data = wecomData客户列表响应(**json_data.json()) return get_data.model_dump() def 获取客户详情(self, external_userid) -> wecomData获取客户详情响应: """通过 external_userid 获取用户的详情,最主要的就是拿到用户的 unionid """ wecom_token = sixToken.get_wecom_token() uri = "https://qyapi.weixin.qq.com/cgi-bin/externalcontact/get" data = requests.get(url=uri, params={"access_token": wecom_token, "external_userid": external_userid}) print("\n🍎 🍏 🍊获取客户详情 🍎 🍏 \n", data.json()) return wecomData获取客户详情响应(**data.json()) def 批量获取客户详情并存储到数据库(self, state企微员工创建时添加的唯一值, userid企微员工账号ID: Literal["YuanBao1975"]): """根据企微员工的ID,获取当前企微员添加的所有的外 部联系人好友,并存储到数据库""" def request_recursive(userid_list=None, cursor="", ): """""" # 初始化参数 wecom_token_inner = sixToken.get_wecom_token() # 构建请求参数 uri = "https://qyapi.weixin.qq.com/cgi-bin/externalcontact/batch/get_by_user" json_data = {"userid_list": userid_list, "limit": 200, "cursor": cursor} # 发送请求 try: data = requests.post(url=uri, params={"access_token": wecom_token_inner}, json=json_data, ) response_json = data.json() # 提取当前页的外部联系人 # 获取下一页游标 external_contact_list = pydash.get(obj=response_json, path="external_contact_list") next_cursor = pydash.get(obj=response_json, path="next_cursor", default="") print(f"当前获取到 {len(external_contact_list)} 个外部联系人，总计 {len(external_contact_list)} 个") # 如果还有下一页，递归调用 for post_data in external_contact_list: userid = pydash.get(obj=post_data, path="follow_info.userid") # 添加了此外部联系人的企业成员userid 如 HuDie_1 remark = pydash.get(obj=post_data, path="follow_info.remark") # 该成员对此微信客户备注的企业名称（仅微信客户有该字段） description = pydash.get(obj=post_data, path="follow_info.description") createtime = pydash.get(obj=post_data, path="follow_info.createtime") state = pydash.get(obj=post_data, path="follow_info.state") # 企业自定义的state参数，用于区分客户具体是通过哪个「联系我」或获客链接添加；由企业通过创建「联系我」或在获客链接中添加customer_channel参数进行指定 tag_id = pydash.get(obj=post_data, path="follow_info.tag_id") remark_mobiles = pydash.get(obj=post_data, path="follow_info.remark_mobiles") # 该成员对此客户备注的手机号码，第三方应用和代开发应用均不可获取，上游企业不可获取下游企业客户该字段 add_way = pydash.get(obj=post_data, path="follow_info.add_way") # 该成员添加此客户的来源add_way为10时，对应的视频号信息 oper_userid = pydash.get(obj=post_data, path="follow_info.oper_userid") # 发起添加的userid，如果成员主动添加，为成员的userid；如果是客户主动添加，则为客户的外部联系人userid；如果是内部成员共享/管理员分配，则为对应的成员/管理员userid external_userid = pydash.get(obj=post_data, path="external_contact.external_userid") # 外部联系人的userid name = pydash.get(obj=post_data, path="external_contact.name") # 外部联系人的名称[注1] type_data = pydash.get(obj=post_data, path="external_contact.type") # 外部联系人的类型，1表示该外部联系人是微信用户，2表示该外部联系人是企业微信用户 avatar = pydash.get(obj=post_data, path="external_contact.avatar") # 外部联系人头像。第三方应用和代开发应用均不可获取，上游企业不可获取下游企业客户该字段 gender = pydash.get(obj=post_data, path="external_contact.gender") # 外部联系人性别 0-未知 1-男性 2-女性。 unionid = pydash.get(obj=post_data, path="external_contact.unionid") # 外部联系人在微信开放平台的唯一身份标识（微信unionid）， print("看看这里的值是多少看看这里的值是多少看看这里的值是多少看看这里的值是多少看看这里的值是多少看看这里的值是多少看看这里的值是多少看看这里的值是多少", state) if avatar and type_data == 1: img_file = requests.get(url=avatar).content src = oss上传腾讯云图片.上传图片(key=f"{userid}/{name}.png", file=io.BytesIO(img_file), ) avatar = src # time.sleep( 1 ) # avatar = "https://hudie.xiao.band/%E6%9C%AA%E5%91%BD%E5%90%8D.png" if bool(avatar) == False and type_data == 1: avatar = "https://hudie.xiao.band/%E6%9C%AA%E5%91%BD%E5%90%8D.png" if type_data == 1: weiuser.objects.update_or_create(unionid=unionid, defaults={"user_头像": avatar, "user_昵称": name, "external_userid": external_userid, }) wecomData = {"state": state企微员工创建时添加的唯一值, "external_userid": external_userid, "unionid": unionid} self.企微原用户处理接口(data=wecomData) wecom企微用户.objects.update_or_create(wecom企微ID=userid, wecom企微对好友添加的手机号码=remark_mobiles, wecom企微好友添加方式=oper_userid, wecom企微状态State=state企微员工创建时添加的唯一值, wecom对好友打的标签=tag_id, wecom对好友的备注信息=description, wecom对好友的备注名称=remark, wecom对好友的添加时间=createtime, wecom好友头像=avatar, wecom好友性别=gender, wecom好友昵称=name, wecom好友的ID=external_userid, wecom好友UNIONID=unionid, ) if next_cursor: print(f"发现下一页，游标: {next_cursor}") # 递归调用 request_recursive(userid_list=userid_list, cursor=next_cursor, ) except requests.exceptions.RequestException as e: print(f"网络请求异常: {e}", ) except Exception as e: print(f"处理异常: {e}") request_recursive(userid_list=[userid企微员工账号ID], ) if __name__ == "__main__": pass wai_external_userid = "wmj4zANAAAGlCZ2zNRhu1qzEMSBF0a9g" # data_wecom_客户详情数据 = wecom客户好友管理().获取客户详情(external_userid=wai_external_userid) wecom客户好友管理().批量获取客户详情并存储到数据库(state企微员工创建时添加的唯一值="oy4Jw6A1T3WkKOOhMxcbQTDSM8rs", userid企微员工账号ID="YuanBao1975") # 此方法会直接更新数据库并且保存数据