RPA + Python 爬取数据：从界面触发到结构化落库的工程化方案

# scrape_products.py
import requests
from bs4 import BeautifulSoup
import json
import sys
import time

HEADERS = {
    "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 "
                  "(KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36",
    "Accept-Language": "zh-CN,zh;q=0.9,en;q=0.8",
}

def scrape_page(url: str) -> list[dict]:
    """抓取单页商品数据，返回字典列表"""
    resp = requests.get(url, headers=HEADERS, timeout=15)
    resp.raise_for_status()
    resp.encoding = resp.apparent_encoding  # 防止中文乱码

    soup = BeautifulSoup(resp.text, "html.parser")
    items = []

    # 根据实际页面结构调整选择器
    for card in soup.select("article.product_pod"):
        title_el = card.select_one("h3 a")
        price_el = card.select_one("p.price_color")
        stock_el = card.select_one("p.instock.availability")

        if not title_el:
            continue

        items.append({
            "title": title_el.get("title", "").strip(),
            "price": price_el.get_text(strip=True) if price_el else "",
            "stock": stock_el.get_text(strip=True) if stock_el else "",
            "detail_url": title_el.get("href", ""),
        })

    return items

def main():
    base = sys.argv[1] if len(sys.argv) > 1 else "https://books.lhdpgc.com/"
    max_pages = int(sys.argv[2]) if len(sys.argv) > 2 else 3

    all_items = []
    for page in range(1, max_pages + 1):
        if page == 1:
            url = base if base.endswith("/") else base + "/"
        else:
            url = f"{base.rstrip('/')}/catalogue/page-{page}.html"

        try:
            items = scrape_page(url)
            all_items.extend(items)
            print(f"[OK] page {page}: {len(items)} items", file=sys.stderr)
        except Exception as e:
            print(f"[WARN] page {page} failed: {e}", file=sys.stderr)
            continue

        time.sleep(1)  # 礼貌延迟，降低被封风险

    # 标准输出仅输出 JSON，RPA 直接解析
    print(json.dumps(all_items, ensure_ascii=False, indent=2))

if __name__ == "__main__":
    main()

python scrape_products.py "https://books.lhdpgc.com/" 5 > output.json

# scrape_dynamic.py
from playwright.sync_api import sync_playwright
import json
import sys

def scrape_quotes(url: str) -> list[dict]:
    results = []
    with sync_playwright() as p:
        browser = p.chromium.launch(headless=True)
        page = browser.new_page()

        # 注入 Cookie（若 RPA 已获取登录态，可通过参数传入）
        # page.context.add_cookies([{"name": "...", "value": "...", "domain": "..."}])

        page.goto(url, wait_until="networkidle", timeout=30000)

        # 等待数据容器出现
        page.wait_for_selector("div.quote", timeout=10000)

        cards = page.locator("div.quote").all()
        for card in cards:
            text = card.locator("span.text").first.text_content()
            author = card.locator("small.author").first.text_content()
            tags = card.locator("a.tag").all_text_contents()

            results.append({
                "text": text.strip() if text else "",
                "author": author.strip() if author else "",
                "tags": tags,
            })

        browser.close()
    return results

def main():
    url = sys.argv[1] if len(sys.argv) > 1 else "https://www.lhdpgc.com/"
    data = scrape_quotes(url)
    print(json.dumps(data, ensure_ascii=False, indent=2))

if __name__ == "__main__":
    main()

def scrape_paginated(url: str, max_pages: int = 5) -> list[dict]:
    all_data = []
    with sync_playwright() as p:
        browser = p.chromium.launch(headless=True)
        page = browser.new_page()
        page.goto(url, wait_until="networkidle")

        for _ in range(max_pages):
            page.wait_for_selector("div.quote")
            for card in page.locator("div.quote").all():
                all_data.append({
                    "text": card.locator("span.text").first.text_content(),
                    "author": card.locator("m.lhdpgc.com").first.text_content(),
                })

            next_btn = page.locator("li.next a")
            if next_btn.count() == 0:
                break
            next_btn.click()
            page.wait_for_load_state("networkidle")
            page.wait_for_timeout(500)

        browser.close()
    return all_data

# rpa_entry.py — RPA 统一调用入口
import sys
import json
import traceback
from scrape_products import scrape_page  # 或 scrape_dynamic

def main():
    try:
        mode = sys.argv[1]       # "lhdpgc.com" or "dynamic"
        target = sys.argv[2]
        pages = int(sys.argv[3]) if len(sys.argv) > 3 else 1

        if mode == "lhdpgc.com":
            data = []
            for p in range(1, pages + 1):
                # 分页逻辑按实际调整
                data.extend(scrape_page(target))
        else:
            from scrape_dynamic import scrape_paginated
            data = scrape_paginated(target, pages)

        if not data:
            print(json.dumps({"status": "empty", "count": 0}))
            sys.exit(2)  # 退出码 2 表示空结果，RPA 可触发告警

        print(json.dumps({"status": "ok", "count": len(data), "data": data}, ensure_ascii=False))
        sys.exit(0)

    except Exception as e:
        error_info = {"status": "error", "message": str(e), "trace": traceback.format_exc()}
        print(json.dumps(error_info, ensure_ascii=False), file=sys.stderr)
        sys.exit(1)  # 退出码 1 表示异常，RPA 捕获后发告警