# -*- coding: utf-8 -*-
import json
import datetime
from optionaldict import optionaldict
from wechatpy.client.api.base import BaseWeChatAPI
[文档]class WeChatMarketing(BaseWeChatAPI):
API_BASE_URL = "https://api.weixin.qq.com/marketing/"
[文档] def add_user_action_sets(self, _type, name, description, version="v1.0"):
"""
创建数据源
https://wximg.qq.com/wxp/pdftool/get.html?id=rkalQXDBM&pa=39
:param _type: 用户行为源类型
:param name: 用户行为源名称 必填
:param description: 用户行为源描述,字段长度最小 1 字节,长度最大 128 字节
:param version: 版本号 v1.0
:return: 数据源唯一ID
"""
return self._post(
"user_action_sets/add",
params={"version": version},
json=optionaldict(type=_type, name=name, description=description, version=version),
result_processor=lambda x: x["data"]["user_action_set_id"],
)
[文档] def get_user_action_sets(self, user_action_set_id, version="v1.0"):
"""
获取数据源信息
:param user_action_set_id: 数据源唯一ID
:param version: 版本号 v1.0
"""
return self._get(
"user_action_sets/get",
params={"version": version, "user_action_set_id": user_action_set_id},
result_processor=lambda x: x["data"]["list"],
)
[文档] def add_user_actions(self, actions=(), version="v1.0"):
"""
回传数据
https://wximg.qq.com/wxp/pdftool/get.html?id=rkalQXDBM&pa=39
:param actions: 用户行为源类型
:param version: 版本号 v1.0
"""
return self._post("user_actions/add", params={"version": version}, json={"actions": actions})
[文档] def get_user_action_set_reports(
self,
user_action_set_id,
start_date=None,
end_date=None,
time_granularity=None,
aggregation=None,
version="v1.0",
):
"""
数据源报表查询
:param user_action_set_id: 数据源唯一ID
:param start_date: 开始日期 默认今天
:param end_date: 结束日期 默认今天
:param time_granularity: 时间粒度 枚举 {'DAILY'(按天) 'HOURLY'(按小时)}
:param aggregation: 聚合纬度 枚举 {'DOMAIN' 'ACTION_TYPE'}
:param version: 版本号 v1.0
:return:
"""
today = datetime.date.today()
if start_date is None:
start_date = today
if end_date is None:
end_date = today
if isinstance(start_date, datetime.date):
start_date = start_date.strftime("%Y-%m-%d")
if isinstance(end_date, datetime.date):
end_date = end_date.strftime("%Y-%m-%d")
if time_granularity not in ("DAILY", "HOURLY"):
time_granularity = None
if aggregation not in ("DOMAIN", "ACTION_TYPE"):
aggregation = None
return self._get(
"user_action_set_reports/get",
params=optionaldict(
user_action_set_id=user_action_set_id,
date_range=json.dumps({"start_date": start_date, "end_date": end_date}),
time_granularity=time_granularity,
aggregation=aggregation,
version=version,
),
result_processor=lambda x: x["data"],
)
[文档] def get_ad_leads(
self,
start_date=None,
end_date=None,
filtering=(),
page=1,
page_size=100,
version="v1.0",
):
"""
获取朋友圈销售线索数据接口
:param start_date: 开始日期 默认今天
:param end_date: 结束日期 默认今天
:param filtering: 过滤条件 [{field: 过滤字段, operator: 操作符, values: 字段取值}]
:param page: 页码,获取指定页数据
:param page_size: 一页获取的数据条数(1-100)
:param version: 版本号 v1.0
"""
today = datetime.date.today()
if start_date is None:
start_date = today
if end_date is None:
end_date = today
if isinstance(start_date, datetime.date):
start_date = start_date.strftime("%Y-%m-%d")
if isinstance(end_date, datetime.date):
end_date = end_date.strftime("%Y-%m-%d")
return self._get(
"wechat_ad_leads/get",
params=optionaldict(
date_range=json.dumps({"start_date": start_date, "end_date": end_date}),
filtering=json.dumps(filtering) if filtering else None,
page=page,
page_size=page_size,
version=version,
),
result_processor=lambda x: x["data"],
)