完成亚马逊搜索主关键词导出工作表

src/excel_tools/file_manager.py

@@ -8,13 +8,13 @@ from utils.file import read_file,save_to_file
 from src.excel_tools.writers import (
     ExcelWriterBase,
     CompetitiveAnalysisWriter,
-    ProductInfoWriter
+    ProductInfoWriter,
 )
 from src.manager import DbManager,StorageManager
 from utils.logu import get_logger
 from config.settings import OUTPUT_DIR
 from src.models.asin_model import AsinExtraResultModel
-from src.models.product_model import Product,CompetitorCrawlData
+from src.models.product_model import AICompetitorAnalyzeMainKeywordsResult, Product,CompetitorCrawlData
 from src.manager.core.db_mongo import BaseMongoManager
 
 logger = get_logger('excel')
@@ -49,6 +49,14 @@ class ExcelFileManager:
         if sheet_name not in self.wb.sheetnames:
             competitive_sheet_writer = CompetitiveAnalysisWriter(self.wb, sheet_name=sheet_name, sheet_index=sheet_index)
             competitive_sheet_writer.add_data(competitor_crawl_data)
+
+    def write_product_info(self, analyze_data: AICompetitorAnalyzeMainKeywordsResult, sheet_name: str = "产品关键词分析", sheet_index: int = 1, overwrite: bool = False):
+        """写入产品关键词分析数据"""
+        if overwrite and sheet_name in self.wb.sheetnames:
+            self.wb.remove(self.wb[sheet_name])
+        if sheet_name not in self.wb.sheetnames:
+            product_info_writer = ProductInfoWriter(self.wb, sheet_name=sheet_name, sheet_index=sheet_index)
+            product_info_writer.add_data(analyze_data)
     def load_s3_extract_data(self) -> list[AsinExtraResultModel]:
         return self.s3_storage_manager.load_s3_complete_extract_data()
 
@@ -59,9 +67,9 @@ async def main():
     product = await Product.find_one(Product.basic_info["name"] == "电线保护套")
 
     logger.info(f"{product}")
-    return
     extract_data_lsit = product.competitor_crawl_data
     self.write_competitive_sheet(product.competitor_crawl_data)
+    self.write_product_info(product.competitor_analyze,overwrite=True)
     self.save_all()
     return
     competi_sheet = CompetitiveAnalysisWriter(excel_file.output_path)

src/excel_tools/writers/product_info.py

@@ -1,57 +1,110 @@
 from openpyxl.worksheet.worksheet import Worksheet
 from openpyxl.styles import Font, PatternFill, Alignment
-from pathlib import Path
-from typing import Dict, Any
 from openpyxl.utils import get_column_letter
-from utils.logu import logger
+from typing import Dict, Any, List
+from openpyxl import Workbook
+from .base_writer import ExcelWriterBase
+from src.models.product_model import AICompetitorAnalyzeMainKeywordsResult
+from utils.logu import get_logger
 
-class ProductInfoWriter():
+logger = get_logger('excel')
+
+class ProductInfoWriter(ExcelWriterBase):
     """产品信息工作表写入器"""
-    SHEET_NAME = "产品信息"
+    HEADER_FILL = PatternFill(start_color='4F81BD', fill_type='solid')
+    HEADER_FONT = Font(bold=True, color='FFFFFF')
     
+    def __init__(self, work_book: Workbook, sheet_index: int=0, sheet_name: str="产品关键词分析"):
+        super().__init__(work_book, sheet_index, sheet_name)
+        self.current_row = 1
+        
     def _init_worksheet(self):
-        """初始化产品信息工作表"""
-        if self.SHEET_NAME not in self.wb.sheetnames:
-            self.ws = self.wb.create_sheet(title=self.SHEET_NAME)
-            logger.warning(f"新建工作表: {self.SHEET_NAME}")
+        """初始化工作表"""
+        if self.sheet_name in self.wb.sheetnames:
+            self.ws = self.wb[self.sheet_name]
+            current_index = self.wb.index(self.ws)
+            offset = self.sheet_index - current_index
+            self.wb.move_sheet(self.ws, offset=offset)
         else:
-            self.ws = self.wb[self.SHEET_NAME]
-            self.ws.delete_rows(1, self.ws.max_row)  # 清空现有数据
+            self.ws = self.wb.create_sheet(self.sheet_name, index=self.sheet_index)
+            logger.info(f"新建工作表: {self.sheet_name}")
+        
+    def add_data(self, analyze_data: AICompetitorAnalyzeMainKeywordsResult):
+        """添加产品分析数据"""
+        if not analyze_data or not analyze_data.results:
+            logger.warning("无有效分析数据可写入")
+            return
             
-        # 初始化表头
-        self._init_headers()
-        
-    def _init_headers(self):
-        """设置固定表头格式"""
-        headers = [
-            ("ASIN", 15),
-            ("产品名称", 30),
-            ("分类节点", 25),
-            ("上架时间", 15),
-            ("评分", 10),
-            ("评论数", 15)
-        ]
-        
-        for col_idx, (header, width) in enumerate(headers, start=1):
-            cell = self.ws.cell(row=1, column=col_idx, value=header)
-            cell.font = Font(bold=True, color="FFFFFF")
-            cell.fill = PatternFill(start_color="4F81BD", fill_type="solid")
-            self.ws.column_dimensions[get_column_letter(col_idx)].width = width
+        # 写入标题
+        self._write_header()
+        
+        # 写入主关键词数据
+        for result in analyze_data.results:
+            self._write_keyword_result(result)
+            
+        # 写入补充说明
+        if analyze_data.supplement:
+            self._write_supplement(analyze_data.supplement)
+            
+        self._apply_formatting()
+    
+    def _write_header(self):
+        """写入表头"""
+        headers = ["ASIN", "主关键词", "月搜索量", "选择理由", "亚马逊推荐搜索关键词", "图片预览"]
+        for col, header in enumerate(headers, start=1):
+            cell = self.ws.cell(self.current_row, col, header)
+            cell.font = self.HEADER_FONT
+            cell.fill = self.HEADER_FILL
+            cell.alignment = Alignment(horizontal='center', vertical='center')
             
-    def add_data(self, product_data: Dict[str, Any]):
-        """添加产品数据"""
-        # 数据验证
-        required_fields = ['asin', 'title', 'category', 'launch_date', 'rating', 'reviews']
-        if not all(field in product_data for field in required_fields):
-            raise ValueError("Missing required product data fields")
+        # 设置列宽
+        self.ws.column_dimensions[get_column_letter(1)].width = 15  # ASIN列
+        self.ws.column_dimensions[get_column_letter(2)].width = 25  # 关键词列
+        self.ws.column_dimensions[get_column_letter(3)].width = 12  # 搜索量列
+        self.ws.column_dimensions[get_column_letter(4)].width = 40  # 理由列
+        self.ws.column_dimensions[get_column_letter(5)].width = 40  # 建议关键词列
+        self.ws.column_dimensions[get_column_letter(6)].width = 40  # 图片预览列
+        
+        self.current_row += 1
+    
+    def _write_keyword_result(self, result):
+        """写入单个关键词结果"""
+        from utils.file import s3_uri_to_http_url
+        
+        # 主数据行
+        self.ws.cell(self.current_row, 1, result.asin)
+        self.ws.cell(self.current_row, 2, result.main_key)
+        self.ws.cell(self.current_row, 3, result.monthly_searches or 0)
+        self.ws.cell(self.current_row, 4, result.reason or "")
+        
+        # 建议关键词
+        if result.crawl_result and result.crawl_result.suggestions:
+            suggestions = ", ".join(result.crawl_result.suggestions[:10])  # 最多显示10个建议
+            self.ws.cell(self.current_row, 5, suggestions)
+        
+        # 图片预览链接
+        if result.crawl_result and result.crawl_result.screenshot:
+            preview_url = s3_uri_to_http_url(result.crawl_result.screenshot)
+            self.ws.cell(self.current_row, 6, preview_url)
+            self.ws.cell(self.current_row, 6).hyperlink = preview_url
+            self.ws.cell(self.current_row, 6).style = "Hyperlink"
             
-        # 写入数据行
-        row = self.ws.max_row + 1
-        self.ws.cell(row=row, column=1, value=product_data['asin'])
-        self.ws.cell(row=row, column=2, value=product_data['title'])
-        self.ws.cell(row=row, column=3, value=product_data['category'])
-        self.ws.cell(row=row, column=4, value=product_data['launch_date']).number_format = 'YYYY-MM-DD'
-        self.ws.cell(row=row, column=5, value=product_data['rating']).number_format = '0.0'
-        self.ws.cell(row=row, column=6, value=product_data['reviews']).number_format = '#,##0'
-        
-        logger.info(f"已写入产品数据: {product_data['asin']}")
+        self.current_row += 1
+    
+    def _write_supplement(self, supplement: str):
+        """写入补充说明"""
+        self.ws.cell(self.current_row, 1, "补充说明:").font = Font(bold=True)
+        self.ws.merge_cells(start_row=self.current_row, start_column=2, 
+                          end_row=self.current_row, end_column=4)
+        cell = self.ws.cell(self.current_row, 2, supplement)
+        cell.alignment = Alignment(wrap_text=True, vertical='top')
+        self.current_row += 2  # 留空一行
+        
+    def _apply_formatting(self):
+        """应用格式"""
+        for row in self.ws.iter_rows():
+            for cell in row:
+                if cell.column in (2,):  # 数字列右对齐
+                    cell.alignment = Alignment(horizontal='right', vertical='center')
+                else:
+                    cell.alignment = Alignment(horizontal='left', vertical='center', wrap_text=True)