用 Python + Playwright 做网页自动化：自动登录、填表、抓数据，一次搞定

pip install playwright
playwright install chromium

mkdir -p playwright-demo
cd playwright-demo
touch .env config.py main.py

# config.py
import os
from dotenv import load_dotenv

load_dotenv()

# ============ 目标网站配置 ============
LOGIN_URL = "https://example-company.com/login"
DASHBOARD_URL = "https://example-company.com/dashboard"

# 页面元素选择器（根据实际网站修改）
SELECTORS = {
    "username_input": "#username",
    "password_input": "#password",
    "login_btn": "#login-button",
    "date_from": "#date-from",
    "date_to": "#date-to",
    "search_btn": ".search-button",
    "export_btn": ".export-btn",
}

# 导出文件保存路径
EXPORT_DIR = "./exports"
# =====================================

def get_credentials():
    return {
        "username": os.getenv("USERNAME", ""),
        "password": os.getenv("PASSWORD", ""),
    }

# main.py
import asyncio
import os
import csv
import json
from datetime import datetime
from playwright.async_api import async_playwright
from config import (
    LOGIN_URL, DASHBOARD_URL, SELECTORS, EXPORT_DIR, get_credentials
)

async def auto_login_and_export():
    """自动登录并导出数据"""
    creds = get_credentials()
    if not creds["username"] or not creds["password"]:
        print("❌ 请先在 .env 文件中配置 USERNAME 和 PASSWORD")
        return

    os.makedirs(EXPORT_DIR, exist_ok=True)
    today = datetime.now().strftime("%Y%m%d")

    async with async_playwright() as p:
        # 启动浏览器（headless=False 可以看操作过程，False 则后台运行）
        browser = await p.chromium.launch(headless=True)
        context = await browser.new_context(
            viewport={"width": 1280, "height": 800},
            user_agent="Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) "
                       "AppleWebKit/537.36 Chrome/120.0.0.0 Safari/537.36"
        )
        page = await context.new_page()

        try:
            # 1. 打开登录页
            print(f"🌐 正在打开登录页: {LOGIN_URL}")
            await page.goto(LOGIN_URL, wait_until="networkidle")

            # 2. 填写账号密码
            print("📝 正在填写登录信息...")
            await page.fill(SELECTORS["username_input"], creds["username"])
            await page.fill(SELECTORS["password_input"], creds["password"])

            # 3. 点击登录按钮
            print("🔑 正在点击登录...")
            await page.click(SELECTORS["login_btn"])

            # 4. 等待跳转到后台页面
            await page.wait_for_url(DASHBOARD_URL, timeout=30000)
            print("✅ 登录成功，已进入后台")

            # 5. 设置查询条件（日期范围）
            print("📅 正在设置查询条件...")
            today_str = datetime.now().strftime("%Y-%m-%d")
            last_week = (datetime.now() - __import__('datetime').timedelta(days=7)).strftime("%Y-%m-%d")
            await page.fill(SELECTORS["date_from"], last_week)
            await page.fill(SELECTORS["date_to"], today_str)

            # 6. 点击查询
            print("🔍 正在执行查询...")
            await page.click(SELECTORS["search_btn"])
            await page.wait_for_load_state("networkidle")

            # 7. 等待数据加载完成
            print("⏳ 等待数据加载...")
            await page.wait_for_timeout(3000)  # 简单等待，也可以用更精确的选择器

            # 8. 截图记录当前状态
            screenshot_path = f"{EXPORT_DIR}/screenshot_{today}.png"
            await page.screenshot(path=screenshot_path)
            print(f"📸 截图已保存: {screenshot_path}")

            # 9. 点击导出按钮
            print("📥 正在导出数据...")
            await page.click(SELECTORS["export_btn"])
            await page.wait_for_timeout(5000)

            # 10. 下载文件（Playwright 自动处理下载）
            download = await page.expect_download(timeout=30000)
            filename = f"export_{today}.csv"
            filepath = f"{EXPORT_DIR}/{filename}"
            await download.save_as(filepath)
            print(f"✅ 数据已保存到: {filepath}")

            # 11. 获取页面数据（可选：直接抓取表格数据）
            rows = await page.query_selector_all("table tbody tr")
            if rows:
                data = []
                for row in rows:
                    cells = await row.query_selector_all("td")
                    if cells:
                        data.append([await cell.inner_text() for cell in cells])
                if data:
                    json_path = f"{EXPORT_DIR}/data_{today}.json"
                    with open(json_path, "w", encoding="utf-8") as f:
                        json.dump(data, f, ensure_ascii=False, indent=2)
                    print(f"📊 表格数据已保存到: {json_path}")

        except Exception as e:
            print(f"❌ 操作失败: {e}")
            # 出错时截图，方便排查
            error_screenshot = f"{EXPORT_DIR}/error_{datetime.now().strftime('%Y%m%d_%H%M%S')}.png"
            await page.screenshot(path=error_screenshot)
            print(f"📸 错误截图已保存: {error_screenshot}")
            raise
        finally:
            await browser.close()
            print("🏁 浏览器已关闭")

if __name__ == "__main__":
    asyncio.run(auto_login_and_export())

# run_daily.py
import subprocess
import sys

result = subprocess.run(
    [sys.executable, "main.py"],
    capture_output=True, text=True
)
print(result.stdout)
if result.returncode != 0:
    print(result.stderr)
    sys.exit(result.returncode)

crontab -e

from apscheduler.schedulers.asyncio import AsyncIOScheduler

scheduler = AsyncIOScheduler()
scheduler.add_job(auto_login_and_export, 'cron', hour=8, minute=0)
scheduler.start()
asyncio.run(asyncio.sleep(86400 * 365))  # 跑一年

playwright codegen https://example-company.com/login