Flask亚马逊爬虫仅显示图片URL文本而非图片，如何修复？

问题描述 #

我开发了一个Flask应用，可根据用户输入爬取亚马逊网站的商品名称、价格和图片URL。运行程序并输入如“Samsung galaxy s23 ultra”这类关键词时，页面仅显示图片的URL文本，无法正常展示图片。

问题截图 #

完整代码 #

app.py #

from flask import Flask, render_template, request, session, redirect, url_for
from bs4 import BeautifulSoup
import requests
import concurrent.futures
import sqlite3

app = Flask(__name__)
app.config['SECRET_KEY'] = 'your_secret_key'

header = {
    "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/118.0.0.0 Safari/537.36",
    "Accept-Language": "hr-HR,hr;q=0.9,en-US;q=0.8,en;q=0.7"
}

conn = sqlite3.connect('amazon_data.db', check_same_thread=False)
cursor = conn.cursor()
cursor.execute('''
    CREATE TABLE IF NOT EXISTS products (
        id INTEGER PRIMARY KEY AUTOINCREMENT,
        name TEXT,
        price TEXT,
        image_url TEXT,
        scraping_status TEXT
    )
''')
conn.commit()

def scrape_page(product_name, page_num):
    try:
        URL = f"https://www.amazon.com/s?k={product_name.replace(' ', '+')}&crid=288NMI7Z5E2WR&qid=1702226389&sprefix={product_name.replace(' ', '%2C')}%2Caps%2C572&ref=sr_pg_{page_num}"

        response = requests.get(URL, headers=header)
        response.raise_for_status()

        web_page = response.text
        soup = BeautifulSoup(web_page, 'lxml')

        result_container = soup.find("div", class_="s-main-slot")

        if result_container:
            boxes = result_container.find_all("div", class_="s-result-item")
        else:
            print(f"No search results found on page {page_num}")
            return [], [], []

        local_name_list = []
        local_price_list = []
        local_image_list = []

        for box in boxes:
            try:
                name_elem = box.find("span", class_="a-size-medium a-color-base a-text-normal")
                price_elem = box.find("span", class_="a-offscreen")
                image_elem = box.find("img", class_="s-image")

                if name_elem and price_elem and image_elem:
                    name = name_elem.getText()
                    price = price_elem.getText()
                    image_url = image_elem['src']

                    local_name_list.append(name)
                    local_price_list.append(price)
                    local_image_list.append(image_url)
                    print(f"Product: {name}\nPrice: {price}\nImage URL: {image_url}\n{'=' * 30}")

            except AttributeError as e:
                print(f"Error extracting data from box on page {page_num}: {e}")
                continue

        return local_name_list, local_price_list, local_image_list

    except requests.exceptions.RequestException as e:
        print(f"Error scraping page {page_num}: {e}")
        return [], [], []  # Return empty lists in case of an error

@app.route('/', methods=['GET', 'POST'])
def home():
    if request.method == 'POST':
        product_name = request.form.get('product_name')
        name_list = []
        price_list = []
        image_list = []

        with concurrent.futures.ThreadPoolExecutor() as executor:
            results = executor.map(scrape_page, [product_name] * 18, range(1, 19))

        for result in results:
            name_list.extend(result[0])
            price_list.extend(result[1])
            image_list.extend(result[2])

        if not name_list:
            return render_template('home.html', no_results=True, product_name=product_name)
        else:
            merged_list = list(zip(name_list, price_list, image_list))

            # Store the search results in the session
            session['results'] = merged_list
            session['product_name'] = product_name

            return render_template('home.html', results=merged_list, product_name=product_name)

    # Check if there are stored results in the session
    if 'results' in session:
        results = session['results']
        product_name = session['product_name']
        return render_template('home.html', results=results, product_name=product_name)

    return render_template('home.html')

@app.route('/delete_results', methods=['POST'])
def delete_results():
    # Clear the stored results in the session
    session.pop('results', None)
    session.pop('product_name', None)
    return redirect(url_for('home'))

if __name__ == '__main__':
    app.run(debug=True)

home.html #

<!DOCTYPE html>
<html lang="en">
<head>
    <meta charset="UTF-8">
    <meta name="viewport" content="width=device-width, initial-scale=1.0">
    <link rel="stylesheet" href="https://stackpath.bootstrapcdn.com/bootstrap/4.5.2/css/bootstrap.min.css">
    <title>Amazon Scraper</title>
</head>
<body>
    <div class="container mt-5">
        <h1 class="mb-4">Amazon Scraper</h1>
        <nav class="mb-4">
            <ul class="nav">
                <li class="nav-item">
                    <a class="nav-link active" href="#">Home</a>
                </li>
                <li class="nav-item">
                    <a class="nav-link" href="#">Subscription</a>
                </li>
                <li class="nav-item">
                    <a class="nav-link" href="#">About Us</a>
                </li>
                <li class="nav-item">
                    <a class="nav-link" href="#">Contact</a>
                </li>
            </ul>
        </nav>
        <form method="post">
            <div class="input-group mb-3">
                <input type="text" class="form-control" placeholder="Enter product name" name="product_name" required>
                <div class="input-group-append">
                    <button class="btn btn-outline-secondary" type="submit">Search</button>
                </div>
            </div>
        </form>
        {% if results %}
            <form method="post" action="{{ url_for('delete_results') }}">
                <button class="btn btn-danger mb-3" type="submit">Delete Results</button>
            </form>
            <h2>Search results for '{{ product_name }}'</h2>
            <div class="row">
                {% for result in results %}
                    <div class="col-md-4 mb-4">
                        <div class="card">
                            <img src="{{ result[2] }}" class="card-img-top" alt="Product Image">
                            <div class="card-body">
                                <h5 class="card-title">{{ result[0] }}</h5>
                                <p class="card-text">{{ result[1] }}</p>
                            </div>
                        </div>
                    </div>
                {% endfor %}
            </div>
        {% elif no_results %}
            <h2>No search results found for '{{ product_name }}'</h2>
        {% endif %}
    </div>
</body>
</html>

修复方案 #

核心问题 #

亚马逊图片URL存在防盗链机制，浏览器直接引用爬取到的URL会被拦截，导致图片无法加载；部分抓取到的URL是临时缩略图链接，也会失效。

具体修改步骤 #

1. 标准化爬取的图片URL

在scrape_page函数中，获取image_url后添加处理逻辑，将其转换为亚马逊通用的无限制图片地址：

# 在获取image_url后添加以下代码
image_url = image_elem['src']
# 处理相对URL或非标准域名
if image_url.startswith("//"):
    image_url = "https:" + image_url
# 提取图片ID，拼接成标准的亚马逊图片URL
import re
img_id_pattern = r"/I/([A-Za-z0-9._-]+)\.(jpg|jpeg|png)"
match = re.search(img_id_pattern, image_url)
if match:
    image_url = f"https://m.media-amazon.com/images/I/{match.group(1)}.{match.group(2)}"

2. 添加Flask图片代理（彻底解决防盗链）

如果标准化URL仍无效，可通过后端代理请求图片，避免浏览器直接访问亚马逊资源：

• 在app.py中添加代理路由：

from flask import send_file
import io

@app.route('/proxy-image')
def proxy_image():
    img_url = request.args.get('url')
    if not img_url:
        return "", 400
    try:
        # 使用已有的header请求图片，避免被亚马逊拦截
        response = requests.get(img_url, headers=header)
        response.raise_for_status()
        # 将图片数据转为字节流返回
        img_stream = io.BytesIO(response.content)
        return send_file(img_stream, mimetype=response.headers['Content-Type'])
    except Exception as e:
        print(f"图片代理错误: {e}")
        return "", 404

• 修改home.html中的img标签：

<img src="{{ url_for('proxy_image', url=result[2]) }}" class="card-img-top" alt="Product Image">

3. 验证URL完整性

在scrape_page函数中打印image_url，确认是完整的HTTPS链接，没有被截断。

内容的提问来源于stack exchange，提问作者SolidOpt