from abc import ABC, abstractmethod
from typing import Any
import logging
class BaseSkill(ABC):
    """技能基類 - 所有技能的抽象基類"""
    def __init__(self, name: str):
        self.name = name
        self.logger = logging.getLogger(f"skill.{name}")
    @abstractmethod
    async def execute(self, *args, **kwargs) -> Any:
        """執(zhí)行技能的核心邏輯"""
        pass
    def __str__(self):
        return f"Skill({self.name})"

class SkillOrchestrator:
    """技能編排器 - 負(fù)責(zé)協(xié)調(diào)多個技能的執(zhí)行"""
    def __init__(self):
        self.skills = {}
        self.logger = logging.getLogger("orchestrator")
    def register_skill(self, name: str, skill: BaseSkill):
        """注冊技能"""
        self.skills[name] = skill
        self.logger.info(f"注冊技能: {name}")
    async def execute_workflow(self, workflow: List[Dict]) -> Dict[str, Any]:
        """執(zhí)行工作流"""
        results = {}
        for step in workflow:
            skill_name = step['skill']
            skill_args = step.get('args', {})
            if skill_name not in self.skills:
                self.logger.error(f"未注冊的技能: {skill_name}")
                continue
            try:
                self.logger.info(f"執(zhí)行技能: {skill_name}")
                result = await self.skills[skill_name].execute(**skill_args)
                results[skill_name] = result
                self.logger.info(f"技能執(zhí)行完成: {skill_name}")
            except Exception as e:
                self.logger.error(f"技能執(zhí)行失敗 {skill_name}: {e}")
                results[skill_name] = None
        return results

class BrowserManagerSkill(BaseSkill):
    """瀏覽器管理技能 - 負(fù)責(zé)瀏覽器生命周期管理"""
    def __init__(self, headless: bool = True):
        super().__init__("browser_manager")
        self.headless = headless
        self.browser = None
        self.context = None
        self.page = None
    async def execute(self, setup_only: bool = True) -> tuple:
        """執(zhí)行瀏覽器設(shè)置"""
        try:
            from playwright.async_api import async_playwright
            self.logger.info("啟動瀏覽器...")
            playwright = await async_playwright().start()
            # 啟動瀏覽器
            self.browser = await playwright.chromium.launch(
                headless=self.headless,
                args=[
                    '--no-sandbox',
                    '--disable-setuid-sandbox',
                    '--disable-blink-features=AutomationControlled'
                ]
            )
            # 創(chuàng)建上下文
            self.context = await self.browser.new_context(
                ignore_https_errors=True
            )
            # 創(chuàng)建頁面
            self.page = await self.context.new_page()
            self.logger.info("瀏覽器啟動成功")
            return self.browser, self.context, self.page
        except Exception as e:
            self.logger.error(f"瀏覽器啟動失敗: {e}")
            raise
    async def cleanup(self):
        """清理瀏覽器資源"""
        try:
            if self.context:
                await self.context.close()
            if self.browser:
                await self.browser.close()
            self.logger.info("瀏覽器資源清理完成")
        except Exception as e:
            self.logger.error(f"資源清理失敗: {e}")

class PageNavigationSkill(BaseSkill):
    """頁面導(dǎo)航技能 - 負(fù)責(zé)頁面導(dǎo)航和等待"""
    def __init__(self, page):
        super().__init__("page_navigation")
        self.page = page
    async def execute(self, url: str, timeout: int = 30000) -> bool:
        """導(dǎo)航到指定URL"""
        try:
            self.logger.info(f"導(dǎo)航到: {url}")
            # 設(shè)置頁面頭信息，避免被識別為爬蟲
            await self.page.set_extra_http_headers({
                'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36'
            })
            # 導(dǎo)航到目標(biāo)頁面
            await self.page.goto(url, timeout=timeout)
            # 等待頁面加載完成
            await self.page.wait_for_load_state('networkidle')
            self.logger.info("頁面導(dǎo)航成功")
            return True
        except Exception as e:
            self.logger.error(f"頁面導(dǎo)航失敗: {e}")
            return False

class DataExtractionSkill(BaseSkill):
    """數(shù)據(jù)提取技能 - 負(fù)責(zé)從頁面提取結(jié)構(gòu)化數(shù)據(jù)"""
    def __init__(self, page):
        super().__init__("data_extraction")
        self.page = page
    async def execute(self, extraction_config: Dict) -> List[Dict]:
        """執(zhí)行數(shù)據(jù)提取"""
        try:
            from bs4 import BeautifulSoup
            self.logger.info("開始數(shù)據(jù)提取...")
            # 獲取頁面內(nèi)容
            html_content = await self.page.content()
            soup = BeautifulSoup(html_content, 'html.parser')
            # 根據(jù)配置提取數(shù)據(jù)
            extracted_data = []
            if 'table_selector' in extraction_config:
                table_data = await self._extract_table_data(soup, extraction_config)
                extracted_data.extend(table_data)
            if 'element_selectors' in extraction_config:
                element_data = await self._extract_elements_data(soup, extraction_config)
                extracted_data.extend(element_data)
            self.logger.info(f"數(shù)據(jù)提取完成，共提取 {len(extracted_data)} 條記錄")
            return extracted_data
        except Exception as e:
            self.logger.error(f"數(shù)據(jù)提取失敗: {e}")
            return []

class DataStorageSkill(BaseSkill):
    """數(shù)據(jù)存儲技能 - 負(fù)責(zé)數(shù)據(jù)持久化"""
    def __init__(self, db_path: str = "weather_data.db"):
        super().__init__("data_storage")
        self.db_path = db_path
        self.connection = None
    async def execute(self, data: List[Dict], table_name: str = "weather_data") -> bool:
        """存儲數(shù)據(jù)到數(shù)據(jù)庫"""
        import sqlite3
        try:
            self.logger.info(f"開始存儲 {len(data)} 條數(shù)據(jù)")
            # 連接數(shù)據(jù)庫
            self.connection = sqlite3.connect(self.db_path)
            cursor = self.connection.cursor()
            # 創(chuàng)建表（如果不存在）
            create_table_sql = f"""
            CREATE TABLE IF NOT EXISTS {table_name} (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                province_name TEXT,
                city_name TEXT,
                collection_date TEXT,
                temperature TEXT,
                weather TEXT,
                created_time DATETIME DEFAULT CURRENT_TIMESTAMP
            )
            """
            cursor.execute(create_table_sql)
            # 插入數(shù)據(jù)
            insert_sql = f"""
            INSERT INTO {table_name} (province_name, city_name, collection_date, temperature, weather)
            VALUES (?, ?, ?, ?, ?)
            """
            success_count = 0
            for record in data:
                try:
                    cursor.execute(insert_sql, (
                        record.get('province_name', ''),
                        record.get('city_name', ''),
                        record.get('date', ''),
                        record.get('temperature', ''),
                        record.get('weather', '')
                    ))
                    success_count += 1
                except Exception as e:
                    self.logger.warning(f"插入數(shù)據(jù)失敗: {e}")
            self.connection.commit()
            self.logger.info(f"數(shù)據(jù)存儲完成，成功 {success_count}/{len(data)} 條")
            return True
        except Exception as e:
            self.logger.error(f"數(shù)據(jù)存儲失敗: {e}")
            if self.connection:
                self.connection.rollback()
            return False
        finally:
            if self.connection:
                self.connection.close()

class AdvancedSkillFeatures:
    """高級技能特性示例"""
    @staticmethod
    def skill_factory(skill_type: str, **kwargs) -> BaseSkill:
        """技能工廠方法"""
        skill_map = {
            "browser": BrowserManagerSkill,
            "navigation": PageNavigationSkill,
            "extraction": DataExtractionSkill,
            "storage": DataStorageSkill
        }
        if skill_type not in skill_map:
            raise ValueError(f"未知的技能類型: {skill_type}")
        return skill_map[skill_type](**kwargs)

@staticmethod
def create_skill_pipeline(skills_config: List[Dict]) -> SkillOrchestrator:
    """創(chuàng)建技能管道"""
    orchestrator = SkillOrchestrator()
    for config in skills_config:
        skill = AdvancedSkillFeatures.skill_factory(
            config['type'], 
            **config.get('params', {})
        )
        orchestrator.register_skill(config['name'], skill)
    return orchestrator

# 注入頁面依賴示例
browser_skill = orchestrator.skills["browser"]
browser, context, page = await browser_skill.execute()
# 更新需要頁面的技能
orchestrator.skills["navigator"].page = page
orchestrator.skills["extractor"].page = page

async def main():
    """Skills架構(gòu)使用示例"""
    # 創(chuàng)建技能編排器
    orchestrator = SkillOrchestrator()
    # 創(chuàng)建并注冊技能
    browser_skill = BrowserManagerSkill(headless=True)
    orchestrator.register_skill("browser_manager", browser_skill)
    # 啟動瀏覽器
    browser, context, page = await browser_skill.execute()
    # 注冊其他需要頁面的技能
    navigation_skill = PageNavigationSkill(page)
    extraction_skill = DataExtractionSkill(page)
    storage_skill = DataStorageSkill()
    orchestrator.register_skill("page_navigation", navigation_skill)
    orchestrator.register_skill("data_extraction", extraction_skill)
    orchestrator.register_skill("data_storage", storage_skill)
    # 定義工作流
    workflow = [
        {
            "skill": "page_navigation",
            "args": {
                "url": "https://example.com/weather",
                "timeout": 30000
            }
        },
        {
            "skill": "data_extraction",
            "args": {
                "extraction_config": {
                    "table_selector": ".weather-table",
                    "column_mapping": {
                        "date": 0,
                        "temperature": 1,
                        "weather": 2
                    }
                }
            }
        },
        {
            "skill": "data_storage",
            "args": {
                "data": [],  # 這里應(yīng)該填充實(shí)際提取的數(shù)據(jù)
                "table_name": "weather_records"
            }
        }
    ]
    # 執(zhí)行工作流
    results = await orchestrator.execute_workflow(workflow)
    # 清理資源
    await browser_skill.cleanup()
    return results

# 技能配置
skills_config = [
    {
        "name": "browser",
        "type": "browser",
        "params": {"headless": True}
    },
    {
        "name": "navigator", 
        "type": "navigation",
        "params": {}
    },
    {
        "name": "extractor",
        "type": "extraction", 
        "params": {}
    },
    {
        "name": "storage",
        "type": "storage",
        "params": {"db_path": "config_driven.db"}
    }
]
# 創(chuàng)建技能管道
orchestrator = AdvancedSkillFeatures.create_skill_pipeline(skills_config)

# 技能級別錯誤處理
async def execute_with_retry(self, max_retries=3):
    for attempt in range(max_retries):
        try:
            return await self.execute()
        except TemporaryError as e:
            if attempt == max_retries - 1:
                raise
            await asyncio.sleep(2 ** attempt)
        except PermanentError as e:
            self.logger.error(f"永久性錯誤: {e}")
            raise