import flet as ft import sqlite3 import datetime from typing import List, Dict, Optional from data_export import DataExporter from compliance import ComplianceManager class SalesAssistant: def __init__(self, page: ft.Page): self.page = page self.page.title = "販売アシスト1号" self.page.theme_mode = ft.ThemeMode.LIGHT self.page.vertical_alignment = ft.MainAxisAlignment.CENTER self.page.horizontal_alignment = ft.CrossAxisAlignment.CENTER # Initialize database self.init_database() # Initialize data exporter and compliance manager self.exporter = DataExporter() self.compliance = ComplianceManager() # Navigation self.navigation_rail = ft.NavigationRail( selected_index=0, label_type=ft.NavigationRailLabelType.ALL, min_width=100, min_extended_width=200, destinations=[ ft.NavigationRailDestination( icon=ft.icons.DASHBOARD_OUTLINED, selected_icon=ft.icons.DASHBOARD, label="ダッシュボード" ), ft.NavigationRailDestination( icon=ft.icons.PEOPLE_OUTLINED, selected_icon=ft.icons.PEOPLE, label="顧客管理" ), ft.NavigationRailDestination( icon=ft.icons.INVENTORY_2_OUTLINED, selected_icon=ft.icons.INVENTORY_2, label="商品管理" ), ft.NavigationRailDestination( icon=ft.icons.RECEIPT_OUTLINED, selected_icon=ft.icons.RECEIPT, label="売上管理" ), ft.NavigationRailDestination( icon=ft.icons.DOWNLOAD_OUTLINED, selected_icon=ft.icons.DOWNLOAD, label="データ出力" ), ft.NavigationRailDestination( icon=ft.icons.GAVEL_OUTLINED, selected_icon=ft.icons.GAVEL, label="コンプライアンス" ), ], on_change=self.navigation_changed ) # Content area self.content_area = ft.Container(expand=True) # Main layout self.main_layout = ft.Row( [ self.navigation_rail, ft.VerticalDivider(width=1), self.content_area ], expand=True ) # Show dashboard initially self.show_dashboard() # Add main layout to page self.page.add(self.main_layout) def init_database(self): """Initialize SQLite database""" conn = sqlite3.connect('sales.db') cursor = conn.cursor() # Create customers table cursor.execute(''' CREATE TABLE IF NOT EXISTS customers ( id INTEGER PRIMARY KEY AUTOINCREMENT, name TEXT NOT NULL, company TEXT, phone TEXT, email TEXT, address TEXT, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ) ''') # Create products table cursor.execute(''' CREATE TABLE IF NOT EXISTS products ( id INTEGER PRIMARY KEY AUTOINCREMENT, name TEXT NOT NULL, code TEXT UNIQUE, price REAL NOT NULL, stock INTEGER DEFAULT 0, description TEXT, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ) ''') # Create sales table cursor.execute(''' CREATE TABLE IF NOT EXISTS sales ( id INTEGER PRIMARY KEY AUTOINCREMENT, customer_id INTEGER, product_id INTEGER, quantity INTEGER NOT NULL, unit_price REAL NOT NULL, total_price REAL NOT NULL, sale_date TIMESTAMP DEFAULT CURRENT_TIMESTAMP, FOREIGN KEY (customer_id) REFERENCES customers (id), FOREIGN KEY (product_id) REFERENCES products (id) ) ''') conn.commit() conn.close() def navigation_changed(self, e): """Handle navigation rail changes""" if e.control.selected_index == 0: self.show_dashboard() elif e.control.selected_index == 1: self.show_customers() elif e.control.selected_index == 2: self.show_products() elif e.control.selected_index == 3: self.show_sales() elif e.control.selected_index == 4: self.show_data_export() elif e.control.selected_index == 5: self.show_compliance() def show_dashboard(self): """Show dashboard view""" # Get statistics conn = sqlite3.connect('sales.db') cursor = conn.cursor() # Total customers cursor.execute("SELECT COUNT(*) FROM customers") total_customers = cursor.fetchone()[0] # Total products cursor.execute("SELECT COUNT(*) FROM products") total_products = cursor.fetchone()[0] # Total sales cursor.execute("SELECT COUNT(*) FROM sales") total_sales = cursor.fetchone()[0] # Total revenue cursor.execute("SELECT SUM(total_price) FROM sales") total_revenue = cursor.fetchone()[0] or 0 conn.close() dashboard_content = ft.Column([ ft.Text("ダッシュボード", size=24, weight=ft.FontWeight.BOLD), ft.Divider(), ft.Row([ ft.Card( content=ft.Container( content=ft.Column([ ft.Text("顧客数", size=16, color=ft.colors.BLUE), ft.Text(str(total_customers), size=32, weight=ft.FontWeight.BOLD) ]), padding=20, width=150 ) ), ft.Card( content=ft.Container( content=ft.Column([ ft.Text("商品数", size=16, color=ft.colors.GREEN), ft.Text(str(total_products), size=32, weight=ft.FontWeight.BOLD) ]), padding=20, width=150 ) ), ft.Card( content=ft.Container( content=ft.Column([ ft.Text("売上件数", size=16, color=ft.colors.ORANGE), ft.Text(str(total_sales), size=32, weight=ft.FontWeight.BOLD) ]), padding=20, width=150 ) ), ft.Card( content=ft.Container( content=ft.Column([ ft.Text("総売上", size=16, color=ft.colors.PURPLE), ft.Text(f"¥{total_revenue:,.0f}", size=32, weight=ft.FontWeight.BOLD) ]), padding=20, width=150 ) ) ], wrap=True) ], scroll=ft.ScrollMode.AUTO) self.content_area.content = ft.Container(dashboard_content, padding=20) self.page.update() def show_customers(self): """Show customers management view""" customers = self.get_customers() # Customer list customer_list = ft.ListView( expand=True, spacing=10, padding=10 ) for customer in customers: customer_list.controls.append( ft.Card( content=ft.Container( content=ft.ListTile( title=ft.Text(customer['name']), subtitle=ft.Text(customer['company'] or ''), trailing=ft.Row([ ft.IconButton( icon=ft.icons.EDIT, on_click=lambda e, c=customer: self.edit_customer(c) ), ft.IconButton( icon=ft.icons.DELETE, on_click=lambda e, c=customer: self.delete_customer(c) ) ]) ), padding=10 ) ) ) # Add customer button add_button = ft.ElevatedButton( "顧客を追加", icon=ft.icons.ADD, on_click=self.add_customer_dialog ) customers_content = ft.Column([ ft.Row([ ft.Text("顧客管理", size=24, weight=ft.FontWeight.BOLD), add_button ], alignment=ft.MainAxisAlignment.SPACE_BETWEEN), ft.Divider(), customer_list ], scroll=ft.ScrollMode.AUTO) self.content_area.content = ft.Container(customers_content, padding=20) self.page.update() def show_products(self): """Show products management view""" products = self.get_products() # Product list product_list = ft.ListView( expand=True, spacing=10, padding=10 ) for product in products: product_list.controls.append( ft.Card( content=ft.Container( content=ft.ListTile( title=ft.Text(product['name']), subtitle=ft.Text(f"¥{product['price']:,.0f} | 在庫: {product['stock']}"), trailing=ft.Row([ ft.IconButton( icon=ft.icons.EDIT, on_click=lambda e, p=product: self.edit_product(p) ), ft.IconButton( icon=ft.icons.DELETE, on_click=lambda e, p=product: self.delete_product(p) ) ]) ), padding=10 ) ) ) # Add product button add_button = ft.ElevatedButton( "商品を追加", icon=ft.icons.ADD, on_click=self.add_product_dialog ) products_content = ft.Column([ ft.Row([ ft.Text("商品管理", size=24, weight=ft.FontWeight.BOLD), add_button ], alignment=ft.MainAxisAlignment.SPACE_BETWEEN), ft.Divider(), product_list ], scroll=ft.ScrollMode.AUTO) self.content_area.content = ft.Container(products_content, padding=20) self.page.update() def show_sales(self): """Show sales management view""" sales = self.get_sales() # Sales list sales_list = ft.ListView( expand=True, spacing=10, padding=10 ) for sale in sales: sales_list.controls.append( ft.Card( content=ft.Container( content=ft.ListTile( title=ft.Text(f"{sale['customer_name']} - {sale['product_name']}"), subtitle=ft.Text(f"{sale['quantity']}個 × ¥{sale['unit_price']:,.0f} = ¥{sale['total_price']:,.0f}"), trailing=ft.Text(sale['sale_date'].split()[0]) ), padding=10 ) ) ) # Add sale button add_button = ft.ElevatedButton( "売上を追加", icon=ft.icons.ADD, on_click=self.add_sale_dialog ) sales_content = ft.Column([ ft.Row([ ft.Text("売上管理", size=24, weight=ft.FontWeight.BOLD), add_button ], alignment=ft.MainAxisAlignment.SPACE_BETWEEN), ft.Divider(), sales_list ], scroll=ft.ScrollMode.AUTO) self.content_area.content = ft.Container(sales_content, padding=20) self.page.update() def get_customers(self) -> List[Dict]: """Get all customers from database""" conn = sqlite3.connect('sales.db') 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]: """Get all products from database""" conn = sqlite3.connect('sales.db') 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]: """Get all sales with customer and product names""" conn = sqlite3.connect('sales.db') 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 add_customer_dialog(self, e): """Show add customer dialog""" name_field = ft.TextField(label="氏名", autofocus=True) company_field = ft.TextField(label="会社名") phone_field = ft.TextField(label="電話番号") email_field = ft.TextField(label="メールアドレス") address_field = ft.TextField(label="住所") def save_customer(e): conn = sqlite3.connect('sales.db') cursor = conn.cursor() cursor.execute(''' INSERT INTO customers (name, company, phone, email, address) VALUES (?, ?, ?, ?, ?) ''', (name_field.value, company_field.value, phone_field.value, email_field.value, address_field.value)) conn.commit() conn.close() dialog.open = False self.show_customers() self.page.update() dialog = ft.AlertDialog( title=ft.Text("顧客を追加"), content=ft.Column([ name_field, company_field, phone_field, email_field, address_field ], tight=True), actions=[ ft.TextButton("キャンセル", on_click=lambda e: self.close_dialog(dialog)), ft.TextButton("保存", on_click=save_customer) ], actions_alignment=ft.MainAxisAlignment.END ) self.page.dialog = dialog dialog.open = True self.page.update() def add_product_dialog(self, e): """Show add product dialog""" name_field = ft.TextField(label="商品名", autofocus=True) code_field = ft.TextField(label="商品コード") price_field = ft.TextField(label="価格", keyboard_type=ft.KeyboardType.NUMBER) stock_field = ft.TextField(label="在庫数", keyboard_type=ft.KeyboardType.NUMBER) description_field = ft.TextField(label="説明", multiline=True) def save_product(e): conn = sqlite3.connect('sales.db') cursor = conn.cursor() cursor.execute(''' INSERT INTO products (name, code, price, stock, description) VALUES (?, ?, ?, ?, ?) ''', (name_field.value, code_field.value, float(price_field.value or 0), int(stock_field.value or 0), description_field.value)) conn.commit() conn.close() dialog.open = False self.show_products() self.page.update() dialog = ft.AlertDialog( title=ft.Text("商品を追加"), content=ft.Column([ name_field, code_field, price_field, stock_field, description_field ], tight=True), actions=[ ft.TextButton("キャンセル", on_click=lambda e: self.close_dialog(dialog)), ft.TextButton("保存", on_click=save_product) ], actions_alignment=ft.MainAxisAlignment.END ) self.page.dialog = dialog dialog.open = True self.page.update() def add_sale_dialog(self, e): """Show add sale dialog""" customers = self.get_customers() products = self.get_products() customer_dropdown = ft.Dropdown( label="顧客", options=[ft.dropdown.Option(c['name'], key=str(c['id'])) for c in customers] ) product_dropdown = ft.Dropdown( label="商品", options=[ft.dropdown.Option(p['name'], key=str(p['id'])) for p in products], on_change=lambda e: self.update_price_display(e, product_dropdown, price_field, quantity_field, total_field) ) quantity_field = ft.TextField( label="数量", value="1", keyboard_type=ft.KeyboardType.NUMBER, on_change=lambda e: self.calculate_total(price_field, quantity_field, total_field) ) price_field = ft.TextField( label="単価", keyboard_type=ft.KeyboardType.NUMBER, on_change=lambda e: self.calculate_total(price_field, quantity_field, total_field) ) total_field = ft.TextField(label="合計", read_only=True) def save_sale(e): conn = sqlite3.connect('sales.db') cursor = conn.cursor() cursor.execute(''' INSERT INTO sales (customer_id, product_id, quantity, unit_price, total_price) VALUES (?, ?, ?, ?, ?) ''', (int(customer_dropdown.value), int(product_dropdown.value), int(quantity_field.value), float(price_field.value), float(total_field.value))) conn.commit() conn.close() dialog.open = False self.show_sales() self.page.update() dialog = ft.AlertDialog( title=ft.Text("売上を追加"), content=ft.Column([ customer_dropdown, product_dropdown, quantity_field, price_field, total_field ], tight=True), actions=[ ft.TextButton("キャンセル", on_click=lambda e: self.close_dialog(dialog)), ft.TextButton("保存", on_click=save_sale) ], actions_alignment=ft.MainAxisAlignment.END ) self.page.dialog = dialog dialog.open = True self.page.update() def update_price_display(self, e, product_dropdown, price_field, quantity_field, total_field): """Update price when product is selected""" if product_dropdown.value: conn = sqlite3.connect('sales.db') cursor = conn.cursor() cursor.execute("SELECT price FROM products WHERE id = ?", (int(product_dropdown.value),)) result = cursor.fetchone() conn.close() if result: price_field.value = str(result[0]) self.calculate_total(price_field, quantity_field, total_field) self.page.update() def calculate_total(self, price_field, quantity_field, total_field): """Calculate total price""" try: price = float(price_field.value or 0) quantity = int(quantity_field.value or 1) total_field.value = str(price * quantity) except ValueError: total_field.value = "0" self.page.update() def close_dialog(self, dialog): """Close dialog""" dialog.open = False self.page.update() def edit_customer(self, customer): """Edit customer (placeholder)""" self.page.snack_bar = ft.SnackBar(content=ft.Text("顧客編集機能は準備中です")) self.page.snack_bar.open = True self.page.update() def delete_customer(self, customer): """Delete customer""" def confirm_delete(e): conn = sqlite3.connect('sales.db') cursor = conn.cursor() cursor.execute("DELETE FROM customers WHERE id = ?", (customer['id'],)) conn.commit() conn.close() dialog.open = False self.show_customers() self.page.update() dialog = ft.AlertDialog( title=ft.Text("確認"), content=ft.Text(f"顧客「{customer['name']}」を削除してもよろしいですか？"), actions=[ ft.TextButton("キャンセル", on_click=lambda e: self.close_dialog(dialog)), ft.TextButton("削除", on_click=confirm_delete) ], actions_alignment=ft.MainAxisAlignment.END ) self.page.dialog = dialog dialog.open = True self.page.update() def edit_product(self, product): """Edit product (placeholder)""" self.page.snack_bar = ft.SnackBar(content=ft.Text("商品編集機能は準備中です")) self.page.snack_bar.open = True self.page.update() def delete_product(self, product): """Delete product""" def confirm_delete(e): conn = sqlite3.connect('sales.db') cursor = conn.cursor() cursor.execute("DELETE FROM products WHERE id = ?", (product['id'],)) conn.commit() conn.close() dialog.open = False self.show_products() self.page.update() dialog = ft.AlertDialog( title=ft.Text("確認"), content=ft.Text(f"商品「{product['name']}」を削除してもよろしいですか？"), actions=[ ft.TextButton("キャンセル", on_click=lambda e: self.close_dialog(dialog)), ft.TextButton("削除", on_click=confirm_delete) ], actions_alignment=ft.MainAxisAlignment.END ) self.page.dialog = dialog dialog.open = True self.page.update() def show_data_export(self): """データ出力画面を表示""" export_content = ft.Column([ ft.Text("データ出力", size=24, weight=ft.FontWeight.BOLD), ft.Divider(), ft.Card( content=ft.Container( content=ft.Column([ ft.Text("JSON形式で出力", size=16, weight=ft.FontWeight.BOLD), ft.Text("全データをJSON形式でエクスポートします"), ft.ElevatedButton( "JSON出力", icon=ft.icons.DOWNLOAD, on_click=self.export_json ) ]), padding=20 ) ), ft.Card( content=ft.Container( content=ft.Column([ ft.Text("CSV形式で出力", size=16, weight=ft.FontWeight.BOLD), ft.Text("各テーブルをCSV形式でエクスポートします"), ft.ElevatedButton( "CSV出力", icon=ft.icons.DOWNLOAD, on_click=self.export_csv ) ]), padding=20 ) ) ], scroll=ft.ScrollMode.AUTO) self.content_area.content = ft.Container(export_content, padding=20) self.page.update() def show_compliance(self): """コンプライアンス画面を表示""" compliance_content = ft.Column([ ft.Text("電子帳簿保存法対応", size=24, weight=ft.FontWeight.BOLD), ft.Divider(), ft.Card( content=ft.Container( content=ft.Column([ ft.Text("データ整合性チェック", size=16, weight=ft.FontWeight.BOLD), ft.Text("データの整合性を検証し、チェックサムを生成します"), ft.ElevatedButton( "整合性チェック", icon=ft.icons.CHECK_CIRCLE, on_click=self.check_integrity ) ]), padding=20 ) ), ft.Card( content=ft.Container( content=ft.Column([ ft.Text("古いデータのアーカイブ", size=16, weight=ft.FontWeight.BOLD), ft.Text("7年以上前のデータをアーカイブします（10年保存）"), ft.ElevatedButton( "データアーカイブ", icon=ft.icons.ARCHIVE, on_click=self.archive_data ) ]), padding=20 ) ), ft.Card( content=ft.Container( content=ft.Column([ ft.Text("コンプライアンスレポート", size=16, weight=ft.FontWeight.BOLD), ft.Text("電子帳簿保存法対応レポートを生成します"), ft.ElevatedButton( "レポート生成", icon=ft.icons.ASSIGNMENT, on_click=self.generate_compliance_report ) ]), padding=20 ) ) ], scroll=ft.ScrollMode.AUTO) self.content_area.content = ft.Container(compliance_content, padding=20) self.page.update() def export_json(self, e): """JSON形式でデータをエクスポート""" try: file_path = self.exporter.export_to_json() self.page.snack_bar = ft.SnackBar( content=ft.Text(f"JSON出力完了: {file_path}"), bgcolor=ft.colors.GREEN ) except Exception as ex: self.page.snack_bar = ft.SnackBar( content=ft.Text(f"JSON出力エラー: {str(ex)}"), bgcolor=ft.colors.RED ) self.page.snack_bar.open = True self.page.update() def export_csv(self, e): """CSV形式でデータをエクスポート""" try: files = self.exporter.export_to_csv() file_list = "\n".join([f"{k}: {v}" for k, v in files.items()]) self.page.snack_bar = ft.SnackBar( content=ft.Text(f"CSV出力完了:\n{file_list}"), bgcolor=ft.colors.GREEN ) except Exception as ex: self.page.snack_bar = ft.SnackBar( content=ft.Text(f"CSV出力エラー: {str(ex)}"), bgcolor=ft.colors.RED ) self.page.snack_bar.open = True self.page.update() def check_integrity(self, e): """データ整合性チェック""" try: results = [] for table in ["customers", "products", "sales"]: result = self.compliance.verify_data_integrity(table) results.append(f"{table}: {result['status']} ({result['record_count']}件)") message = "整合性チェック完了:\n" + "\n".join(results) self.page.snack_bar = ft.SnackBar( content=ft.Text(message), bgcolor=ft.colors.GREEN ) except Exception as ex: self.page.snack_bar = ft.SnackBar( content=ft.Text(f"整合性チェックエラー: {str(ex)}"), bgcolor=ft.colors.RED ) self.page.snack_bar.open = True self.page.update() def archive_data(self, e): """古いデータをアーカイブ""" try: result = self.compliance.archive_old_data() if result["status"] == "SUCCESS": message = f"アーカイブ完了: {result['archived_count']}件" bgcolor = ft.colors.GREEN else: message = f"アーカイブエラー: {result['error']}" bgcolor = ft.colors.RED self.page.snack_bar = ft.SnackBar( content=ft.Text(message), bgcolor=bgcolor ) except Exception as ex: self.page.snack_bar = ft.SnackBar( content=ft.Text(f"アーカイブエラー: {str(ex)}"), bgcolor=ft.colors.RED ) self.page.snack_bar.open = True self.page.update() def generate_compliance_report(self, e): """コンプライアンスレポートを生成""" try: report = self.compliance.generate_compliance_report() # レポート内容を表示 dialog = ft.AlertDialog( title=ft.Text("コンプライアンスレポート"), content=ft.Container( content=ft.Column([ ft.Text(f"生成日時: {report['generated_date']}"), ft.Text(f"データベース: {report['database_path']}"), ft.Divider(), ft.Text("テーブル状況:", weight=ft.FontWeight.BOLD), *[ft.Text(f" {table}: {info['record_count']}件") for table, info in report['tables'].items()], ft.Divider(), ft.Text("監査ログ:", weight=ft.FontWeight.BOLD), *[ft.Text(f" {op}: {count}件") for op, count in report['audit_summary'].items()] ], scroll=ft.ScrollMode.AUTO), height=400 ), actions=[ ft.TextButton("閉じる", on_click=lambda e: self.close_dialog(dialog)) ] ) self.page.dialog = dialog dialog.open = True self.page.update() except Exception as ex: self.page.snack_bar = ft.SnackBar( content=ft.Text(f"レポート生成エラー: {str(ex)}"), bgcolor=ft.colors.RED ) self.page.snack_bar.open = True self.page.update() def main(page: ft.Page): app = SalesAssistant(page) if __name__ == "__main__": ft.app(target=main)