百度统计API接口Python简易SDK
2021/10/10 17:17:37
本文主要是介绍百度统计API接口Python简易SDK，对大家解决编程问题具有一定的参考价值，需要的程序猿们随着小编来一起学习吧！
由于太久没有动笔所以决定瞎发点东西最近在接入百度统计API，虽然没有对响应结果做进一步处理，还是希望能够减少大家的工作量
百度商业账号接口未对接，百度普通账号所有接口已接入，基于第三方库httpx，所有方法都是异步调用的
如果想要快速改成同步调用，只需替换如下几个关键字
- "AsyncClient" -> "Client"
- "async " -> ""
- "await " -> ""
from httpx import AsyncClient, Response
class BaiduConfig:
# docs https://tongji.baidu.com/api/manual/Chapter2/openapi.html
CLIENT_ID = '自行申请'
CLIENT_SECRET = '自行申请'
REFRESH_TOKEN = '见百度授权文档'
ACCESS_TOKEN = '见百度授权文档'
class BaiduStat(AsyncClient):
# docs https://tongji.baidu.com/api/manual/
site_id: int = None
fields = {
'pv_count': '浏览量(PV)', 'pv_ratio': '浏览量占比', 'visit_count': '来源',
'visitor_count': '访客数(UV)', 'new_visitor_count': '新访客数',
'new_visitor_ratio': '新访客比率', 'ip_count': 'IP 数',
'bounce_ratio': '跳出率', 'avg_visit_time': '平均访问时长',
'avg_visit_pages': '平均访问页数', 'trans_count': '转化次数',
'trans_ratio': '转化率', 'visit1_count': '入口页次数',
'outward_count': '贡献下游浏览量', 'exit_count': '退出页次数',
'average_stay_time': '平均停留时长', 'exit_ratio': '退出率',
'out_pv_count': '贡献浏览量',
}
area = {
'全国': 'china', '北京': 'province, 1', '上海': 'province, 2',
'天津': 'province, 3', '广东': 'province, 4', '福建': 'province, 5',
'海南': 'province, 8', '安徽': 'province, 9', '贵州': 'province, 10',
'甘肃': 'province, 11', '广西': 'province, 12', '河北': 'province, 13',
'河南': 'province, 14', '黑龙江': 'province, 15', '湖北': 'province, 16',
'湖南': 'province, 17', '吉林': 'province, 18', '江苏': 'province, 19',
'江西': 'province, 20', '辽宁': 'province, 21', '内蒙古': 'province, 22',
'宁夏': 'province, 23', '青海': 'province, 24', '山东': 'province, 25',
'山西': 'province, 26', '陕西': 'province, 27', '四川': 'province, 28',
'西藏': 'province, 29', '新疆': 'province, 30', '云南': 'province, 31',
'浙江': 'province, 32', '重庆': 'province, 33', '香港': 'province, 34',
'台湾': 'province, 35', '澳门': 'province, 36'
}
async def get(self, url, *args, **kwargs) -> Response:
# 增加token过期判断逻辑
resp = await super().get(url, *args, **kwargs)
if resp.json().get('error_code') == 111: # Access token expired
await self.refresh_token()
resp = await super().get(url, *args, **kwargs)
return resp
async def refresh_token(self):
url = 'https://openapi.baidu.com/oauth/2.0/token'
params = {
'grant_type': 'refresh_token', 'client_id': BaiduConfig.CLIENT_ID,
'client_secret': BaiduConfig.CLIENT_SECRET,
'refresh_token': BaiduConfig.REFRESH_TOKEN,
}
resp = (await self.get(url, params=params)).json()
access_token = resp.get('access_token')
refresh_token = resp.get('refresh_token')
if access_token and refresh_token:
with open(__file__, 'r+') as f:
content = f.read()
f.seek(0)
f.truncate()
content = content.replace(BaiduConfig.ACCESS_TOKEN, access_token)
content = content.replace(BaiduConfig.REFRESH_TOKEN, refresh_token)
f.write(content)
BaiduConfig.ACCESS_TOKEN = access_token
BaiduConfig.REFRESH_TOKEN = refresh_token
async def get_site_list(self):
url = 'https://openapi.baidu.com/rest/2.0/tongji/config/getSiteList'
params = {'access_token': BaiduConfig.ACCESS_TOKEN}
resp = await self.get(url, params=params)
return resp.json()
async def get_site_id(self, domain=None):
"""获取站点id
:param domain: 不传入时 选择第一个站点
:return:
"""
if self.site_id is None:
site_resp = await self.get_site_list()
for site in site_resp.get('list', []):
if domain is None:
self.site_id = site['site_id']
break
if site['domain'] == domain:
self.site_id = site['site_id']
break
return self.site_id
async def get_time_trend_report(self, site_id=None, date_range=None):
"""获取站点趋势数据 浏览量PV、访客数UV、IP数
:param site_id: 站点ID
:param date_range: 时间范围 如"20211001-20211007"
"""
metric = [
'pv_count', 'visitor_count', 'ip_count', 'bounce_ratio',
'avg_visit_time', 'trans_count'
]
if site_id is None:
site_id = await self.get_site_id()
url = 'https://openapi.baidu.com/rest/2.0/tongji/report/getData'
params = {
'site_id': site_id, 'method': 'overview/getTimeTrendRpt',
'access_token': BaiduConfig.ACCESS_TOKEN,
'metrics': ','.join(metric)
}
if date_range:
params['start_date'], params['end_date'] = date_range.split('-')
resp = await self.get(url, params=params)
return resp.json()
async def get_district_report(self, site_id=None, date_range=None):
"""获取访客地域分布 浏览量PV、访客数UV、IP数
:param site_id: 站点ID
:param date_range: 时间范围 如"20211001-20211007"
:return:
"""
if site_id is None:
site_id = await self.get_site_id()
url = 'https://openapi.baidu.com/rest/2.0/tongji/report/getData'
params = {
'site_id': site_id, 'method': 'overview/getDistrictRpt',
'access_token': BaiduConfig.ACCESS_TOKEN,
'metrics': 'pv_count'
}
if date_range:
params['start_date'], params['end_date'] = date_range.split('-')
resp = await self.get(url, params=params)
return resp.json()
async def get_common_track_report(self, site_id=None, date_range=None):
"""获取访客来源
:param site_id: 站点ID
:param date_range: 时间范围 如"20211001-20211007"
:return:
"""
if site_id is None:
site_id = await self.get_site_id()
url = 'https://openapi.baidu.com/rest/2.0/tongji/report/getData'
params = {
'site_id': site_id, 'method': 'overview/getCommonTrackRpt',
'access_token': BaiduConfig.ACCESS_TOKEN,
'metrics': 'pv_count'
}
if date_range:
params['start_date'], params['end_date'] = date_range.split('-')
resp = await self.get(url, params=params)
return resp.json()
async def get_trend_analyse(self, site_id=None, date_range=None, gran=None,
source=None, clientDevice=None, area=None,
visitor=None):
"""获取趋势分析
:param site_id: 站点ID
:param date_range: 时间范围 如"20211001-20211007"
:param gran: 时间粒度 可选值 day/hour/week/month
:param source: 来源过滤 可选值
through 直接访问
search,0 搜索引擎全部
link 外部链接
:param clientDevice: 设备过滤 可选值
pc 计算机
mobile 移动设备
:param area: 地域过滤 百度文档支持省级行政区 传入"广东"或"province,4"等效
:param visitor: 访客过滤 可选值
new 新访客
old 老访客
:return:
"""
metric = [
'pv_count', 'pv_ratio', 'visit_count', 'visitor_count',
'new_visitor_count', 'new_visitor_ratio', 'ip_count',
'avg_visit_time', 'avg_visit_pages', 'trans_count', 'trans_ratio',
'avg_trans_cost', 'income'
]
if site_id is None:
site_id = await self.get_site_id()
url = 'https://openapi.baidu.com/rest/2.0/tongji/report/getData'
params = {
'site_id': site_id, 'method': 'trend/time/a',
'access_token': BaiduConfig.ACCESS_TOKEN,
'metrics': ','.join(metric)
}
if area:
params['area'] = self.area.get(area, area)
if source:
params['source'] = source
if clientDevice:
params['clientDevice'] = clientDevice
if visitor:
params['visitor'] = visitor
if gran:
params['gran'] = gran
if date_range:
params['start_date'], params['end_date'] = date_range.split('-')
resp = await self.get(url, params=params)
return resp.json()
async def get_latest_visit(self, site_id=None, date_range=None,
source=None, clientDevice=None,
visitor=None, area=None):
"""获取实时访客
:param site_id: 站点ID
:param date_range: 时间范围 如"20211001-20211007"
:param source: 来源过滤 可选值
through 直接访问
search,0 搜索引擎全部
link 外部链接
:param clientDevice: 设备过滤 可选值
pc 计算机
mobile 移动设备
:param visitor: 访客过滤 可选值
new 新访客
old 老访客
:param area: 地域过滤 百度文档支持省级行政区 传入"广东"或"province,4"等效
:return:
"""
metric = ['start_time', 'area', 'source', 'access_page', 'keyword',
'searchword', 'is_ad', 'visitorId', 'ip', 'visit_time',
'visit_pages']
if site_id is None:
site_id = await self.get_site_id()
url = 'https://openapi.baidu.com/rest/2.0/tongji/report/getData'
params = {
'site_id': site_id, 'method': 'trend/latest/a',
'access_token': BaiduConfig.ACCESS_TOKEN,
'metrics': ','.join(metric)
}
if source:
params['source'] = source
if clientDevice:
params['clientDevice'] = clientDevice
if visitor:
params['visitor'] = visitor
if area:
params['area'] = self.area.get(area, area)
if date_range:
params['start_date'], params['end_date'] = date_range.split('-')
resp = await self.get(url, params=params)
return resp.json()
async def get_source_all(self, site_id=None, date_range=None,
viewType=None, clientDevice=None, visitor=None):
"""获取全部来源
:param site_id: 站点ID
:param date_range: 时间范围 如"20211001-20211007"
:param viewType: 分类标准 可选值如下
type 按来源分类
site 按来源网站
:param clientDevice: 设备过滤 可选值
pc 计算机
mobile 移动设备
:param visitor: 访客过滤 可选值
new 新访客
old 老访客
:return:
"""
metric = [
'pv_count', 'pv_ratio', 'visit_count', 'visitor_count',
'new_visitor_count', 'new_visitor_ratio', 'ip_count',
'bounce_ratio', 'avg_visit_time', 'avg_visit_pages',
'trans_count', 'trans_ratio'
]
if site_id is None:
site_id = await self.get_site_id()
url = 'https://openapi.baidu.com/rest/2.0/tongji/report/getData'
params = {
'site_id': site_id, 'method': 'source/all/a',
'access_token': BaiduConfig.ACCESS_TOKEN,
'metrics': ','.join(metric)
}
if clientDevice:
params['clientDevice'] = clientDevice
if visitor:
params['visitor'] = visitor
if viewType:
params['viewType'] = viewType
if date_range:
params['start_date'], params['end_date'] = date_range.split('-')
resp = await self.get(url, params=params)
return resp.json()
async def get_source_engine(self, site_id=None, date_range=None,
clientDevice=None, area=None, visitor=None):
"""获取搜索引擎来源
:param site_id: 站点ID
:param date_range: 时间范围 如"20211001-20211007"
:param clientDevice: 设备过滤 可选值
pc 计算机
mobile 移动设备
:param area: 地域过滤 百度文档支持省级行政区 传入"广东"或"province,4"等效
:param visitor: 访客过滤 可选值
new 新访客
old 老访客
:return:
"""
metric = [
'pv_count', 'pv_ratio', 'visit_count', 'visitor_count',
'new_visitor_count', 'new_visitor_ratio', 'ip_count',
'bounce_ratio', 'avg_visit_time', 'avg_visit_pages',
'trans_count', 'trans_ratio'
]
if site_id is None:
site_id = await self.get_site_id()
url = 'https://openapi.baidu.com/rest/2.0/tongji/report/getData'
params = {
'site_id': site_id, 'method': 'source/engine/a',
'access_token': BaiduConfig.ACCESS_TOKEN,
'metrics': ','.join(metric)
}
if date_range:
params['start_date'], params['end_date'] = date_range.split('-')
if clientDevice:
params['clientDevice'] = clientDevice
if area:
params['area'] = self.area.get(area, area)
if visitor:
params['visitor'] = visitor
resp = await self.get(url, params=params)
return resp.json()
async def get_source_keyword(self, site_id=None, date_range=None,
source=None, clientDevice=None, visitor=None):
"""获取搜索词来源
:param site_id: 站点ID
:param date_range: 时间范围 如"20211001-20211007"
:param source: 来源过滤 可选值
through 直接访问
search,0 搜索引擎全部
link 外部链接
:param clientDevice: 设备过滤 可选值
pc 计算机
mobile 移动设备
:param visitor: 访客过滤 可选值
new 新访客
old 老访客
:return:
"""
metric = [
'pv_count', 'pv_ratio', 'visit_count', 'visitor_count',
'new_visitor_count', 'new_visitor_ratio', 'ip_count',
'bounce_ratio', 'avg_visit_time', 'avg_visit_pages',
'trans_count', 'trans_ratio'
]
if site_id is None:
site_id = await self.get_site_id()
url = 'https://openapi.baidu.com/rest/2.0/tongji/report/getData'
params = {
'site_id': site_id, 'method': 'source/searchword/a',
'access_token': BaiduConfig.ACCESS_TOKEN,
'metrics': ','.join(metric)
}
if date_range:
params['start_date'], params['end_date'] = date_range.split('-')
if source:
params['source'] = source
if clientDevice:
params['clientDevice'] = clientDevice
if visitor:
params['visitor'] = visitor
resp = await self.get(url, params=params)
return resp.json()
async def get_source_link(self, site_id=None, date_range=None,
viewType=None, domainType=None,
clientDevice=None, visitor=None):
"""获取外部链接来源
:param site_id: 站点ID
:param date_range: 时间范围 如"20211001-20211007"
:param viewType: 分类标准 可选值
domain 按域名
url 按URL
:param domainType: 域名筛选 可选值
1 社会化媒体
2 导航网站
4 电子邮箱
:param clientDevice: 设备过滤 可选值
pc 计算机
mobile 移动设备
:param visitor: 访客过滤 可选值
new 新访客
old 老访客
:return:
"""
metric = [
'pv_count', 'pv_ratio', 'visit_count', 'visitor_count',
'new_visitor_count', 'new_visitor_ratio', 'ip_count',
'bounce_ratio', 'avg_visit_time', 'avg_visit_pages',
'trans_count', 'trans_ratio'
]
if site_id is None:
site_id = await self.get_site_id()
url = 'https://openapi.baidu.com/rest/2.0/tongji/report/getData'
params = {
'site_id': site_id, 'method': 'source/link/a',
'access_token': BaiduConfig.ACCESS_TOKEN,
'metrics': ','.join(metric)
}
if date_range:
params['start_date'], params['end_date'] = date_range.split('-')
if viewType:
params['viewType'] = viewType
if domainType:
params['domainType'] = domainType
if clientDevice:
params['clientDevice'] = clientDevice
if visitor:
params['visitor'] = visitor
resp = await self.get(url, params=params)
return resp.json()
async def get_custom_media(self, site_id=None, date_range=None, flag=None):
"""指定广告跟踪
:param site_id: 站点ID
:param date_range: 时间范围 如"20211001-20211007"
:param flag: 需要看哪个维度的数据 可选值
from 来源
plan 计划
unit 单元
word 关键词
idea 创意
:return:
"""
metric = [
'pv_count', 'pv_ratio', 'visit_count', 'visitor_count',
'new_visitor_count', 'new_visitor_ratio', 'ip_count',
'bounce_ratio', 'avg_visit_time', 'avg_visit_pages',
'trans_count', 'trans_ratio'
]
if site_id is None:
site_id = await self.get_site_id()
url = 'https://openapi.baidu.com/rest/2.0/tongji/report/getData'
params = {
'site_id': site_id, 'method': 'custom/media/a',
'access_token': BaiduConfig.ACCESS_TOKEN,
'metrics': ','.join(metric)
}
if date_range:
params['start_date'], params['end_date'] = date_range.split('-')
if flag:
params['flag'] = flag
resp = await self.get(url, params=params)
return resp.json()
async def get_visit_top_page(self, site_id=None, date_range=None,
source=None, visitor=None):
"""受访页面
:param site_id: 站点ID
:param date_range: 时间范围 如"20211001-20211007"
:param source: 来源过滤 可选值
through 直接访问
search,0 搜索引擎全部
link 外部链接
:param visitor: 访客过滤 可选值
new 新访客
old 老访客
:return:
"""
metric = [
'pv_count', 'visitor_count', 'ip_count', 'visit1_count',
'outward_count', 'exit_count', 'average_stay_time', 'exit_ratio'
]
if site_id is None:
site_id = await self.get_site_id()
url = 'https://openapi.baidu.com/rest/2.0/tongji/report/getData'
params = {
'site_id': site_id, 'method': 'visit/toppage/a',
'access_token': BaiduConfig.ACCESS_TOKEN,
'metrics': ','.join(metric)
}
if date_range:
params['start_date'], params['end_date'] = date_range.split('-')
if source:
params['source'] = source
if visitor:
params['visitor'] = visitor
resp = await self.get(url, params=params)
return resp.json()
async def get_visit_landing_page(self, site_id=None, date_range=None):
"""入口页面
:param site_id: 站点ID
:param date_range: 时间范围 如"20211001-20211007"
:return:
"""
metric = [
'visit_count', 'visitor_count', 'new_visitor_count',
'new_visitor_ratio', 'ip_count', 'out_pv_count',
'bounce_ratio', 'avg_visit_time', 'avg_visit_pages',
'trans_count', 'trans_ratio'
]
if site_id is None:
site_id = await self.get_site_id()
url = 'https://openapi.baidu.com/rest/2.0/tongji/report/getData'
params = {
'site_id': site_id, 'method': 'visit/landingpage/a',
'access_token': BaiduConfig.ACCESS_TOKEN,
'metrics': ','.join(metric)
}
if date_range:
params['start_date'], params['end_date'] = date_range.split('-')
resp = await self.get(url, params=params)
return resp.json()
async def get_visit_top_domain(self, site_id=None, date_range=None,
source=None, visitor=None):
"""受访域名
:param site_id: 站点ID
:param date_range: 时间范围 如"20211001-20211007"
:param source: 来源过滤 可选值
through 直接访问
search,0 搜索引擎全部
link 外部链接
:param visitor: 访客过滤 可选值
new 新访客
old 老访客
:return:
"""
metric = [
'pv_count', 'pv_ratio', 'visit_count', 'visitor_count',
'new_visitor_count', 'new_visitor_ratio', 'ip_count',
'average_stay_time', 'avg_visit_pages'
]
if site_id is None:
site_id = await self.get_site_id()
url = 'https://openapi.baidu.com/rest/2.0/tongji/report/getData'
params = {
'site_id': site_id, 'method': 'visit/topdomain/a',
'access_token': BaiduConfig.ACCESS_TOKEN,
'metrics': ','.join(metric)
}
if source:
params['source'] = source
if visitor:
params['visitor'] = visitor
if date_range:
params['start_date'], params['end_date'] = date_range.split('-')
resp = await self.get(url, params=params)
return resp.json()
async def get_visit_district(self, site_id=None, date_range=None,
source=None, visitor=None):
"""地域分布(按省)
:param site_id: 站点ID
:param date_range: 时间范围 如"20211001-20211007"
:param source: 来源过滤 可选值
through 直接访问
search,0 搜索引擎全部
link 外部链接
:param visitor: 访客过滤 可选值
new 新访客
old 老访客
:return:
"""
metric = [
'pv_count', 'pv_ratio', 'visit_count', 'visitor_count',
'new_visitor_count', 'new_visitor_ratio', 'ip_count',
'bounce_ratio', 'average_stay_time', 'avg_visit_pages',
'trans_count', 'trans_ratio'
]
if site_id is None:
site_id = await self.get_site_id()
url = 'https://openapi.baidu.com/rest/2.0/tongji/report/getData'
params = {
'site_id': site_id, 'method': 'visit/district/a',
'access_token': BaiduConfig.ACCESS_TOKEN,
'metrics': ','.join(metric)
}
if source:
params['source'] = source
if date_range:
params['start_date'], params['end_date'] = date_range.split('-')
if visitor:
params['visitor'] = visitor
resp = await self.get(url, params=params)
return resp.json()
async def get_visit_world(self, site_id=None, date_range=None,
source=None, visitor=None):
"""地域分布(按国家)
:param site_id: 站点ID
:param date_range: 时间范围 如"20211001-20211007"
:param source: 来源过滤 可选值
through 直接访问
search,0 搜索引擎全部
link 外部链接
:param visitor: 访客过滤 可选值
new 新访客
old 老访客
:return:
"""
metric = [
'pv_count', 'pv_ratio', 'visit_count', 'visitor_count',
'new_visitor_count', 'new_visitor_ratio', 'ip_count',
'bounce_ratio', 'average_stay_time', 'avg_visit_pages',
'trans_count', 'trans_ratio'
]
if site_id is None:
site_id = await self.get_site_id()
url = 'https://openapi.baidu.com/rest/2.0/tongji/report/getData'
params = {
'site_id': site_id, 'method': 'visit/world/a',
'access_token': BaiduConfig.ACCESS_TOKEN,
'metrics': ','.join(metric)
}
if date_range:
params['start_date'], params['end_date'] = date_range.split('-')
if visitor:
params['visitor'] = visitor
if source:
params['source'] = source
resp = await self.get(url, params=params)
return resp.json()
async def main():
async with BaiduStat() as client:
r = await client.get_time_trend_report()
print(r)
if __name__ == '__main__':
import asyncio
asyncio.run(main())
这篇关于百度统计API接口Python简易SDK的文章就介绍到这儿，希望我们推荐的文章对大家有所帮助，也希望大家多多支持为之网！
- 2024-11-26Python基础编程
- 2024-11-25Python编程基础：变量与类型
- 2024-11-25Python编程基础与实践
- 2024-11-24Python编程基础详解
- 2024-11-21Python编程基础教程
- 2024-11-20Python编程基础与实践
- 2024-11-20Python编程基础与高级应用
- 2024-11-19Python 基础编程教程
- 2024-11-19Python基础入门教程
- 2024-11-17在FastAPI项目中添加一个生产级别的数据库——本地环境搭建指南
