copywriting_production/src/excel_tools/file_manager.py

import asyncio
import json
from pathlib import Path
import shutil
from typing import Dict, Type, Any
from openpyxl import load_workbook,Workbook
from utils.file import read_file,save_to_file
from src.excel_tools.writers import (
    ExcelWriterBase,
    CompetitiveAnalysisWriter,
    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,CompetitorAnalysis
from src.manager.core.db_mongo import BaseMongoManager

logger = get_logger('excel')

class ExcelFileManager:
    TEMPLATE_PATH = Path(f"{OUTPUT_DIR}/resource/文案制作-template.xlsx")
    """Excel文件协调管理器"""
    def __init__(self, output_path: str=None, template_path: str = None):
        self.output_path = Path(output_path)
        self.template_path = template_path or self.TEMPLATE_PATH
        self.writers: Dict[str, ExcelWriterBase] = {}
        self.s3_storage_manager = StorageManager()
        self.wb:Workbook = self._prepare_workbook()
        logger.info(f"{self.wb.sheetnames}")
        logger.info(f"{self.wb.worksheets}")

    def _prepare_workbook(self):
        """准备工作簿"""
        if not self.output_path.exists():
            shutil.copy(self.template_path, self.output_path)
            
        return load_workbook(self.output_path)

    def save_all(self):
        self.wb.save(self.output_path)
        self.wb.close()
    
    def write_competitive_sheet(self, competitor_analysis:Dict[str, CompetitorAnalysis], sheet_name: str = "竞品关键词调研", sheet_index: int = 0, 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:
            competitive_sheet_writer = CompetitiveAnalysisWriter(self.wb, sheet_name=sheet_name, sheet_index=sheet_index)
            competitive_sheet_writer.add_data(competitor_analysis)
    def load_s3_extract_data(self) -> list[AsinExtraResultModel]:
        return self.s3_storage_manager.load_s3_complete_extract_data()

async def main():
    self = ExcelFileManager(r"G:\code\amazone\copywriting_production\output\resource\multi_data.xlsx")
    db_mongo = BaseMongoManager()
    asyncio.run(db_mongo.initialize())
    product = await Product.find_one(Product.basic_info.name == "电线保护套")
    extract_data_lsit = product.competitor_analysis
    self.write_competitive_sheet(product.competitor_analysis)
    # self.save_all()
    return
    competi_sheet = CompetitiveAnalysisWriter(excel_file.output_path)

    list_model = excel_file.db.get_asin_completed()
    competi_sheet.add_data(list_dict)
    competi_sheet.save()
    return
    logger.info(f"{list_dict}")
    for model in list_dict:
        json_path = model.extra_result_path
        asin = model.asin
        data = json.loads(read_file(json_path))
    return
    output_path = r"G:\code\amazone\copywriting_production\output\multi_data.xlsx"
    
    generator = CompetitiveAnalysisWriter(output_path)
    data = json.loads(read_file(json_path))
    for json_path, asin in json_files:
        generator.add_product(json_path, asin)
    
    generator.apply_formatting()
    generator.save()



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