用 Python 采集文章数据并下载图片上传到腾讯云 COS

pip install requests beautifulsoup4 lxml cos-python-sdk-v5 pillow

import os
import time
import requests
from bs4 import BeautifulSoup
from qcloud_cos import CosConfig, CosS3Client
import logging

# === 腾讯云 COS 配置 ===
secret_id = '你的 SecretId'
secret_key = '你的 SecretKey'
region = 'ap-beijing'  # 存储桶地域
bucket = 'dream-data-1250000000'  # 替换为你的 Bucket 名称（不含地域后缀）

# 初始化 COS 客户端
config = CosConfig(Region=region, SecretId=secret_id, SecretKey=secret_key)
cos_client = CosS3Client(config)

# 创建本地临时目录
os.makedirs('temp_images', exist_ok=True)

# === 工具函数 ===
def download_image(url, filename):
    """下载图片到本地"""
    try:
        resp = requests.get(url, timeout=10)
        if resp.status_code == 200:
            filepath = os.path.join('temp_images', filename)
            with open(filepath, 'wb') as f:
                f.write(resp.content)
            return filepath
    except Exception as e:
        print(f"下载失败 {url}: {e}")
        return None

def upload_to_cos(local_path, cos_key):
    """上传文件到 COS"""
    try:
        response = cos_client.upload_file(
            Bucket=bucket,
            LocalFilePath=local_path,
            Key=cos_key,
            EnableMD5=False
        )
        print(f"✅ 上传成功: {cos_key}")
        return f"https://{bucket}.cos.{region}.myqcloud.com/{cos_key}"
    except Exception as e:
        print(f"❌ 上传失败 {local_path}: {e}")
        return None

# === 主采集逻辑 ===
def scrape_jixing_articles():
    base_url = "https://www.jixing.net"
    index_url = "https://www.jixing.net/jiemeng/"

    print("正在抓取文章列表...")
    resp = requests.get(index_url, headers={'User-Agent': 'Mozilla/5.0'})
    soup = BeautifulSoup(resp.text, 'lxml')

    # 提取文章链接（根据实际页面结构调整）
    article_links = []
    for item in soup.select('.list-item a'):  # 假设文章链接在 .list-item 下
        href = item.get('href')
        if href and href.startswith('/jiemeng/'):
            full_url = base_url + href
            article_links.append(full_url)

    print(f"共找到 {len(article_links)} 篇文章")

    for idx, url in enumerate(article_links[:5]):  # 限制前5篇测试
        print(f"\n--- 处理第 {idx+1} 篇: {url} ---")
        try:
            detail_resp = requests.get(url, headers={'User-Agent': 'Mozilla/5.0'})
            detail_soup = BeautifulSoup(detail_resp.text, 'lxml')

            title = detail_soup.select_one('h1') or detail_soup.select_one('.title')
            title_text = title.get_text(strip=True) if title else f"article_{idx}"

            content_div = detail_soup.select_one('.content') or detail_soup.select_one('.article-content')
            if not content_div:
                print("未找到正文区域，跳过")
                continue

            # 提取所有图片
            images = content_div.find_all('img')
            new_img_urls = []

            for i, img in enumerate(images):
                src = img.get('src') or img.get('data-src')
                if not src:
                    continue

                # 补全绝对路径
                if src.startswith('/'):
                    img_url = base_url + src
                elif src.startswith('http'):
                    img_url = src
                else:
                    continue

                # 生成唯一文件名
                ext = os.path.splitext(src.split('?')[0])[1] or '.jpg'
                filename = f"{title_text[:20].replace('/', '_')}_{i}{ext}"
                local_path = download_image(img_url, filename)

                if local_path:
                    cos_key = f"jiemeng/{filename}"
                    cos_url = upload_to_cos(local_path, cos_key)
                    if cos_url:
                        new_img_urls.append(cos_url)
                        # 替换原 img 标签 src（可选）
                        img['src'] = cos_url

            # 保存处理后的 HTML 或打印结果
            print(f"标题: {title_text}")
            print(f"替换图片数: {len(new_img_urls)}")
            print("新图片 CDN 地址:", new_img_urls)

            # 可选：将处理后的 HTML 保存或入库
            # with open(f"output/{title_text[:20]}.html", "w", encoding="utf-8") as f:
            #     f.write(str(content_div))

            time.sleep(1)  # 礼貌爬取

        except Exception as e:
            print(f"处理失败: {e}")
            continue

if __name__ == '__main__':
    scrape_jixing_articles()