Browse Source

feat: add Excel processing and translation service modules

mrh (aider) 1 year ago
parent
commit
789525e120

+ 47 - 0
function_calling/excel_processor.py

@@ -0,0 +1,47 @@
+from typing import List, Dict
+import pandas as pd
+from urllib.parse import quote
+from mylib.logging_config import setup_logging
+import logging
+
+# Setup logging
+setup_logging()
+logger = logging.getLogger('excel_tool')
+
+class ExcelProcessor:
+    def __init__(self, file_path: str):
+        """Initialize Excel processor with file path"""
+        self.file_path = file_path
+        self.df = pd.read_excel(file_path)
+        logger.info(f"Loaded Excel file: {file_path}")
+
+    def add_translation_column(self, column_name: str, translations: Dict[str, str]):
+        """Add translated column next to specified column"""
+        try:
+            new_col = f"{column_name}_中文"
+            self.df[new_col] = self.df[column_name].map(translations)
+            logger.info(f"Added translation column for {column_name}")
+        except Exception as e:
+            logger.error(f"Error adding translation column: {str(e)}")
+            raise
+
+    def add_hyperlink_column(self, column_name: str, base_url: str):
+        """Add hyperlink column for specified column"""
+        try:
+            new_col = f"{column_name}_链接"
+            self.df[new_col] = self.df[column_name].apply(
+                lambda x: f'=HYPERLINK("{base_url}{quote(x)}", "{x}")'
+            )
+            logger.info(f"Added hyperlink column for {column_name}")
+        except Exception as e:
+            logger.error(f"Error adding hyperlink column: {str(e)}")
+            raise
+
+    def save(self, output_path: str):
+        """Save processed Excel file"""
+        try:
+            self.df.to_excel(output_path, index=False)
+            logger.info(f"Saved processed file to {output_path}")
+        except Exception as e:
+            logger.error(f"Error saving file: {str(e)}")
+            raise

+ 51 - 0
function_calling/main.py

@@ -0,0 +1,51 @@
+from typing import List
+from excel_processor import ExcelProcessor
+from translation_service import TranslationService
+from mylib.logging_config import setup_logging
+import logging
+
+# Setup logging
+setup_logging()
+logger = logging.getLogger('main')
+
+def process_excel_file(input_path: str, output_path: str, columns_to_translate: List[str]):
+    """Main function to process Excel file"""
+    try:
+        logger.info("Starting Excel processing")
+        
+        # Initialize services
+        excel_processor = ExcelProcessor(input_path)
+        translation_service = TranslationService()
+
+        # Process each column
+        for column in columns_to_translate:
+            # Get unique values for translation
+            unique_values = excel_processor.df[column].unique().tolist()
+            
+            # Translate values
+            translations = translation_service.translate_batch(unique_values)
+            
+            # Add translated column
+            excel_processor.add_translation_column(column, translations)
+            
+            # Add hyperlink column
+            if column == "搜索词":
+                excel_processor.add_hyperlink_column(column, "https://www.amazon.co.jp/s?k=")
+            elif column == "ASIN":
+                excel_processor.add_hyperlink_column(column, "https://www.amazon.co.jp/dp/")
+
+        # Save processed file
+        excel_processor.save(output_path)
+        logger.info("Excel processing completed successfully")
+        
+    except Exception as e:
+        logger.error(f"Error processing Excel file: {str(e)}")
+        raise
+
+if __name__ == "__main__":
+    # Example usage
+    process_excel_file(
+        input_path="input.xlsx",
+        output_path="output.xlsx",
+        columns_to_translate=["搜索词", "品牌", "点击量最高的类别"]
+    )

+ 34 - 0
function_calling/translation_service.py

@@ -0,0 +1,34 @@
+from typing import Dict
+from mylib.logging_config import setup_logging
+import logging
+
+# Setup logging
+setup_logging()
+logger = logging.getLogger('translation_service')
+
+class TranslationService:
+    def __init__(self):
+        """Initialize translation service"""
+        # TODO: Add actual translation API initialization
+        logger.info("Translation service initialized")
+
+    def translate_text(self, text: str) -> str:
+        """Translate single text string"""
+        try:
+            # TODO: Implement actual translation logic
+            translated = f"Translated {text}"  # Placeholder
+            logger.debug(f"Translated: {text} -> {translated}")
+            return translated
+        except Exception as e:
+            logger.error(f"Translation error: {str(e)}")
+            raise
+
+    def translate_batch(self, texts: List[str]) -> Dict[str, str]:
+        """Translate batch of texts"""
+        try:
+            translations = {text: self.translate_text(text) for text in texts}
+            logger.info(f"Translated {len(translations)} items")
+            return translations
+        except Exception as e:
+            logger.error(f"Batch translation error: {str(e)}")
+            raise