""" Database service classes for CRUD operations """ from sqlalchemy.orm import Session from sqlalchemy import and_, or_, desc from typing import List, Optional, Dict, Any from datetime import datetime, timedelta import logging from database import User, Order, Product, OrderItem, ProductCode, ProductGallery, ProductAttribute, Sale, SaleItem, get_db_session, engine from sqlalchemy.orm import sessionmaker # Configure logging logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) # Create session factory SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine) class UserService: """Service class for User operations""" @staticmethod def get_user_by_id(user_id: int) -> Optional[User]: """Get user by ID""" db = SessionLocal() try: return db.query(User).filter(User.id == user_id).first() finally: db.close() @staticmethod def get_user_by_username(username: str) -> Optional[User]: """Get user by username""" db = SessionLocal() try: return db.query(User).filter(User.username == username).first() finally: db.close() @staticmethod def get_user_by_email(email: str) -> Optional[User]: """Get user by email""" db = SessionLocal() try: return db.query(User).filter(User.email == email).first() finally: db.close() @staticmethod def create_user(username: str, email: str, password: str, role: str = "user") -> User: """Create a new user""" db = SessionLocal() try: # Check if user already exists existing_user = db.query(User).filter( or_(User.username == username, User.email == email) ).first() if existing_user: raise ValueError("Username or email already exists") user = User(username=username, email=email, role=role) user.set_password(password) db.add(user) db.commit() db.refresh(user) logger.info(f"Created new user: {username}") return user except Exception as e: db.rollback() logger.error(f"Failed to create user {username}: {e}") raise finally: db.close() @staticmethod def update_last_login(username: str) -> None: """Update user's last login time""" db = SessionLocal() try: user = db.query(User).filter(User.username == username).first() if user: user.last_login = datetime.utcnow() db.commit() logger.info(f"Updated last login for user: {username}") except Exception as e: db.rollback() logger.error(f"Failed to update last login for {username}: {e}") finally: db.close() @staticmethod def get_all_users() -> List[User]: """Get all users""" db = SessionLocal() try: return db.query(User).all() finally: db.close() @staticmethod def delete_user(user_id: int) -> bool: """Delete user by ID""" db = SessionLocal() try: user = db.query(User).filter(User.id == user_id).first() if user: db.delete(user) db.commit() logger.info(f"Deleted user: {user.username}") return True return False except Exception as e: db.rollback() logger.error(f"Failed to delete user {user_id}: {e}") return False finally: db.close() class OrderService: """Service class for Order operations""" @staticmethod def get_order_by_id(order_id: int) -> Optional[Order]: """Get order by ID with items""" db = SessionLocal() try: return db.query(Order).filter(Order.id == order_id).first() finally: db.close() @staticmethod def get_orders_by_platform(platform: str, limit: int = 100) -> List[Order]: """Get orders by platform""" db = SessionLocal() try: return db.query(Order).filter(Order.platform == platform).limit(limit).all() finally: db.close() @staticmethod def get_orders_by_status(status: str, limit: int = 100) -> List[Order]: """Get orders by status""" db = SessionLocal() try: return db.query(Order).filter(Order.status == status).limit(limit).all() finally: db.close() @staticmethod def create_order(order_number: str, customer_name: str, customer_email: str = None, total_amount: float = 0.0, freight_amount: float = 0.0, platform: str = None) -> Order: """Create a new order""" db = SessionLocal() try: order = Order( order_number=order_number, customer_name=customer_name, customer_email=customer_email, total_amount=total_amount, freight_amount=freight_amount, platform=platform ) db.add(order) db.commit() db.refresh(order) logger.info(f"Created new order: {order_number}") return order except Exception as e: db.rollback() logger.error(f"Failed to create order {order_number}: {e}") raise finally: db.close() @staticmethod def get_daily_sales_data(date: str) -> Dict[str, Any]: """Get sales data for a specific date""" db = SessionLocal() try: # Parse date target_date = datetime.strptime(date, '%Y-%m-%d') # Get orders for the specific date orders = db.query(Order).filter( Order.created_at >= target_date, Order.created_at < target_date + timedelta(days=1) ).all() # Aggregate data by platform platform_totals = {} total_sales = 0 total_freight = 0 for order in orders: platform = order.platform or 'Other' if platform not in platform_totals: platform_totals[platform] = { 'orders': 0, 'total': 0.0, 'freight': 0.0 } platform_totals[platform]['orders'] += 1 platform_totals[platform]['total'] += order.total_amount platform_totals[platform]['freight'] += order.freight_amount total_sales += order.total_amount total_freight += order.freight_amount return { 'date': date, 'formatted_date': target_date.strftime('%d/%m/%Y'), 'summary': { 'valor_pedidos': total_sales, 'valor_fretes': total_freight, 'valor_total': total_sales + total_freight, 'num_pedidos': len(orders) }, 'platform_totals': platform_totals } finally: db.close() @staticmethod def get_sales_period_data(start_date: str, end_date: str) -> Dict[str, Any]: """Get sales data for a date range with 7-day averages""" db = SessionLocal() try: start = datetime.strptime(start_date, '%Y-%m-%d') end = datetime.strptime(end_date, '%Y-%m-%d') # Extend range to include previous week for 7-day averages extended_start = start - timedelta(days=7) # Get orders in extended date range orders = db.query(Order).filter( Order.created_at >= extended_start, Order.created_at <= end ).all() # Group by date and platform daily_data = {} current_date = extended_start while current_date <= end: date_str = current_date.strftime('%Y-%m-%d') daily_data[date_str] = { 'geral': 0, 'MercadoLivre': 0, 'Shopee': 0, 'Magalu': 0, 'Amazon': 0, 'Site': 0, 'BelezaNaWeb': 0, 'WebContinental': 0, 'Enjoei': 0 } current_date += timedelta(days=1) # Populate with actual data for order in orders: date_str = order.created_at.strftime('%Y-%m-%d') platform = order.platform or 'Site' if platform in daily_data[date_str]: daily_data[date_str][platform] += order.total_amount daily_data[date_str]['geral'] += order.total_amount # Calculate 7-day moving averages moving_averages = {} platforms = ['geral', 'MercadoLivre', 'Shopee', 'Magalu', 'Amazon', 'Site', 'BelezaNaWeb', 'WebContinental', 'Enjoei'] current_date = start while current_date <= end: date_str = current_date.strftime('%Y-%m-%d') moving_averages[date_str] = {} for platform in platforms: moving_averages[date_str][platform] = calculate_7day_average(date_str, daily_data, platform) current_date += timedelta(days=1) # Filter daily_data to only include requested date range filtered_daily_data = {k: v for k, v in daily_data.items() if k >= start_date and k <= end_date} filtered_averages = {k: v for k, v in moving_averages.items() if k >= start_date and k <= end_date} return { 'date_range': { 'start': start_date, 'end': end_date }, 'daily_data': filtered_daily_data, 'moving_averages': filtered_averages } finally: db.close() def calculate_7day_average(target_date: str, data_values: Dict[str, Any], platform: str) -> float: """ Calculate 7-day average for a specific platform - replicates PHP function """ try: target_dt = datetime.strptime(target_date, '%Y-%m-%d') sum_values = 0 days_counted = 0 # Calculate average of last 7 days (including target date) for i in range(7): check_date = (target_dt - timedelta(days=i)).strftime('%Y-%m-%d') if check_date in data_values and platform in data_values[check_date]: sum_values += data_values[check_date][platform] days_counted += 1 return round(sum_values / days_counted, 2) if days_counted > 0 else 0 except Exception as e: logger.error(f"Error calculating 7-day average for {target_date}, {platform}: {e}") return 0 class ProductService: """Service class for Product operations""" @staticmethod def get_product_by_id(product_id: int) -> Optional[Product]: """Get product by ID""" db = SessionLocal() try: return db.query(Product).filter(Product.id == product_id).first() finally: db.close() @staticmethod def get_product_by_sku(sku: str) -> Optional[Product]: """Get product by SKU""" db = SessionLocal() try: return db.query(Product).filter(Product.sku == sku).first() finally: db.close() @staticmethod def get_products_by_category(category: str, limit: int = 100) -> List[Product]: """Get products by category""" db = SessionLocal() try: return db.query(Product).filter(Product.category == category).limit(limit).all() finally: db.close() @staticmethod def create_product(sku: str, name: str, cost_price: float = 0.0, sale_price: float = 0.0, stock_quantity: int = 0, category: str = None, description: str = None) -> Product: """Create a new product""" db = SessionLocal() try: # Check if product already exists existing_product = db.query(Product).filter(Product.sku == sku).first() if existing_product: raise ValueError(f"Product with SKU {sku} already exists") product = Product( sku=sku, name=name, cost_price=cost_price, sale_price=sale_price, stock_quantity=stock_quantity, category=category, description=description ) db.add(product) db.commit() db.refresh(product) logger.info(f"Created new product: {sku} - {name}") return product except Exception as e: db.rollback() logger.error(f"Failed to create product {sku}: {e}") raise finally: db.close() @staticmethod def update_stock(sku: str, new_quantity: int) -> bool: """Update product stock quantity""" db = SessionLocal() try: product = db.query(Product).filter(Product.sku == sku).first() if product: product.stock_quantity = new_quantity db.commit() logger.info(f"Updated stock for {sku}: {new_quantity}") return True return False except Exception as e: db.rollback() logger.error(f"Failed to update stock for {sku}: {e}") return False finally: db.close() class DashboardService: """Service class for dashboard data""" @staticmethod def get_dashboard_summary() -> Dict[str, Any]: """Get dashboard summary data""" db = SessionLocal() try: # Get today's sales today = datetime.utcnow().replace(hour=0, minute=0, second=0, microsecond=0) today_orders = db.query(Order).filter(Order.created_at >= today).all() daily_sales = sum(order.total_amount for order in today_orders) total_orders = len(today_orders) average_ticket = daily_sales / total_orders if total_orders > 0 else 0 # Get record sales (mock data for now) record_sales = 44671.48 record_date = "07/07/2025" return { "daily_sales": daily_sales, "total_orders": total_orders, "average_ticket": average_ticket, "refund_amount": 0.00, "record_sales": record_sales, "record_date": record_date, "refresh_token_remaining": "28 dias", "refresh_token_expire": "17/09/2025" } finally: db.close() @staticmethod def get_orders_status() -> Dict[str, int]: """Get orders status breakdown""" db = SessionLocal() try: total_orders = db.query(Order).count() # Mock data for now - in real implementation, this would be based on actual order statuses return { "to_separate": int(total_orders * 0.2), "unlabeled": int(total_orders * 0.5), "to_send": int(total_orders * 0.15), "ready_to_ship": int(total_orders * 0.1), "with_problems": int(total_orders * 0.03), "freight_combine": int(total_orders * 0.02) } finally: db.close() @staticmethod def get_marketplace_breakdown() -> List[Dict[str, Any]]: """Get sales breakdown by marketplace""" db = SessionLocal() try: # Group orders by platform from sqlalchemy import func platform_data = db.query( Order.platform, func.count(Order.id).label('order_count'), func.sum(Order.total_amount).label('total_sales') ).group_by(Order.platform).all() result = [] for platform, order_count, total_sales in platform_data: platform_name = platform or 'Outros' logo_map = { 'MercadoLivre': 'https://cdn.worldvectorlogo.com/logos/mercado-livre-2.svg', 'Shopee': 'https://img.icons8.com/m_outlined/512/shopee.png', 'Magalu': 'https://sys.lojalemis.com/wp-content/uploads/icons/magalu.png', 'Amazon': 'https://www.svgrepo.com/show/112049/amazon-logo.svg', 'Site': 'https://sys.lojalemis.com/wp-content/uploads/icons/lemis.png' } result.append({ "name": platform_name, "orders": int(order_count or 0), "revenue": float(total_sales or 0), "logo": logo_map.get(platform_name, "") }) return result finally: db.close() class ExtendedProductService: """Extended service class for WordPress-like product management""" @staticmethod def create_product_with_details(product_data: Dict[str, Any]) -> Product: """Create a product with codes, galleries, and attributes""" db = SessionLocal() try: # Create main product product = Product( sku=product_data.get('sku', ''), name=product_data.get('title', ''), description=product_data.get('description', ''), cost_price=product_data.get('cost', 0.0), sale_price=product_data.get('price', 0.0), category=product_data.get('category', ''), is_active=not product_data.get('fora_de_linha', False) ) db.add(product) db.flush() # Get the product ID # Add product codes for code_data in product_data.get('codes', []): product_code = ProductCode( product_id=product.id, code=code_data['code'], size=code_data.get('size', ''), ean=code_data.get('ean', ''), stock_store=code_data.get('stock_store', 0), stock_one=code_data.get('stock_one', 0) ) db.add(product_code) # Add gallery images for gallery_type in ['tray', 'enjoei']: if gallery_type in product_data.get('galleries', {}): for idx, image_url in enumerate(product_data['galleries'][gallery_type]): gallery = ProductGallery( product_id=product.id, gallery_type=gallery_type, image_url=image_url, image_order=idx ) db.add(gallery) # Add attributes attributes = product_data.get('attributes', {}) for attr_type, attr_value in attributes.items(): if attr_value: attr = ProductAttribute( product_id=product.id, attribute_type=attr_type, attribute_value=str(attr_value) ) db.add(attr) db.commit() db.refresh(product) logger.info(f"Created extended product: {product.name} with ID {product.id}") return product except Exception as e: db.rollback() logger.error(f"Failed to create extended product: {e}") raise finally: db.close() @staticmethod def search_products(query: str, filters: Dict[str, Any] = None, limit: int = 50) -> List[Dict[str, Any]]: """Search products with filters (replicating WordPress search)""" db = SessionLocal() try: query_obj = db.query(Product) # Text search if query: search_term = f"%{query}%" query_obj = query_obj.filter( or_( Product.name.ilike(search_term), Product.sku.ilike(search_term), Product.description.ilike(search_term) ) ) # Apply filters if filters: if filters.get('category'): query_obj = query_obj.filter(Product.category == filters['category']) if filters.get('min_price'): query_obj = query_obj.filter(Product.sale_price >= filters['min_price']) if filters.get('max_price'): query_obj = query_obj.filter(Product.sale_price <= filters['max_price']) if filters.get('in_stock_only'): # This would need to check product codes for total stock pass products = query_obj.limit(limit).all() # Format results like WordPress results = [] for product in products: # Get codes and calculate total stock codes = db.query(ProductCode).filter(ProductCode.product_id == product.id).all() total_stock = sum(code.stock_store + code.stock_one for code in codes) # Get gallery images galleries = db.query(ProductGallery).filter(ProductGallery.product_id == product.id).all() gallery_urls = {} for gallery in galleries: if gallery.gallery_type not in gallery_urls: gallery_urls[gallery.gallery_type] = [] gallery_urls[gallery.gallery_type].append(gallery.image_url) # Get attributes attributes = db.query(ProductAttribute).filter(ProductAttribute.product_id == product.id).all() attr_dict = {attr.attribute_type: attr.attribute_value for attr in attributes} results.append({ 'id': product.id, 'sku': product.sku, 'title': product.name, 'description': product.description, 'price': product.sale_price, 'cost': product.cost_price, 'category': product.category, 'codes': [{'code': code.code, 'size': code.size, 'ean': code.ean, 'stock_store': code.stock_store, 'stock_one': code.stock_one} for code in codes], 'galleries': gallery_urls, 'attributes': attr_dict, 'total_stock': total_stock, 'is_active': product.is_active }) return results finally: db.close() class SalesService: """Service class for WordPress-like sales/vendas management""" @staticmethod def get_product_analysis_data(start_date: str, end_date: str) -> Dict[str, Any]: """Get product analysis data for CSV export""" db = SessionLocal() try: # Mock data for now - in real implementation, this would query actual sales data lojas = ['Geral', 'Shopee Store', 'Shopee One', 'Shopee KS', 'MercadoLivre One', 'MercadoLivre Store', 'Magalu One', 'Magalu Store', 'Site'] product_analysis = {} for loja in lojas: product_analysis[loja] = { "items": { "101528": { "nome": "Produto Exemplo 1", "custo": 25.50, "unidades": 85, "valor": 599.90 }, "103170": { "nome": "Produto Exemplo 2", "custo": 15.75, "unidades": 120, "valor": 899.25 }, "104285": { "nome": "Produto Exemplo 3", "custo": 35.00, "unidades": 45, "valor": 1249.50 } }, "totals": { "custo_total": 0, "venda_total": 0, "margem_total": 0, "lucro_total": 0 } } # Calculate totals for the store custo_loja = 0 venda_loja = 0 for item in product_analysis[loja]["items"].values(): custo_loja += item['custo'] * item['unidades'] venda_loja += item['valor'] margem_total = ((venda_loja - custo_loja) / venda_loja) * 100 if venda_loja > 0 else 0 lucro_total = venda_loja - custo_loja product_analysis[loja]["totals"] = { "custo_total": round(custo_loja, 2), "venda_total": round(venda_loja, 2), "margem_total": round(margem_total, 2), "lucro_total": round(lucro_total, 2) } return { "date_range": { "start": start_date, "end": end_date }, "product_analysis": product_analysis } finally: db.close() @staticmethod def create_sale(sale_data: Dict[str, Any]) -> Sale: """Create a sale with items (replicating WordPress sale creation)""" db = SessionLocal() try: # Create main sale sale = Sale( id_venda=sale_data['id_venda'], data=datetime.strptime(sale_data['data'], '%Y-%m-%d'), vendedor=sale_data['vendedor'], cliente=sale_data.get('cliente', ''), frete=sale_data.get('frete', 0.0), plataforma=sale_data.get('plataforma', ''), outro_plataforma=sale_data.get('outro_plataforma', ''), taxas=sale_data.get('taxas', 0.0), id_venda_ml=sale_data.get('id_venda_ml', ''), obs=sale_data.get('obs', ''), devolucao=sale_data.get('devolucao', False) ) db.add(sale) db.flush() subtotal = 0.0 # Add sale items for item_data in sale_data.get('items', []): item = SaleItem( sale_id=sale.id, product_id=item_data.get('product_id'), product_name=item_data['product_name'], code=item_data['code'], size=item_data.get('size', ''), quantity=item_data['quantity'], price=item_data['price'], total=item_data['total'] ) db.add(item) subtotal += item_data['total'] # Update totals sale.subtotal = subtotal sale.total = subtotal + sale.frete + sale.taxas db.commit() db.refresh(sale) logger.info(f"Created sale: {sale.id_venda} with total {sale.total}") return sale except Exception as e: db.rollback() logger.error(f"Failed to create sale: {e}") raise finally: db.close() @staticmethod def get_sales(filters: Dict[str, Any] = None, limit: int = 100) -> List[Dict[str, Any]]: """Get sales with filters (replicating WordPress sales queries)""" db = SessionLocal() try: query = db.query(Sale) # Apply filters if filters: if filters.get('start_date'): start_date = datetime.strptime(filters['start_date'], '%Y-%m-%d') query = query.filter(Sale.data >= start_date) if filters.get('end_date'): end_date = datetime.strptime(filters['end_date'], '%Y-%m-%d') query = query.filter(Sale.data <= end_date) if filters.get('vendedor'): query = query.filter(Sale.vendedor.ilike(f"%{filters['vendedor']}%")) if filters.get('cliente'): query = query.filter(Sale.cliente.ilike(f"%{filters['cliente']}%")) if filters.get('plataforma'): query = query.filter(Sale.plataforma == filters['plataforma']) sales = query.order_by(Sale.data.desc()).limit(limit).all() # Format results results = [] for sale in sales: items = db.query(SaleItem).filter(SaleItem.sale_id == sale.id).all() results.append({ 'id': sale.id, 'id_venda': sale.id_venda, 'data': sale.data.strftime('%Y-%m-%d'), 'vendedor': sale.vendedor, 'cliente': sale.cliente, 'plataforma': sale.plataforma, 'total': sale.total, 'frete': sale.frete, 'status_pacote': sale.status_pacote, 'devolucao': sale.devolucao, 'items': [{ 'product_name': item.product_name, 'code': item.code, 'size': item.size, 'quantity': item.quantity, 'price': item.price, 'total': item.total } for item in items] }) return results finally: db.close() # Legacy WordPress data access (for existing data) class WordPressService: """Service class for accessing existing WordPress data""" @staticmethod def get_wordpress_posts(limit: int = 100) -> List[Dict[str, Any]]: """Get posts from WordPress wp_posts table""" db = SessionLocal() try: # This would use raw SQL to access WordPress tables # For now, return mock data return [] finally: db.close() @staticmethod def get_wordpress_postmeta(post_id: int) -> List[Dict[str, Any]]: """Get postmeta for a specific post""" db = SessionLocal() try: # This would use raw SQL to access WordPress wp_postmeta table return [] finally: db.close()