h-1.flet.3/data_export.py

"""データエクスポート機能"""
import sqlite3
import json
import csv
import datetime
from typing import List, Dict

class DataExporter:
    """販売アシスト1号のデータエクスポート担当"""
    
    def __init__(self, db_path: str = "sales.db"):
        self.db_path = db_path
    
    def export_to_json(self, output_path: str = None) -> str:
        """全データをJSON形式でエクスポート"""
        if not output_path:
            timestamp = datetime.datetime.now().strftime("%Y%m%d_%H%M%S")
            output_path = f"sales_backup_{timestamp}.json"
        
        data = {
            "export_date": datetime.datetime.now().isoformat(),
            "customers": self._get_customers(),
            "products": self._get_products(),
            "sales": self._get_sales()
        }
        
        with open(output_path, 'w', encoding='utf-8') as f:
            json.dump(data, f, ensure_ascii=False, indent=2)
        
        return output_path
    
    def export_to_csv(self, output_dir: str = ".") -> Dict[str, str]:
        """各テーブルをCSV形式でエクスポート"""
        timestamp = datetime.datetime.now().strftime("%Y%m%d_%H%M%S")
        
        files = {}
        
        # 顧客データ
        customers_file = f"{output_dir}/customers_{timestamp}.csv"
        self._export_table_to_csv("customers", customers_file)
        files["customers"] = customers_file
        
        # 商品データ
        products_file = f"{output_dir}/products_{timestamp}.csv"
        self._export_table_to_csv("products", products_file)
        files["products"] = products_file
        
        # 売上データ
        sales_file = f"{output_dir}/sales_{timestamp}.csv"
        self._export_table_to_csv("sales", sales_file)
        files["sales"] = sales_file
        
        return files
    
    def _get_customers(self) -> List[Dict]:
        """顧客データ取得"""
        conn = sqlite3.connect(self.db_path)
        cursor = conn.cursor()
        cursor.execute("SELECT * FROM customers ORDER BY created_at DESC")
        
        customers = []
        for row in cursor.fetchall():
            customers.append({
                'id': row[0],
                'name': row[1],
                'company': row[2],
                'phone': row[3],
                'email': row[4],
                'address': row[5],
                'created_at': row[6]
            })
        
        conn.close()
        return customers
    
    def _get_products(self) -> List[Dict]:
        """商品データ取得"""
        conn = sqlite3.connect(self.db_path)
        cursor = conn.cursor()
        cursor.execute("SELECT * FROM products ORDER BY created_at DESC")
        
        products = []
        for row in cursor.fetchall():
            products.append({
                'id': row[0],
                'name': row[1],
                'code': row[2],
                'price': row[3],
                'stock': row[4],
                'description': row[5],
                'created_at': row[6]
            })
        
        conn.close()
        return products
    
    def _get_sales(self) -> List[Dict]:
        """売上データ取得"""
        conn = sqlite3.connect(self.db_path)
        cursor = conn.cursor()
        cursor.execute('''
            SELECT s.*, c.name as customer_name, p.name as product_name
            FROM sales s
            LEFT JOIN customers c ON s.customer_id = c.id
            LEFT JOIN products p ON s.product_id = p.id
            ORDER BY s.sale_date DESC
        ''')
        
        sales = []
        for row in cursor.fetchall():
            sales.append({
                'id': row[0],
                'customer_id': row[1],
                'product_id': row[2],
                'quantity': row[3],
                'unit_price': row[4],
                'total_price': row[5],
                'sale_date': row[6],
                'customer_name': row[7] or '不明',
                'product_name': row[8] or '不明'
            })
        
        conn.close()
        return sales
    
    def _export_table_to_csv(self, table_name: str, output_path: str):
        """テーブルをCSVにエクスポート"""
        conn = sqlite3.connect(self.db_path)
        cursor = conn.cursor()
        
        # テーブルデータ取得
        if table_name == "sales":
            cursor.execute('''
                SELECT s.id, c.name as customer_name, p.name as product_name,
                       s.quantity, s.unit_price, s.total_price, s.sale_date
                FROM sales s
                LEFT JOIN customers c ON s.customer_id = c.id
                LEFT JOIN products p ON s.product_id = p.id
                ORDER BY s.sale_date DESC
            ''')
            headers = ['ID', '顧客名', '商品名', '数量', '単価', '合計', '日時']
        else:
            cursor.execute(f"SELECT * FROM {table_name}")
            headers = [description[0] for description in cursor.description]
        
        rows = cursor.fetchall()
        
        # CSV書き込み
        with open(output_path, 'w', newline='', encoding='utf-8') as csvfile:
            writer = csv.writer(csvfile)
            writer.writerow(headers)
            writer.writerows(rows)
        
        conn.close()

if __name__ == "__main__":
    exporter = DataExporter()
    
    print("📦 データエクスポート中...")
    
    # JSONエクスポート
    json_file = exporter.export_to_json()
    print(f"✅ JSONエクスポート完了: {json_file}")
    
    # CSVエクスポート
    csv_files = exporter.export_to_csv()
    print("✅ CSVエクスポート完了:")
    for table, file in csv_files.items():
        print(f"  {table}: {file}")