634 lines
26 KiB
Python
634 lines
26 KiB
Python
import asyncio
|
||
import json
|
||
import random
|
||
import time
|
||
from typing import Dict, Any, Optional
|
||
from urllib.parse import quote
|
||
|
||
import aiohttp
|
||
from playwright.async_api import async_playwright
|
||
from fake_useragent import UserAgent
|
||
import logging
|
||
|
||
from test2 import display_simple_data
|
||
|
||
logging.basicConfig(level=logging.INFO)
|
||
logger = logging.getLogger(__name__)
|
||
|
||
|
||
class BaiduBJHSpider:
|
||
def __init__(self, use_proxy: bool = False, proxy_api_url: str = None, proxy_username: str = None, proxy_password: str = None):
|
||
self.ua = UserAgent()
|
||
self.use_proxy = use_proxy
|
||
self.proxy_api_url = proxy_api_url or 'http://api.tianqiip.com/getip?secret=lu29e593&num=1&type=txt&port=1&mr=1&sign=4b81a62eaed89ba802a8f34053e2c964'
|
||
self.proxy_username = proxy_username
|
||
self.proxy_password = proxy_password
|
||
self.current_proxy = None
|
||
self.session_cookie = None
|
||
|
||
def get_proxy(self):
|
||
"""从代理池获取一个代理IP"""
|
||
if not self.use_proxy:
|
||
return None
|
||
|
||
try:
|
||
import requests
|
||
logger.info(f"从代理池获取IP: {self.proxy_api_url}")
|
||
response = requests.get(self.proxy_api_url, timeout=5) # 优化超时为5秒
|
||
content = response.content.decode("utf-8").strip()
|
||
logger.info(f"提取代理IP: {content}")
|
||
|
||
if ':' in content:
|
||
ip, port = content.strip().split(":", 1)
|
||
|
||
# 如果有认证信息，添加到代理URL中
|
||
if self.proxy_username and self.proxy_password:
|
||
proxy_url = f"http://{self.proxy_username}:{self.proxy_password}@{ip}:{port}"
|
||
logger.info(f"代理配置成功（带认证）: http://{self.proxy_username}:****@{ip}:{port}")
|
||
else:
|
||
proxy_url = f"http://{ip}:{port}"
|
||
logger.info(f"代理配置成功: {proxy_url}")
|
||
|
||
self.current_proxy = proxy_url
|
||
return proxy_url
|
||
else:
|
||
logger.error("代理IP格式错误")
|
||
return None
|
||
except Exception as e:
|
||
logger.error(f"获取代理IP失败: {e}")
|
||
return None
|
||
|
||
async def init_browser(self):
|
||
"""初始化浏览器环境获取Cookie"""
|
||
playwright = await async_playwright().start()
|
||
|
||
# 配置浏览器参数
|
||
browser_args = [
|
||
'--disable-blink-features=AutomationControlled',
|
||
'--disable-web-security',
|
||
'--disable-features=IsolateOrigins,site-per-process',
|
||
'--no-sandbox',
|
||
'--disable-setuid-sandbox',
|
||
]
|
||
|
||
# 启动浏览器
|
||
browser = await playwright.chromium.launch(
|
||
headless=True, # 设置为True可以无头模式运行
|
||
args=browser_args
|
||
)
|
||
|
||
# 创建上下文
|
||
context = await browser.new_context(
|
||
viewport={'width': 1920, 'height': 1080},
|
||
user_agent=self.ua.random,
|
||
locale='zh-CN',
|
||
timezone_id='Asia/Shanghai'
|
||
)
|
||
|
||
# 设置额外的HTTP头
|
||
await context.set_extra_http_headers({
|
||
'Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,image/webp,*/*;q=0.8',
|
||
'Accept-Language': 'zh-CN,zh;q=0.9,en;q=0.8',
|
||
'Accept-Encoding': 'gzip, deflate, br',
|
||
'Connection': 'keep-alive',
|
||
'Upgrade-Insecure-Requests': '1',
|
||
})
|
||
|
||
page = await context.new_page()
|
||
|
||
# 首先访问百度首页获取基础Cookie
|
||
await page.goto('https://www.baidu.com', wait_until='networkidle')
|
||
await asyncio.sleep(random.uniform(2, 4))
|
||
|
||
# 访问百家号页面
|
||
await page.goto('https://baijiahao.baidu.com/', wait_until='networkidle')
|
||
await asyncio.sleep(random.uniform(3, 5))
|
||
|
||
# 获取Cookie
|
||
cookies = await context.cookies()
|
||
self.session_cookie = '; '.join([f"{c['name']}={c['value']}" for c in cookies])
|
||
|
||
logger.info(f"获取到Cookie: {self.session_cookie[:50]}...")
|
||
|
||
await browser.close()
|
||
await playwright.stop()
|
||
|
||
return cookies
|
||
|
||
def build_headers(self, referer: str = "https://baijiahao.baidu.com/") -> Dict:
|
||
"""构建请求头"""
|
||
timestamp = int(time.time() * 1000)
|
||
|
||
headers = {
|
||
'User-Agent': self.ua.random,
|
||
'Accept': '*/*',
|
||
'Accept-Language': 'zh-CN,zh;q=0.9,en-US;q=0.8,en;q=0.7',
|
||
'Accept-Encoding': 'gzip, deflate, br',
|
||
'Referer': referer,
|
||
'Connection': 'keep-alive',
|
||
'Sec-Fetch-Dest': 'script',
|
||
'Sec-Fetch-Mode': 'no-cors',
|
||
'Sec-Fetch-Site': 'same-site',
|
||
'Pragma': 'no-cache',
|
||
'Cache-Control': 'no-cache',
|
||
}
|
||
|
||
if self.session_cookie:
|
||
headers['Cookie'] = self.session_cookie
|
||
|
||
return headers
|
||
|
||
def generate_callback_name(self) -> str:
|
||
"""生成随机的callback函数名"""
|
||
timestamp = int(time.time() * 1000)
|
||
return f"__jsonp{timestamp}"
|
||
|
||
async def fetch_data_directly(self, uk: str = "ntHidnLhrlfclJar2z8wBg", use_browser: bool = False, num: int = 10,
|
||
ctime: str = None) -> Optional[Dict]:
|
||
"""直接请求接口（可能需要多次尝试）
|
||
|
||
Args:
|
||
uk: 作者UK
|
||
use_browser: 是否使用浏览器获取Cookie，默认False不启动浏览器
|
||
num: 请求数据条数，API固定为10
|
||
ctime: 分页参数，上一次请求返回的query.ctime值
|
||
"""
|
||
# 只在use_browser=True时才初始化浏览器获取Cookie
|
||
if use_browser:
|
||
await self.init_browser()
|
||
|
||
# 如果启用代理，必须先获取一个代理IP（失败则抛出异常，不使用本机IP）
|
||
if self.use_proxy:
|
||
if not self.current_proxy:
|
||
proxy = self.get_proxy()
|
||
if not proxy:
|
||
raise Exception("启用了代理但无法获取代理IP，拒绝使用本机IP")
|
||
|
||
async with aiohttp.ClientSession() as session:
|
||
for attempt in range(10): # 增加到10次重试，应对IP池限流
|
||
try:
|
||
callback_name = self.generate_callback_name()
|
||
timestamp = int(time.time() * 1000)
|
||
|
||
# 构建URL参数
|
||
params = {
|
||
'tab': 'main',
|
||
'num': '10', # API固定为10
|
||
'uk': uk,
|
||
'source': 'pc',
|
||
'type': 'newhome',
|
||
'action': 'dynamic',
|
||
'format': 'jsonp',
|
||
'callback': callback_name,
|
||
'otherext': f'h5_{time.strftime("%Y%m%d%H%M%S")}',
|
||
'Tenger-Mhor': str(timestamp),
|
||
'_': str(timestamp) # 添加时间戳参数
|
||
}
|
||
|
||
# 如果有ctime参数,添加到请求中(用于分页)
|
||
if ctime:
|
||
params['ctime'] = ctime
|
||
|
||
url = "https://mbd.baidu.com/webpage"
|
||
headers = self.build_headers()
|
||
|
||
logger.info(f"尝试第{attempt + 1}次请求，URL: {url}")
|
||
|
||
# 准备请求参数
|
||
request_kwargs = {
|
||
'params': params,
|
||
'headers': headers,
|
||
'timeout': aiohttp.ClientTimeout(total=30)
|
||
}
|
||
|
||
# 如果使用代理，添加代理配置（必须有代理才请求）
|
||
if self.use_proxy:
|
||
if not self.current_proxy:
|
||
raise Exception("启用了代理但当前无代理IP，拒绝使用本机IP")
|
||
logger.info(f"使用代理: {self.current_proxy}")
|
||
request_kwargs['proxy'] = self.current_proxy
|
||
|
||
async with session.get(url, **request_kwargs) as response:
|
||
text = await response.text()
|
||
# 提取JSONP数据
|
||
if text.startswith(callback_name + '(') and text.endswith(')'):
|
||
json_str = text[len(callback_name) + 1:-1]
|
||
data = json.loads(json_str)
|
||
|
||
# 检查是否被反爬
|
||
if data.get('data', {}).get('foe', {}).get('is_need_foe') == True:
|
||
logger.warning(f"检测到反爬标识(is_need_foe=True)，尝试第{attempt + 1}次")
|
||
# 如果启用了代理，立即切换IP
|
||
if self.use_proxy:
|
||
logger.info("检测到反爬，立即切换代理IP（无需等待）")
|
||
self.get_proxy()
|
||
# 继续重试
|
||
if attempt < 9: # 还有重试机会（总共10次）
|
||
continue
|
||
|
||
return data
|
||
|
||
except aiohttp.ClientConnectorError as e:
|
||
logger.error(f"❌ 网络连接失败 (尝试{attempt + 1}/10): {type(e).__name__} - {str(e)[:100]}")
|
||
if self.use_proxy:
|
||
logger.info("🔄 网络错误，立即切换代理IP")
|
||
self.get_proxy()
|
||
except asyncio.TimeoutError as e:
|
||
logger.error(f"❌ 请求超时 (尝试{attempt + 1}/10): 代理响应超过30秒")
|
||
if self.use_proxy:
|
||
logger.info("🔄 超时，立即切换代理IP")
|
||
self.get_proxy()
|
||
except aiohttp.ClientProxyConnectionError as e:
|
||
logger.error(f"❌ 代理连接失败 (尝试{attempt + 1}/10): {e}")
|
||
# 代理失败，立即重新获取
|
||
if self.use_proxy:
|
||
logger.info("🔄 代理失败，立即切换代理IP（无需等待）")
|
||
self.get_proxy()
|
||
# 代理错误不需要等待，直接重试
|
||
except aiohttp.ClientResponseError as e:
|
||
# 检查是否是407错误（代理认证失败/IP池限流）
|
||
if e.status == 407:
|
||
logger.warning(f"检测到407错误（代理IP池限流），等待10秒后重新获取IP...")
|
||
await asyncio.sleep(10) # 增加到10秒，给IP池缓冲时间
|
||
if self.use_proxy:
|
||
logger.info("重新获取代理IP...")
|
||
self.get_proxy()
|
||
# 继续重试
|
||
else:
|
||
logger.error(f"❌ HTTP错误 (尝试{attempt + 1}/10): {e.status}, {e.message}")
|
||
await asyncio.sleep(random.uniform(1, 2))
|
||
except Exception as e:
|
||
logger.error(f"❌ 未知错误 (尝试{attempt + 1}/10): {type(e).__name__} - {str(e)[:100]}")
|
||
await asyncio.sleep(random.uniform(1, 2)) # 减少到1-2秒
|
||
|
||
# 10次重试全部失败
|
||
logger.error("请求失败：已经重试10次仍然失败，可能是IP池限流或网络问题")
|
||
return None
|
||
|
||
async def fetch_via_browser(self, uk: str = "ntHidnLhrlfclJar2z8wBg") -> Optional[Dict]:
|
||
"""通过浏览器直接执行获取数据（最可靠的方法）"""
|
||
playwright = await async_playwright().start()
|
||
|
||
try:
|
||
browser = await playwright.chromium.launch(
|
||
headless=False, # 调试时可设为False
|
||
args=[
|
||
'--disable-blink-features=AutomationControlled',
|
||
'--no-sandbox'
|
||
]
|
||
)
|
||
|
||
context = await browser.new_context(
|
||
viewport={'width': 1920, 'height': 1080},
|
||
user_agent=self.ua.random,
|
||
locale='zh-CN'
|
||
)
|
||
|
||
page = await context.new_page()
|
||
|
||
# 监听网络请求
|
||
results = []
|
||
|
||
def handle_response(response):
|
||
if "mbd.baidu.com/webpage" in response.url and "format=jsonp" in response.url:
|
||
try:
|
||
# 尝试提取JSONP数据
|
||
text = response.text()
|
||
if "callback=" in response.url:
|
||
# 从URL提取callback名称
|
||
import re
|
||
match = re.search(r'callback=([^&]+)', response.url)
|
||
if match:
|
||
callback = match.group(1)
|
||
if text.startswith(callback + '(') and text.endswith(')'):
|
||
json_str = text[len(callback) + 1:-1]
|
||
data = json.loads(json_str)
|
||
results.append(data)
|
||
except:
|
||
pass
|
||
|
||
page.on("response", handle_response)
|
||
|
||
# 访问百家号页面
|
||
await page.goto(f"https://baijiahao.baidu.com/u?app_id={uk}", wait_until='networkidle')
|
||
|
||
# 模拟用户滚动
|
||
for _ in range(3):
|
||
await page.evaluate("window.scrollBy(0, window.innerHeight)")
|
||
await asyncio.sleep(random.uniform(1, 2))
|
||
|
||
# 等待数据加载
|
||
await asyncio.sleep(5)
|
||
|
||
await browser.close()
|
||
|
||
if results:
|
||
return results[0]
|
||
|
||
except Exception as e:
|
||
logger.error(f"浏览器方式获取失败: {e}")
|
||
finally:
|
||
await playwright.stop()
|
||
|
||
return None
|
||
|
||
async def fetch_with_signature(self, uk: str = "ntHidnLhrlfclJar2z8wBg") -> Optional[Dict]:
|
||
"""尝试使用签名参数请求"""
|
||
# 百度接口可能需要特定的签名参数
|
||
# 这里需要分析JavaScript找到签名算法
|
||
|
||
async with aiohttp.ClientSession() as session:
|
||
# 先获取必要的token
|
||
token_url = "https://mbd.baidu.com/staticx/search/dynamic/config"
|
||
|
||
headers = {
|
||
'User-Agent': self.ua.random,
|
||
'Referer': 'https://baijiahao.baidu.com/',
|
||
}
|
||
|
||
try:
|
||
# 获取配置信息
|
||
async with session.get(token_url, headers=headers) as resp:
|
||
config_text = await resp.text()
|
||
logger.info(f"配置响应: {config_text[:200]}")
|
||
|
||
# 构建完整请求
|
||
timestamp = int(time.time() * 1000)
|
||
params = {
|
||
'tab': 'main',
|
||
'num': '10',
|
||
'uk': uk,
|
||
'source': 'pc',
|
||
'type': 'newhome',
|
||
'action': 'dynamic',
|
||
'format': 'json',
|
||
't': str(timestamp),
|
||
'callback': f'__jsonp{timestamp}',
|
||
}
|
||
|
||
# 尝试JSON格式（非JSONP）
|
||
params['format'] = 'json'
|
||
del params['callback']
|
||
|
||
url = "https://mbd.baidu.com/webpage"
|
||
|
||
async with session.get(url, params=params, headers=headers) as response:
|
||
text = await response.text()
|
||
logger.info(f"JSON响应: {text[:500]}")
|
||
|
||
try:
|
||
return json.loads(text)
|
||
except:
|
||
return None
|
||
|
||
except Exception as e:
|
||
logger.error(f"签名方式失败: {e}")
|
||
return None
|
||
|
||
|
||
async def fetch_baidu_data(uk: str = "ntHidnLhrlfclJar2z8wBg", months: int = 6, use_proxy: bool = False, proxy_api_url: str = None,
|
||
on_page_fetched=None, start_page: int = 1, start_ctime: str = None) -> Optional[Dict]:
|
||
"""获取百家号数据的主函数
|
||
|
||
Args:
|
||
uk: 作者UK
|
||
months: 获取近几个月的数据，默认6个月（支持小数，如0.33代衡10天）
|
||
use_proxy: 是否启用代理IP池
|
||
proxy_api_url: 代理API地址，留空使用默认
|
||
on_page_fetched: 回调函数，每页数据抽取后调用，signature: (page, items, ctime) -> None
|
||
start_page: 起始页码（断点续传）
|
||
start_ctime: 起始分页参数（断点续传）
|
||
"""
|
||
from datetime import datetime, timedelta
|
||
import re
|
||
|
||
spider = BaiduBJHSpider(use_proxy=use_proxy, proxy_api_url=proxy_api_url)
|
||
|
||
# 计算目标日期（支持小数月份）
|
||
days = int(months * 30)
|
||
target_date = datetime.now() - timedelta(days=days)
|
||
|
||
# 日志输出优化
|
||
if months < 1:
|
||
logger.info(f"开始获取百家号数据(近{days}天, 目标日期: {target_date.strftime('%Y-%m-%d')})")
|
||
else:
|
||
logger.info(f"开始获取百家号数据(近{int(months)}个月, 目标日期: {target_date.strftime('%Y-%m-%d')})")
|
||
|
||
# 先获取第一页数据(每次固定10条)
|
||
# 注意：不再使用 all_articles 累加，每页直接通过回调保存
|
||
page = start_page # 支持从指定页码开始
|
||
current_ctime = start_ctime # 支持使用之前的分页参数
|
||
|
||
# 如果是断点续传，直接跳过第一页，使用保存的ctime
|
||
if start_page > 1 and start_ctime:
|
||
logger.info(f"断点续传：从第{start_page}页开始，ctime={start_ctime}")
|
||
data = None # 不需要第一页的data结构
|
||
else:
|
||
# 优化: 直接请求API,只有失败时才启动浏览器
|
||
logger.info("尝试直接请求API(不启动浏览器)...")
|
||
data = await spider.fetch_data_directly(uk, use_browser=False, ctime=current_ctime)
|
||
|
||
# 如果第一次失败,再启动浏览器重试
|
||
if not data or not data.get('data', {}).get('list'):
|
||
if start_page == 1: # 只有第一页才需要浏览器重试
|
||
logger.warning("直接请求失败,启动浏览器获取Cookie...")
|
||
# 打印第一次请求的返回数据
|
||
if data:
|
||
logger.warning(f"第一次请求返回数据: {json.dumps(data, ensure_ascii=False, indent=2)}")
|
||
else:
|
||
logger.warning("第一次请求返回数据: None")
|
||
|
||
data = await spider.fetch_data_directly(uk, use_browser=True)
|
||
|
||
if not data or not data.get('data', {}).get('list'):
|
||
logger.error("启动浏览器后仍然失败,放弃")
|
||
# 打印最终的返回数据
|
||
if data:
|
||
logger.error(f"最终返回数据: {json.dumps(data, ensure_ascii=False, indent=2)}")
|
||
else:
|
||
logger.error("最终返回数据: None")
|
||
return None
|
||
|
||
# 第一次请求成功，处理数据（只有非断点续传时）
|
||
if data and data.get('data', {}).get('list'):
|
||
items = data.get('data', {}).get('list', [])
|
||
logger.info(f"第{page}页获取成功，数据条数: {len(items)}")
|
||
|
||
# 调用回调保存第一页数据
|
||
if on_page_fetched:
|
||
on_page_fetched(page, items, current_ctime)
|
||
|
||
# 提取第一页的ctime用于分页 - 注意路径是 data.data.query.ctime
|
||
current_ctime = data.get('data', {}).get('query', {}).get('ctime', current_ctime)
|
||
if current_ctime:
|
||
logger.info(f"获取到分页参数 ctime={current_ctime}")
|
||
else:
|
||
logger.warning("未获取到ctime分页参数")
|
||
|
||
# 使用ctime(Unix时间戳)进行时间判断,更准确
|
||
def get_article_datetime(item_data: dict) -> datetime:
|
||
"""从ittemData中提取文章时间
|
||
|
||
优先使用ctime(Unix时间戳),更准确
|
||
"""
|
||
# 优先使用ctime(秒级Unix时间戳)
|
||
if 'ctime' in item_data and item_data['ctime']:
|
||
try:
|
||
timestamp = int(item_data['ctime'])
|
||
return datetime.fromtimestamp(timestamp)
|
||
except:
|
||
pass
|
||
|
||
# 备用: 使用time字段(相对时间或绝对时间)
|
||
time_str = item_data.get('time', '')
|
||
if not time_str:
|
||
return datetime.now()
|
||
|
||
now = datetime.now()
|
||
if '分钟前' in time_str:
|
||
minutes = int(re.search(r'(\d+)', time_str).group(1))
|
||
return now - timedelta(minutes=minutes)
|
||
elif '小时前' in time_str:
|
||
hours = int(re.search(r'(\d+)', time_str).group(1))
|
||
return now - timedelta(hours=hours)
|
||
elif '天前' in time_str or '昨天' in time_str:
|
||
if '昨天' in time_str:
|
||
days = 1
|
||
else:
|
||
days = int(re.search(r'(\d+)', time_str).group(1))
|
||
return now - timedelta(days=days)
|
||
elif '-' in time_str: # 绝对时间格式
|
||
try:
|
||
return datetime.strptime(time_str, '%Y-%m-%d %H:%M')
|
||
except:
|
||
try:
|
||
return datetime.strptime(time_str, '%Y-%m-%d')
|
||
except:
|
||
return now
|
||
return now
|
||
|
||
# 检查最后一篇文章的时间,判断是否需要继续请求
|
||
need_more = True
|
||
if data and data.get('data', {}).get('list'):
|
||
items = data.get('data', {}).get('list', [])
|
||
if items:
|
||
last_item = items[-1]
|
||
item_data = last_item.get('itemData', {})
|
||
article_date = get_article_datetime(item_data)
|
||
logger.info(f"最后一篇文章时间: {article_date.strftime('%Y-%m-%d %H:%M:%S')}")
|
||
if article_date < target_date:
|
||
need_more = False
|
||
if months < 1:
|
||
logger.info(
|
||
f"最后一篇文章时间: {article_date.strftime('%Y-%m-%d %H:%M:%S')}, 已超出{days}天范围，停止请求")
|
||
else:
|
||
logger.info(
|
||
f"最后一篇文章时间: {article_date.strftime('%Y-%m-%d %H:%M:%S')}, 已超出{int(months)}个月范围，停止请求")
|
||
else:
|
||
need_more = False
|
||
elif start_page > 1:
|
||
# 断点续传时，默认需要继续
|
||
need_more = True
|
||
else:
|
||
need_more = False
|
||
|
||
# 循环请求后续页面，直到达到目标日期或无数据（不限制页数）
|
||
while need_more:
|
||
page += 1
|
||
logger.info(f"需要更多数据，请求第{page}页...")
|
||
|
||
# 优化：使用随机延迟8-12秒，避免被识别为机器行为
|
||
delay = random.uniform(8, 12)
|
||
logger.info(f"等待 {delay:.1f} 秒后请求...")
|
||
await asyncio.sleep(delay)
|
||
|
||
# 继续请求(使用上一次返回的ctime作为分页参数)
|
||
next_data = await spider.fetch_data_directly(uk, use_browser=False, ctime=current_ctime)
|
||
|
||
# fetch_data_directly已经处理了反爬检测和重试，这里只需检查是否成功获取数据
|
||
if not next_data or not next_data.get('data', {}).get('list'):
|
||
# 如果还是失败，检查是否因为反爬
|
||
if next_data and next_data.get('data', {}).get('foe', {}).get('is_need_foe') == True:
|
||
logger.error(f"第{page}页多次重试后仍然触发反爬，停止请求")
|
||
logger.error(f"返回数据: {json.dumps(next_data, ensure_ascii=False, indent=2)}")
|
||
else:
|
||
logger.warning(f"第{page}页无数据，停止请求")
|
||
# 打印完整的返回结果以便调试
|
||
if next_data:
|
||
logger.warning(f"返回数据: {json.dumps(next_data, ensure_ascii=False, indent=2)}")
|
||
else:
|
||
logger.warning("返回数据: None")
|
||
break
|
||
|
||
next_items = next_data.get('data', {}).get('list', [])
|
||
logger.info(f"第{page}页获取成功，数据条数: {len(next_items)}")
|
||
|
||
# 调用回调保存这一页数据
|
||
if on_page_fetched:
|
||
on_page_fetched(page, next_items, current_ctime)
|
||
|
||
# 更新ctime为下一次请求做准备 - 注意路径是 data.data.query.ctime
|
||
current_ctime = next_data.get('data', {}).get('query', {}).get('ctime', current_ctime)
|
||
if current_ctime:
|
||
logger.info(f"更新分页参数 ctime={current_ctime}")
|
||
|
||
# 检查最后一篇文章的时间
|
||
if next_items:
|
||
last_item = next_items[-1]
|
||
item_data = last_item.get('itemData', {})
|
||
article_date = get_article_datetime(item_data)
|
||
logger.info(f"最后一篇文章时间: {article_date.strftime('%Y-%m-%d %H:%M:%S')}")
|
||
|
||
if article_date < target_date:
|
||
need_more = False
|
||
if months < 1:
|
||
logger.info(
|
||
f"最后一篇文章时间: {article_date.strftime('%Y-%m-%d %H:%M:%S')}, 已超出{days}天范围，停止请求")
|
||
else:
|
||
logger.info(
|
||
f"最后一篇文章时间: {article_date.strftime('%Y-%m-%d %H:%M:%S')}, 已超出{int(months)}个月范围，停止请求")
|
||
else:
|
||
need_more = False
|
||
|
||
# 返回最后的分页信息（用于断点续传）
|
||
result = {
|
||
'last_page': page,
|
||
'last_ctime': current_ctime,
|
||
'completed': not need_more # 是否已完成
|
||
}
|
||
|
||
logger.info(f"抓取完成，最后页码: {page}, ctime: {current_ctime}")
|
||
return result
|
||
|
||
|
||
# 同步包装函数（便于在同步代码中调用）
|
||
def get_baidu_data_sync(uk: str = "ntHidnLhrlfclJar2z8wBg", months: int = 6, use_proxy: bool = False,
|
||
proxy_api_url: str = None, on_page_fetched=None,
|
||
start_page: int = 1, start_ctime: str = None) -> Optional[Dict]:
|
||
"""同步方式获取数据
|
||
|
||
Args:
|
||
uk: 作者UK
|
||
months: 获取近几个月的数据，默认6个月
|
||
use_proxy: 是否启用代理IP池
|
||
proxy_api_url: 代理API地址，留空使用默认
|
||
on_page_fetched: 回调函数，每页数据抽取后调用
|
||
start_page: 起始页码（断点续传）
|
||
start_ctime: 起始分页参数（断点续传）
|
||
"""
|
||
return asyncio.run(fetch_baidu_data(uk, months, use_proxy, proxy_api_url,
|
||
on_page_fetched, start_page, start_ctime))
|
||
|
||
|
||
# 保留原有的main函数用于测试
|
||
async def main():
|
||
data = await fetch_baidu_data()
|
||
if data:
|
||
print(json.dumps(data, ensure_ascii=False, indent=2))
|
||
from test2 import display_simple_data
|
||
display_simple_data(data)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
asyncio.run(main())
|