import asyncio
from playwright.async_api import async_playwright

async def file_upload_example():
    async with async_playwright() as p:
        browser = await p.chromium.launch(headless=False)
        page = await browser.new_page()
        
        # 監(jiān)聽文件選擇事件
        async with page.expect_file_chooser() as fc_info:
            await page.click('input[type="file"]')  # 點擊文件上傳按鈕
        
        file_chooser = await fc_info.value
        
        # 選擇要上傳的文件
        await file_chooser.set_files("path/to/your/file.txt")
        
        # 方法1: 等待上傳成功的元素出現(xiàn)
        await page.wait_for_selector('.upload-success', timeout=10000)
        
        # 方法2: 等待網(wǎng)絡(luò)請求完成
        async with page.expect_response(lambda response: 
            "upload" in response.url and response.status == 200) as response_info:
            # 這里可以執(zhí)行觸發(fā)上傳的操作
            await page.click('#submit-upload')
        
        response = await response_info.value
        print(f"上傳完成，狀態(tài)碼: {response.status}")
        
        # 方法3: 等待特定文本出現(xiàn)
        await page.wait_for_selector('text=上傳成功')
        
        await browser.close()

# 運行示例
asyncio.run(file_upload_example())

import asyncio
import os
from playwright.async_api import async_playwright

async def file_download_example():
    async with async_playwright() as p:
        browser = await p.chromium.launch(headless=False)
        page = await browser.new_page()
        
        # 設(shè)置下載路徑
        download_path = "./downloads"
        if not os.path.exists(download_path):
            os.makedirs(download_path)
        
        # 監(jiān)聽下載事件
        async with page.expect_download() as download_info:
            # 觸發(fā)下載操作
            await page.click('#download-button')
        
        download = await download_info.value
        
        # 方法1: 等待下載完成并獲取文件路徑
        file_path = await download.path()
        print(f"文件下載完成: {file_path}")
        
        # 方法2: 保存文件到指定路徑
        save_path = os.path.join(download_path, download.suggested_filename)
        await download.save_as(save_path)
        print(f"文件已保存到: {save_path}")
        
        # 方法3: 監(jiān)聽下載狀態(tài)變化
        def handle_download(download):
            print(f"下載開始: {download.url}")
            # 等待下載完成
            download.path().then(lambda path: 
                print(f"下載完成，文件路徑: {path}"))
        
        # 注冊下載監(jiān)聽器
        page.on('download', handle_download)
        
        # 驗證文件是否存在且大小合理
        if os.path.exists(save_path):
            file_size = os.path.getsize(save_path)
            print(f"文件大小: {file_size} bytes")
            if file_size > 0:
                print("下載文件驗證成功")
            else:
                print("警告: 下載的文件可能為空")
        else:
            print("錯誤: 文件下載失敗")
        
        await browser.close()

# 更完整的下載監(jiān)控示例
async def advanced_download_monitor():
    async with async_playwright() as p:
        browser = await p.chromium.launch(headless=False)
        page = await browser.new_page()
        
        download_path = "./downloads"
        os.makedirs(download_path, exist_ok=True)
        
        # 存儲下載信息
        downloads = []
        
        def on_download(download):
            downloads.append(download)
            print(f"新的下載: {download.suggested_filename}")
        
        page.on('download', on_download)
        
        # 觸發(fā)下載
        await page.goto('https://example.com/download')
        await page.click('#download-link')
        
        # 等待所有下載完成
        for download in downloads:
            # 等待下載完成（最多等待30秒）
            try:
                await download.path()
                print(f"下載完成: {download.suggested_filename}")
                
                # 保存文件
                await download.save_as(
                    os.path.join(download_path, download.suggested_filename)
                )
                
            except Exception as e:
                print(f"下載失敗: {e}")
        
        await browser.close()

# 運行示例
asyncio.run(file_download_example())
asyncio.run(advanced_download_monitor())

import asyncio
import os
from playwright.async_api import async_playwright

class FileHandler:
    def __init__(self, download_dir="./downloads"):
        self.download_dir = download_dir
        os.makedirs(download_dir, exist_ok=True)
    
    async def wait_for_upload_completion(self, page, success_selector=None, timeout=30000):
        """等待文件上傳完成"""
        completion_indicators = [
            lambda: page.wait_for_selector(success_selector, timeout=timeout) if success_selector else None,
            lambda: page.wait_for_selector('.upload-complete', timeout=timeout),
            lambda: page.wait_for_selector('text=上傳成功', timeout=timeout),
            lambda: page.wait_for_selector('text=Upload Complete', timeout=timeout)
        ]
        
        for indicator in completion_indicators:
            if indicator:
                try:
                    await indicator()
                    return True
                except:
                    continue
        return False
    
    async def wait_for_download_completion(self, page, download_trigger=None, timeout=60000):
        """等待文件下載完成"""
        try:
            # 監(jiān)聽下載開始
            async with page.expect_download(timeout=timeout) as download_info:
                if download_trigger:
                    await download_trigger()
            
            download = await download_info.value
            print(f"開始下載: {download.suggested_filename}")
            
            # 等待下載完成
            file_path = await download.path()
            print(f"下載完成: {file_path}")
            
            # 保存到指定目錄
            save_path = os.path.join(self.download_dir, download.suggested_filename)
            await download.save_as(save_path)
            
            # 驗證文件
            if self.validate_downloaded_file(save_path):
                print(f"文件驗證成功: {save_path}")
                return save_path
            else:
                print(f"文件驗證失敗: {save_path}")
                return None
                
        except Exception as e:
            print(f"下載過程出錯: {e}")
            return None
    
    def validate_downloaded_file(self, file_path):
        """驗證下載的文件"""
        if not os.path.exists(file_path):
            return False
        
        file_size = os.path.getsize(file_path)
        if file_size == 0:
            return False
        
        # 可以根據(jù)需要添加更多驗證邏輯
        # 例如文件類型、內(nèi)容校驗等
        
        return True

# 使用示例
async def main():
    handler = FileHandler()
    
    async with async_playwright() as p:
        browser = await p.chromium.launch(headless=False)
        page = await browser.new_page()
        
        # 文件上傳示例
        await page.goto('https://example.com/upload')
        await page.set_input_files('input[type="file"]', 'test_file.txt')
        await page.click('#upload-button')
        
        # 等待上傳完成
        upload_success = await handler.wait_for_upload_completion(
            page, 
            success_selector='.upload-success-message'
        )
        
        if upload_success:
            print("文件上傳成功")
        else:
            print("文件上傳失敗或超時")
        
        # 文件下載示例
        await page.goto('https://example.com/download')
        
        download_path = await handler.wait_for_download_completion(
            page,
            download_trigger=lambda: page.click('#download-btn')
        )
        
        if download_path:
            print(f"文件下載并保存到: {download_path}")
        
        await browser.close()

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