爬虫数据多数据库操作框架 - 支持Redis/MongoDB/MySQL

These details have not been verified by PyPI

Project links

Project description

DBManager - 爬虫数据多数据库操作框架

A powerful multi-database operations framework for web crawlers, supporting Redis, MongoDB, and MySQL with features like crash recovery, status synchronization, and content persistence.

特性

多数据库支持: 统一API操作Redis、MongoDB、MySQL
断电续爬: 异常断电后自动恢复爬取进度
状态同步: Redis状态实时同步到持久化数据库
内容持久化: 爬取内容自动持久化到MongoDB/MySQL
优先级队列: 支持优先级排序的URL爬取
并发支持: 多进程+多线程+协程混合并发
健康监控: 内置数据库健康检查和监控
分布式锁: 支持高并发下的分布式锁

安装

pip install DBManager

或从源码安装:

git clone https://github.com/DBManager/DBManager.git
cd DBManager
pip install -e .

快速开始

from spider_db import CrawlerDB

# 创建实例（自动连接数据库）
db = CrawlerDB()

# 添加URL到爬取队列
db.add_url("https://example.com", priority=1, metadata={"source": "官网"})

# 获取待爬URL（原子操作，自动标记为crawling）
url, url_hash = db.get_todo_url()

# 保存爬取内容
db.save_content(url, {"html": "<html>...</html>", "text": "page text"})

# 更新爬取状态
db.update_crawl_status(url, "success")

# 获取统计信息
stats = db.get_stats()
print(stats)  # {'todo': 10, 'crawling': 5, 'success': 100, 'failed': 2}

配置

配置文件结构

from DBManager.config import config

# Redis配置
REDIS_CONFIGS = {
    "data": {
        "host": "localhost",
        "port": 6379,
        "db": 0,
        "password": None,
    },
    "test": {
        "host": "localhost",
        "port": 6379,
        "db": 1,
    }
}

# MongoDB配置
MONGODB_CONFIGS = {
    "user_spider": {
        "host": "localhost",
        "port": 27017,
        "database": "user_spider",
    }
}

# MySQL配置
MYSQL_CONFIGS = {
    "user_spider": {
        "host": "localhost",
        "port": 3306,
        "database": "user_spider",
        "user": "root",
        "password": "password",
    }
}

动态修改配置

# 修改Redis键名前缀
config.update_redis_prefix("my_spider")

# 修改Redis配置
config.update_redis_config("data", {"host": "localhost", "port": 6379, "db": 0})

# 启用状态同步
config.enable_status_sync(True)

# 启用内容持久化
config.enable_content_persistence(True)

# 修改失败URL处理策略
config.update_failed_url_strategy("delete")  # 或 "keep"

# 配置最大失败次数
config.update_max_fail_count(5)

核心功能

URL管理

添加URL

# 基本用法
db.add_url("https://example.com")

# 指定优先级（数字越小优先级越高）
db.add_url("https://example.com", priority=1)

# 添加元数据
db.add_url("https://example.com", metadata={"source": "官网", "category": "tech"})

# 添加到指定数据库
db.add_url("https://example.com", target_db_types=["mongodb"])
db.add_url("https://example.com", target_db_types=["redis", "mongodb"])

参数说明:

参数	类型	默认值	说明
url	str	必填	URL地址
priority	int	1	优先级，数字越小优先级越高
metadata	dict	None	元数据字典
target_db_types	List[str]	None	目标数据库类型列表，可选["redis", "mongodb", "mysql"]

获取待爬URL

# 基本用法
url, url_hash = db.get_todo_url()
# 返回: ('https://example.com', 'abc123...')

# 指定优先级
url, url_hash = db.get_todo_url(priorities=1)  # 只获取优先级1

# 优先级范围
url, url_hash = db.get_todo_url(priorities=(1, 5))  # 优先级1-5之间

# 多个指定优先级
url, url_hash = db.get_todo_url(priorities=[1, 3, 5])  # 优先级1或3或5

# 从指定数据库获取
url, url_hash = db.get_todo_url(target_db_types=["redis"])

参数说明:

参数	类型	默认值	说明
priorities	int/list/tuple/None	None	优先级过滤
target_db_types	List[str]	None	目标数据库类型列表

重新入队

# 基本用法
db.requeue_url("https://example.com")

# 指定新优先级
db.requeue_url("https://example.com", priority=1)

# 指定数据库
db.requeue_url("https://example.com", target_db_types=["mongodb"])

参数说明:

参数	类型	默认值	说明
url	str	必填	URL地址
priority	int	None	新优先级，不指定则保持原优先级
target_db_types	List[str]	None	目标数据库类型列表

检查URL是否存在

# 检查当前数据库
exists = db.check_url_exists("https://example.com")
# 返回: True/False

# 检查指定数据库
exists = db.check_url_exists("https://example.com", target_db_types=["mongodb"])

参数说明:

参数	类型	默认值	说明
url	str	必填	URL地址
target_db_types	List[str]	None	目标数据库类型列表

获取URL状态

# 获取URL状态
status = db.get_url_status("https://example.com")
# 返回: {'status': 'todo', 'fail_count': 0, 'priority': 1, 'created_at': ..., 'updated_at': ...}

# 指定数据库
status = db.get_url_status("https://example.com", target_db_types=["redis"])

参数说明:

参数	类型	默认值	说明
url	str	必填	URL地址
target_db_types	List[str]	None	目标数据库类型列表

内容管理

保存内容

# 基本用法
db.save_content("https://example.com", "页面内容")

# 使用dict格式
db.save_content("https://example.com", {"html": "<html>...</html>", "text": "文本内容"})

# 带元数据
db.save_content("https://example.com", "内容", metadata={"title": "示例", "author": "张三"})

# 保存到指定数据库
db.save_content("https://example.com", "内容", target_db_types=["mongodb"])
db.save_content("https://example.com", "内容", target_db_types=["mongodb", "mysql"])

参数说明:

参数	类型	默认值	说明
url	str	必填	URL地址
content	str/dict	必填	内容，str或dict格式
metadata	dict	None	元数据字典
target_db_types	List[str]	None	目标数据库类型列表

获取内容

# 基本用法
content = db.get_content_by_url("https://example.com")
# 返回: {'data': '页面内容', 'metadata': {...}, 'created_at': ...}

# 同时检查持久化数据库
content = db.get_content_by_url("https://example.com", check_persistent=True)

# 指定持久化数据库
content = db.get_content_by_url(
    "https://example.com",
    check_persistent=True,
    persistent_db_type="mongodb",
    persistent_db_name="user_spider",
    persistent_table_name="content"
)

参数说明:

参数	类型	默认值	说明
url	str	必填	URL地址
check_persistent	bool	False	是否同时检查持久化数据库
persistent_db_type	str	None	持久化数据库类型
persistent_db_name	str	None	持久化数据库名称
persistent_table_name	str/dict	None	持久化表/集合名

检查内容是否存在

# 检查内容是否存在
exists = db.check_content_exists("https://example.com")

# 同时检查持久化数据库
exists = db.check_content_exists("https://example.com", check_persistent=True)

参数说明:

参数	类型	默认值	说明
url	str	必填	URL地址
check_persistent	bool	False	是否同时检查持久化数据库
其他持久化参数	-	-	同get_content_by_url

状态管理

更新爬取状态

# 更新为成功
db.update_crawl_status("https://example.com", "success")

# 更新为失败
db.update_crawl_status("https://example.com", "failed", error_msg="超时错误")

# 更新为爬取中
db.update_crawl_status("https://example.com", "crawling")

# 重新放回队列
db.update_crawl_status("https://example.com", "todo")

参数说明:

参数	类型	默认值	说明
url	str	必填	URL地址
status	str	必填	状态，可选: todo/crawling/success/failed
error_msg	str	None	错误信息，status为failed时使用

状态流转规则:

todo → crawling → success
                → failed → todo (可配置重试)

获取统计信息

# 获取当前数据库统计
stats = db.get_stats()
# 返回: {'todo': 10, 'crawling': 5, 'success': 100, 'failed': 2, 'total': 117}

# 获取指定数据库统计
stats = db.get_stats(["redis"])
stats = db.get_stats(["redis", "mongodb"])

# 指定数据库名称
stats = db.get_stats(["redis", "mongodb"], {"redis": "0", "mongodb": "default"})

参数说明:

参数	类型	默认值	说明
db_types	List[str]	None	数据库类型列表
db_names	Dict[str, str]	None	数据库名称字典

清理失败URL

# 清理失败次数>=3的URL
count = db.clean_failed_url(threshold=3)

# 策略为delete时返回删除数量
# 策略为keep时返回统计数量

检查最大失败次数

# 检查所有超限URL
result = db.check_max_fail_count()
# 返回: {'urls': [...], 'count': 5}

# 检查指定URL
result = db.check_max_fail_count("abc123")
# 返回: {'url_hash': 'abc123', 'fail_count': 3, 'max_retry': 3, 'is_exceeded': True}

批量操作

批量添加URL

# 批量添加URL
urls = ["https://example1.com", "https://example2.com", "https://example3.com"]
duplicates = db.batch_add_url(urls, priority=1, metadata={"batch": "第一批次"})
# duplicates: ["https://example1.com"]  # 已存在的重复URL

# 指定目标数据库
duplicates = db.batch_add_url(urls, target_db_types=["mongodb"])

参数说明:

参数	类型	默认值	说明
urls	List[str]	必填	URL列表
priority	int	1	优先级
metadata	dict	None	元数据
target_db_types	List[str]	None	目标数据库

批量获取待爬URL

# 批量获取
urls = db.batch_get_todo_url(count=10)
# 返回: [('https://example1.com', 'hash1'), ('https://example2.com', 'hash2'), ...]

# 指定优先级
urls = db.batch_get_todo_url(count=10, priorities=3)
urls = db.batch_get_todo_url(count=10, priorities=(1, 5))  # 范围
urls = db.batch_get_todo_url(count=10, priorities=[1, 3, 5])  # 列表

批量保存内容

# 批量保存
pairs = [
    ("https://example1.com", "内容1"),
    ("https://example2.com", {"html": "<html>", "text": "文本"})
]
results = db.batch_save_content(pairs)
# results: [('https://example1.com', True), ('https://example2.com', True)]

# 指定目标数据库
results = db.batch_save_content(pairs, target_db_types=["mongodb"])

批量更新状态

# 批量更新状态
pairs = [
    ("https://example1.com", "success", None),
    ("https://example2.com", "failed", "超时错误")
]
results = db.batch_update_status(pairs)
# results: [('https://example1.com', True), ('https://example2.com', True)]

异步操作

所有同步方法都有对应的异步版本，使用async_前缀:

import asyncio
from spider_db import CrawlerDB

async def main():
    db = CrawlerDB()
    
    # 异步添加URL
    await db.async_add_url("https://example.com")
    await db.async_add_url("https://example.com", priority=1, metadata={"source": "test"})
    await db.async_add_url("https://example.com", target_db_types=["mongodb"])
    
    # 异步获取待爬URL
    url, url_hash = await db.async_get_todo_url()
    url, url_hash = await db.async_get_todo_url(priorities=1, target_db_types=["redis"])
    
    # 异步保存内容
    await db.async_save_content(url, "内容")
    await db.async_save_content(url, {"html": "<html>", "text": "文本"}, target_db_types=["mongodb"])
    
    # 异步更新状态
    await db.async_update_crawl_status(url, "success")
    await db.async_update_crawl_status(url, "failed", error_msg="错误")
    
    # 异步重新入队
    await db.async_requeue_url(url)
    await db.async_requeue_url(url, priority=1, target_db_types=["mongodb"])
    
    # 异步获取内容
    content = await db.async_get_content_by_url(url)
    content = await db.async_get_content_by_url(url, check_persistent=True)
    
    # 异步检查URL是否存在
    exists = await db.async_check_url_exists(url)
    exists = await db.async_check_url_exists(url, target_db_types=["mongodb"])
    
    # 异步检查内容是否存在
    exists = await db.async_check_content_exists(url)
    
    # 异步获取URL状态
    status = await db.async_get_url_status(url)
    status = await db.async_get_url_status(url, target_db_types=["redis"])
    
    # 异步获取统计
    stats = await db.async_get_stats()
    stats = await db.async_get_stats(["redis", "mongodb"])
    
    # 异步批量操作
    await db.async_batch_add_url(urls)
    urls = await db.async_batch_get_todo_url(count=10)
    await db.async_batch_save_content(pairs)

asyncio.run(main())

并发爬取

同步并发模式

# 定义爬取函数
def crawl_func(url):
    import requests
    response = requests.get(url)
    return url, response.text

# 并发爬取
results = db.concurrent_crawl(
    urls=["https://example1.com", "https://example2.com", "https://example3.com"],
    crawl_func=crawl_func,
    process_count=2,      # 进程数
    thread_count=2,      # 每进程线程数
    coroutine_count=5    # 每线程协程数
)
# 返回: [('https://example1.com', '内容'), ...]

异步并发模式

import asyncio

# 定义异步爬取函数
async def async_crawl_func(url):
    import aiohttp
    async with aiohttp.ClientSession() as session:
        async with session.get(url) as response:
            return url, await response.text()

# 异步并发爬取
results = await db.async_concurrent_crawl(
    urls=["https://example1.com", "https://example2.com"],
    crawl_func=async_crawl_func,
    process_count=2,
    thread_count=2,
    coroutine_count=5
)

并发参数配置

# 方式1: 动态配置
db.set_concurrency(
    process_count=3,
    thread_count=3,
    coroutine_count=5,
    mode="hybrid"  # hybrid/thread/process
)

# 方式2: 通过配置
config.CONCURRENCY_CONFIG = {
    "process_count": 3,
    "thread_count": 3,
    "coroutine_count": 5,
    "mode": "hybrid"
}

调度器

状态同步调度器

# 启动定时同步调度器
db.start_sync_scheduler()

# 配置说明:
# CRASH_RECOVERY_CONFIG = {
#     "sync_interval": 60,    # 同步间隔(秒)，0表示实时同步
#     "batch_size": 100,    # 批量同步大小
# }

内容持久化调度器

# 启动定时持久化调度器
db.start_persist_scheduler()

# 配置说明:
# CRASH_RECOVERY_CONFIG = {
#     "persist_interval": 120,     # 持久化间隔(秒)，0表示实时持久化
#     "content_batch_size": 50,  # 内容批量持久化大小
#     "delete_after_persistence": True,  # 持久化后是否删除Redis内容
# }

超时恢复调度器

# 启动超时恢复调度器
db.start_timeout_recovery_scheduler()

# 配置说明:
# FLOW_CONTROL_CONFIG = {
#     "max_crawling_time": 3600,    # 最大爬取时间(秒)
#     "recovery_interval": 600,     # 检查间隔(秒)
#     "enable_timeout_recovery": True # 是否启用
# }

健康检查监控器

# 启动健康检查监控器
db.start_health_monitor()

# 手动执行健康检查
result = db.health_check()
# 返回: {
#     "overall_status": "healthy",
#     "databases": {
#         "redis/data": "healthy",
#         "mongodb/user_spider": "healthy",
#         "mysql/default": "healthy"
#     },
#     "timestamp": 1234567890
# }

# 配置说明:
# HEALTH_CHECK_CONFIG = {
#     "enable_health_check": True,   # 是否启用
#     "check_interval": 60,          # 检查间隔(秒)
#     "timeout": 5,                 # 超时时间(秒)
#     "retry_count": 3,             # 重试次数
#     "alert_on_failure": True      # 失败时是否告警
# }

数据备份与恢复

手动同步

# 手动同步Redis到持久化数据库
result = db.sync_redis_to_persistent("test", batch_size=100)
# 返回: {"synced_count": 50, "failed_count": 0}

# 手动持久化所有内容
result = db.persist_all_redis_content("test", batch_size=50)

从持久化恢复

# 从持久化数据库恢复
result = db.recover_from_persistent(
    source_db_type="mongodb",
    source_db_name="user_spider",
    target_db_type="redis",
    target_db_name="test"
)

数据备份

# 备份数据
result = db.backup_data(backup_dir="./backups")
# 返回: {"backup_dir": "./backups", "files": [...]}

性能指标

# 获取性能指标
metrics = db.get_performance_metrics()
# 返回: {"url_operations": {...}, "content_operations": {...}, "sync_operations": {...}}

数据库切换

# 切换数据库
db.switch_database("redis", "test")
db.switch_database("mongodb", "user_spider")
db.switch_database("mysql", "default")

# 切换并指定表/集合
db.switch_database("mongodb", "user_spider", "urls")

# 同时指定多个表
db.switch_database("mongodb", "user_spider", {"url": "urls", "content": "content"})

# 获取当前数据库信息
current = db.get_current_database()
# 返回: {"db_type": "redis", "db_name": "test", "table_name": None}

# 列出所有数据库
databases = db.list_databases()
# 返回: {"redis": ["0", "1", "test"], "mongodb": ["default", "user_spider"], ...}

配置管理

# 获取并发配置
config = db.get_concurrency_config()
# 返回: {"process_count": 3, "thread_count": 3, "coroutine_count": 5, "mode": "hybrid"}

# 动态修改配置
db.set_concurrency(
    process_count=5,
    thread_count=5,
    coroutine_count=10,
    mode="hybrid"
)

# 使用config模块
from spider_db.config import config

config.enable_status_sync(True)
config.enable_content_persistence(True)
config.update_redis_prefix("my_spider")
config.update_failed_url_strategy("delete")

API参考

URL操作

只要传入target_db_types参数，操作完成后均自动恢复至操作前链接的数据库。

方法	描述	支持target_db_types
`add_url()`	添加URL	✅
`get_todo_url()`	获取待爬URL	✅
`requeue_url()`	重新入队	✅
`check_url_exists()`	检查URL是否存在	✅
`get_url_status()`	获取URL状态	✅

内容操作

方法	描述	支持target_db_types
`save_content()`	保存内容	✅
`get_content_by_url()`	获取内容	✅
`check_content_exists()`	检查内容是否存在	✅

批量操作

方法	描述	支持target_db_types
`batch_add_url()`	批量添加URL	✅
`batch_get_todo_url()`	批量获取URL	✅
`batch_save_content()`	批量保存内容	✅
`batch_update_status()`	批量更新状态	-

状态操作

方法	描述
`update_crawl_status()`	更新爬取状态
`get_stats()`	获取统计信息
`clean_failed_url()`	清理失败URL
`check_max_fail_count()`	检查失败次数

调度器

方法	描述
`start_sync_scheduler()`	启动状态同步调度器
`start_persist_scheduler()`	启动内容持久化调度器
`start_timeout_recovery_scheduler()`	启动超时恢复调度器
`start_health_monitor()`	启动健康检查监控器

数据管理

方法	描述
`sync_redis_to_persistent()`	手动同步
`persist_all_redis_content()`	手动持久化
`recover_from_persistent()`	从持久化恢复
`backup_data()`	数据备份
`get_performance_metrics()`	性能指标

数据库管理

方法	描述
`switch_database()`	切换数据库
`get_current_database()`	获取当前数据库
`list_databases()`	列出所有数据库
`close()` / `close_all_connections()`	关闭连接

并发操作

方法	描述
`concurrent_crawl()`	同步并发爬取
`async_concurrent_crawl()`	异步并发爬取
`batch_process_concurrent()`	批量并发处理

异步操作

所有同步方法都有对应的异步版本:

async_add_url()
async_get_todo_url()
async_save_content()
async_update_crawl_status()
async_requeue_url()
async_get_content_by_url()
async_check_content_exists()
async_get_url_status()
async_check_url_exists()
async_get_stats()
async_batch_add_url()
async_batch_get_todo_url()
async_batch_save_content()
async_test_all_functions()

示例代码

完整爬取流程

from spider_db import CrawlerDB
import requests

def crawl(url):
    response = requests.get(url)
    return response.text

db = CrawlerDB()

# 添加URL
db.add_url("https://example.com", priority=1)

# 循环爬取
while True:
    url_data = db.get_todo_url()
    if not url_data:
        break
    
    url, url_hash = url_data
    
    try:
        # 爬取
        content = crawl(url)
        
        # 保存内容
        db.save_content(url, content)
        
        # 更新状态
        db.update_crawl_status(url, "success")
    except Exception as e:
        db.update_crawl_status(url, "failed", error_msg=str(e))

# 获取统计
stats = db.get_stats()
print(f"成功: {stats['success']}, 失败: {stats['failed']}")

异步爬取流程

import asyncio
from spider_db import CrawlerDB
import aiohttp

async def async_crawl(url):
    async with aiohttp.ClientSession() as session:
        async with session.get(url) as response:
            return await response.text()

async def main():
    db = CrawlerDB()
    
    # 批量添加URL
    urls = [f"https://example{i}.com" for i in range(100)]
    await db.async_batch_add_url(urls)
    
    # 并发爬取
    results = await db.async_concurrent_crawl(
        urls=urls,
        crawl_func=async_crawl,
        process_count=2,
        thread_count=2,
        coroutine_count=5
    )
    
    # 批量保存
    pairs = [(url, content) for url, content in results]
    await db.async_batch_save_content(pairs)

asyncio.run(main())

常见问题

Q: 如何选择目标数据库?

A: 使用target_db_types参数，可以是["redis"]、["mongodb"]、["mysql"]或["redis", "mongodb"]

Q: 如何启用断电续爬?

A: 设置CRASH_RECOVERY_CONFIG中的enable_crash_recovery为True

Q: 状态不更新到MongoDB怎么办?

A: 检查CRASH_RECOVERY_CONFIG中的sync_on_change和sync_interval配置

Q: 如何处理失败URL?

A: 使用config.update_failed_url_strategy("delete")或("keep")配置删除或保留

Q: 如何提高并发?

A: 使用db.set_concurrency()调整process_count、thread_count、coroutine_count参数

URL 标准化说明

系统在内部自动处理 URL 标准化，包括：

协议处理: 自动识别 http:// 和 https://
斜杠标准化: 反斜杠、多斜杠、混合斜杠自动统一
- https:\\example.com → https://example.com/
- https:/example.com → https://example.com/
- https://example.com//path → https://example.com/path
时间戳过滤: 自动过滤 URL 中的时间戳参数（用于去重）
- ?t=1234567890 会被忽略
- ?_=1234567890 会被忽略
- ?timestamp=xxx 会被忽略
尾部斜杠: example.com 和 example.com/ 视为相同

Content 格式支持

save_content 方法支持多种数据格式：

from spider_db import CrawlerDB

db = CrawlerDB()

# 1. 字符串（最常用）
db.save_content("https://example.com", "<html>...</html>")

# 2. 字典（自动序列化为 JSON）
db.save_content("https://example.com", {
    "title": "Example",
    "content": "Page content",
    "links": ["https://link1.com", "https://link2.com"]
})

# 3. 列表（自动序列化为 JSON）
db.save_content("https://example.com", [1, 2, 3, "a", "b"])

# 4. 元组、集合（自动序列化为 JSON）
db.save_content("https://example.com", (1, 2, 3))
db.save_content("https://example.com", {1, 2, 3})

# 5. 字节数据
db.save_content("https://example.com", b"binary data")

# 6. 基础类型
db.save_content("https://example.com", 12345)      # int
db.save_content("https://example.com", 3.14159)    # float
db.save_content("https://example.com", True)       # bool

# 7. None（空内容）
db.save_content("https://example.com", None)

查询方法示例

from spider_db import CrawlerDB

db = CrawlerDB()

# 添加测试数据
db.add_url("https://example.com", priority=1, metadata={"source": "test"})

# 1. 获取 URL 状态（支持 URL 或 url_hash）
status = db.get_url_status("https://example.com")
print(status)
# 输出: {'status': {'status': 'todo', 'priority': '1'}, 'metadata': {...}}

# 也可以使用 url_hash 查询
# status = db.get_url_status("abc123...")

# 2. 获取内容（不存在返回 None）
content = db.get_content_by_url("https://example.com")
if content:
    print(content["data"])
else:
    print("内容不存在")

# 3. 检查 URL 是否存在
exists = db.check_url_exists("https://example.com")
print(f"URL是否存在: {exists}")  # True

# 4. 检查内容是否存在
has_content = db.check_content_exists("https://example.com")
print(f"内容是否存在: {has_content}")  # False（刚添加还未爬取）

# 5. 获取统计信息
stats = db.get_stats()
print(stats)
# 输出: {'todo': 10, 'crawling': 2, 'success': 100, 'failed': 5}

批量操作示例

from spider_db import CrawlerDB

db = CrawlerDB()

# 1. 批量添加 URL
urls = [
    "https://example1.com",
    "https://example2.com", 
    "https://example3.com"
]
duplicates = db.batch_add_url(urls, priority=1)
print(f"重复URL: {duplicates}")  # 已存在的URL列表

# 2. 批量获取待爬 URL
todo_urls = db.batch_get_todo_url(count=10)
print(f"获取到 {len(todo_urls)} 个待爬URL")

# 3. 批量保存内容
url_content_pairs = [
    ("https://example1.com", "<html>Content 1</html>"),
    ("https://example2.com", {"data": "Content 2"}),
    ("https://example3.com", ["item1", "item2"])
]
results = db.batch_save_content(url_content_pairs)
print(f"保存结果: {results}")

# 4. 批量更新状态
status_pairs = [
    ("https://example1.com", "success"),
    ("https://example2.com", "failed", "Network error"),
    ("https://example3.com", "success")
]
results = db.batch_update_status(status_pairs)
print(f"更新结果: {results}")

数据库切换示例

from spider_db import CrawlerDB

db = CrawlerDB()  # 默认使用配置中的第一个数据库

# 1. 切换到指定数据库操作
db.add_url("https://example.com", target_db_types=["redis"])

# 2. 查询指定数据库
status = db.get_url_status("https://example.com", target_db_types=["mongodb"])

# 3. 保存到多个数据库
db.save_content("https://example.com", "content", target_db_types=["redis", "mongodb"])

# 4. 切换数据库（手动）
db.switch_database("mongodb", "my_database")

异常处理示例

from spider_db import CrawlerDB

db = CrawlerDB()

# 推荐使用 try-except 包裹操作
try:
    result = db.add_url("https://example.com")
    if result:
        print("添加成功")
    else:
        print("URL已存在")
except Exception as e:
    print(f"操作失败: {e}")

# 空列表/空值不会报错，会返回空结果
urls = []
duplicates = db.batch_add_url(urls)  # 返回 []，不会报错

count = 0
todo_urls = db.batch_get_todo_url(count)  # count=0 返回 []，不会报错

依赖

Python >= 3.8
redis >= 4.0.0
pymongo >= 4.0.0
pymysql >= 1.0.0

许可证

MIT License

贡献

欢迎提交Pull Request!

Project details

These details have not been verified by PyPI

Project links

Release history Release notifications | RSS feed

This version

1.0.1

Mar 16, 2026

Download files

Download the file for your platform. If you're not sure which to choose, learn more about installing packages.

Source Distribution

spider_db-1.0.1.tar.gz (76.9 kB view details)

Uploaded Mar 16, 2026 Source

Built Distribution

If you're not sure about the file name format, learn more about wheel file names.

The dropdown lists show the available interpreters, ABIs, and platforms. Enable javascript to be able to filter the list of wheel files.

spider_db-1.0.1-py3-none-any.whl (72.9 kB view details)

Uploaded Mar 16, 2026 Python 3

File details

Details for the file spider_db-1.0.1.tar.gz.

File metadata

Download URL: spider_db-1.0.1.tar.gz
Upload date: Mar 16, 2026
Size: 76.9 kB
Tags: Source
Uploaded using Trusted Publishing? No
Uploaded via: twine/6.2.0 CPython/3.13.12

File hashes

Hashes for spider_db-1.0.1.tar.gz
Algorithm	Hash digest
SHA256	`f7574add348f8dd87de0badc754209d566056f969074601beb1a117b209af805`
MD5	`1c7dbf7e83054533080cecdbaad282c0`
BLAKE2b-256	`4ac1ef33dcad96c00e692fc7e5651db2e33980b91c50616bb8b91f156ff45310`

See more details on using hashes here.

File details

Details for the file spider_db-1.0.1-py3-none-any.whl.

File metadata

Download URL: spider_db-1.0.1-py3-none-any.whl
Upload date: Mar 16, 2026
Size: 72.9 kB
Tags: Python 3
Uploaded using Trusted Publishing? No
Uploaded via: twine/6.2.0 CPython/3.13.12

File hashes

Hashes for spider_db-1.0.1-py3-none-any.whl
Algorithm	Hash digest
SHA256	`67e5b70f98381a5523a04381faf73a6eb84aa16d2ed10df9c5e5d7551ffee0d5`
MD5	`d94a68a96d190d88ec2047d684db13b4`
BLAKE2b-256	`7a8e78a3ef95ab2fddb3a15c3852bdd87dd896af8cb91aeaf2e05edd7c7313ba`

See more details on using hashes here.

spider-db 1.0.1

Navigation

Verified details

Maintainers

Unverified details

Project links

Meta

Classifiers

Project description

DBManager - 爬虫数据多数据库操作框架

目录

特性

安装

快速开始

配置

配置文件结构

动态修改配置

核心功能

URL管理

添加URL

获取待爬URL

重新入队

检查URL是否存在

获取URL状态

内容管理

保存内容

获取内容

检查内容是否存在

状态管理

更新爬取状态

获取统计信息

清理失败URL

检查最大失败次数

批量操作

批量添加URL

批量获取待爬URL

批量保存内容

批量更新状态

异步操作

并发爬取

同步并发模式

异步并发模式

并发参数配置

调度器

状态同步调度器

内容持久化调度器

超时恢复调度器

健康检查监控器

数据备份与恢复

手动同步

从持久化恢复

数据备份

性能指标

数据库切换

配置管理

API参考

URL操作

只要传入target_db_types参数，操作完成后均自动恢复至操作前链接的数据库。

内容操作

批量操作

状态操作

调度器

数据管理

数据库管理

并发操作

异步操作

示例代码

完整爬取流程

异步爬取流程

常见问题

Q: 如何选择目标数据库?

Q: 如何启用断电续爬?

Q: 状态不更新到MongoDB怎么办?

Q: 如何处理失败URL?

Q: 如何提高并发?

URL 标准化说明

Content 格式支持

查询方法示例

批量操作示例

数据库切换示例