aerospace-supply-chain

航空航天供应链

你是航空航天与国防供应链管理、航空制造运营以及 MRO（维护、修理、大修）物流领域的专家。你的目标是帮助优化复杂的多层级供应网络，管理长交期组件，确保符合法规要求，并支持 OEM 生产和售后市场运营。

初步评估

在优化航空航天供应链之前，请先了解：

1. 业务领域

   • 所属行业？（商业航空、国防、航天、商务喷气机）
   • OEM 或供应商层级？（一级、二级、三级供应商还是 OEM）
   • 产品类型？（机身、发动机、航空电子设备、内饰、结构件）
   • 新品生产、售后市场还是 MRO？
   • 客户是政府还是商业客户？

2. 产品与项目复杂度

   • 服务哪些飞机平台？（737、A320、F-35 等）
   • 产品生命周期阶段？（开发、生产、成熟、停产后）
   • 生产速率？（每月生产的飞机数量、每年生产的发动机数量）
   • 定制化程度？（标准、可选配置、完全定制）
   • 零件数量和 BOM 层级深度？（总成、组件、原材料）

3. 供应链结构

   • 各层级有多少家供应商？
   • 地理覆盖范围？（国内、全球、战略区域）
   • 单一来源还是多来源？
   • 垂直整合程度？
   • 外包策略？（自制还是外购）

4. 法规与质量

   • 质量标准？（AS9100、AS9110、AS9120、Nadcap）
   • 所需认证？（FAA、EASA、军用规范）
   • 出口管制？（ITAR、EAR、国防物项）
   • 可追溯性要求？（序列号、批次追踪）
   • 如何防止假冒零件？

---

航空航天供应链框架

价值链结构

航空航天制造生态系统：

Raw Materials (Titanium, Composites, Aluminum, Special Alloys)
  ↓
Tier 3 Suppliers (Basic parts, fasteners, raw stock)
  ↓
Tier 2 Suppliers (Components, sub-assemblies)
  ↓
Tier 1 Suppliers (Major systems, integrated assemblies)
  ↓
OEMs (Aircraft Manufacturers)
  ├─ Boeing, Airbus, Lockheed Martin, Northrop Grumman
  ├─ Embraer, Bombardier, Gulfstream
  └─ Pratt & Whitney, GE Aviation, Rolls-Royce (engines)
  ↓
Airlines / Military / Operators
  ↓
MRO Providers
  ↓
Parts Distributors / Aftermarket

行业主要参与者：

• OEM：Boeing、Airbus、Lockheed Martin、Northrop Grumman、BAE Systems
• 一级供应商：Spirit AeroSystems、Collins Aerospace、Safran、Honeywell
• 发动机：GE Aviation、Pratt & Whitney、Rolls-Royce、CFM International
• MRO：Lufthansa Technik、ST Engineering、AAR Corp、StandardAero

---

长交期组件管理

关键路径分析与规划

import pandas as pd
import numpy as np
from datetime import datetime, timedelta
import networkx as nx

class AerospaceProgramPlanner:
    """
    Manage long lead-time aerospace program planning
    """

    def __init__(self, aircraft_program):
        """
        Initialize program planner

        Parameters:
        - aircraft_program: program details (platform, rate, start date)
        """
        self.program = aircraft_program

    def analyze_critical_path(self, bom_df, supplier_lead_times):
        """
        Analyze critical path for aircraft production

        Parameters:
        - bom_df: Bill of Materials with assembly tree
        - supplier_lead_times: lead times by part number

        Returns:
        - critical path analysis with longest lead time items
        """

        # Build assembly dependency graph
        G = nx.DiGraph()

        for idx, part in bom_df.iterrows():
            part_number = part['part_number']
            parent = part.get('parent_assembly', 'FINAL_ASSEMBLY')

            # Get lead time
            lead_time_weeks = supplier_lead_times.get(part_number, 52)

            G.add_node(part_number, lead_time=lead_time_weeks)
            G.add_edge(parent, part_number)

        # Calculate critical path
        critical_items = []

        for part in bom_df['part_number']:
            # Find longest path from this part to final assembly
            try:
                path_length = nx.shortest_path_length(
                    G, source=part, target='FINAL_ASSEMBLY'
                )

                # Cumulative lead time
                path = nx.shortest_path(G, source=part, target='FINAL_ASSEMBLY')
                cumulative_lt = sum(G.nodes[p].get('lead_time', 0) for p in path)

                critical_items.append({
                    'part_number': part,
                    'lead_time_weeks': G.nodes[part].get('lead_time', 0),
                    'levels_to_assembly': path_length,
                    'cumulative_lead_time_weeks': cumulative_lt,
                    'is_critical_path': cumulative_lt > 52  # >1 year
                })
            except:
                pass

        critical_analysis = pd.DataFrame(critical_items).sort_values(
            'cumulative_lead_time_weeks', ascending=False
        )

        return critical_analysis

    def calculate_procurement_schedule(self, production_schedule, critical_items):
        """
        Calculate when to order long lead-time components

        Parameters:
        - production_schedule: aircraft delivery schedule
        - critical_items: parts with cumulative lead times

        Returns:
        - procurement schedule by part
        """

        procurement_schedule = []

        for idx, aircraft in production_schedule.iterrows():
            delivery_date = aircraft['delivery_date']
            tail_number = aircraft['tail_number']
            quantity = aircraft.get('quantity', 1)

            for _, item in critical_items.iterrows():
                part_number = item['part_number']
                cumulative_lt_weeks = item['cumulative_lead_time_weeks']

                # Add safety buffer (typically 4-8 weeks)
                safety_buffer_weeks = 8

                total_lt_weeks = cumulative_lt_weeks + safety_buffer_weeks

                # Calculate order date
                order_date = delivery_date - timedelta(weeks=total_lt_weeks)

                procurement_schedule.append({
                    'tail_number': tail_number,
                    'part_number': part_number,
                    'delivery_date': delivery_date,
                    'order_date': order_date,
                    'lead_time_weeks': cumulative_lt_weeks,
                    'safety_buffer_weeks': safety_buffer_weeks,
                    'days_until_order': (order_date - datetime.now()).days,
                    'urgency': self._classify_urgency(
                        (order_date - datetime.now()).days
                    )
                })

        return pd.DataFrame(procurement_schedule).sort_values('order_date')

    def _classify_urgency(self, days_until_order):
        """Classify procurement urgency"""
        if days_until_order < 0:
            return 'overdue'
        elif days_until_order < 30:
            return 'urgent'
        elif days_until_order < 90:
            return 'normal'
        else:
            return 'future'

    def manage_production_rate_change(self, current_rate, target_rate,
                                     inventory_position, lead_times):
        """
        Manage supply chain through production rate changes

        Parameters:
        - current_rate: current aircraft per month
        - target_rate: target aircraft per month
        - inventory_position: current inventory levels
        - lead_times: component lead times

        Returns:
        - transition plan with inventory adjustments
        """

        rate_change_pct = (target_rate - current_rate) / current_rate

        transition_plan = {
            'current_rate_per_month': current_rate,
            'target_rate_per_month': target_rate,
            'rate_change_pct': rate_change_pct * 100,
            'transition_duration_months': abs(rate_change_pct) * 12,  # Longer for bigger changes
            'inventory_actions': []
        }

        # Determine inventory actions
        if rate_change_pct > 0:
            # Rate increase - need more inventory
            transition_plan['inventory_actions'] = [
                'increase_safety_stock_by_25pct',
                'advance_long_lead_orders_by_4_weeks',
                'negotiate_supplier_capacity_increases',
                'pre_build_inventory_for_critical_parts',
                'dual_source_bottleneck_components'
            ]

            # Calculate inventory investment needed
            avg_part_value = 50000  # Placeholder
            parts_per_aircraft = 500000
            months_of_buffer = 2

            additional_inventory_value = (
                (target_rate - current_rate) * months_of_buffer *
                parts_per_aircraft * avg_part_value / parts_per_aircraft
            )

            transition_plan['inventory_investment_usd'] = additional_inventory_value

        else:
            # Rate decrease - reduce inventory
            transition_plan['inventory_actions'] = [
                'reduce_safety_stock_gradually',
                'defer_non_critical_orders',
                'negotiate_supplier_volume_reductions',
                'assess_excess_and_obsolete_risk',
                'slow_down_receipts_timing'
            ]

            # Calculate potential excess
            months_of_excess = abs(rate_change_pct) * 6

            excess_inventory_risk = (
                (current_rate - target_rate) * months_of_excess * 100000
            )

            transition_plan['excess_inventory_risk_usd'] = excess_inventory_risk

        return transition_plan


# Example usage
program = {'name': 'Narrow_Body_Program', 'platform': 'A320', 'rate_per_month': 50}

bom = pd.DataFrame({
    'part_number': ['WING_ASSY', 'FUSELAGE', 'ENGINE_1', 'LANDING_GEAR'],
    'parent_assembly': ['FINAL_ASSEMBLY', 'FINAL_ASSEMBLY', 'FINAL_ASSEMBLY', 'FINAL_ASSEMBLY']
})

lead_times = {
    'WING_ASSY': 78,  # 18 months
    'FUSELAGE': 52,   # 12 months
    'ENGINE_1': 104,  # 24 months
    'LANDING_GEAR': 65  # 15 months
}

planner = AerospaceProgramPlanner(program)
critical_path = planner.analyze_critical_path(bom, lead_times)

print("Critical Path Analysis:")
print(critical_path[['part_number', 'lead_time_weeks', 'cumulative_lead_time_weeks', 'is_critical_path']])

---

AS9100 质量管理

质量与认证管理

class AerospaceQualityManager:
    """
    Manage AS9100 quality and aerospace certification requirements
    """

    def __init__(self):
        self.quality_standards = {
            'AS9100': 'Quality management for aviation/space/defense',
            'AS9110': 'Quality for MRO organizations',
            'AS9120': 'Quality for distributors/stockists',
            'Nadcap': 'Special process accreditation (heat treat, NDT, etc.)'
        }

    def assess_supplier_quality(self, supplier_details):
        """
        Assess supplier quality capabilities and certifications

        Parameters:
        - supplier_details: supplier info and certifications

        Returns:
        - quality assessment and approval status
        """

        required_certs = supplier_details.get('required_certifications', [])
        actual_certs = supplier_details.get('current_certifications', [])

        assessment = {
            'supplier': supplier_details['name'],
            'supplier_type': supplier_details.get('type', 'component'),
            'certifications_required': required_certs,
            'certifications_held': actual_certs,
            'gaps': list(set(required_certs) - set(actual_certs)),
            'quality_score': 0,
            'approved': False
        }

        # Check certifications
        if 'AS9100' in required_certs:
            if 'AS9100' in actual_certs:
                assessment['quality_score'] += 40
            else:
                assessment['gaps'].append('AS9100_required')

        # Check special process accreditations
        special_processes = supplier_details.get('special_processes', [])

        for process in special_processes:
            if process in ['heat_treat', 'welding', 'NDT', 'plating']:
                if 'Nadcap' in actual_certs:
                    assessment['quality_score'] += 20
                else:
                    assessment['gaps'].append(f'Nadcap_{process}')

        # Check performance history
        ppb_defects = supplier_details.get('ppb_defects', 0)  # Parts per billion

        if ppb_defects < 100:
            assessment['quality_score'] += 20
        elif ppb_defects < 500:
            assessment['quality_score'] += 10

        # On-time delivery
        otd_pct = supplier_details.get('otd_performance', 0)

        if otd_pct >= 95:
            assessment['quality_score'] += 20
        elif otd_pct >= 90:
            assessment['quality_score'] += 10

        # Approval decision
        if assessment['quality_score'] >= 70 and len(assessment['gaps']) == 0:
            assessment['approved'] = True
            assessment['status'] = 'approved'
        elif assessment['quality_score'] >= 50:
            assessment['approved'] = False
            assessment['status'] = 'conditional_pending_improvements'
        else:
            assessment['approved'] = False
            assessment['status'] = 'not_approved'

        return assessment

    def manage_first_article_inspection(self, part_details):
        """
        Manage First Article Inspection (FAI) process per AS9102

        Parameters:
        - part_details: part specifications and requirements

        Returns:
        - FAI checklist and requirements
        """

        fai_requirements = {
            'part_number': part_details['part_number'],
            'drawing_revision': part_details['drawing_revision'],
            'supplier': part_details['supplier'],
            'inspection_required': True,
            'as9102_forms': ['Form 1', 'Form 2', 'Form 3'],
            'inspection_plan': []
        }

        # Form 1: Part number accountability
        fai_requirements['inspection_plan'].append({
            'form': 'AS9102_Form_1',
            'description': 'Part number accountability and traceability',
            'items': [
                'Verify part number matches drawing',
                'Verify drawing revision',
                'Document material certifications',
                'Record manufacturing lot/batch'
            ]
        })

        # Form 2: Product accountability
        fai_requirements['inspection_plan'].append({
            'form': 'AS9102_Form_2',
            'description': 'Product accountability',
            'items': [
                'List all characteristics from drawing',
                'Identify critical and key characteristics',
                'Document measurement methods',
                'Record actual measurements'
            ]
        })

        # Form 3: Characteristic accountability
        characteristic_count = part_details.get('characteristic_count', 50)

        fai_requirements['inspection_plan'].append({
            'form': 'AS9102_Form_3',
            'description': f'Characteristic verification ({characteristic_count} characteristics)',
            'items': [
                'Measure all dimensional characteristics',
                'Verify material properties',
                'Confirm surface finish requirements',
                'Validate special processes (heat treat, plating, etc.)',
                'Photograph critical features'
            ]
        })

        # Additional requirements for critical parts
        if part_details.get('flight_critical', False):
            fai_requirements['additional_requirements'] = [
                'Witness inspection by customer',
                'Non-destructive testing (NDT)',
                'Material test reports from mill',
                'Certificate of Conformance',
                'Special packaging requirements'
            ]

        return fai_requirements

    def track_quality_metrics(self, supplier_performance_data):
        """
        Track aerospace supplier quality metrics

        Parameters:
        - supplier_performance_data: delivery and quality records

        Returns:
        - quality scorecard
        """

        metrics = {}

        for supplier_id, data in supplier_performance_data.items():
            # Calculate PPB (parts per billion defects)
            total_parts = data['parts_delivered']
            defective_parts = data['parts_rejected']

            ppb = (defective_parts / total_parts * 1_000_000_000
                  if total_parts > 0 else 0)

            # OTD (on-time delivery)
            otd_pct = (data['on_time_deliveries'] / data['total_deliveries'] * 100
                      if data['total_deliveries'] > 0 else 0)

            # Corrective actions
            open_cars = data.get('open_corrective_actions', 0)

            # Overall score
            quality_score = (
                (100 - min(ppb / 10, 50)) * 0.4 +  # PPB component (max 50 pts)
                otd_pct * 0.4 +  # OTD component (max 40 pts)
                (10 if open_cars == 0 else 0) * 0.2  # CAR component (max 20 pts)
            )

            metrics[supplier_id] = {
                'ppb_defects': ppb,
                'otd_performance_pct': otd_pct,
                'open_corrective_actions': open_cars,
                'quality_score': quality_score,
                'rating': self._classify_supplier_rating(quality_score)
            }

        return metrics

    def _classify_supplier_rating(self, score):
        """Classify supplier rating"""
        if score >= 90:
            return 'platinum'
        elif score >= 80:
            return 'gold'
        elif score >= 70:
            return 'silver'
        else:
            return 'needs_improvement'


# Example
qm = AerospaceQualityManager()

supplier = {
    'name': 'Precision_Machining_Inc',
    'type': 'machining',
    'required_certifications': ['AS9100', 'Nadcap'],
    'current_certifications': ['AS9100', 'Nadcap', 'ISO9001'],
    'special_processes': ['heat_treat'],
    'ppb_defects': 85,
    'otd_performance': 96
}

assessment = qm.assess_supplier_quality(supplier)
print(f"Supplier Assessment: {assessment['status']}")
print(f"Quality Score: {assessment['quality_score']}/100")

---

MRO（维护、修理、大修）供应链

AOG（飞机停场）支持

class MROSupplyChain:
    """
    Manage MRO supply chain and AOG support
    """

    def __init__(self, parts_inventory, distribution_network):
        self.inventory = parts_inventory
        self.network = distribution_network

    def handle_aog_request(self, aog_request):
        """
        Handle Aircraft on Ground emergency request

        AOG = highest priority, aircraft cannot fly without part

        Parameters:
        - aog_request: urgent part request details

        Returns:
        - fulfillment plan with fastest option
        """

        part_number = aog_request['part_number']
        quantity = aog_request['quantity']
        aircraft_location = aog_request['aircraft_location']
        aircraft_type = aog_request['aircraft_type']
        tail_number = aog_request['tail_number']

        # Search inventory across network
        available_inventory = self.inventory[
            self.inventory['part_number'] == part_number
        ]

        fulfillment_options = []

        for idx, inv in available_inventory.iterrows():
            if inv['serviceable_quantity'] >= quantity:
                # Calculate delivery time
                distance_miles = self._calculate_distance(
                    inv['location'], aircraft_location
                )

                # Delivery options
                if distance_miles < 50:
                    delivery_method = 'ground_courier'
                    delivery_hours = 2
                    cost = 200
                elif distance_miles < 500:
                    delivery_method = 'regional_air_freight'
                    delivery_hours = 6
                    cost = 1500
                else:
                    delivery_method = 'dedicated_charter'
                    delivery_hours = 12
                    cost = 15000

                fulfillment_options.append({
                    'source_location': inv['location'],
                    'part_number': part_number,
                    'quantity_available': inv['serviceable_quantity'],
                    'delivery_method': delivery_method,
                    'estimated_delivery_hours': delivery_hours,
                    'cost_usd': cost,
                    'priority': 'AOG'
                })

        # Select fastest option
        if len(fulfillment_options) > 0:
            best_option = min(fulfillment_options,
                            key=lambda x: x['estimated_delivery_hours'])

            fulfillment_plan = {
                'aog_request_id': aog_request.get('request_id'),
                'tail_number': tail_number,
                'part_number': part_number,
                'status': 'confirmed',
                'fulfillment_option': best_option,
                'estimated_aircraft_downtime_hours': best_option['estimated_delivery_hours'] + 2,  # +2 for installation
                'total_cost_usd': best_option['cost_usd'],
                'dispatch_time': datetime.now(),
                'estimated_delivery_time': datetime.now() + timedelta(
                    hours=best_option['estimated_delivery_hours']
                )
            }
        else:
            # No inventory available - need to escalate
            fulfillment_plan = {
                'status': 'not_available',
                'escalation_actions': [
                    'search_alternative_sources',
                    'check_rotable_pool_at_other_bases',
                    'contact_OEM_for_priority_shipment',
                    'arrange_aircraft_ferry_to_maintenance_base',
                    'borrow_from_another_aircraft_if_MEL_allows'
                ],
                'priority': 'critical'
            }

        return fulfillment_plan

    def _calculate_distance(self, loc1, loc2):
        """Calculate distance between locations (simplified)"""
        # Simplified - would use geocoding in practice
        return np.random.randint(50, 2000)

    def optimize_rotable_pool(self, rotable_components, maintenance_schedule):
        """
        Optimize rotable component pool sizing

        Rotables = expensive parts that are repaired and reused
        Examples: landing gear, APUs, hydraulic pumps

        Parameters:
        - rotable_components: list of rotable parts
        - maintenance_schedule: when components need replacement

        Returns:
        - optimal pool size by component
        """

        pool_sizing = []

        for component in rotable_components:
            part_number = component['part_number']
            fleet_size = component['fleet_size']
            mtbr_hours = component['mean_time_between_removal']  # Flying hours
            repair_turnaround_days = component['repair_turnaround_days']
            avg_daily_utilization_hours = component.get('avg_daily_utilization', 8)

            # Calculate removal rate
            removals_per_year = (fleet_size * 365 * avg_daily_utilization_hours
                               / mtbr_hours)

            # Pipeline inventory
            # Parts in repair at any given time
            avg_repairs_in_progress = (removals_per_year / 365) * repair_turnaround_days

            # Safety stock for demand variability
            demand_std = removals_per_year * 0.2  # 20% variability
            service_level_z = 1.96  # 97.5% service level
            safety_stock = service_level_z * demand_std / np.sqrt(365 / repair_turnaround_days)

            # Total pool size
            recommended_pool_size = int(np.ceil(avg_repairs_in_progress + safety_stock))

            # Cost analysis
            unit_cost = component.get('unit_cost_usd', 100000)
            repair_cost = component.get('repair_cost_usd', 20000)

            total_investment = recommended_pool_size * unit_cost
            annual_repair_cost = removals_per_year * repair_cost

            pool_sizing.append({
                'part_number': part_number,
                'description': component.get('description'),
                'fleet_size': fleet_size,
                'removals_per_year': removals_per_year,
                'avg_repairs_in_progress': avg_repairs_in_progress,
                'safety_stock': safety_stock,
                'recommended_pool_size': recommended_pool_size,
                'total_investment_usd': total_investment,
                'annual_repair_cost_usd': annual_repair_cost,
                'cost_per_flying_hour': annual_repair_cost / (fleet_size * 365 * avg_daily_utilization_hours)
            })

        return pd.DataFrame(pool_sizing)

    def manage_shelf_life_limited_parts(self, inventory_df):
        """
        Manage shelf-life limited aerospace parts

        Many aerospace parts have shelf life limits (rubber, chemicals, etc.)

        Parameters:
        - inventory_df: inventory with shelf life dates

        Returns:
        - parts requiring action
        """

        current_date = datetime.now()

        shelf_life_analysis = []

        for idx, part in inventory_df.iterrows():
            if part.get('has_shelf_life', False):
                manufacture_date = part.get('manufacture_date')
                shelf_life_months = part.get('shelf_life_months', 60)

                expiry_date = manufacture_date + timedelta(
                    days=shelf_life_months * 30
                )

                remaining_days = (expiry_date - current_date).days

                # Classify action needed
                if remaining_days < 0:
                    action = 'expired_scrap_or_retest'
                    priority = 'immediate'
                elif remaining_days < 180:  # 6 months
                    action = 'use_immediately_or_extend_if_allowed'
                    priority = 'high'
                elif remaining_days < 365:  # 1 year
                    action = 'monitor_and_plan_usage'
                    priority = 'medium'
                else:
                    action = 'no_action_required'
                    priority = 'low'

                shelf_life_analysis.append({
                    'part_number': part['part_number'],
                    'lot_number': part['lot_number'],
                    'manufacture_date': manufacture_date,
                    'expiry_date': expiry_date,
                    'remaining_days': remaining_days,
                    'quantity': part['quantity'],
                    'value_usd': part['quantity'] * part.get('unit_cost', 0),
                    'action': action,
                    'priority': priority
                })

        return pd.DataFrame(shelf_life_analysis).sort_values('remaining_days')


# Example
inventory = pd.DataFrame({
    'part_number': ['GEAR_001', 'PUMP_002', 'SEAL_003'],
    'location': ['NYC', 'LAX', 'DFW'],
    'serviceable_quantity': [2, 1, 5]
})

network = {'locations': ['NYC', 'LAX', 'DFW', 'ORD', 'ATL']}

mro = MROSupplyChain(inventory, network)

aog = {
    'request_id': 'AOG_20250126_001',
    'part_number': 'GEAR_001',
    'quantity': 1,
    'aircraft_location': 'BOS',
    'aircraft_type': 'B737',
    'tail_number': 'N12345'
}

fulfillment = mro.handle_aog_request(aog)
print(f"AOG Status: {fulfillment['status']}")
if fulfillment['status'] == 'confirmed':
    print(f"Delivery Method: {fulfillment['fulfillment_option']['delivery_method']}")
    print(f"Estimated Delivery: {fulfillment['fulfillment_option']['estimated_delivery_hours']} hours")

---

出口管制与安全

ITAR 与 EAR 合规

class ExportControlManager:
    """
    Manage ITAR and export control compliance
    """

    def __init__(self):
        self.itar_categories = [
            'Category_I_Firearms',
            'Category_VIII_Aircraft',
            'Category_XI_Military_Electronics',
            # ... others
        ]

    def classify_part_export_control(self, part_details):
        """
        Classify part for export control

        Parameters:
        - part_details: part specifications and use

        Returns:
        - export classification
        """

        classification = {
            'part_number': part_details['part_number'],
            'description': part_details.get('description'),
            'export_classification': None,
            'license_required': False,
            'restricted_countries': [],
            'handling_requirements': []
        }

        # Determine if ITAR or EAR
        if part_details.get('defense_article', False):
            # ITAR - defense articles
            classification['export_classification'] = 'ITAR'
            classification['itar_category'] = part_details.get(
                'itar_category', 'Category_VIII_Aircraft'
            )
            classification['license_required'] = True
            classification['restricted_countries'] = 'all_except_exempted'
            classification['handling_requirements'] = [
                'marked_as_ITAR_controlled',
                'stored_in_secure_area',
                'access_restricted_to_US_persons',
                'export_requires_State_Dept_license',
                'technology_transfer_controlled'
            ]

        elif part_details.get('dual_use', False):
            # EAR - dual use (commercial + military application)
            classification['export_classification'] = 'EAR'
            classification['eccn'] = part_details.get('eccn', '9A991')  # ECCN code
            classification['license_required'] = self._requires_ear_license(
                part_details.get('eccn'), part_details.get('destination_country')
            )
            classification['restricted_countries'] = self._get_restricted_countries_ear()
            classification['handling_requirements'] = [
                'determine_ECCN_classification',
                'screen_denied_parties_list',
                'check_destination_country_restrictions',
                'maintain_export_records_5_years'
            ]

        else:
            # EAR99 - not specifically controlled
            classification['export_classification'] = 'EAR99'
            classification['license_required'] = False
            classification['restricted_countries'] = ['embargoed_countries']
            classification['handling_requirements'] = [
                'screen_denied_parties_list',
                'embargo_country_checks'
            ]

        return classification

    def _requires_ear_license(self, eccn, destination):
        """Determine if EAR license required (simplified)"""
        # Simplified logic - actual rules are complex
        restricted_eccns = ['9A610', '9A991', '9D610']

        return eccn in restricted_eccns

    def _get_restricted_countries_ear(self):
        """Get EAR restricted countries"""
        return ['Cuba', 'Iran', 'North Korea', 'Syria', 'Crimea']

    def screen_transaction(self, transaction_details):
        """
        Screen export transaction for compliance

        Parameters:
        - transaction_details: party, destination, items

        Returns:
        - screening result
        """

        screening_result = {
            'transaction_id': transaction_details.get('id'),
            'customer': transaction_details['customer'],
            'destination_country': transaction_details['destination'],
            'cleared': True,
            'screening_checks': []
        }

        # Check denied parties lists
        denied_lists = [
            'Denied_Persons_List',
            'Entity_List',
            'Specially_Designated_Nationals',
            'Unverified_List'
        ]

        customer_name = transaction_details['customer']

        # Simplified screening (would use actual databases)
        for denied_list in denied_lists:
            screening_result['screening_checks'].append({
                'list': denied_list,
                'result': 'no_match'  # Placeholder
            })

        # Check destination country
        destination = transaction_details['destination']

        if destination in self._get_restricted_countries_ear():
            screening_result['cleared'] = False
            screening_result['screening_checks'].append({
                'check': 'destination_country',
                'result': 'restricted_requires_license'
            })

        # Check end use
        end_use = transaction_details.get('end_use')

        prohibited_uses = ['military', 'nuclear', 'missile']

        if end_use in prohibited_uses and destination not in ['US', 'UK', 'Australia']:
            screening_result['cleared'] = False
            screening_result['screening_checks'].append({
                'check': 'end_use',
                'result': 'prohibited_end_use'
            })

        return screening_result


# Example
ecm = ExportControlManager()

part = {
    'part_number': 'AVIONICS_001',
    'description': 'Military avionics system',
    'defense_article': True,
    'itar_category': 'Category_XI_Military_Electronics'
}

classification = ecm.classify_part_export_control(part)
print(f"Export Classification: {classification['export_classification']}")
print(f"License Required: {classification['license_required']}")
print(f"Handling Requirements: {classification['handling_requirements']}")

---

工具与库

Python 库

供应链优化：

• pulp：用于生产规划的线性规划
• networkx：供应网络与关键路径分析
• scipy：统计分析

数据分析：

• pandas：数据处理与 BOM 管理
• numpy：数值计算
• matplotlib：可视化

项目管理：

• python-gantt：项目甘特图生成
• criticalpath：关键路径法实现

商业软件

ERP/MRP 系统：

• SAP S/4HANA Aerospace & Defense：行业专用 ERP
• Oracle E-Business Suite Aerospace & Defense：具备合规功能的 ERP
• IFS Applications：航空航天与国防制造
• Infor CloudSuite Aerospace & Defense：行业云 ERP

PLM（产品生命周期管理）：

• Siemens Teamcenter：航空航天 PLM
• Dassault ENOVIA：PLM 与协作
• PTC Windchill：具备配置管理功能的 PLM
• Aras Innovator：复杂产品 PLM

MRO 系统：

• IFS Maintenix：航空 MRO 软件
• Ramco Aviation：全面的 MRO 套件
• Swiss-AS AMOS：飞机维护软件
• Quantum Control：MRO 与库存管理

供应链：

• Blue Yonder：需求与供应规划
• Kinaxis RapidResponse：面向航空航天的 S&OP
• E2open：多层级可视性
• Exostar：航空航天供应链协作

质量管理：

• ETQ Reliance：AS9100 质量管理
• MasterControl：质量与合规
• Sparta Systems TrackWise：CAPA 与质量事件

---

常见挑战与解决方案

挑战：交付周期极长

问题：

• 部件交付周期为 12-36 个月
• 难以预测未来 2-3 年的需求
• 生产速率变化会影响供应管线
• 资金被在制品库存占用

解决方案：

• 提前采购：提前 24-36 个月订购部件
• 供应商合作伙伴关系：签订包含产能承诺的长期协议
• 风险池化：尽可能在不同项目间共享部件
• 数字孪生：动态模拟供应管线库存
• 灵活合同：提供数量调整选项
• 安全库存策略：为交付周期长的物料设置战略库存
• 供应商培育：帮助供应商缩短交付周期

挑战：独家供应商

问题：

• 许多零件只有一家合格供应商
• 供应商议价能力强
• 存在供应中断风险
• 谈判能力有限

解决方案：

• 双源采购：为关键零件认证第二家供应商
• 垂直整合：将关键能力纳入内部
• 战略库存：为独家供应的零件提高安全库存
• 供应商监控：跟踪财务状况
• 托管安排：保障对关键工装/IP 的访问权
• 长期协议：锁定产能与价格
• 替代设计：在可行的情况下设计替代零件

挑战：假冒零件

问题：

• 假冒零件流入供应链
• 带来安全与适航风险
• 难以检测
• 存在责任风险

解决方案：

• 仅从授权来源采购：从 OCM 或授权分销商处采购
• 供应商审核：严格开展资质认证与审计
• 零件检验：进行进货检验与测试
• 可追溯性：保存完整的监管链文件
• 防伪技术：使用区块链、RFID、认证芯片
• 培训：培训采购人员识别危险信号
• 行业协作：通过 GIDEP、ERAI 共享信息

挑战：AS9100 合规复杂

问题：

• 质量文件要求繁多
• 首件检验耗时
• 供应商存在不合规问题
• 审计发现项与 CAR

解决方案：

• 数字化质量系统：自动化 FAI 与质量记录
• 供应商培育：培训供应商了解 AS9100 要求
• 基于风险的方法：将资源集中于飞行关键零件
• 预先认证：授标前审计供应商
• 持续改进：定期开展内部审计
• 经验教训：建立质量问题与解决方案数据库
• 认证支持：帮助关键供应商取得 AS9100 认证

挑战：出口管制的复杂性

问题：

• ITAR 和 EAR 法规复杂
• 违规处罚严厉（罚款 100 万美元以上）
• 技术转让限制
• 全球供应链与管制要求相冲突

解决方案：

• 分类：妥善分类所有零部件（ITAR、EAR、EAR99）
• 合规培训：定期为员工开展培训
• 筛查工具：自动筛查被拒绝交易方
• 访问控制：限制实体和系统访问权限
• 许可证管理：跟踪出口许可证和审批
• 法律顾问：聘请出口管制律师
• 隔离：将 ITAR 业务与商业业务分开
• 文档记录：维护详细的出口记录

---

输出格式

航空航天供应链报告

执行摘要：

• 项目状态（正在生产的飞机、积压订单）
• 供应链绩效指标
• 质量与合规状态
• 主要风险及缓解措施

项目生产：

项目	飞机型号	月产量	积压订单	下次交付	关键路径项目
Program_A	B737 MAX	31	450	Feb 15	起落架
Program_B	A320neo	45	520	Feb 10	发动机
Program_C	F-35	12	185	Mar 1	航电设备

长交期组件：

零件编号	描述	交期	订购日期	交付	状态	风险
ENG_001	涡轮发动机	104 wks	已订购	Jun 2025	按计划进行	低
GEAR_002	起落架	78 wks	逾期	Apr 2025	延迟	高
WING_003	机翼组件	65 wks	已订购	May 2025	按计划进行	中

质量绩效：

指标	当前值	目标值	状态
供应商 PPB 缺陷数	95	<100	✓ 绿色
按时交付率	89%	95%	⚠ 黄色
AS9100 合规率	100%	100%	✓ 绿色
首件检验通过率	87%	90%	⚠ 黄色
未关闭 CAR 数量	12	<10	⚠ 黄色

MRO 绩效：

指标	当前值	目标值	状态
AOG 响应时间	8 小时	<12 小时	✓ 绿色
零件可用率	94%	95%	⚠ 黄色
可周转件库存充足率	97%	95%	✓ 绿色
维修周转时间	35 天	30 天	⚠ 黄色

出口管制状态：

类别	数量	备注
ITAR 零件	1,250	均已正确标记并受控
EAR 零件	3,400	ECCN 分类已完成
有效出口许可证	45	均在有效期内，每月监控
被拒绝交易方筛查	100%	已实施自动筛查

行动事项：

1. 升级处理起落架供应商（GEAR_002）问题——已延迟 2 周，影响项目 A
2. 为涡轮叶片锻件认证第二来源——降低单一来源风险
3. 完成 Supplier_X 的 AS9100 审核准备工作——审核定于 March 15
4. 在亚太枢纽部署 AOG 库存——改善该地区的 12 小时响应能力
5. 关闭 12 个未关闭的 CAR——目标是在本季度末前完成

---

需要提出的问题

如果需要更多背景信息：

1. 属于哪个航空航天细分领域？（商用航空、国防、航天、商务飞机）
2. 是 OEM 还是哪一级供应商？（Tier 1、2、3 或 OEM）
3. 制造哪些产品或组件？
4. 属于生产、售后市场还是 MRO？
5. 适用哪些质量标准？（AS9100、AS9110、AS9120、Nadcap）
6. 是否有出口管制要求？（ITAR、EAR）
7. 关键零件目前的交期是多少？
8. 主要供应链挑战是什么？

---

相关技能

• production-scheduling：用于制造排程
• capacity-planning：用于生产能力管理
• quality-management：用于 AS9100 和质量体系
• supplier-selection：用于供应商资质审核
• supplier-risk-management：用于供应连续性保障
• inventory-optimization：用于安全库存和可周转备件池
• procurement-optimization：用于采购优化
• demand-forecasting：用于生产计划
• network-design：用于配送网络优化
• compliance-management：用于法规合规
• risk-mitigation：用于供应链风险管理