# """ # Viewflow workflows for inventory app. # Provides inventory management, procurement, and supply chain workflows. # """ # # from viewflow import Flow, lock # from viewflow.base import this, flow_func # from viewflow.contrib import celery # from viewflow.decorators import flow_view # from viewflow.fields import CharField, ModelField # from viewflow.forms import ModelForm # from viewflow.views import CreateProcessView, UpdateProcessView # from viewflow.models import Process, Task # from django.contrib.auth.models import User # from django.urls import reverse_lazy # from django.utils import timezone # from django.db import transaction # from django.core.mail import send_mail # # from .models import InventoryItem, PurchaseOrder, PurchaseOrderItem, Supplier, InventoryStock, InventoryLocation # from .views import ( # PurchaseRequestView, VendorSelectionView, PurchaseOrderCreationView, # ApprovalView, OrderSubmissionView, ReceivingView, InspectionView, # StockUpdateView, InvoiceMatchingView, PaymentProcessingView, # StockReplenishmentView, StockTransferView, StockAdjustmentView, # CycleCountView, InventoryAuditView # ) # # # class ProcurementProcess(Process): # """ # Viewflow process model for procurement # """ # purchase_order = ModelField(PurchaseOrder, help_text='Associated purchase order') # # # Process status tracking # request_submitted = models.BooleanField(default=False) # vendor_selected = models.BooleanField(default=False) # order_created = models.BooleanField(default=False) # order_approved = models.BooleanField(default=False) # order_sent = models.BooleanField(default=False) # goods_received = models.BooleanField(default=False) # invoice_processed = models.BooleanField(default=False) # payment_completed = models.BooleanField(default=False) # procurement_closed = models.BooleanField(default=False) # # class Meta: # verbose_name = 'Procurement Process' # verbose_name_plural = 'Procurement Processes' # # # class ProcurementFlow(Flow): # """ # Procurement Workflow # # This flow manages the complete procurement process from purchase # request through payment and order closure. # """ # # process_class = ProcurementProcess # # # Flow definition # start = ( # flow_func(this.start_procurement) # .Next(this.submit_request) # ) # # submit_request = ( # flow_view(PurchaseRequestView) # .Permission('inventory.can_submit_purchase_requests') # .Next(this.select_vendor) # ) # # select_vendor = ( # flow_view(VendorSelectionView) # .Permission('inventory.can_select_vendors') # .Next(this.create_order) # ) # # create_order = ( # flow_view(PurchaseOrderCreationView) # .Permission('inventory.can_create_purchase_orders') # .Next(this.approve_order) # ) # # approve_order = ( # flow_view(ApprovalView) # .Permission('inventory.can_approve_purchase_orders') # .Next(this.send_order) # ) # # send_order = ( # flow_view(OrderSubmissionView) # .Permission('inventory.can_send_purchase_orders') # .Next(this.receive_goods) # ) # # receive_goods = ( # flow_view(ReceivingView) # .Permission('inventory.can_receive_goods') # .Next(this.inspect_goods) # ) # # inspect_goods = ( # flow_view(InspectionView) # .Permission('inventory.can_inspect_goods') # .Next(this.update_stock) # ) # # update_stock = ( # flow_view(StockUpdateView) # .Permission('inventory.can_update_stock') # .Next(this.process_invoice) # ) # # process_invoice = ( # flow_view(InvoiceMatchingView) # .Permission('inventory.can_process_invoices') # .Next(this.process_payment) # ) # # process_payment = ( # flow_view(PaymentProcessingView) # .Permission('inventory.can_process_payments') # .Next(this.close_order) # ) # # close_order = ( # flow_func(this.complete_procurement) # .Next(this.end) # ) # # end = flow_func(this.end_procurement) # # # Flow functions # def start_procurement(self, activation): # """Initialize the procurement process""" # process = activation.process # order = process.purchase_order # # # Update order status # order.status = 'DRAFT' # order.save() # # # Send notification to procurement staff # self.notify_procurement_staff(order) # # # Check for urgent orders # if order.priority in ['HIGH', 'URGENT']: # self.notify_urgent_procurement(order) # # def complete_procurement(self, activation): # """Finalize the procurement process""" # process = activation.process # order = process.purchase_order # # # Update order status # order.status = 'CLOSED' # order.save() # # # Mark process as completed # process.procurement_closed = True # process.save() # # # Send completion notifications # self.notify_procurement_completion(order) # # # Update supplier performance metrics # self.update_supplier_performance(order) # # # Update procurement metrics # self.update_procurement_metrics(order) # # def end_procurement(self, activation): # """End the procurement workflow""" # process = activation.process # # # Generate procurement summary report # self.generate_procurement_summary(process.purchase_order) # # # Helper methods # def notify_procurement_staff(self, order): # """Notify procurement staff of new order""" # from django.contrib.auth.models import Group # # procurement_staff = User.objects.filter( # groups__name='Procurement Staff' # ) # # for staff in procurement_staff: # send_mail( # subject=f'New Purchase Order: {order.po_number}', # message=f'New purchase order for {order.supplier.name} requires processing.', # from_email='procurement@hospital.com', # recipient_list=[staff.email], # fail_silently=True # ) # # def notify_urgent_procurement(self, order): # """Notify of urgent procurement""" # procurement_managers = User.objects.filter( # groups__name='Procurement Managers' # ) # # for manager in procurement_managers: # send_mail( # subject=f'URGENT Purchase Order: {order.po_number}', # message=f'{order.get_priority_display()} purchase order requires immediate attention.', # from_email='procurement@hospital.com', # recipient_list=[manager.email], # fail_silently=True # ) # # def notify_procurement_completion(self, order): # """Notify procurement completion""" # # Notify requestor # if order.requested_by and order.requested_by.email: # send_mail( # subject=f'Purchase Order Complete: {order.po_number}', # message=f'Your purchase order has been completed and goods received.', # from_email='procurement@hospital.com', # recipient_list=[order.requested_by.email], # fail_silently=True # ) # # def update_supplier_performance(self, order): # """Update supplier performance metrics""" # supplier = order.supplier # # # Calculate on-time delivery # if order.actual_delivery_date and order.promised_delivery_date: # if order.actual_delivery_date <= order.promised_delivery_date: # # Update on-time delivery rate # pass # # # This would update comprehensive supplier metrics # pass # # def update_procurement_metrics(self, order): # """Update procurement performance metrics""" # # This would update procurement cycle time and other metrics # pass # # def generate_procurement_summary(self, order): # """Generate procurement summary report""" # # This would generate a comprehensive procurement report # pass # # # class InventoryReplenishmentProcess(Process): # """ # Viewflow process model for inventory replenishment # """ # inventory_item = ModelField(InventoryItem, help_text='Associated inventory item') # # # Process status tracking # reorder_triggered = models.BooleanField(default=False) # demand_analyzed = models.BooleanField(default=False) # quantity_calculated = models.BooleanField(default=False) # supplier_contacted = models.BooleanField(default=False) # order_placed = models.BooleanField(default=False) # delivery_scheduled = models.BooleanField(default=False) # stock_replenished = models.BooleanField(default=False) # # class Meta: # verbose_name = 'Inventory Replenishment Process' # verbose_name_plural = 'Inventory Replenishment Processes' # # # class InventoryReplenishmentFlow(Flow): # """ # Inventory Replenishment Workflow # # This flow manages automatic and manual inventory replenishment # including demand analysis and supplier coordination. # """ # # process_class = InventoryReplenishmentProcess # # # Flow definition # start = ( # flow_func(this.start_replenishment) # .Next(this.trigger_reorder) # ) # # trigger_reorder = ( # flow_func(this.check_reorder_point) # .Next(this.analyze_demand) # ) # # analyze_demand = ( # flow_view(DemandAnalysisView) # .Permission('inventory.can_analyze_demand') # .Next(this.calculate_quantity) # ) # # calculate_quantity = ( # flow_view(QuantityCalculationView) # .Permission('inventory.can_calculate_quantities') # .Next(this.contact_supplier) # ) # # contact_supplier = ( # flow_view(SupplierContactView) # .Permission('inventory.can_contact_suppliers') # .Next(this.place_order) # ) # # place_order = ( # flow_view(OrderPlacementView) # .Permission('inventory.can_place_orders') # .Next(this.schedule_delivery) # ) # # schedule_delivery = ( # flow_view(DeliverySchedulingView) # .Permission('inventory.can_schedule_deliveries') # .Next(this.replenish_stock) # ) # # replenish_stock = ( # flow_func(this.complete_replenishment) # .Next(this.end) # ) # # end = flow_func(this.end_replenishment) # # # Flow functions # def start_replenishment(self, activation): # """Initialize the replenishment process""" # process = activation.process # item = process.inventory_item # # # Send notification to inventory staff # self.notify_replenishment_needed(item) # # def check_reorder_point(self, activation): # """Check if item has reached reorder point""" # process = activation.process # item = process.inventory_item # # # Check current stock levels # current_stock = item.total_stock # # if current_stock <= item.reorder_point: # process.reorder_triggered = True # process.save() # # # Send urgent notification if below safety stock # if current_stock <= item.safety_stock: # self.notify_critical_stock(item) # # def complete_replenishment(self, activation): # """Finalize the replenishment process""" # process = activation.process # item = process.inventory_item # # # Mark process as completed # process.stock_replenished = True # process.save() # # # Send completion notifications # self.notify_replenishment_completion(item) # # # Update inventory metrics # self.update_inventory_metrics(item) # # def end_replenishment(self, activation): # """End the replenishment workflow""" # process = activation.process # # # Generate replenishment summary # self.generate_replenishment_summary(process.inventory_item) # # # Helper methods # def notify_replenishment_needed(self, item): # """Notify inventory staff of replenishment need""" # inventory_staff = User.objects.filter( # groups__name='Inventory Staff' # ) # # for staff in inventory_staff: # send_mail( # subject=f'Replenishment Needed: {item.item_name}', # message=f'Item {item.item_code} has reached reorder point.', # from_email='inventory@hospital.com', # recipient_list=[staff.email], # fail_silently=True # ) # # def notify_critical_stock(self, item): # """Notify of critical stock levels""" # inventory_managers = User.objects.filter( # groups__name='Inventory Managers' # ) # # for manager in inventory_managers: # send_mail( # subject=f'CRITICAL STOCK: {item.item_name}', # message=f'Item {item.item_code} is below safety stock level.', # from_email='inventory@hospital.com', # recipient_list=[manager.email], # fail_silently=True # ) # # def notify_replenishment_completion(self, item): # """Notify replenishment completion""" # # This would notify relevant staff # pass # # def update_inventory_metrics(self, item): # """Update inventory performance metrics""" # # This would update inventory turnover and other metrics # pass # # def generate_replenishment_summary(self, item): # """Generate replenishment summary""" # # This would generate replenishment summary # pass # # # class StockMovementProcess(Process): # """ # Viewflow process model for stock movements # """ # movement_type = CharField(max_length=20, help_text='Type of stock movement') # item_id = CharField(max_length=50, help_text='Item identifier') # # # Process status tracking # movement_initiated = models.BooleanField(default=False) # authorization_verified = models.BooleanField(default=False) # stock_reserved = models.BooleanField(default=False) # movement_executed = models.BooleanField(default=False) # documentation_completed = models.BooleanField(default=False) # movement_verified = models.BooleanField(default=False) # # class Meta: # verbose_name = 'Stock Movement Process' # verbose_name_plural = 'Stock Movement Processes' # # # class StockMovementFlow(Flow): # """ # Stock Movement Workflow # # This flow manages stock transfers, adjustments, and other # inventory movements with proper authorization and tracking. # """ # # process_class = StockMovementProcess # # # Flow definition # start = ( # flow_func(this.start_movement) # .Next(this.initiate_movement) # ) # # initiate_movement = ( # flow_view(MovementInitiationView) # .Permission('inventory.can_initiate_movements') # .Next(this.verify_authorization) # ) # # verify_authorization = ( # flow_view(AuthorizationVerificationView) # .Permission('inventory.can_verify_authorization') # .Next(this.reserve_stock) # ) # # reserve_stock = ( # flow_view(StockReservationView) # .Permission('inventory.can_reserve_stock') # .Next(this.execute_movement) # ) # # execute_movement = ( # flow_view(MovementExecutionView) # .Permission('inventory.can_execute_movements') # .Next(this.complete_documentation) # ) # # complete_documentation = ( # flow_view(DocumentationView) # .Permission('inventory.can_complete_documentation') # .Next(this.verify_movement) # ) # # verify_movement = ( # flow_view(MovementVerificationView) # .Permission('inventory.can_verify_movements') # .Next(this.finalize_movement) # ) # # finalize_movement = ( # flow_func(this.complete_movement) # .Next(this.end) # ) # # end = flow_func(this.end_movement) # # # Flow functions # def start_movement(self, activation): # """Initialize the stock movement process""" # process = activation.process # # # Send notification to inventory staff # self.notify_movement_initiated(process.movement_type, process.item_id) # # def complete_movement(self, activation): # """Finalize the stock movement process""" # process = activation.process # # # Mark process as completed # process.movement_verified = True # process.save() # # # Send completion notifications # self.notify_movement_completion(process.movement_type, process.item_id) # # # Update inventory records # self.update_inventory_records(process.movement_type, process.item_id) # # def end_movement(self, activation): # """End the stock movement workflow""" # process = activation.process # # # Generate movement summary # self.generate_movement_summary(process.movement_type, process.item_id) # # # Helper methods # def notify_movement_initiated(self, movement_type, item_id): # """Notify inventory staff of movement initiation""" # inventory_staff = User.objects.filter( # groups__name='Inventory Staff' # ) # # for staff in inventory_staff: # send_mail( # subject=f'Stock Movement Initiated: {item_id}', # message=f'{movement_type} movement initiated for item {item_id}.', # from_email='inventory@hospital.com', # recipient_list=[staff.email], # fail_silently=True # ) # # def notify_movement_completion(self, movement_type, item_id): # """Notify movement completion""" # # This would notify relevant staff # pass # # def update_inventory_records(self, movement_type, item_id): # """Update inventory records""" # # This would update inventory records # pass # # def generate_movement_summary(self, movement_type, item_id): # """Generate movement summary""" # # This would generate movement summary # pass # # # class InventoryAuditProcess(Process): # """ # Viewflow process model for inventory audits # """ # audit_type = CharField(max_length=20, help_text='Type of audit') # location_id = CharField(max_length=50, help_text='Location identifier') # # # Process status tracking # audit_scheduled = models.BooleanField(default=False) # audit_team_assigned = models.BooleanField(default=False) # physical_count_completed = models.BooleanField(default=False) # discrepancies_identified = models.BooleanField(default=False) # adjustments_made = models.BooleanField(default=False) # audit_report_generated = models.BooleanField(default=False) # audit_completed = models.BooleanField(default=False) # # class Meta: # verbose_name = 'Inventory Audit Process' # verbose_name_plural = 'Inventory Audit Processes' # # # class InventoryAuditFlow(Flow): # """ # Inventory Audit Workflow # # This flow manages inventory audits including cycle counts, # physical inventories, and discrepancy resolution. # """ # # process_class = InventoryAuditProcess # # # Flow definition # start = ( # flow_func(this.start_audit) # .Next(this.schedule_audit) # ) # # schedule_audit = ( # flow_view(AuditSchedulingView) # .Permission('inventory.can_schedule_audits') # .Next(this.assign_team) # ) # # assign_team = ( # flow_view(TeamAssignmentView) # .Permission('inventory.can_assign_audit_teams') # .Next(this.conduct_count) # ) # # conduct_count = ( # flow_view(CycleCountView) # .Permission('inventory.can_conduct_counts') # .Next(this.identify_discrepancies) # ) # # identify_discrepancies = ( # flow_func(this.analyze_discrepancies) # .Next(this.make_adjustments) # ) # # make_adjustments = ( # flow_view(InventoryAdjustmentView) # .Permission('inventory.can_make_adjustments') # .Next(this.generate_report) # ) # # generate_report = ( # flow_view(AuditReportView) # .Permission('inventory.can_generate_audit_reports') # .Next(this.complete_audit) # ) # # complete_audit = ( # flow_func(this.finalize_audit) # .Next(this.end) # ) # # end = flow_func(this.end_audit) # # # Flow functions # def start_audit(self, activation): # """Initialize the audit process""" # process = activation.process # # # Send notification to audit team # self.notify_audit_scheduled(process.audit_type, process.location_id) # # def analyze_discrepancies(self, activation): # """Analyze inventory discrepancies""" # process = activation.process # # # Check for discrepancies # discrepancies = self.check_discrepancies(process.location_id) # # if discrepancies: # process.discrepancies_identified = True # process.save() # # # Alert audit supervisor # self.alert_audit_supervisor(process.location_id, discrepancies) # # def finalize_audit(self, activation): # """Finalize the audit process""" # process = activation.process # # # Mark audit as completed # process.audit_completed = True # process.save() # # # Send completion notifications # self.notify_audit_completion(process.audit_type, process.location_id) # # # Update audit metrics # self.update_audit_metrics(process.audit_type, process.location_id) # # def end_audit(self, activation): # """End the audit workflow""" # process = activation.process # # # Generate audit summary # self.generate_audit_summary(process.audit_type, process.location_id) # # # Helper methods # def notify_audit_scheduled(self, audit_type, location_id): # """Notify audit team of scheduled audit""" # audit_staff = User.objects.filter( # groups__name='Audit Staff' # ) # # for staff in audit_staff: # send_mail( # subject=f'Audit Scheduled: {location_id}', # message=f'{audit_type} audit scheduled for location {location_id}.', # from_email='audit@hospital.com', # recipient_list=[staff.email], # fail_silently=True # ) # # def check_discrepancies(self, location_id): # """Check for inventory discrepancies""" # # This would implement discrepancy checking logic # return [] # # def alert_audit_supervisor(self, location_id, discrepancies): # """Alert audit supervisor of discrepancies""" # supervisors = User.objects.filter( # groups__name='Audit Supervisors' # ) # # for supervisor in supervisors: # send_mail( # subject=f'Audit Discrepancies Found: {location_id}', # message=f'Inventory discrepancies identified during audit.', # from_email='audit@hospital.com', # recipient_list=[supervisor.email], # fail_silently=True # ) # # def notify_audit_completion(self, audit_type, location_id): # """Notify audit completion""" # # This would notify relevant parties # pass # # def update_audit_metrics(self, audit_type, location_id): # """Update audit metrics""" # # This would update audit performance metrics # pass # # def generate_audit_summary(self, audit_type, location_id): # """Generate audit summary""" # # This would generate audit summary # pass # # # class SupplierManagementProcess(Process): # """ # Viewflow process model for supplier management # """ # supplier = ModelField(Supplier, help_text='Associated supplier') # # # Process status tracking # supplier_onboarded = models.BooleanField(default=False) # qualifications_verified = models.BooleanField(default=False) # contracts_negotiated = models.BooleanField(default=False) # performance_monitored = models.BooleanField(default=False) # relationship_maintained = models.BooleanField(default=False) # # class Meta: # verbose_name = 'Supplier Management Process' # verbose_name_plural = 'Supplier Management Processes' # # # class SupplierManagementFlow(Flow): # """ # Supplier Management Workflow # # This flow manages supplier onboarding, qualification, # performance monitoring, and relationship management. # """ # # process_class = SupplierManagementProcess # # # Flow definition # start = ( # flow_func(this.start_supplier_management) # .Next(this.onboard_supplier) # ) # # onboard_supplier = ( # flow_view(SupplierOnboardingView) # .Permission('inventory.can_onboard_suppliers') # .Next(this.verify_qualifications) # ) # # verify_qualifications = ( # flow_view(QualificationVerificationView) # .Permission('inventory.can_verify_qualifications') # .Next(this.negotiate_contracts) # ) # # negotiate_contracts = ( # flow_view(ContractNegotiationView) # .Permission('inventory.can_negotiate_contracts') # .Next(this.monitor_performance) # ) # # monitor_performance = ( # flow_view(PerformanceMonitoringView) # .Permission('inventory.can_monitor_performance') # .Next(this.maintain_relationship) # ) # # maintain_relationship = ( # flow_func(this.complete_supplier_management) # .Next(this.end) # ) # # end = flow_func(this.end_supplier_management) # # # Flow functions # def start_supplier_management(self, activation): # """Initialize the supplier management process""" # process = activation.process # supplier = process.supplier # # # Send notification to procurement team # self.notify_supplier_onboarding(supplier) # # def complete_supplier_management(self, activation): # """Finalize the supplier management process""" # process = activation.process # supplier = process.supplier # # # Mark process as completed # process.relationship_maintained = True # process.save() # # # Send completion notifications # self.notify_supplier_management_completion(supplier) # # def end_supplier_management(self, activation): # """End the supplier management workflow""" # process = activation.process # # # Generate supplier management summary # self.generate_supplier_summary(process.supplier) # # # Helper methods # def notify_supplier_onboarding(self, supplier): # """Notify procurement team of supplier onboarding""" # procurement_staff = User.objects.filter( # groups__name='Procurement Staff' # ) # # for staff in procurement_staff: # send_mail( # subject=f'Supplier Onboarding: {supplier.name}', # message=f'New supplier {supplier.name} requires onboarding.', # from_email='procurement@hospital.com', # recipient_list=[staff.email], # fail_silently=True # ) # # def notify_supplier_management_completion(self, supplier): # """Notify supplier management completion""" # # This would notify relevant parties # pass # # def generate_supplier_summary(self, supplier): # """Generate supplier management summary""" # # This would generate supplier summary # pass # # # # Celery tasks for background processing # @celery.job # def auto_reorder_items(): # """Background task to automatically reorder items""" # try: # # This would check reorder points and create orders # return True # except Exception: # return False # # # @celery.job # def monitor_expiring_items(): # """Background task to monitor expiring inventory items""" # try: # # This would identify items nearing expiration # return True # except Exception: # return False # # # @celery.job # def generate_inventory_reports(): # """Background task to generate inventory reports""" # try: # # This would generate daily inventory reports # return True # except Exception: # return False # # # @celery.job # def update_supplier_performance(): # """Background task to update supplier performance metrics""" # try: # # This would calculate supplier performance metrics # return True # except Exception: # return False # # # @celery.job # def schedule_cycle_counts(): # """Background task to schedule cycle counts""" # try: # # This would schedule regular cycle counts # return True # except Exception: # return False #