PDD/pdd1/process_shop/get_pdd_shop_name.py

import os
import json
from dotenv import load_dotenv
import pymysql
from typing import List, Dict, Optional
import time
from playwright.sync_api import (
    sync_playwright,
    TimeoutError as PlaywrightTimeoutError,
    BrowserContext
)
import requests
from urllib.parse import urlparse
# 补充百度OCR所需依赖
import base64
from PIL import Image
import io

import asyncio

# 加载环境变量
load_dotenv()

# ===================== 全局常量配置（集中管理，方便修改）=====================
# 数据库默认配置
DEFAULT_DB_CONFIG = {
    "host": "localhost",
    "port": 3306,
    "user": "root",
    "password": "",
    "db_name": "",
    "table_name": ""
}


# Playwright配置
PLAYWRIGHT_CONFIG = {
    "headless": False,
    "slow_mo": 300,
    "browser_args": [
        "--start-maximized",
        "--disable-blink-features=AutomationControlled",  # 核心防检测
        "--no-sandbox",  # 兼容Windows/Linux
        "--disable-dev-shm-usage",  # 解决内存不足
        "--disable-popup-blocking",  # 禁用弹窗拦截
        "--disable-extensions",  # 禁用扩展
        "--disable-gpu",  # 禁用GPU加速
        "--lang=zh-CN,zh",  # 中文语言
        "--user-agent=Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/122.0.0.0 Safari/537.36"
    ],
    "viewport": {"width": 2050, "height": 1200},
    "locale": "zh-CN",
    "timezone_id": "Asia/Shanghai",
    "default_timeout": 15000,
    "navigation_timeout": 30000,
    "login_state_path": "pdd_login_state.json",  # 登录状态持久化文件
    "tianyancha_login_state": "tianyancha_login_state.json"  # 天眼查登录状态保存路径
}


# 百度OCR配置
BAIDU_OCR_CONFIG = {
    "api_key": os.getenv('APP_KEY'),
    "secret_key": os.getenv('APP_SECRET'),
    "scale": 1.5  # OCR图片放大倍数
}




# 图片保存配置
IMAGE_CONFIG = {
    "save_dir": "pdd_goods_images",  # 独立文件夹名（项目根目录下）
    "timeout": 10,  # 图片下载超时时间（秒）
    "retry": 1  # 下载失败重试次数
}

# 1. 日期变量赋值SQL
SET_DATE_SQL = "SET @date_constant = CURDATE();"


# 2. 核心查询SQL
QUERY_SQL = """
    SELECT 
      product, 
      shop, 
      product_link, 
      scrape_date, 
      business_license_company, 
      search_key
    FROM (
      SELECT 
        pd.product,
        pd.shop,
        pd.product_link,
        pd.scrape_date,
        psi.business_license_company,
        pd.search_key,
        ROW_NUMBER() OVER (PARTITION BY pd.shop ORDER BY pd.search_key ASC) AS rn
      FROM pdd_drug_middle pd
      LEFT JOIN pdd_shop_info_middle psi ON psi.shop = pd.shop
      WHERE pd.scrape_date >= @date_constant 
        AND psi.business_license_company IS NULL
    ) AS sub
    WHERE rn = 1
    ORDER BY search_key;
"""



# ===================== 百度OCR类（完整整合）=====================
class BaiduOCR:
    """百度 OCR 文字识别封装类"""
    def __init__(self, api_key: str, secret_key: str):
        """
        初始化百度 OCR
        :param api_key: 百度智能云应用的 API Key
        :param secret_key: 百度智能云应用的 Secret Key
        """
        self.api_key = api_key
        self.secret_key = secret_key
        self.access_token: Optional[str] = None
        # 获取 access_token（有效期30天，建议缓存）
        self._get_access_token()

    def _get_access_token(self) -> bool:
        """
        获取百度 OCR 的 access_token（有效期30天）
        :return: 是否获取成功
        """
        url = "https://aip.baidubce.com/oauth/2.0/token"
        params = {
            "grant_type": "client_credentials",
            "client_id": self.api_key,
            "client_secret": self.secret_key
        }
        try:
            response = requests.post(url, params=params, timeout=10)
            response.raise_for_status()
            result = response.json()
            if "access_token" in result:
                self.access_token = result["access_token"]
                print(f"✅ 成功获取 access_token：{self.access_token[:20]}...")
                return True
            else:
                print(f"❌ 获取 access_token 失败：{result}")
                return False
        except Exception as e:
            print(f"❌ 获取 access_token 异常：{e}")
            return False

    def _enlarge_and_crop_image(self, image_path: str, scale: float=1.5, crop_ratio: float=0.5) -> bytes:
        """
        先裁剪图片上半部分（保留有效内容），再放大图片（解决OCR尺寸错误）
        :param image_path: 原图路径
        :param scale: 放大倍数（推荐1.5~3.0）
        :param crop_ratio: 裁剪比例（0.5=保留上50%，0.6=保留上60%，可根据图片调整）
        :return: 处理后的图片二进制数据
        """
        try:
            with Image.open(image_path) as img:
                # ========== 步骤1：裁剪上半部分（核心修复） ==========
                # 计算裁剪区域：左=0，上=0，右=原图宽度，下=原图高度×裁剪比例
                crop_box = (
                    0,  # 左边界
                    0,  # 上边界
                    img.width,  # 右边界
                    int(img.height * crop_ratio)  # 下边界（只保留上半部分）
                )
                img_cropped = img.crop(crop_box)  # 执行裁剪
                print(f"✅ 图片裁剪完成：保留上{int(crop_ratio * 100)}%区域，尺寸={img_cropped.size}")

                # ========== 步骤2：放大裁剪后的图片 ==========
                new_width = int(img_cropped.width * scale)
                new_height = int(img_cropped.height * scale)
                # 高质量放大（Lanczos算法，最清晰）
                img_resized = img_cropped.resize(
                    (new_width, new_height),
                    Image.Resampling.LANCZOS
                )

                # ========== 仅5行，强制缩到4096×4096以内（核心微调） ==========
                MAX_OCR_SIZE = 4096  # 百度OCR最大允许宽度/高度
                if img_resized.width > MAX_OCR_SIZE:
                    ratio = MAX_OCR_SIZE / img_resized.width  # 计算缩放比例
                    img_resized = img_resized.resize(
                        (MAX_OCR_SIZE, int(img_resized.height * ratio)),
                        Image.Resampling.LANCZOS
                    )



                if img_resized.mode == 'RGBA':
                    # 创建白色背景的RGB画布
                    rgb_img = Image.new('RGB', img_resized.size, (255, 255, 255))
                    # 将RGBA图片粘贴到RGB画布（透明区域显示白色）
                    rgb_img.paste(img_resized, mask=img_resized.split()[3])  # mask=alpha通道
                    img_resized = rgb_img

                # # ========== 保存处理后图片到本地 ==========
                # # 1. 确保pdd_goods_images文件夹存在（不存在则创建）
                # save_dir = "pdd_goods_images"
                # if not os.path.exists(save_dir):
                #     os.makedirs(save_dir)
                # # 2. 提取原图片文件名（比如从image_path中拿到"鸿祥堂大药房旗舰店_1773649991220.jpeg"）
                # file_name = os.path.basename(image_path)
                # # 3. 拼接保存路径
                # save_path = os.path.join(save_dir, file_name)
                # # 4. 保存图片到本地（质量和OCR用的一致）
                # img_resized.save(save_path, format='JPEG', quality=95)
                # print(f"✅ 处理后图片已保存到：{save_path}")
                # # ======================================================


                # 保存到内存（不生成本地文件）
                img_byte_arr = io.BytesIO()
                # 保存为 JPG，保证清晰度
                img_resized.save(img_byte_arr, format='JPEG', quality=95)
                img_byte_arr = img_byte_arr.getvalue()

                # 校验文件大小（超4MB则再次压缩）
                file_size = len(img_byte_arr) / 1024 / 1024  # 转MB
                if file_size > 4:
                    print(f"⚠️ 文件超4MB（{file_size:.2f}MB），二次压缩...")
                    img_byte_arr = io.BytesIO()
                    img_resized.save(img_byte_arr, format='JPEG', quality=70, optimize=True)
                    img_byte_arr = img_byte_arr.getvalue()

                # 打印最终尺寸（方便调试）
                print(f"✅ 图片放大完成：最终尺寸={img_resized.size}")
                return img_byte_arr
        except Exception as e:
            print(f"❌ 图片裁剪/放大失败：{str(e)}")
            return b''

    def general_ocr(self, image_path: str, scale: float = 1.5) -> Optional[Dict]:
        """
        调用百度通用文字识别（支持图片放大）
        :param image_path: 本地图片路径
        :param scale: 放大倍数，默认2倍
        :return: OCR识别结果
        """
        if not self.access_token:
            print("❌ access_token 无效，请先初始化")
            return None

        try:
            """
            百度OCR通用识别（整合裁剪+放大）
            """
            # 替换原放大逻辑为「裁剪+放大」
            image_data = self._enlarge_and_crop_image(image_path, scale=scale, crop_ratio=0.5)
            if not image_data:
                print("❌ 图片处理失败，无法识别")
                return {}
            image_base64 = base64.b64encode(image_data).decode("utf-8")

        except Exception as e:
            print(f"❌ 图片放大/读取失败：{e}")
            return None

        # 调用 OCR 接口
        url = f"https://aip.baidubce.com/rest/2.0/ocr/v1/general_basic?access_token={self.access_token}"
        headers = {"Content-Type": "application/x-www-form-urlencoded"}
        data = {"image": image_base64}

        try:
            response = requests.post(url, headers=headers, data=data, timeout=10)
            response.raise_for_status()
            result = response.json()
            if "words_result" in result:
                print(f"✅ 识别成功，共识别到 {len(result['words_result'])} 行文字")
                return result
            else:
                print(f"❌ 识别失败：{result}")
                return None
        except Exception as e:
            print(f"❌ 调用 OCR 接口异常：{e}")
            return None

    def extract_enterprise_info(self, ocr_result: Dict) -> Dict:
        """
        从OCR识别结果中提取企业名称和社会信用代码
        :param ocr_result: general_ocr 返回的识别结果字典
        :return: 包含企业名称和社会信用代码的JSON格式字典
                 格式：{"enterprise_name": "企业名称", "credit_code": "社会信用代码"}
        """
        # 初始化返回结果（默认空值）
        enterprise_info = {
            "enterprise_name": "",
            "credit_code": ""
        }

        if not ocr_result or "words_result" not in ocr_result:
            print("❌ OCR识别结果为空，无法提取企业信息")
            return enterprise_info

        all_text_lines = []
        # 遍历所有识别的文字行，匹配关键词
        for item in ocr_result["words_result"]:
            line_text = item["words"].strip()  # 去除首尾空格
            if line_text and line_text not in all_text_lines:  # 去空+去重
                all_text_lines.append(line_text)
        print(f"📝 OCR识别的有效行：{all_text_lines}")

        # ==================== 1. 提取并清洗企业名称 ====================
        enterprise_name = ""
        # 名称匹配关键词（覆盖所有场景）
        name_keywords = ["企业名称", "名称：", "名：", "称："]
        # 常见错别字修正映射
        name_correction = {
            "人药房": "大药房",
            "有松司": "有限公司",
            "松司": "公司",
            "关药房": "大药房"
        }

        for idx, line_text in enumerate(all_text_lines):
            # 场景1：包含"企业名称"（处理括号冗余，如"企业名称（名称xxx"）
            if "企业名称" in line_text:
                # 移除所有括号及内部内容，再提取名称
                import re
                # 正则移除括号（()/（））及内容
                clean_line = re.sub(r'\([^)]*\)|\（[^）]*\）', '', line_text)
                # 提取"企业名称"后的所有内容
                name_part = clean_line.split("企业名称")[-1].strip()
                # 若还有"名称"前缀，继续拆分
                if "名称" in name_part:
                    name_part = name_part.split("名称")[-1].strip()
                enterprise_name = name_part
                break
            # 场景2：单行包含"名称："/"名："/"称："
            elif any(key in line_text for key in ["名称：", "名：", "称："]):
                name_part = line_text.split("：")[-1].strip() if "：" in line_text else line_text.split(":")[-1].strip()
                enterprise_name = name_part
                break
            # 场景3：跨行拆分（前一行是"名"，当前行以"称："开头）
            elif idx > 0 and all_text_lines[idx - 1] == "名" and line_text.startswith("称："):
                name_part = line_text.split("：")[-1].strip()
                enterprise_name = name_part
                break

        # 清洗企业名称：修正错别字、移除多余空格
        if enterprise_name:
            for wrong, right in name_correction.items():
                enterprise_name = enterprise_name.replace(wrong, right)

            # 步骤2：移除开头/结尾的冒号（中文+英文）、空格、特殊符号
            enterprise_name = enterprise_name.strip("：: \t\n\r")
            # 步骤3：移除中间多余空格
            enterprise_name = enterprise_name.replace(" ", "")  # 移除所有空格

        # ==================== 2. 提取并清洗社会信用代码 ====================
        credit_code = ""
        # 信用代码匹配关键词（兼容错别字+多格式）
        code_keywords = ["社会信用代码：", "统一社会信用代码：", "社会震用代码："]
        for line_text in all_text_lines:
            # 匹配任意关键词
            for keyword in code_keywords:
                if keyword in line_text:
                    code_part = line_text.split(keyword)[-1].strip()
                    credit_code = code_part.replace(" ", "")  # 移除空格（如"91360105 MAEGBDKMXF"）
                    break
            if credit_code:  # 找到后跳出循环
                break

        # ==================== 赋值并打印结果 ====================
        enterprise_info["enterprise_name"] = enterprise_name
        enterprise_info["credit_code"] = credit_code
        # enterprise_info["address"] = address

        # 打印提取结果
        if enterprise_name:
            print(f"✅ 提取到企业名称：{enterprise_name}")
        else:
            print("⚠️ 未识别到企业名称字段")

        if credit_code:
            print(f"✅ 提取到社会信用代码：{credit_code}")
        else:
            print("⚠️ 未识别到社会信用代码字段")

        return enterprise_info






# ===================== 数据库读取类 =====================
class DBGoodsReader:
    """数据库商品链接读取器"""

    def __init__(
            self,
            host: str = DEFAULT_DB_CONFIG["host"],
            port: int = DEFAULT_DB_CONFIG["port"],
            user: str = DEFAULT_DB_CONFIG["user"],
            password: str = DEFAULT_DB_CONFIG["password"],
            db_name: str = DEFAULT_DB_CONFIG["db_name"],
            charset: str = "utf8mb4"
    ):
        self.host = host
        self.port = port
        self.user = user
        self.password = password
        self.db_name = db_name
        self.charset = charset
        self.conn: Optional[pymysql.connections.Connection] = None
        self.cursor: Optional[pymysql.cursors.DictCursor] = None




    def connect_db(self) -> bool:
        """连接数据库（带重试机制）"""
        max_retry = 2
        for retry in range(max_retry + 1):
            try:
                self.conn = pymysql.connect(
                    host=self.host,
                    port=self.port,
                    user=self.user,
                    password=self.password,
                    database=self.db_name,
                    charset=self.charset,
                    cursorclass=pymysql.cursors.DictCursor,
                    connect_timeout=10  # 连接超时
                )
                self.cursor = self.conn.cursor()
                print(f"✅ 成功连接数据库：{self.db_name}")
                return True
            except pymysql.MySQLError as e:
                if retry < max_retry:
                    print(f"❌ 数据库连接失败（重试{retry + 1}/{max_retry}）：{e}")
                    time.sleep(1)
                    continue
                print(f"❌ 数据库连接最终失败：{e}")
                return False

    def get_shop_and_goods(self) -> List[Dict]:
        """
        读取待补充企业信息的店铺数据（每个店铺仅取1条）
        返回：包含product/shop/product_link等字段的字典列表
        """
        if not self.conn or not self.cursor:
            print("❌ 未连接数据库，请先调用 connect_db()")
            return []

        try:
            # 步骤1：执行日期变量赋值
            self.cursor.execute(SET_DATE_SQL)

            # 步骤2：执行核心查询
            self.cursor.execute(QUERY_SQL)

            # 步骤3：获取结果（DictCursor返回字典格式，字段名对应SQL列名）
            results = self.cursor.fetchall()
            print(f"✅ 成功读取 {len(results)} 条待补充企业信息的店铺数据")
            return results

        except pymysql.MySQLError as e:
            print(f"❌ 读取数据失败：{e}")
            return []

    def _get_next_id(self) -> int:
        """获取表中最大ID并+1，用于生成新插入数据的ID（若ID非自增）"""
        try:
            sql = "SELECT IFNULL(MAX(id), 0) + 1 AS next_id FROM pdd_shop_info_middle"
            self.cursor.execute(sql)
            result = self.cursor.fetchone()
            next_id = result.get("next_id", 9078)  # 默认初始值9078（兼容示例）
            print(f"✅ 获取到下一个可用ID：{next_id}")
            return next_id
        except pymysql.MySQLError as e:
            print(f"❌ 获取自增ID失败，使用默认值9078：{e}")
            return 9078

    def insert_enterprise_info(self, shop_name: str, enterprise_info: Dict) -> bool:
        """
        向pdd_shop_info_middle表插入企业信息（替代原更新逻辑）
        :param shop_name: 店铺名称（关联表的shop字段）
        :param enterprise_info: 包含tyc_company_name/tyc_company_code/tyc_company_address的字典
        :return: 插入是否成功
        """
        if not self.conn or not self.cursor:
            print("❌ 未连接数据库，请先调用 connect_db()")
            return False

        if not shop_name:
            print("❌ 店铺名称为空，无法更新")
            return False

        business_company_name = enterprise_info.get("tyc_company_name", "").strip()
        qualification_number = enterprise_info.get("tyc_company_code", "").strip()
        contact_address = enterprise_info.get("tyc_company_address", "").strip()
        business_license_address = contact_address  # 两个地址字段都用同一个值

        # 空值校验提示
        empty_fields = []
        if not business_company_name:
            empty_fields.append("企业名称")
        if not qualification_number:
            empty_fields.append("统一信用代码")
        if not contact_address:
            empty_fields.append("企业地址")
        if empty_fields:
            print(f"⚠️ 店铺[{shop_name}]以下字段为空：{','.join(empty_fields)}，仍继续插入（空值）")


        # 生成插入ID（若表ID为自增主键，可删除ID相关逻辑，SQL中也去掉id字段）
        next_id = self._get_next_id()

        insert_sql = """
                    INSERT INTO `test2`.`pdd_shop_info_middle` (
                        `id`, 
                        `shop`, 
                        `contact_address`, 
                        `qualification_number`, 
                        `business_license_company`, 
                        `business_license_address`, 
                        `scrape_date`, 
                        `platform`, 
                        `province`, 
                        `city`, 
                        `create_time`, 
                        `update_time`
                    ) VALUES (%s, %s, %s, %s, %s, %s, CURDATE(), '拼多多', '', '', NOW(), NOW())
                """

        # 组装插入参数
        insert_params = [
            next_id,
            shop_name,
            contact_address,
            qualification_number,
            business_company_name,
            business_license_address
        ]

        try:
            # 执行插入
            self.cursor.execute(insert_sql, insert_params)
            self.conn.commit()

            # 检查影响行数
            affected_rows = self.cursor.rowcount
            if affected_rows > 0:
                print(f"✅ 店铺[{shop_name}]成功插入1条数据（ID：{next_id}）")
                print(f"   插入内容：企业名称={business_company_name} | 信用代码={qualification_number} | 地址={contact_address}")
                return True
            else:
                print(f"⚠️ 店铺[{shop_name}]插入0行数据，无数据变更")
                return False
        except pymysql.MySQLError as e:
            print(f"❌ 店铺[{shop_name}]插入失败：{e}")
            self.conn.rollback()  # 回滚事务
            return False
        except Exception as e:
            print(f"❌ 店铺[{shop_name}]插入异常：{e}")
            self.conn.rollback()
            return False





    def close_db(self) -> None:
        """安全关闭数据库连接"""
        if self.cursor:
            try:
                self.cursor.close()
            except Exception:
                pass
        if self.conn:
            try:
                self.conn.close()
                print("✅ 数据库连接已关闭")
            except Exception:
                pass







# ===================== 天眼查浏览器类 =====================
class TianyanchaBrowser:
    """天眼查浏览器：自动打开、登录、搜索企业名"""
    def __init__(self):
        self.pw = None
        self.browser = None
        self.context: Optional[BrowserContext] = None
        self.page = None
        self.login_state_path = PLAYWRIGHT_CONFIG["tianyancha_login_state"]



    def check_scan_login_prompt(self):
        """
        检查是否出现「扫码登录」提示，若出现则暂停并提示手动扫码
        """
        try:
            # 定位扫码登录提示文本（结合父div，避免误匹配其他页面文本）
            scan_prompt_locator = self.page.locator(
                "div.scan-title",
                has_text="扫码登录 更快 更安全"
            )

            scan_prompt_locator.wait_for(
                state="visible",
                timeout=10000  # 超时10秒，可根据网络调整
            )

            # 提示出现，暂停脚本让你手动扫码
            print("⚠️ 检测到天眼查扫码登录提示！")
            input("请打开天眼查APP扫码完成登录后，按回车键继续执行脚本...")

        except PlaywrightTimeoutError:
            # 超时未出现，说明无需扫码，直接继续
            print("✅ 未检测到扫码登录提示，跳过扫码步骤")


    def _load_login_state(self) -> Optional[Dict]:
        """加载本地登录状态"""
        if os.path.exists(self.login_state_path):
            try:
                with open(self.login_state_path, "r", encoding="utf-8") as f:
                    return json.load(f)
            except json.JSONDecodeError:
                print(f"⚠️ 天眼查登录状态文件损坏，将重新登录")
                os.remove(self.login_state_path)
        return None


    def _save_login_state(self) -> None:
        """保存登录状态"""
        if self.context:
            try:
                self.context.storage_state(path=self.login_state_path) # 同步保存
                print(f"✅ 天眼查登录状态已保存到：{self.login_state_path}")
            except Exception as e:
                print(f"⚠️ 天眼查登录状态保存失败：{e}")

    def init_browser(self, pw) -> bool:  # ✅ 保持async
        """初始化天眼查浏览器"""
        try:
            self.pw = pw  # ✅ 核心修改：直接使用传进来的 playwright 引擎，不再自己 start()
            # 启动防检测浏览器
            self.browser = self.pw.chromium.launch(
                headless=PLAYWRIGHT_CONFIG["headless"],
                slow_mo=PLAYWRIGHT_CONFIG["slow_mo"],
                args=PLAYWRIGHT_CONFIG["browser_args"],
                ignore_default_args=["--enable-automation"],
                timeout=60000
            )


            # 加载登录状态或手动登录
            login_state = self._load_login_state()
            if login_state:
                self.context = self.browser.new_context(
                    viewport=None,  # ✅ 设为None，适配最大化窗口
                    locale=PLAYWRIGHT_CONFIG["locale"],
                    timezone_id=PLAYWRIGHT_CONFIG["timezone_id"],
                    ignore_https_errors=True,
                    storage_state=login_state  # ✅ 加载已保存的登录状态
                )
                print("✅ 已加载天眼查本地登录状态")
            else:  # ✅ 无登录状态：提示手动登录
                self.context = self.browser.new_context(
                    viewport=None,
                    locale=PLAYWRIGHT_CONFIG["locale"],
                    timezone_id=PLAYWRIGHT_CONFIG["timezone_id"],
                    ignore_https_errors=True,
                )



            # 初始化页面
            self.page = self.context.new_page()

            # self.page.window_maximize()  # ✅ 强制窗口最大化（兜底）

            self.page.set_default_timeout(PLAYWRIGHT_CONFIG['default_timeout'])
            self.page.set_default_navigation_timeout(PLAYWRIGHT_CONFIG['navigation_timeout'])
            return True
        except Exception as e:
            print(f"❌ 天眼查浏览器初始化失败：{e}")
            self.close()  # ✅ await关闭
            return False


    def search_enterprise(self, enterprise_name: str) -> bool:
        """
        打开天眼查并搜索指定企业名
        :param enterprise_name: 从OCR提取的企业名称
        :return: 搜索是否成功
        """
        if not self.page:
            print("❌ 天眼查浏览器未初始化！")
            return False
        if not enterprise_name or enterprise_name.strip() == '':
            print("❌ 企业名称为空！无法搜索")
            return False

        try:
            #设置一个计数器，去往官网只运行一次


            # 1. 打开天眼查首页（替换你指定的链接）
            print(f"\n📌 打开天眼查：https://www.tianyancha.com/")
            self.page.goto(
                "https://www.tianyancha.com/",
                wait_until="networkidle",
                timeout=30000
            )
            # input("天眼查登录")

            # 2. 定位天眼查搜索框（适配最新页面结构）
            # 搜索框selector：优先用placeholder匹配，兼容不同版本
            # 先检查是否需要扫码登录
            # self.check_scan_login_prompt()

            # ========== 核心：自动检测并处理登录 ==========
            # 定位「登录/注册」按钮（完全匹配你提供的HTML结构）
            login_button = self.page.locator(
                "div.tyc-header-nav-item.tyc-nav-user span.tyc-nav-user-btn",
                has_text="登录/注册"
            ).nth(0)

            try:
                # 等待按钮出现（最多10秒），如果出现说明未登录
                login_button.wait_for(state="visible", timeout=10000)
                print("⚠️ 检测到未登录状态，正在点击「登录/注册」按钮...")
                login_button.click()  # 点击按钮，唤起扫码登录弹窗

                # 提示你手动扫码登录
                print("\n🔔 请打开天眼查APP，扫描页面上的登录二维码，只有四十秒，登录完成后按回车键继续...")

                # 等待登录完成：等待「登录/注册」按钮消失（说明已成功登录）
                self.page.wait_for_selector(
                    "div.tyc-header-nav-item.tyc-nav-user span.tyc-nav-user-btn",
                    state="hidden",  # 等待元素隐藏
                    timeout=40000  # 最多等30秒，给足扫码时间
                )
                print("✅ 扫码登录成功！")
            except PlaywrightTimeoutError:
                # 10秒内没找到「登录/注册」按钮 → 说明已经处于登录状态
                print("✅ 检测到已登录状态，无需重复登录")

            print("\n⚠️ 请先完成天眼查登录！")
            # self.page = self.context.new_page()  # ✅ await创建页面
            # self.page.goto("https://www.tianyancha.com", timeout=30000)  # ✅ await跳转
            # input("请在浏览器中完成天眼查登录，登录后按回车继续...")
            self.context.storage_state(path=self.login_state_path)  # ✅ await保存状态
            print(f"✅ 天眼查登录状态已保存到：{self.login_state_path}")


            search_locator = None
            try:
                # 优先定位：placeholder匹配
                search_locator = self.page.locator('input[placeholder="请输入公司名称、老板姓名、品牌名称等"]')
                if search_locator.count() > 1:
                    search_locator = self.page.locator('input[placeholder="请输入公司名称、老板姓名、品牌名称等"]').nth(1)
                else:
                    search_locator = self.page.locator('input[placeholder="请输入公司名称、老板姓名、品牌名称等"]')
                # # 备用定位：ID匹配
                # if not search_locator.count():
                #     search_locator = self.page.locator('input#header-company-search')

                # 等待搜索框加载（超时会触发TimeoutError）
                search_locator.wait_for(timeout=10000, state="visible")
                print("✅ 定位到天眼查搜索框")
            except PlaywrightTimeoutError:
                print(f"❌ 搜索框定位超时：页面加载过慢或搜索框元素不存在")
                return False
            except Exception as e:
                print(f"❌ 搜索框定位失败：{str(e)}")
                return False



            # 3. 清空搜索框 + 输入企业名 + 回车搜索
            search_locator.click()
            search_locator.clear()
            print(f"📌 输入企业名：{enterprise_name}")
            # 模拟真人输入延迟
            search_locator.fill(enterprise_name)
            self.page.wait_for_timeout(1000)

            # 推荐：直接用键盘回车触发搜索，这在大部分前端框架中最稳定
            search_locator.press("Enter")
            print("🖱️ 已触发回车搜索")
            #点击搜索按钮
            # search_btn = self.page.locator("button.50ab4.tyc-header-suggest-button_52bf6")
            # await  search_btn.click()  # 回车搜索

            # 4. 等待搜索结果加载
            self.page.wait_for_load_state("networkidle", timeout=20000)
            print(f"✅ 天眼查搜索完成！已搜索：{enterprise_name}")
            return True

        except PlaywrightTimeoutError:
            print(f"❌ 天眼查搜索超时（企业名：{enterprise_name}）")
            return False
        except Exception as e:
            print(f"❌ 天眼查搜索异常：{e}")
            return False

    def get_enterprise_info(self) -> Dict:
        """
        从天眼查搜索结果页提取核心字段（可自定义字段）
        返回：包含三个字段的字典（示例：法定代表人、注册资本、成立日期）
        """
        enterprise_detail = {
            "tyc_company_name": "",     # 公司名
            "tyc_company_code": "",             # 统一社会信用代码
            "tyc_company_address": ""           # 成立日期
        }
        if not self.page:
            print("❌ 天眼查页面未初始化")
            return enterprise_detail
        try:
            # 等待详情页加载
            # self.page.('div.company-header-container', timeout=8000)
            self.page.wait_for_timeout(timeout=4000)
            try:
                # 提取公司名，可能会出现很多个结果，但路径都一样，一般取第一个。
                company_name_locator = self.page.locator("div.index_name__qEdWi span").nth(0)
                if company_name_locator.count():
                    company_name = company_name_locator.inner_text()
                    enterprise_detail['tyc_company_name'] = company_name.strip()
                    print(f"获取到公司名：{enterprise_detail['tyc_company_name']}")
                else:
                    print(f"没有获取到企业名，网页路径有问题")
            except Exception as e:
                input("提取企业元素发生问题，检查一下")
                print(f"提取企业名时发生异常：{str(e)}，网页路径或元素定位异常")
                enterprise_detail['tyc_company_name'] = ""

            try:
                # 提取统一社会信用代码
                code_locator = self.page.locator("div.index_info-col__UVcZb.index_credit-code__kWuDZ span").nth(0)
                if code_locator.count():
                    code = code_locator.inner_text()
                    enterprise_detail['tyc_company_code'] = code.strip()
                    print(f"获取到企业信用代码：{enterprise_detail['tyc_company_code']}")
                else:
                    print(f"没有获取到企业信用代码，网页路径有问题")
            except Exception as e:
                print(f"提取统一社会信用代码时发生异常：{str(e)}，网页路径或元素定位异常")
                enterprise_detail['tyc_company_code'] = ""

            try:
                address_locator = self.page.locator("div.index_contact-col__7AboU.index_address__mHjQD .index_value__Pl0Nh").nth(0)
                if address_locator.count():
                    address = address_locator.inner_text()
                    enterprise_detail['tyc_company_address'] = address.strip()
                    print(f"获取到企业地址：{enterprise_detail['tyc_company_address']}")
                else:
                    print(f"没有获取到企业地址，网页路径有问题")
            except Exception as e:
                print(f"提取企业地址时发生异常：{str(e)}，网页路径或元素定位异常")
                enterprise_detail['tyc_company_address'] = ""



            print("\n📌 提取的企业核心信息：")
            print(f"公司名：{enterprise_detail['tyc_company_name']}")
            print(f"企业信用代码：{enterprise_detail['tyc_company_code']}")
            print(f"企业地址：{enterprise_detail['tyc_company_address']}")
            return enterprise_detail

        except Exception as e:
            print(f"❌ 提取企业信息失败：{e}")
            return enterprise_detail



    def close(self) -> None:
        """关闭浏览器"""
        if self.page:
            try:
                self.page.close()
            except Exception:
                pass
        if self.context:
            try:
                self.context.close()
            except Exception:
                pass
        if self.browser:
            try:
                self.browser.close()
                print("✅ 天眼查浏览器已关闭")
            except Exception:
                pass










class PddLinkBrowser:
    """拼多多链接浏览器（支持登录持久化+图片下载+OCR识别）"""

    def __init__(self, login_state_path: str = PLAYWRIGHT_CONFIG["login_state_path"]):
        self.login_state_path = login_state_path
        self.browser = None
        self.context: Optional[BrowserContext] = None
        self.page = None
        # 初始化图片保存文件夹
        self._init_image_dir()

        # 初始化百度OCR实例
        self.ocr_client: Optional[BaiduOCR] = None
        self._init_ocr_client()

        # ========== 初始化天眼查浏览器 ==========
        self.tyc_browser = TianyanchaBrowser()

    def _init_ocr_client(self):
        """初始化百度OCR客户端"""
        api_key = BAIDU_OCR_CONFIG["api_key"]
        secret_key = BAIDU_OCR_CONFIG["secret_key"]
        if not api_key or not secret_key:
            print("⚠️ 未配置百度OCR的API_KEY/SECRET_KEY，将跳过OCR识别")
            return
        self.ocr_client = BaiduOCR(api_key=api_key, secret_key=secret_key)
        print("✅ 百度OCR客户端初始化完成")



    # ========== 检测手机登录框并等待手动登录 ==========
    def _check_login_box(self) -> bool:
        """
        检测是否出现「手机登录」框（div.phone-login 包含"手机登录"文本）
        :return: True=检测到并完成登录；False=未检测到登录框
        """
        if not self.page:
            print("❌ 页面未初始化，无法检测登录框")
            return False
        try:
            # 精准定位登录框元素：div.phone-login 下的 span（包含"手机登录"文本）
            login_locator = self.page.locator("div.phone-login span")
            # 等待元素可见（最多5秒，超时则认为无登录框）
            login_locator.wait_for(timeout=5000, state="visible")
            # 获取元素文本（包含伪元素的"手机登录"）
            login_text = login_locator.inner_text().strip()

            if "手机登录" in login_text:
                print("\n⚠️ 检测到【手机登录】框，请手动完成登录！")
                input("登录完成后，请按回车键继续执行脚本...")
                # 登录后等待页面重新加载完成（确保登录状态生效）
                self.page.wait_for_load_state("networkidle", timeout=15000)
                print("✅ 登录已完成，继续处理当前商品")
                return True
            return False
        except PlaywrightTimeoutError:
            # 超时说明页面没有登录框，直接返回False
            return False
        except Exception as e:
            print(f"⚠️ 检测登录框时发生异常：{str(e)[:60]}，继续执行")
            return False
    # ========== 登录检测方法结束 ==========




    # ========== 自定义向下滑动方法 ==========
    def _scroll_down(self, distance: int = 500, step: int = 50, interval: int =100):
        """
        分步向下滑动指定距离（解决图片懒加载，避免一次性跳转）
        :param distance: 总滑动距离（像素，默认500）
        :param step: 每次滑动的步长（像素，默认50，越小越平缓）
        :param interval: 每次滑动后的间隔时间（毫秒，默认100）
        """
        if not self.page:
            print("❌ 浏览器页面未初始化，无法滑动")
            return

        # 容错处理：步长/总距离为非正数时直接返回
        if step <= 0 or distance <= 0:
            print(f"⚠️ 无效的滑动参数（总距离：{distance}，步长：{step}），跳过滑动")
            return


        try:
            remaining = distance  # 剩余未滑动的距离
            print(f"📝 开始分步滑动：总距离{distance}像素，每次滑{step}像素，间隔{interval}ms")

            while remaining > 0:
                current_step = min(step, remaining)
                self.page.evaluate(f"window.scrollBy(0, {current_step})")
                remaining -= current_step
                self.page.wait_for_timeout(interval)

            self.page.wait_for_timeout(2000)
            print(f"✅ 分步滑动完成，总滑动距离：{distance}像素")
        except Exception as e:
            print(f"⚠️ 分步滑动失败：{str(e)[:50]}")

    # ========== 滑动方法结束 ==========



    def _init_image_dir(self):
        """创建图片保存文件夹（不存在则创建）"""
        if not os.path.exists(IMAGE_CONFIG["save_dir"]):
            os.makedirs(IMAGE_CONFIG['save_dir'])
            print(f"✅ 图片保存文件夹已创建：{os.path.abspath(IMAGE_CONFIG['save_dir'])}")
        else:
            print(f"✅ 图片保存文件夹已存在：{os.path.abspath(IMAGE_CONFIG['save_dir'])}")


    def _get_image_filename(self, img_src: str, shop_name: str) -> str:
        """生成唯一的图片文件名（避免重复）"""
        # 提取原始文件名后缀（如.png/.jpg）
        parsed_url = urlparse(img_src)
        ext = os.path.splitext(parsed_url.path)[-1] or '.png'
        # 清洗店铺名（避免特殊字符）
        clean_shop = "".join([c for c in shop_name if c.isalnum() or c in ["_", "-"]])[:20]
        # 时间戳+店铺名+随机数，确保唯一
        timestamp = str(int(time.time() * 1000))
        filename = f"{clean_shop}_{timestamp}{ext}"
        return filename

    def _download_image(self, img_src: str, shop_name: str) -> Optional[str]:
        """
        下载图片到指定文件夹
        :return: 成功返回保存路径，失败返回None
        """
        if not img_src:
            print("⚠️ 图片链接为空，跳过下载")
            return None

        # 生成唯一文件名
        filename = self._get_image_filename(img_src, shop_name)
        save_path = os.path.join(IMAGE_CONFIG["save_dir"], filename)

        # 下载重试逻辑
        for retry in range(IMAGE_CONFIG["retry"] + 1):
            try:
                # 发送请求下载图片（添加headers模拟浏览器）
                headers = {
                    "User-Agent": PLAYWRIGHT_CONFIG["browser_args"][-1].split("=")[1],
                    "Referer": "https://www.pinduoduo.com/",
                    "Accept": "image/avif,image/webp,image/apng,image/svg+xml,image/*,*/*;q=0.8"  # 新增：支持jpeg格式
                }
                response = requests.get(
                    img_src,
                    headers=headers,
                    timeout=IMAGE_CONFIG["timeout"],
                    stream=True, # 流式下载，避免内存溢出
                    allow_redirects=True  # 显式开启重定向（拼多多签名链接可能302）
                )
                response.raise_for_status()  # 抛出HTTP错误（4xx/5xx）

                # 保存图片到文件
                with open(save_path, "wb") as f:
                    for chunk in response.iter_content(chunk_size=8192):
                        f.write(chunk)

                # 校验文件是否保存成功
                if os.path.getsize(save_path) > 0:
                    print(f"✅ 图片下载成功：{save_path}")
                    return save_path
                else:
                    os.remove(save_path)  # 删除空文件
                    print(f"⚠️ 图片下载为空，重试{retry+1}/{IMAGE_CONFIG['retry']}")
            except requests.exceptions.HTTPError as e:
                if e.response.status_code == 403:
                    print(f"❌ 图片签名过期/无权限：{img_src[:50]}...")
                    return None  # 403无需重试，直接跳过
                elif retry < IMAGE_CONFIG["retry"]:
                    print(f"⚠️ HTTP错误（重试{retry + 1}/{IMAGE_CONFIG['retry']}）：{e}")
                    time.sleep(1)
                    continue
                print(f"❌ 图片下载失败：{e}")
                return None
            except Exception as e:
                if retry < IMAGE_CONFIG["retry"]:
                    print(f"⚠️ 下载失败（重试{retry + 1}/{IMAGE_CONFIG['retry']}）：{str(e)[:50]}")
                    time.sleep(1)
                    continue
                print(f"❌ 图片最终下载失败：{str(e)[:50]}")
                return None



    def _process_ocr(self, image_path: str) -> Optional[Dict]:
        """OCR识别后立即调用天眼查搜索+提取企业信息"""
        """
        调用OCR识别并提取企业信息
        :param image_path: 图片路径
        :return: 企业信息字典
        """
        if not self.ocr_client:
            print("⚠️ OCR客户端未初始化，跳过识别")
            return None

        if not os.path.exists(image_path):
            print(f"❌ 图片文件不存在：{image_path}")
            return None

        # 调用OCR识别
        ocr_result = self.ocr_client.general_ocr(
            image_path=image_path,
            scale=BAIDU_OCR_CONFIG["scale"]
        )

        print(f"识别结果{ocr_result}")
        if not ocr_result:
            return None

        # 提取企业信息
        enterprise_info = self.ocr_client.extract_enterprise_info(ocr_result)
        print("\n📌 提取的企业信息：")
        enterprise_name = enterprise_info.get("enterprise_name", "")
        if not enterprise_name:
            print("⚠️ 未提取到企业名称，跳过天眼查")
            return enterprise_info

        # 2. 调用天眼查搜索+提取字段
        if self.tyc_browser.search_enterprise(enterprise_name):
            # 提取三个核心字段
            tyc_info = self.tyc_browser.get_enterprise_info()
            # 合并OCR结果和天眼查字段
            enterprise_info.update(tyc_info)



        print("\n📌 最终整合结果：")
        print(json.dumps(enterprise_info, ensure_ascii=False, indent=4))
        return enterprise_info





    def _load_login_state(self) -> Optional[Dict]:
        """加载本地登录状态"""
        if os.path.exists(self.login_state_path):
            try:
                with open(self.login_state_path, "r", encoding="utf-8") as f:
                    return json.load(f)
            except json.JSONDecodeError:
                print(f"⚠️ 登录状态文件损坏：{self.login_state_path}，将重新登录")
                os.remove(self.login_state_path)
        return None


    def _save_login_state(self) -> None:
        """保存登录状态到本地"""
        if self.context:
            try:
                self.context.storage_state(path=self.login_state_path)
                print(f"✅ 登录状态已保存到：{self.login_state_path}")
            except Exception as e:
                print(f"⚠️ 保存登录状态失败：{e}")

    def init_browser(self) -> bool:
        """初始化浏览器（加载登录状态/提示登录）"""
        try:
            # ✅ 核心修改 1：全局只启动【一次】 Playwright 引擎！存到 self.pw 中
            self.pw = sync_playwright().start()

            # ✅ 核心修改 2：把启动好的引擎传给天眼查去用
            self.tyc_browser.init_browser(self.pw)

            # 启动优化后的浏览器
            # ✅ 核心修改 3：拼多多也用这同一个引擎启动浏览器
            self.browser = self.pw.chromium.launch(
                headless=PLAYWRIGHT_CONFIG["headless"],
                slow_mo=PLAYWRIGHT_CONFIG["slow_mo"],
                args=PLAYWRIGHT_CONFIG["browser_args"],
                ignore_default_args=["--enable-automation"],  # 隐藏自动化标识
                timeout=60000
            )

            # 加载登录状态或创建新上下文
            login_state = self._load_login_state()
            if login_state:
                self.context = self.browser.new_context(
                    viewport=PLAYWRIGHT_CONFIG["viewport"],
                    locale=PLAYWRIGHT_CONFIG["locale"],
                    timezone_id=PLAYWRIGHT_CONFIG["timezone_id"],
                    ignore_https_errors=True,
                    storage_state=login_state  # 加载登录状态
                )
                print("✅ 已加载本地登录状态")
            else:
                self.context = self.browser.new_context(
                    viewport=PLAYWRIGHT_CONFIG["viewport"],
                    locale=PLAYWRIGHT_CONFIG["locale"],
                    timezone_id=PLAYWRIGHT_CONFIG["timezone_id"],
                    ignore_https_errors=True
                )
                print("\n⚠️ 未检测到登录状态，请先完成拼多多登录！")
                self.page = self.context.new_page()
                self.page.goto("https://www.pinduoduo.com", timeout=30000)
                input("请在浏览器中完成登录，登录后按回车继续...")
                self.context.storage_state(path=self.login_state_path)  # ✅ 加await保存状态

            # 初始化页面
            self.page = self.context.new_page()
            self.page.set_default_timeout(PLAYWRIGHT_CONFIG["default_timeout"])
            self.page.set_default_navigation_timeout(PLAYWRIGHT_CONFIG["navigation_timeout"])
            return True
        except Exception as e:
            print(f"❌ 浏览器初始化失败：{e}")
            self.close()  # ✅ 核心修复：加await
            return False

    def open_links(self, goods_data: List[Dict], db_reader: DBGoodsReader) -> List[Dict]:
        """依次打开商品链接（支持店名过滤+图片下载+OCR识别+天眼查搜索提取）"""
        if not self.page:
            print("❌ 浏览器未初始化")
            return []

        total = len(goods_data)
        if total == 0:
            print("⚠️ 无商品链接可处理")
            return []

        print(f"\n📋 共待处理 {total} 条商品链接")

        # 收集所有抓取到的结果（可选，如果想最后统一保存的话）
        all_results = []

        for idx, item in enumerate(goods_data, 1):
            shop = item.get("shop", "未知店铺").strip()
            link = (item.get("product_link") or "").strip()

            if not link:
                print(f"\n⚠️ 第{idx}/{total}条：店铺【{shop}】链接为空，跳过")
                continue

            print(f"\n{'=' * 15} 第 {idx}/{total} 条 {'=' * 15}")
            print(f"🏪 数据库店名：{shop}")
            print(f"🔗 商品链接：{link}")

            # ========== 判断店名是否包含“旗舰店” ==========
            if "旗舰店" not in shop:
                print(f"⚠️ 第{idx}/{total}条：店铺【{shop}】名称不含“旗舰店”，跳过")
                #涉及突破滑块验证的部分了。


                continue
            else:
                print(f"⚠️ 第{idx}/{total}条：店铺【{shop}】名称包含“旗舰店”，打开商品链接")

            try:
                # 1. 打开商品链接
                self.page.goto(
                    link,
                    wait_until="load",
                    timeout=PLAYWRIGHT_CONFIG["navigation_timeout"]
                )
                self.page.wait_for_load_state("networkidle", timeout=15000)
                print(f"✅ 页面加载成功：{self.page.title()}...")

                # 检测登录框
                self._check_login_box()


                #如果已售罄，不跳过。
                # sold_out_locator  = self.page.locator("")
                # if sold_out_locator.count() > 0 :
                #     print("该商品已售罄，跳过这次采集")
                #

                # 2. 店名匹配判断
                page_shop_locator = self.page.locator("div.BAq4Lzv7")
                try:
                    page_shop_locator.wait_for(timeout=5000)

                    page_shop_text = (page_shop_locator.inner_text()).strip().lower()
                except PlaywrightTimeoutError:
                    print(f"❌ 未找到页面店名元素，可能页面结构改变或被风控，跳过")
                    continue

                db_shop_text = shop.lower()
                print(f"🏪 页面元素店铺名：{page_shop_text}")
                if page_shop_text != db_shop_text:
                    print(f"❌ 店名不匹配（数据库：{db_shop_text} | 页面：{page_shop_text}），跳过")

                    self.page.wait_for_timeout(2000)
                    continue
                print(f"✅ 店名匹配成功！")

                # 自定义滑动距离，触发图片懒加载
                self._scroll_down(distance=2100)

                # ========== 获取图片src并下载 ==========
                # shop_name = 'pdd_shop_info_middle_back'
                final_enterprise_info = None
                try:
                    # ========== 原定位策略（优先使用） ==========
                    img_locators = self.page.locator("img[role='img'][aria-label='查看图片']")

                    img_count = img_locators.count()

                    # ========== 原定位不足时，切换到备用定位 ==========
                    if img_count < 2:
                        print(f"⚠️ 原定位仅匹配到{img_count}个图片，尝试备用定位（拼多多懒加载图片）...")
                        input("请手动检查页面图片元素，按回车继续...")
                        continue
                        # 备用定位：匹配截图里的「pdd-lazy-image」类资质图片（带水印的营业执照）
                        # backup_img_locators = self.page.locator(
                        #     "img.pdd-lazy-image.loaded"  # 精准匹配已加载的懒加载图片
                        # )
                        #
                        # backup_count = backup_img_locators.count()
                        #
                        # if backup_count >= 2:
                        #     img_locators = backup_img_locators
                        #     img_count = backup_count
                        #     print(f"✅ 备用定位生效，匹配到图片元素：{img_count} 个")
                        # else:
                        #     print(f"⚠️ 原定位({img_count}个) + 备用定位({backup_count}个)均不足2个，跳过下载")
                        #     input("请手动检查页面图片元素，按回车继续...")
                        #     continue  # 跳过当前店铺，避免卡死


                    print(f"📸 匹配到图片元素：{img_count} 个")
                    # 3. 定位第二个元素
                    target_img_locator = img_locators.nth(1)

                    target_img_locator.wait_for(timeout=5000, state="visible")

                    # 4. 获取第二个图片的src
                    img_src = target_img_locator.get_attribute("src")
                    if img_src:
                        print(f"🖼️ 第2个图片 src：{img_src[:80]}...")
                        image_path = self._download_image(img_src, shop)

                        if image_path:
                            # ========== 核心：调用OCR并获取最终的天眼查数据 ==========
                            final_enterprise_info = self._process_ocr(image_path)

                    else:
                        print(f"⚠️ 第2个图片的src为空")
                except Exception as e:
                    print(f"❌ 获取图片/识别失败：{str(e)[:100]}")

                # 3. 收集数据并自动循环
                if final_enterprise_info:
                    # 将原数据库的店名也塞进去，方便后续入库对比
                    print(f"天眼查---查出来的数据为{final_enterprise_info}")

                    # final_enterprise_info['pdd_shop_name'] = shop
                    all_results.append(final_enterprise_info)

                    # 获取到的数据回填数据库
                    update_success = db_reader.insert_enterprise_info(
                        shop_name=shop,
                        enterprise_info=final_enterprise_info,  # 直接传入天眼查返回的字典
                    )
                    if update_success:
                        print(f"✅ 店铺[{shop}]数据回填成功")
                    else:
                        print(f"❌ 店铺[{shop}]数据回填失败")


                    print(f"\n🎉 成功获取数据，准备进入下一条...")
                else:
                    print(f"\n⚠️ 本条未获取到有效企业信息，准备进入下一条...")
                self.page.wait_for_timeout(5000)

            except PlaywrightTimeoutError:
                print(f"⏰ 页面加载/元素定位超时：{link}")
                input("排查问题")
                continue
            except Exception as e:
                print(f"❌ 第{idx}条处理异常：{str(e)[:100]}...，跳过")
                continue

        return all_results

    def close(self) -> None:
        """关闭浏览器（异步版，补全所有await）"""
        # 先关闭天眼查浏览器
        if hasattr(self, 'tyc_browser') and self.tyc_browser:
            self.tyc_browser.close()

        if hasattr(self, 'pw') and self.pw:
            try:
                self.pw.stop()
                print("✅ Playwright 驱动已彻底停止")
            except Exception:
                pass

        # 关闭拼多多浏览器
        if self.page:
            try:
                self.page.close()
            except Exception:
                pass
        if self.context:
            try:
                self.context.close()
            except Exception:
                pass
        if self.browser:
            try:
                self.browser.close()
                print("✅ 拼多多浏览器已关闭")
            except Exception:
                pass

def main():
    """主函数：整合数据库读取+链接浏览"""
    # 1. 读取环境变量并补全默认值
    db_config = {
        "host": os.getenv("DB_HOST", DEFAULT_DB_CONFIG["host"]),
        "port": int(os.getenv("DB_PORT", DEFAULT_DB_CONFIG["port"])),
        "user": os.getenv("DB_USERNAME", DEFAULT_DB_CONFIG["user"]),
        "password": os.getenv("DB_PASSWORD", DEFAULT_DB_CONFIG["password"]),
        "db_name": os.getenv("DB_DATABASE", DEFAULT_DB_CONFIG["db_name"]),
        "table_name": os.getenv("DB_TABLENAME", DEFAULT_DB_CONFIG["table_name"])
    }

    # 2. 初始化数据库读取器
    db_reader = DBGoodsReader(
        host=db_config["host"],
        port=db_config["port"],
        user=db_config["user"],
        password=db_config["password"],
        db_name=db_config["db_name"]
    )


    if not db_reader.connect_db():
        return

    # 3. 读取商品链接
    goods_data = db_reader.get_shop_and_goods()
    # 预览前5条数据
    if goods_data:
        print("\n📌 数据预览（前5条）：")
        for idx, item in enumerate(goods_data[:5], 1):
            print(f"第{idx}条 | 店铺：{item['shop'][:20]} | 链接：{item['product_link'][:50]}...")

    # 4. 初始化浏览器并打开链接
    # 初始化拼多多浏览器
    pdd_browser = PddLinkBrowser()
    if not pdd_browser.init_browser():
        return

    # 接收返回的所有提取结果
    extracted_data = pdd_browser.open_links(goods_data, db_reader)

    # 打印最终统计
    print(f"\n📊 爬取任务结束，共成功提取 {len(extracted_data)} 条企业信息！")
    if extracted_data:
        # 这里你可以将 extracted_data 写入数据库，或者存为 json/csv
        # 例如打印第一条看看：
        print("💡 最终数据示例：", json.dumps(extracted_data[0], ensure_ascii=False, indent=2))

    pdd_browser.close()
    db_reader.close_db()


if __name__ == "__main__":
    try:
        main()
    except KeyboardInterrupt:
        print("\n⚠️ 程序被用户中断")
    except Exception as e:
        print(f"\n❌ 程序运行出错：{e}")