AI中医12经脉诊断系统
开发指南

import requests
import base64

# API配置
API_URL = "https://api.tcm-ai.org/api/v1/diagnosis/tongue"
API_KEY = "your_api_key_here"

# 读取舌相图片
with open("tongue_image.jpg", "rb") as image_file:
    encoded_image = base64.b64encode(image_file.read()).decode('utf-8')

# 构建请求
headers = {
    "Authorization": f"Bearer {API_KEY}",
    "Content-Type": "application/json"
}

payload = {
    "image_base64": encoded_image,
    "image_type": "jpg",
    "analysis_type": "full",  # full | quick | detail
    "include_meridian": True
}

# 发送请求
response = requests.post(API_URL, json=payload, headers=headers)

if response.status_code == 200:
    result = response.json()
    print(f"舌质颜色: {result['tongue_color']}")
    print(f"舌苔状况: {result['coating']}")
    print(f"经脉关联: {result['meridian_correlation']}")
else:
    print(f"错误: {response.status_code}")
    print(response.text)

{
    "success": true,
    "analysis_id": "tongue_analysis_001",
    "tongue_color": "淡紫色",
    "coating": "薄白苔",
    "tongue_shape": "略有齿痕",
    "meridian_correlation": [
        {
            "meridian": "足太阴脾经",
            "correlation_score": 0.85,
            "status": "虚弱",
            "description": "舌边齿痕明显，脾经气血不足"
        },
        {
            "meridian": "手少阴心经", 
            "correlation_score": 0.72,
            "status": "阻滞",
            "description": "舌质偏紫，心经血瘀"
        }
    ],
    "timestamp": "2026-01-21T10:30:00Z"
}

// 12经脉诊断API调用
const diagnoseMeridian = async (healthData) => {
    const API_URL = 'https://api.tcm-ai.org/api/v1/diagnosis/meridian';
    const API_KEY = process.env.TCM_API_KEY;
    
    try {
        const response = await fetch(API_URL, {
            method: 'POST',
            headers: {
                'Authorization': `Bearer ${API_KEY}`,
                'Content-Type': 'application/json'
            },
            body: JSON.stringify({
                patient_data: {
                    vital_signs: {
                        heart_rate: healthData.heartRate,
                        blood_pressure: {
                            systolic: healthData.bpSystolic,
                            diastolic: healthData.bpDiastolic
                        },
                        blood_oxygen: healthData.spo2,
                        temperature: healthData.temperature
                    },
                    tongue_analysis_id: healthData.tongueAnalysisId,
                    symptoms: healthData.symptoms,
                    age: healthData.age,
                    gender: healthData.gender
                },
                analysis_depth: 'comprehensive',
                include_treatment: true,
                include_prognosis: true
            })
        });
        
        if (!response.ok) {
            throw new Error(`API错误: ${response.status}`);
        }
        
        const result = await response.json();
        return result;
        
    } catch (error) {
        console.error('经脉诊断失败:', error);
        throw error;
    }
};

// 使用示例
const healthData = {
    heartRate: 72,
    bpSystolic: 118,
    bpDiastolic: 78,
    spo2: 98,
    temperature: 36.5,
    tongueAnalysisId: 'tongue_analysis_001',
    symptoms: '倦怠乏力，心悸气短',
    age: 45,
    gender: 'male'
};

diagnoseMeridian(healthData).then(result => {
    console.log('经脉诊断结果:', result);
});

class MeridianDiagnosis:
    """12经脉诊断核心算法类"""
    
    def __init__(self):
        # 经脉-体征关联权重矩阵
        self.meridian_weights = {
            '手太阴肺经': {
                'heart_rate': 0.15,
                'blood_oxygen': 0.35,
                'temperature': 0.25,
                'tongue_color': 0.25
            },
            '手少阴心经': {
                'heart_rate': 0.40,
                'blood_pressure': 0.30,
                'tongue_color': 0.30
            },
            '足太阴脾经': {
                'temperature': 0.30,
                'tongue_shape': 0.40,
                'symptoms': 0.30
            },
            '足厥阴肝经': {
                'blood_pressure': 0.40,
                'heart_rate_variability': 0.35,
                'tongue_color': 0.25
            }
            # ... 其他经脉权重定义
        }
    
    def calculate_meridian_status(self, health_data, tongue_analysis):
        """计算经脉状态"""
        
        meridian_status = {}
        
        for meridian, weights in self.meridian_weights.items():
            status_score = 0
            total_weight = 0
            
            for feature, weight in weights.items():
                if feature in health_data:
                    # 标准化体征数据
                    normalized_value = self.normalize_feature(
                        feature, 
                        health_data[feature]
                    )
                    status_score += normalized_value * weight
                    total_weight += weight
            
            # 考虑舌相分析结果
            tongue_correlation = self.get_tongue_correlation(
                meridian, 
                tongue_analysis
            )
            status_score += tongue_correlation * 0.3
            
            # 计算经脉状态
            meridian_status[meridian] = {
                'status_score': status_score / (total_weight + 0.3),
                'status': self.get_status_label(status_score),
                'confidence': self.calculate_confidence(status_score)
            }
        
        return meridian_status
    
    def normalize_feature(self, feature, value):
        """体征数据标准化"""
        if feature == 'heart_rate':
            # 心率标准化 (60-100为正常范围)
            if 60 <= value <= 100:
                return 1.0
            elif value < 60:
                return max(0, 1 - (60 - value) / 30)
            else:
                return max(0, 1 - (value - 100) / 40)
        
        elif feature == 'blood_pressure':
            # 血压标准化 (收缩压)
            systolic = value['systolic']
            if 90 <= systolic <= 140:
                return 1.0
            elif systolic < 90:
                return max(0, 1 - (90 - systolic) / 30)
            else:
                return max(0, 1 - (systolic - 140) / 60)
        
        elif feature == 'blood_oxygen':
            # 血氧标准化
            if value >= 95:
                return 1.0
            else:
                return max(0, value / 95)
        
        return 0.5  # 默认值
    
    def get_tongue_correlation(self, meridian, tongue_analysis):
        """计算舌相与经脉的相关性"""
        correlation_map = {
            '手太阴肺经': ['淡白色', '薄白苔'],
            '手少阴心经': ['红色', '绛红色', '裂纹'],
            '足太阴脾经': ['胖大', '齿痕', '淡白色'],
            '足厥阴肝经': ['青紫色', '瘀斑', '暗红色']
        }
        
        tongue_features = tongue_analysis.get('features', {})
        correlation = 0
        
        for feature, values in correlation_map.items():
            if meridian == feature:
                for value in values:
                    if value in str(tongue_features):
                        correlation += 0.3
        
        return min(correlation, 1.0)
    
    def get_status_label(self, score):
        """根据分数获取状态标签"""
        if score >= 0.8:
            return "旺盛"
        elif score >= 0.6:
            return "正常"
        elif score >= 0.4:
            return "虚弱"
        elif score >= 0.2:
            return "阻滞"
        else:
            return "严重阻滞"
    
    def calculate_confidence(self, score):
        """计算诊断置信度"""
        # 中间值置信度较低，极端值置信度较高
        distance_from_mid = abs(score - 0.5)
        return 0.5 + distance_from_mid

class MeridianNetwork:
    """经脉网络关系分析"""
    
    def __init__(self):
        # 经脉表里关系
        self.exterior_interior_pairs = {
            ('手太阴肺经', '手阳明大肠经'),
            ('足阳明胃经', '足太阴脾经'),
            ('手少阴心经', '手太阳小肠经'),
            ('足太阳膀胱经', '足少阴肾经'),
            ('手厥阴心包经', '手少阳三焦经'),
            ('足少阳胆经', '足厥阴肝经')
        }
        
        # 五行相生关系
        self.generation_cycle = [
            '足厥阴肝经',  # 木
            '手少阴心经',  # 火
            '足太阴脾经',  # 土
            '手太阴肺经',  # 金
            '足少阴肾经'   # 水
        ]
        
        # 经脉流注次序
        self.flow_order = [
            '手太阴肺经', '手阳明大肠经', '足阳明胃经', '足太阴脾经',
            '手少阴心经', '手太阳小肠经', '足太阳膀胱经', '足少阴肾经',
            '手厥阴心包经', '手少阳三焦经', '足少阳胆经', '足厥阴肝经'
        ]
    
    def analyze_network_effects(self, meridian_status):
        """分析经脉网络效应"""
        
        network_analysis = {
            'exterior_interior_effects': [],
            'generation_cycle_effects': [],
            'flow_disruptions': []
        }
        
        # 分析表里关系影响
        for meridian1, meridian2 in self.exterior_interior_pairs:
            if meridian1 in meridian_status and meridian2 in meridian_status:
                status1 = meridian_status[meridian1]['status']
                status2 = meridian_status[meridian2]['status']
                
                if status1 != status2:
                    network_analysis['exterior_interior_effects'].append({
                        'pair': (meridian1, meridian2),
                        'discrepancy': f"{status1} vs {status2}",
                        'interpretation': self.interpret_exterior_interior(status1, status2)
                    })
        
        # 分析五行相生影响
        for i, meridian in enumerate(self.generation_cycle):
            if meridian in meridian_status:
                current_status = meridian_status[meridian]['status']
                next_meridian = self.generation_cycle[(i + 1) % 5]
                
                if next_meridian in meridian_status:
                    next_status = meridian_status[next_meridian]['status']
                    
                    if current_status == '虚弱' and next_status == '虚弱':
                        network_analysis['generation_cycle_effects'].append({
                            'cycle': '五行相生',
                            'weak_link': f"{meridian} -> {next_meridian}",
                            'effect': '相生无力，整体虚损'
                        })
        
        # 分析流注次序
        disruptions = self.check_flow_disruptions(meridian_status)
        network_analysis['flow_disruptions'] = disruptions
        
        return network_analysis
    
    def check_flow_disruptions(self, meridian_status):
        """检查经脉流注次序中的异常"""
        disruptions = []
        
        for i in range(len(self.flow_order) - 1):
            current = self.flow_order[i]
            next_meridian = self.flow_order[i + 1]
            
            if current in meridian_status and next_meridian in meridian_status:
                current_score = meridian_status[current]['status_score']
                next_score = meridian_status[next_meridian]['status_score']
                
                # 如果相邻经脉状态差异过大
                if abs(current_score - next_score) > 0.4:
                    disruptions.append({
                        'from': current,
                        'to': next_meridian,
                        'disruption_level': 'high' if abs(current_score - next_score) > 0.6 else 'medium',
                        'description': f"从{current}到{next_meridian}的气血流动可能受阻"
                    })
        
        return disruptions
    
    def interpret_exterior_interior(self, status1, status2):
        """解释表里关系差异"""
        interpretations = {
            ('虚弱', '旺盛'): "里虚表实，正气不足于内",
            ('旺盛', '虚弱'): "里实表虚，邪气郁闭于内",
            ('阻滞', '正常'): "里经阻滞，表经代偿",
            ('正常', '阻滞'): "表经阻滞，里经未受影响"
        }
        
        return interpretations.get((status1, status2), "表里关系需要进一步观察")

# 基础镜像
FROM python:3.9-slim

# 设置工作目录
WORKDIR /app

# 安装系统依赖
RUN apt-get update && apt-get install -y \
    build-essential \
    libopencv-dev \
    && rm -rf /var/lib/apt/lists/*

# 复制依赖文件
COPY requirements.txt .

# 安装Python依赖
RUN pip install --no-cache-dir -r requirements.txt

# 复制应用代码
COPY . .

# 创建非root用户
RUN useradd -m -u 1000 tcmuser && chown -R tcmuser:tcmuser /app
USER tcmuser

# 暴露端口
EXPOSE 8000

# 启动命令
CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "8000"]

class DataAnonymizer:
    """医疗数据脱敏处理"""
    
    def __init__(self):
        self.salt = os.getenv('ANONYMIZATION_SALT')
    
    def anonymize_health_data(self, health_data):
        """脱敏健康数据"""
        anonymized = {}
        
        # 保留诊断所需数据
        anonymized['vital_signs'] = {
            'heart_rate': health_data.get('heart_rate'),
            'blood_pressure': health_data.get('blood_pressure'),
            'blood_oxygen': health_data.get('blood_oxygen'),
            'temperature': health_data.get('temperature')
        }
        
        # 脱敏个人信息
        anonymized['patient_id'] = self.hash_id(health_data.get('patient_id'))
        anonymized['age_group'] = self.get_age_group(health_data.get('age'))
        anonymized['gender'] = health_data.get('gender')  # 性别对诊断重要
        
        # 移除直接标识信息
        # 不存储：姓名、身份证号、电话、地址等
        
        return anonymized
    
    def hash_id(self, original_id):
        """哈希处理ID"""
        if not original_id:
            return None
        
        # 使用加盐哈希
        salted_id = f"{original_id}{self.salt}"
        hashed = hashlib.sha256(salted_id.encode()).hexdigest()
        
        return hashed[:16]  # 返回前16位
    
    def get_age_group(self, age):
        """获取年龄段"""
        if not age:
            return None
        
        if age < 18:
            return 'child'
        elif age < 40:
            return 'young_adult'
        elif age < 60:
            return 'middle_age'
        else:
            return 'senior'
    
    def prepare_training_data(self, health_records):
        """准备训练数据（完全匿名化）"""
        training_data = []
        
        for record in health_records:
            # 移除所有个人标识信息
            training_record = {
                'features': self.extract_features(record['health_data']),
                'labels': record['diagnosis_labels'],
                'metadata': {
                    'data_source': 'anonymized',
                    'collection_date': record.get('timestamp'),
                    'region': 'generalized'  # 泛化地区信息
                }
            }
            training_data.append(training_record)
        
        return training_data
    
    def extract_features(self, health_data):
        """提取特征（不包含个人身份信息）"""
        features = {}
        
        # 体征特征
        if 'vital_signs' in health_data:
            vital = health_data['vital_signs']
            features['heart_rate_norm'] = self.normalize(vital.get('heart_rate'), 60, 100)
            features['bp_systolic_norm'] = self.normalize(vital.get('blood_pressure', {}).get('systolic'), 90, 140)
            # ... 其他体征标准化
        
        # 舌相特征（已经过处理）
        if 'tongue_analysis' in health_data:
            tongue = health_data['tongue_analysis']
            features['tongue_color'] = tongue.get('color_encoding')
            features['coating_type'] = tongue.get('coating_encoding')
            # ... 其他舌相特征
        
        return features
    
    def normalize(self, value, min_val, max_val):
        """数值标准化"""
        if value is None:
            return 0.5
        
        normalized = (value - min_val) / (max_val - min_val)
        return max(0, min(1, normalized))

#!/bin/bash

# 克隆项目
git clone https://github.com/ai-tcm/meridian-diagnosis.git
cd meridian-diagnosis

# 创建虚拟环境
python -m venv venv
source venv/bin/activate  # Linux/Mac
# venv\Scripts\activate   # Windows

# 安装依赖
pip install -r requirements.txt

# 设置环境变量
cp .env.example .env
# 编辑.env文件，添加API密钥等配置

# 初始化数据库
python scripts/init_database.py

# 下载预训练模型
python scripts/download_models.py

# 启动开发服务器
uvicorn app.main:app --reload --host 0.0.0.0 --port 8000

# 访问API文档
# http://localhost:8000/docs

# 测试脚本 test_api.py
import requests
import json

# 本地开发服务器
BASE_URL = "http://localhost:8000"

def test_meridian_diagnosis():
    """测试经脉诊断API"""
    
    test_data = {
        "patient_data": {
            "vital_signs": {
                "heart_rate": 72,
                "blood_pressure": {"systolic": 118, "diastolic": 78},
                "blood_oxygen": 98,
                "temperature": 36.5
            },
            "tongue_analysis": {
                "color": "淡紫色",
                "coating": "薄白苔",
                "shape": "略有齿痕"
            },
            "symptoms": "倦怠乏力，心悸气短",
            "age": 45,
            "gender": "male"
        }
    }
    
    response = requests.post(
        f"{BASE_URL}/api/v1/diagnosis/meridian",
        json=test_data,
        headers={"Content-Type": "application/json"}
    )
    
    if response.status_code == 200:
        result = response.json()
        print("诊断成功!")
        print(f"总体评估: {result['overall_assessment']}")
        print(f"主要异常经脉: {result['primary_meridian_issues']}")
        
        # 保存结果
        with open('diagnosis_result.json', 'w') as f:
            json.dump(result, f, indent=2, ensure_ascii=False)
        
        return result
    else:
        print(f"诊断失败: {response.status_code}")
        print(response.text)
        return None

if __name__ == "__main__":
    result = test_meridian_diagnosis()
    if result:
        print("\n诊断结果已保存至 diagnosis_result.json")