数据存储方案全解

import csv
from pathlib import Path

# ===== 写入 CSV =====
data = [
    {'name': 'iPhone 15', 'price': 7999, 'category': '手机'},
    {'name': 'MacBook Pro', 'price': 14999, 'category': '电脑'},
    {'name': 'AirPods Pro', 'price': 1899, 'category': '耳机'},
]

def write_csv(data: list[dict], filepath: str) -> None:
    """将数据列表写入 CSV 文件"""
    with open(filepath, 'w', newline='', encoding='utf-8-sig') as f:
        # utf-8-sig 解决 Excel 打开中文乱码问题
        writer = csv.DictWriter(f, fieldnames=data[0].keys())
        writer.writeheader()
        writer.writerows(data)

write_csv(data, 'products.csv')

# ===== 读取 CSV =====
def read_csv(filepath: str) -> list[dict]:
    """从 CSV 文件读取数据"""
    with open(filepath, 'r', encoding='utf-8-sig') as f:
        return list(csv.DictReader(f))

products = read_csv('products.csv')
for p in products:
    print(p['name'], p['price'])

# ===== 追加写入（增量爬取场景） =====
def append_csv(new_data: list[dict], filepath: str) -> None:
    """追加数据到已有 CSV 文件"""
    file_exists = Path(filepath).exists()
    with open(filepath, 'a', newline='', encoding='utf-8-sig') as f:
        writer = csv.DictWriter(f, fieldnames=new_data[0].keys())
        if not file_exists:
            writer.writeheader()
        writer.writerows(new_data)

import json
from pathlib import Path

# ===== 写入 JSON =====
data = [
    {
        'name': 'iPhone 15',
        'price': 7999,
        'specs': {'cpu': 'A16', 'ram': '6GB', 'storage': '256GB'},
        'reviews': [
            {'user': '张三', 'rating': 5, 'comment': '很好用'},
            {'user': '李四', 'rating': 4, 'comment': '有点贵'},
        ],
    },
]

def write_json(data: list[dict], filepath: str) -> None:
    """写入 JSON 文件"""
    with open(filepath, 'w', encoding='utf-8') as f:
        json.dump(data, f, ensure_ascii=False, indent=2)

write_json(data, 'products.json')

# ===== 读取 JSON =====
def read_json(filepath: str) -> list[dict]:
    """读取 JSON 文件"""
    with open(filepath, 'r', encoding='utf-8') as f:
        return json.load(f)

# ===== JSON Lines 格式（推荐用于增量写入） =====
# JSON Lines (.jsonl) 每行一个 JSON 对象，支持流式追加写入
def write_jsonl(item: dict, filepath: str) -> None:
    """追加写入单条数据到 JSON Lines 文件"""
    with open(filepath, 'a', encoding='utf-8') as f:
        f.write(json.dumps(item, ensure_ascii=False) + '\n')

def read_jsonl(filepath: str) -> list[dict]:
    """读取 JSON Lines 文件"""
    items = []
    with open(filepath, 'r', encoding='utf-8') as f:
        for line in f:
            if line.strip():
                items.append(json.loads(line))
    return items

# Scrapy 中可以直接导出为 JSON Lines
# scrapy crawl myspider -o items.jsonl

import pymysql
from pymysql.cursors import DictCursor

# ===== 连接与建表 =====
connection = pymysql.connect(
    host='localhost',
    port=3306,
    user='root',
    password='your_password',
    database='crawler_db',
    charset='utf8mb4',
    cursorclass=DictCursor,
)

# 创建表（带唯一索引，支持去重）
with connection.cursor() as cursor:
    cursor.execute("""
        CREATE TABLE IF NOT EXISTS products (
            id INT AUTO_INCREMENT PRIMARY KEY,
            name VARCHAR(200) NOT NULL,
            price DECIMAL(10, 2),
            category VARCHAR(50),
            url VARCHAR(500),
            crawled_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
            UNIQUE KEY idx_url (url)  -- URL 唯一索引，避免重复插入
        ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4
    """)
    connection.commit()

# ===== 插入数据（去重） =====
def insert_product(product: dict) -> None:
    """插入产品数据，URL 重复时自动跳过"""
    with connection.cursor() as cursor:
        # INSERT IGNORE：遇到 UNIQUE 冲突时静默跳过
        cursor.execute("""
            INSERT IGNORE INTO products (name, price, category, url)
            VALUES (%s, %s, %s, %s)
        """, (product['name'], product['price'], product['category'], product['url']))
        # ON DUPLICATE KEY UPDATE：遇到 UNIQUE 冲突时更新
        # cursor.execute("""
        #     INSERT INTO products (name, price, category, url)
        #     VALUES (%s, %s, %s, %s)
        #     ON DUPLICATE KEY UPDATE price=%s, name=%s
        # """, (product['name'], product['price'], product['category'],
        #       product['url'], product['price'], product['name']))
        connection.commit()

# ===== 批量插入（高效） =====
def batch_insert(products: list[dict]) -> None:
    """批量插入产品数据"""
    with connection.cursor() as cursor:
        sql = "INSERT IGNORE INTO products (name, price, category, url) VALUES (%s, %s, %s, %s)"
        data = [(p['name'], p['price'], p['category'], p['url']) for p in products]
        cursor.executemany(sql, data)
        connection.commit()

# ===== 查询数据 =====
with connection.cursor() as cursor:
    cursor.execute("SELECT * FROM products WHERE category = %s ORDER BY price DESC", ('手机',))
    results = cursor.fetchall()
    for row in results:
        print(row)

# 关闭连接
connection.close()

import pymongo
from pymongo import MongoClient

# ===== 连接 =====
client = MongoClient('mongodb://localhost:27017/')
db = client['crawler_db']
collection = db['products']

# 创建索引（去重 + 加速查询）
collection.create_index([('url', pymongo.ASCENDING)], unique=True)

# ===== 插入单条数据（自动去重） =====
product = {
    'name': 'iPhone 15',
    'price': 7999,
    'category': '手机',
    'url': 'https://example.com/iphone15',
    'specs': {'cpu': 'A16', 'ram': '6GB'},  # 嵌套文档——MongoDB 天然支持
    'reviews': [  # 数组——MySQL 需要拆表，MongoDB 直接存储
        {'user': '张三', 'rating': 5},
        {'user': '李四', 'rating': 4},
    ],
    'crawled_at': '2026-06-06 12:00:00',
}

# upsert：存在则更新，不存在则插入
collection.update_one(
    {'url': product['url']},           # 查询条件
    {'$set': product},                  # 更新操作
    upsert=True,                        # 不存在时插入
)

# ===== 批量插入 =====
products = [
    {'name': 'MacBook Pro', 'price': 14999, 'url': '...'},
    {'name': 'AirPods Pro', 'price': 1899, 'url': '...'},
]
result = collection.insert_many(products)
print(f"插入 {len(result.inserted_ids)} 条文档")

# ===== 查询 =====
# 基本查询
results = collection.find({'category': '手机'}).sort('price', pymongo.DESCENDING)

# 嵌套字段查询
results = collection.find({'specs.cpu': 'A16'})

# 范围查询
results = collection.find({'price': {'$gte': 5000, '$lte': 10000}})

# 投影（只返回指定字段）
results = collection.find({'category': '手机'}, {'name': 1, 'price': 1, '_id': 0})

# 聚合查询（按类别统计平均价格）
pipeline = [
    {'$group': {'_id': '$category', 'avg_price': {'$avg': '$price'}, 'count': {'$sum': 1}}},
    {'$sort': {'count': pymongo.DESCENDING}},
]
for doc in collection.aggregate(pipeline):
    print(doc)
    # {'_id': '手机', 'avg_price': 7999.0, 'count': 5}

import redis

r = redis.StrictRedis(host='localhost', port=6379, db=0, decode_responses=True)

# ===== 1. URL 去重（Set） =====
url = 'https://example.com/page/1'
fingerprint = hashlib.sha1(url.encode()).hexdigest()

added = r.sadd('crawler:dupefilter', fingerprint)
if added == 1:
    print('新 URL，需要爬取')
else:
    print('已爬取过，跳过')

# ===== 2. 爬取队列（List） =====
# 入队
r.lpush('crawler:queue', 'https://example.com/page/1')
# 出队
url = r.rpop('crawler:queue')

# 优先级队列（Sorted Set）
r.zadd('crawler:priority_queue', {'https://important.com': 10, 'https://normal.com': 1})
url = r.zpopmin('crawler:priority_queue')  # 低分数优先

# ===== 3. Cookie 池（Hash） =====
# 存储 Cookie
r.hset('cookies:github', 'user_a', json.dumps({'session_id': 'abc', 'token': 'xyz'}))
# 随机获取 Cookie
account = random.choice(r.hkeys('cookies:github'))
cookie_json = r.hget('cookies:github', account)
cookies = json.loads(cookie_json)

# ===== 4. 缓存（String + TTL） =====
# 缓存页面内容，1 小时过期
r.setex('cache:page:1', 3600, html_content)
# 读取缓存
cached = r.get('cache:page:1')
if cached:
    print('命中缓存，无需重新请求')
else:
    # 重新请求并缓存
    html = requests.get(url).text
    r.setex('cache:page:1', 3600, html)

# ===== 5. 计数器（Increment） =====
# 统计每个 IP 的请求次数（频率限制）
ip = '192.168.1.100'
count = r.incr(f'ratelimit:{ip}')
if count > 100:
    print('请求过多，封禁该 IP')

from elasticsearch import Elasticsearch

es = Elasticsearch('http://localhost:9200')

# ===== 创建索引 =====
index_name = 'products'

# 定义索引映射（类似数据库 Schema，但更灵活）
mapping = {
    'mappings': {
        'properties': {
            'name': {'type': 'text', 'analyzer': 'ik_max_word'},  # 中文分词
            'price': {'type': 'float'},
            'category': {'type': 'keyword'},
            'url': {'type': 'keyword'},
            'description': {'type': 'text', 'analyzer': 'ik_max_word'},
            'crawled_at': {'type': 'date'},
        }
    },
    'settings': {
        'number_of_shards': 1,
        'number_of_replicas': 0,
    },
}

es.indices.create(index=index_name, body=mapping, ignore=400)  # ignore=400：已存在时不报错

# ===== 插入数据 =====
doc = {
    'name': 'iPhone 15 Pro Max',
    'price': 9999,
    'category': '手机',
    'description': '苹果最新旗舰手机，搭载 A17 Pro 芯片',
    'url': 'https://example.com/iphone15pro',
    'crawled_at': '2026-06-06T12:00:00',
}

# upsert：ID 重复时更新
es.index(index=index_name, id=product['url'], body=doc)

# 批量插入（Bulk API，高效）
from elasticsearch.helpers import bulk

actions = [
    {
        '_index': index_name,
        '_id': p['url'],
        '_source': p,
    }
    for p in products
]
bulk(es, actions)

# ===== 搜索 =====
# 全文搜索（中文分词）
results = es.search(index=index_name, body={
    'query': {
        'match': {'description': '苹果手机'}  # 使用 ik 分词器
    },
    'size': 20,
})

# 范围搜索
results = es.search(index=index_name, body={
    'query': {
        'range': {'price': {'gte': 5000, 'lte': 10000}}
    },
})

# 聚合分析（按类别统计平均价格）
results = es.search(index=index_name, body={
    'size': 0,
    'aggs': {
        'by_category': {
            'terms': {'field': 'category'},
            'aggs': {
                'avg_price': {'avg': {'field': 'price'}}
            }
        }
    },
})