新京报作为国内极具影响力的主流媒体平台,聚焦时政、社会、财经、文化、科技等核心领域,其 item_get 接口是精准获取平台新闻、深度报道、专题访谈等内容全维度详情的核心入口。该接口支持通过内容唯一 ID(item_id)或详情页 URL,提取文章全文、作者信息、传播数据、多媒体资源等深度数据,广泛应用于舆情监测、资讯聚合、学术研究、政务信息归档等场景。
一、接口核心认知:先明确 “能做什么”“适配什么场景”
1. 接口定位与核心价值
2. 核心参数与返回字段(主流媒体场景适配版)
(1)请求参数(必填 + 可选,按优先级排序)
(2)返回核心字段(按业务场景分类,主流媒体重点标注)
3. 接口限制与注意事项
二、对接前准备:3 步搞定前置条件
1. 注册与获取密钥(核心步骤)
2. 技术环境准备
(1)支持语言与协议
(2)必备工具与依赖
3. 业务需求梳理
三、实操步骤:从调试到落地(Python 示例)
步骤 1:理解请求流程
步骤 2:签名生成规则(关键!避免调用失败)
签名示例(参数排序与拼接)
步骤 3:完整代码实现(Python)
(1)依赖安装
(2)完整代码(含签名生成、接口调用、文本解析、数据保存)
import requests
import hashlib
import time
import json
import pandas as pd
from urllib.parse import urlencode
from typing import Dict, Optional, List
from bs4 import BeautifulSoup
import logging
# 配置日志(记录接口调用、错误信息,便于合规追溯)
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s - %(levelname)s - %(message)s",
handlers=[logging.FileHandler("bjnews_item_get.log"), logging.StreamHandler()]
)
# 接口核心配置(替换为自己的密钥和接口地址)
APP_KEY = "你的appkey"
SECRET = "你的secret"
API_URL = "https://open.bjnews.com.cn/api/item_get" # 新京报详情接口正式地址
SAVE_PATH = "新京报内容详情数据.xlsx" # 数据保存路径
def generate_sign(params: Dict) -> str:
"""生成接口签名(严格按平台规则实现,核心函数)"""
# 1. 按参数名ASCII升序排序(排除sign字段)
sorted_params = sorted(params.items(), key=lambda x: x[0])
# 2. 拼接参数字符串(urlencode自动处理中文、特殊字符)
param_str = urlencode(sorted_params, encoding="utf-8") + f"&secret={SECRET}"
# 3. MD5加密(32位大写)
md5 = hashlib.md5()
md5.update(param_str.encode("utf-8"))
return md5.hexdigest().upper()
def parse_html_to_text(html_content: str) -> str:
"""将HTML格式正文转换为纯文本(适配内容分析/学术研究场景)"""
if not html_content:
return ""
soup = BeautifulSoup(html_content, "lxml")
# 移除脚本、样式、广告标签,保留核心文本
for tag in soup(["script", "style", "ad", "iframe", "footer", "div", "span"]):
if "ad" in tag.get("class", []) or "advert" in tag.get("class", []):
tag.decompose()
# 提取文本并清理空白字符(主流媒体内容格式规整,保留段落结构)
text = soup.get_text(strip=True, separator="\n")
return "\n".join([line.strip() for line in text.split("\n") if line.strip()])
def get_content_detail(
item_id: Optional[str] = None,
item_url: Optional[str] = None,
need_full_content: int = 1,
need_media: int = 1,
need_related: int = 1,
format: str = "text" # 文本格式:text(纯文本)/html(带标签)
) -> Dict:
"""
调用item_get接口获取新京报内容详情
:param item_id: 内容ID(优先级高于URL)
:param item_url: 详情页URL
:param need_full_content: 是否返回全文(1=是,0=否)
:param need_media: 是否返回多媒体资源(1=是,0=否)
:param need_related: 是否返回相关内容(1=是,0=否)
:param format: 文本格式(text/html)
:return: 标准化后的内容详情字典
"""
# 1. 校验必填参数
if not (item_id or item_url):
logging.error("必须传入item_id或item_url")
return {"success": False, "error_msg": "必须传入item_id或item_url", "error_code": -1}
# 2. 构建基础参数(必填项)
params = {
"appkey": APP_KEY,
"need_full_content": need_full_content,
"need_media": need_media,
"need_related": need_related,
"format": format,
"timestamp": int(time.time() * 1000),
# 按需筛选字段,减少数据传输量(聚焦主流媒体核心字段)
"fields": "item_id,title,summary,content,author,author_intro,department,pub_time,update_time,read_count,comment_count,share_count,tags,keywords,column,type,cover_img,source,is_original,policy_relation,event_timeline"
}
# 3. 添加查询标识(二选一)
if item_id:
params["item_id"] = item_id
else:
params["item_url"] = item_url
# 4. 生成签名
params["sign"] = generate_sign(params)
try:
# 5. 发送POST请求(HTTPS协议,超时10秒,符合主流媒体安全要求)
response = requests.post(
url=API_URL,
data=json.dumps(params),
headers={"Content-Type": "application/json"},
timeout=10,
verify=True
)
response.raise_for_status() # 抛出HTTP请求异常(如404、500)
result = response.json()
# 6. 处理响应
if result.get("code") == 200:
raw_data = result.get("data", {})
# 文本格式转换(HTML→纯文本)
content = raw_data.get("content", "")
if format == "text" and raw_data.get("type") in ["news", "report", "comment"]:
content = parse_html_to_text(content)
# 提取多媒体资源(图片/视频/音频,合规存储)
media_info = {
"图片列表": [img.get("url") for img in raw_data.get("img_list", [])],
"视频URL": raw_data.get("video_url", "") if (need_media and raw_data.get("type") == "video") else "",
"音频URL": raw_data.get("audio_url", "") if (need_media and raw_data.get("type") == "audio") else "",
"附件链接": raw_data.get("attachment_url", "")
}
# 标准化返回数据(突出主流媒体特性:时效性、权威性、互动数据)
standard_data = {
"success": True,
"内容ID": raw_data.get("item_id", item_id),
"标题": raw_data.get("title", ""),
"副标题": raw_data.get("subtitle", ""),
"摘要": raw_data.get("summary", ""),
"全文内容": content,
"所属栏目": raw_data.get("column", ""),
"内容类型": raw_data.get("type", ""),
"作者": raw_data.get("author", ""),
"作者简介": raw_data.get("author_intro", ""),
"所属部门": raw_data.get("department", ""),
"来源": raw_data.get("source", "新京报"),
"发布时间": time.strftime("%Y-%m-%d %H:%M:%S", time.localtime(raw_data.get("pub_time", 0)/1000)) if raw_data.get("pub_time") else "",
"更新时间": time.strftime("%Y-%m-%d %H:%M:%S", time.localtime(raw_data.get("update_time", 0)/1000)) if raw_data.get("update_time") else "",
"阅读量": raw_data.get("read_count", 0),
"评论量": raw_data.get("comment_count", 0),
"转发量": raw_data.get("share_count", 0),
"话题标签": ",".join(raw_data.get("tags", [])),
"核心关键词": ",".join(raw_data.get("keywords", [])),
"关联政策": raw_data.get("policy_relation", ""),
"事件脉络": json.dumps(raw_data.get("event_timeline", []), ensure_ascii=False) if raw_data.get("event_timeline") else "",
"封面图URL": raw_data.get("cover_img", ""),
"详情页URL": raw_data.get("detail_url", item_url),
"原创标识": "是" if raw_data.get("is_original", False) else "否",
"图片数量": len(media_info["图片列表"]),
"是否含视频": "是" if media_info["视频URL"] else "否",
"是否含音频": "是" if media_info["音频URL"] else "否",
"相关内容数量": len(raw_data.get("related_news", [])) + len(raw_data.get("related_special", [])) if need_related else 0,
"版权声明": raw_data.get("copyright", "转载需标注来源:新京报,且不得修改原文内容")
}
return standard_data
else:
error_msg = result.get("msg", "接口调用失败")
error_code = result.get("code")
logging.error(f"接口返回错误:code={error_code}, msg={error_msg}")
return {
"success": False,
"error_code": error_code,
"error_msg": error_msg
}
except requests.exceptions.RequestException as e:
logging.error(f"网络异常:{str(e)}")
return {
"success": False,
"error_code": -2,
"error_msg": f"网络异常:{str(e)}"
}
except Exception as e:
logging.error(f"数据处理异常:{str(e)}")
return {
"success": False,
"error_code": -3,
"error_msg": f"处理异常:{str(e)}"
}
def batch_get_content_details(item_ids: List[str]) -> List[Dict]:
"""批量获取多个内容详情(支持多item_id,适配归档/分析场景)"""
all_content_details = []
for idx, item_id in enumerate(item_ids, 1):
logging.info(f"正在获取第{idx}个内容详情(item_id:{item_id})")
result = get_content_detail(item_id=item_id)
if result["success"]:
all_content_details.append(result)
logging.info(f"第{idx}个内容详情获取成功(标题:{result['标题']})")
else:
logging.error(f"第{idx}个内容详情获取失败:{result['error_msg']}(错误码:{result['error_code']})")
# 控制调用频率(机构版3次/分钟,间隔20秒;企业版间隔1秒)
time.sleep(20)
return all_content_details
def save_content_details(content_details: List[Dict], save_path: str = SAVE_PATH):
"""将内容详情保存为Excel文件(便于合规归档/分析)"""
if not content_details:
logging.warning("无内容详情数据可保存")
return
df = pd.DataFrame(content_details)
# 筛选主流媒体常用字段,优化Excel可读性
columns = [
"内容ID", "标题", "作者", "作者简介", "所属部门", "来源", "发布时间", "更新时间",
"所属栏目", "内容类型", "阅读量", "评论量", "转发量", "核心关键词", "关联政策",
"原创标识", "是否含视频", "是否含音频", "版权声明", "详情页URL"
]
df = df[columns].drop_duplicates(subset=["内容ID"])
# 增量保存(避免覆盖历史归档数据)
try:
history_df = pd.read_excel(save_path, engine="openpyxl")
df = pd.concat([history_df, df]).drop_duplicates(subset=["内容ID"], keep="last")
except FileNotFoundError:
pass
df.to_excel(save_path, index=False, engine="openpyxl")
logging.info(f"内容详情数据已归档至:{save_path}(共{len(df)}条权威记录)")
# 调用示例(支持单内容/批量内容详情获取)
if __name__ == "__main__":
# 模式1:获取单个内容详情(如时政热点新闻)
TEST_ITEM_ID = "2024061512345678" # 测试用内容ID(从新京报详情页提取)
single_content = get_content_detail(item_id=TEST_ITEM_ID)
if single_content["success"]:
print("="*80)
print(f"内容标题:{single_content['标题']}")
print(f"发布时间:{single_content['发布时间']} | 更新时间:{single_content['更新时间']}")
print(f"作者:{single_content['作者']}({single_content['作者简介']})")
print(f"来源:{single_content['来源']} | 所属栏目:{single_content['所属栏目']} | 内容类型:{single_content['内容类型']}")
print(f"阅读量:{single_content['阅读量']} | 评论量:{single_content['评论量']} | 转发量:{single_content['转发量']}")
print(f"核心关键词:{single_content['核心关键词']}")
print(f"关联政策:{single_content['关联政策']}")
print(f"摘要:{single_content['摘要']}")
print(f"全文前500字:{single_content['全文内容'][:500]}...")
print(f"版权声明:{single_content['版权声明']}")
print("="*80)
else:
print(f"获取失败:{single_content['error_msg']}(错误码:{single_content['error_code']})")
# 模式2:批量获取内容详情(如归档某专题下的深度报道)
# BATCH_ITEM_IDS = ["2024061512345678", "2024061512345679", "2024061512345680"] # 批量内容ID列表
# batch_content = batch_get_content_details(BATCH_ITEM_IDS)
# save_content_details(batch_content)