"""发票信息提取器：从网站提取发票信息。""" from __future__ import annotations import json import logging import re from typing import Any, Dict, List, Optional from jiangchang_skill_core.activity import interruptible_sleep logger = logging.getLogger(__name__) # 发票信息提取选择器（需要根据实际页面调整） INVOICE_SELECTORS = { # 发票列表容器 "invoice_list_container": [ ".invoice-list", ".invoice-table", "[data-testid='invoice-list']", "table tbody", ], # 单个发票行 "invoice_row": [ "tr.invoice-item", ".invoice-row", "[data-testid='invoice-row']", "tbody tr", ], # 发票详情页面元素 "invoice_detail": { "invoice_code": [ ".invoice-code", "[data-testid='invoice-code']", "td:has-text('发票代码') + td", ], "invoice_number": [ ".invoice-number", "[data-testid='invoice-number']", "td:has-text('发票号码') + td", ], "invoice_date": [ ".invoice-date", "[data-testid='invoice-date']", "td:has-text('开票日期') + td", ], "total_amount": [ ".total-amount", "[data-testid='total-amount']", "td:has-text('金额') + td", ], "tax_amount": [ ".tax-amount", "[data-testid='tax-amount']", "td:has-text('税额') + td", ], "buyer_name": [ ".buyer-name", "[data-testid='buyer-name']", "td:has-text('购买方') + td", ], "buyer_tax_no": [ ".buyer-tax-no", "[data-testid='buyer-tax-no']", "td:has-text('购买方税号') + td", ], "seller_name": [ ".seller-name", "[data-testid='seller-name']", "td:has-text('销售方') + td", ], "seller_tax_no": [ ".seller-tax-no", "[data-testid='seller-tax-no']", "td:has-text('销售方税号') + td", ], "items": [ ".invoice-items", "[data-testid='invoice-items']", ".item-list", ], }, # 分页控件 "pagination": { "next_button": [ ".pagination .next", "button:has-text('下一页')", "[data-testid='next-page']", ], "page_info": [ ".page-info", ".pagination-info", "[data-testid='page-info']", ], }, # 搜索/筛选控件 "search": { "search_input": [ "input[placeholder*='搜索']", "input[placeholder*='查询']", ".search-input", ], "search_button": [ "button:has-text('搜索')", "button:has-text('查询')", ".search-btn", ], }, } async def extract_invoice_list(page: Any) -> List[Dict[str, Any]]: """提取发票列表信息。 Args: page: Playwright页面对象 Returns: 发票信息列表 """ invoices = [] try: # 等待页面加载 await interruptible_sleep(2.0) # 查找发票列表容器 container = await _find_element(page, INVOICE_SELECTORS["invoice_list_container"]) if not container: logger.warning("invoice_list_container_not_found") return [] # 查找所有发票行 rows = await _find_elements(page, INVOICE_SELECTORS["invoice_row"]) logger.info("found_invoice_rows count=%s", len(rows)) # 提取每行发票信息 for i, row in enumerate(rows): try: invoice_data = await _extract_invoice_from_row(page, row, i) if invoice_data: invoices.append(invoice_data) except Exception as e: logger.warning("extract_row_failed row=%s error=%s", i, str(e)) continue # 处理分页 has_next = await _has_next_page(page) while has_next: # 点击下一页 next_clicked = await _click_next_page(page) if not next_clicked: break await interruptible_sleep(2.0) # 提取下一页数据 rows = await _find_elements(page, INVOICE_SELECTORS["invoice_row"]) for i, row in enumerate(rows): try: invoice_data = await _extract_invoice_from_row(page, row, i) if invoice_data: invoices.append(invoice_data) except Exception as e: logger.warning("extract_row_failed page_row=%s error=%s", i, str(e)) continue has_next = await _has_next_page(page) logger.info("extraction_complete total_invoices=%s", len(invoices)) return invoices except Exception as e: logger.error("extract_invoice_list_failed error=%s", str(e)) return [] async def extract_invoice_detail(page: Any, invoice_index: int = 0) -> Optional[Dict[str, Any]]: """提取单个发票的详细信息。 Args: page: Playwright页面对象 invoice_index: 发票索引（从0开始） Returns: 发票详细信息字典 """ try: # 查找发票行 rows = await _find_elements(page, INVOICE_SELECTORS["invoice_row"]) if invoice_index >= len(rows): logger.warning("invoice_index_out_of_range index=%s total=%s", invoice_index, len(rows)) return None # 点击发票行查看详情 row = rows[invoice_index] await row.click() await interruptible_sleep(2.0) # 提取详细信息 detail = {} for field, selectors in INVOICE_SELECTORS["invoice_detail"].items(): value = await _extract_text_by_selectors(page, selectors) if value: detail[field] = value # 提取商品明细 items = await _extract_invoice_items(page) if items: detail["items"] = items logger.info("extract_invoice_detail success index=%s fields=%s", invoice_index, list(detail.keys())) return detail except Exception as e: logger.error("extract_invoice_detail_failed error=%s", str(e)) return None async def _extract_invoice_from_row(page: Any, row: Any, index: int) -> Optional[Dict[str, Any]]: """从行元素提取发票信息。""" try: # 提取行文本 row_text = await row.inner_text() # 尝试从文本中提取关键信息 invoice_data = { "row_index": index, "row_text": row_text.strip(), "raw_text": row_text, } # 尝试提取发票号码（通常是数字） invoice_match = re.search(r'(\d{8,20})', row_text) if invoice_match: invoice_data["invoice_number"] = invoice_match.group(1) # 尝试提取金额（包含小数点的数字） amount_match = re.search(r'(\d+\.?\d*)', row_text) if amount_match: invoice_data["amount"] = amount_match.group(1) return invoice_data except Exception as e: logger.warning("extract_from_row_failed index=%s error=%s", index, str(e)) return None async def _extract_invoice_items(page: Any) -> List[Dict[str, str]]: """提取发票商品明细。""" items = [] try: item_containers = await _find_elements(page, INVOICE_SELECTORS["invoice_detail"]["items"]) if not item_containers: return [] for container in item_containers: try: item_text = await container.inner_text() items.append({"text": item_text.strip()}) except Exception: continue return items except Exception: return [] async def _has_next_page(page: Any) -> bool: """检查是否有下一页。""" try: next_button = await _find_element(page, INVOICE_SELECTORS["pagination"]["next_button"]) if next_button: # 检查按钮是否可用 is_disabled = await next_button.get_attribute("disabled") if is_disabled is None: # 没有disabled属性 return True return False except Exception: return False async def _click_next_page(page: Any) -> bool: """点击下一页按钮。""" try: next_button = await _find_element(page, INVOICE_SELECTORS["pagination"]["next_button"]) if next_button: await next_button.click() return True return False except Exception as e: logger.warning("click_next_page_failed error=%s", str(e)) return False async def _find_element(page: Any, selectors: list) -> Optional[Any]: """根据选择器列表查找元素。""" for selector in selectors: try: element = page.locator(selector) if await element.count() > 0 and await element.is_visible(): return element except Exception: continue return None async def _find_elements(page: Any, selectors: list) -> List[Any]: """根据选择器列表查找多个元素。""" elements = [] for selector in selectors: try: found_elements = await page.locator(selector).all() if found_elements: elements.extend(found_elements) break # 找到一组就停止 except Exception: continue return elements async def _extract_text_by_selectors(page: Any, selectors: list) -> Optional[str]: """根据选择器列表提取文本。""" for selector in selectors: try: element = page.locator(selector) if await element.count() > 0: text = await element.inner_text() if text and text.strip(): return text.strip() except Exception: continue return None def format_invoice_data(invoices: List[Dict[str, Any]]) -> Dict[str, Any]: """格式化发票数据为结构化格式。""" return { "total_count": len(invoices), "invoices": invoices, "extraction_complete": True, }