暗号化されたデータAPIにおけるデータ品質管理：欠損値と異常値の安全な処理

ECサイトのAIカスタマーサービスが急速に拡大する現代において、顧客メッセージの暗号化とデータ品質の両立は避けて通れない課題です。私は以前、東証上場企業のECプラットフォームでAIチャットボット開発を担当していた際、GDPR準拠のために顧客メッセージの大部分が暗号化され、欠損値や異常値が頻出する環境に取り組みました。本稿では、暗号化されたデータを送受信するAPIにおいて、データ品質を担保するための実践的なアプローチを解説します。

暗号化されたAPI呼び出しの基本構造

HolySheep AIのAPIは<50msのレイテンシを実現しており、暗号化データの送受信においても高速な処理が可能です。まずは基本的な実装を見てみましょう。

import requests
import json
import hashlib
from cryptography.fernet import Fernet
from typing import Optional, Dict, Any

class HolySheepSecureClient:
    """HolySheep AI API用セキュアクライアント - 暗号化されたデータに対応"""
    
    def __init__(self, api_key: str, encryption_key: str):
        self.base_url = "https://api.holysheep.ai/v1"
        self.api_key = api_key
        self.cipher = Fernet(encryption_key.encode())
    
    def _encrypt_data(self, data: str) -> str:
        """データを暗号化する"""
        encrypted = self.cipher.encrypt(data.encode())
        return encrypted.decode('utf-8')
    
    def _decrypt_data(self, encrypted_data: str) -> str:
        """データを復号化する"""
        decrypted = self.cipher.decrypt(encrypted_data.encode())
        return decrypted.decode('utf-8')
    
    def analyze_secure_message(
        self, 
        encrypted_message: str,
        handle_missing: str = "skip",
        detect_anomalies: bool = True
    ) -> Dict[str, Any]:
        """
        暗号化されたメッセージを分析
        
        Args:
            encrypted_message: 暗号化された入力メッセージ
            handle_missing: 欠損値処理 ('skip', 'placeholder', 'error')
            detect_anomalies: 異常値検出を有効にするか
        """
        # 復号化
        try:
            message = self._decrypt_data(encrypted_message)
        except Exception as e:
            return {
                "status": "error",
                "error": f"復号化失敗: {str(e)}",
                "error_code": "DECRYPTION_FAILED"
            }
        
        # 欠損値チェック
        if not message or message.strip() == "":
            if handle_missing == "error":
                return {
                    "status": "error",
                    "error": "メッセージが空です",
                    "error_code": "MISSING_VALUE"
                }
            elif handle_missing == "placeholder":
                message = "[入力なし - 欠損値]"
            else:  # skip
                return {
                    "status": "skipped",
                    "reason": "欠損値をスキップ"
                }
        
        # 異常値検出
        anomalies = []
        if detect_anomalies:
            anomalies = self._detect_anomalies(message)
        
        # HolySheep AI API呼び出し
        headers = {
            "Authorization": f"Bearer {self.api_key}",
            "Content-Type": "application/json"
        }
        
        payload = {
            "model": "gpt-4.1",
            "messages": [
                {"role": "user", "content": f"分析対象: {message}"}
            ],
            "temperature": 0.7,
            "max_tokens": 500
        }
        
        if anomalies:
            payload["metadata"] = {"anomalies": anomalies}
        
        try:
            response = requests.post(
                f"{self.base_url}/chat/completions",
                headers=headers,
                json=payload,
                timeout=30
            )
            response.raise_for_status()
            result = response.json()
            
            return {
                "status": "success",
                "original_message": message,
                "anomalies_detected": anomalies,
                "response": result["choices"][0]["message"]["content"],
                "usage": result.get("usage", {})
            }
        except requests.exceptions.RequestException as e:
            return {
                "status": "error",
                "error": str(e),
                "error_code": "API_REQUEST_FAILED"
            }
    
    def _detect_anomalies(self, message: str) -> list:
        """異常値を検出する"""
        anomalies = []
        
        # 長さの異常
        if len(message) > 5000:
            anomalies.append({
                "type": "LENGTH_OUTLIER",
                "detail": f"メッセージが長すぎます ({len(message)}文字)"
            })
        elif len(message) < 3:
            anomalies.append({
                "type": "LENGTH_OUTLIER",
                "detail": f"メッセージが短すぎます ({len(message)}文字)"
            })
        
        # 特殊文字の過度な使用
        special_char_ratio = sum(1 for c in message if not c.isalnum()) / max(len(message), 1)
        if special_char_ratio > 0.5:
            anomalies.append({
                "type": "SPECIAL_CHAR_ABUSE",
                "detail": f"特殊文字比率过高: {special_char_ratio:.2%}"
            })
        
        # 反復パターンの検出
        if self._has_repetitive_pattern(message):
            anomalies.append({
                "type": "REPETITIVE_PATTERN",
                "detail": "同一文字の反復を検出"
            })
        
        return anomalies
    
    def _has_repetitive_pattern(self, text: str, threshold: int = 5) -> bool:
        """反復パターンを検出"""
        for i in range(len(text) - threshold):
            chunk = text[i:i+threshold]
            if text.count(chunk) >= 3:
                return True
        return False


使用例
if __name__ == "__main__":
    client = HolySheepSecureClient(
        api_key="YOUR_HOLYSHEEP_API_KEY",
        encryption_key="your-32-byte-encryption-key-here"
    )
    
    # テスト用暗号化メッセージ
    test_encrypted = client._encrypt_data("商品の配送状況を教えてください")
    
    result = client.analyze_secure_message(
        encrypted_message=test_encrypted,
        handle_missing="skip",
        detect_anomalies=True
    )
    print(json.dumps(result, indent=2, ensure_ascii=False))

企業RAGシステムにおける欠損値処理の実例

企业内部のナレッジベースをRAG（Retrieval-Augmented Generation）形式で活用する際、暗号化された社内文書からの欠損値処理は極めて重要です。HolySheep AIのAPIはレートが¥1=$1という破格のコスト効率いため、大規模なRAG処理でも費用を抑えられます。

import asyncio
import aiohttp
from dataclasses import dataclass
from typing import List, Optional, Dict, Any
from enum import Enum
import numpy as np

class MissingValueStrategy(Enum):
    """欠損値処理戦略"""
    OMIT = "omit"           # 完全除外
    PLACEHOLDER = "placeholder"  # プレースホルダーで置換
    INTERPOLATE = "interpolate"   # 補間
    INFER = "infer"         # 前後の文脈から推論

@dataclass
class DataQualityReport:
    """データ品質レポート"""
    total_records: int
    missing_count: int
    anomaly_count: int
    clean_records: int
    missing_rate: float
    anomaly_rate: float
    quality_score: float  # 0-100

class SecureRAGProcessor:
    """暗号化されたデータ向けRAGプロセッサ"""
    
    def __init__(self, api_key: str):
        self.api_key = api_key
        self.base_url = "https://api.holysheep.ai/v1"
        self.missing_threshold = 0.3  # 30%以上の欠損率で除外
    
    async def process_documents_batch(
        self,
        encrypted_documents: List[str],
        decryption_key: str,
        missing_strategy: MissingValueStrategy = MissingValueStrategy.OMIT,
        quality_threshold: float = 70.0
    ) -> Dict[str, Any]:
        """
        暗号化されたドキュメントのバッチ処理
        
        Args:
            encrypted_documents: 暗号化されたドキュメントリスト
            decryption_key: 復号化キー
            missing_strategy: 欠損値処理戦略
            quality_threshold: 品質スコア閾値
        """
        from cryptography.fernet import Fernet
        
        cipher = Fernet(decryption_key.encode())
        processed_docs = []
        quality_reports = []
        
        for idx, enc_doc in enumerate(encrypted_documents):
            try:
                # 復号化
                decrypted = cipher.decrypt(enc_doc.encode()).decode()
                
                # 品質チェック
                quality = self._assess_quality(decrypted)
                
                if quality["missing_rate"] > self.missing_threshold:
                    # 欠損率が閾値を超える
                    quality_reports.append({
                        "index": idx,
                        "status": "rejected",
                        "reason": "欠損率超過",
                        "quality": quality
                    })
                    continue
                
                # 異常値処理
                cleaned = self._clean_document(
                    decrypted, 
                    missing_strategy
                )
                
                processed_docs.append({
                    "index": idx,
                    "content": cleaned,
                    "quality": quality
                })
                quality_reports.append({
                    "index": idx,
                    "status": "processed",
                    "quality": quality
                })
                
            except Exception as e:
                quality_reports.append({
                    "index": idx,
                    "status": "error",
                    "reason": str(e)
                })
        
        # 全体レポート生成
        overall_report = self._generate_report(quality_reports)
        
        # RAG用のベクトル化（Embedding API呼び出し）
        if processed_docs:
            embeddings = await self._generate_embeddings(processed_docs)
            return {
                "documents": processed_docs,
                "embeddings": embeddings,
                "report": overall_report
            }
        
        return {"documents": [], "embeddings": [], "report": overall_report}
    
    def _assess_quality(self, text: str) -> Dict[str, Any]:
        """ドキュメントの品質を評価"""
        total_fields = 10
        missing_count = 0
        
        # フィールド欠損チェック
        required_fields = ["title", "content", "date", "author", "category"]
        for field in required_fields:
            if not self._field_exists(text, field):
                missing_count += 1
        
        # 異常値検出
        anomaly_count = self._count_anomalies(text)
        
        missing_rate = missing_count / max(total_fields, 1)
        anomaly_rate = anomaly_count / max(len(text), 1)
        quality_score = max(0, 100 - (missing_rate * 50) - (anomaly_rate * 1000))
        
        return {
            "missing_count": missing_count,
            "anomaly_count": anomaly_count,
            "missing_rate": missing_rate,
            "anomaly_rate": anomaly_rate,
            "quality_score": quality_score,
            "char_count": len(text)
        }
    
    def _field_exists(self, text: str, field: str) -> bool:
        """フィールドの存在確認（簡易実装）"""
        # 実際の実装ではJSONや構造化データのパースを使用
        field_markers = {
            "title": ["# ", "タイトル", "title:", "Title:"],
            "content": ["\n\n", "本文", "content:", "Content:"],
            "date": ["2024", "2025", "date:", "Date:"],
            "author": ["著者", "author:", "Author:", "By "],
            "category": ["カテゴリ", "category:", "Category:", "タグ"]
        }
        
        markers = field_markers.get(field, [])
        return any(marker in text for marker in markers)
    
    def _count_anomalies(self, text: str) -> int:
        """異常値の数をカウント"""
        anomalies = 0
        
        # null/Noneパターンの検出
        if "null" in text.lower() or "none" in text.lower():
            anomalies += 1
        
        # 空文字の連続
        if "\n\n\n" in text:
            anomalies += 1
        
        # エンコードエラーの兆候
        if "�" in text or "\ufffd" in text:
            anomalies += 1
        
        return anomalies
    
    def _clean_document(
        self, 
        text: str, 
        strategy: MissingValueStrategy
    ) -> str:
        """ドキュメントをクリーン化"""
        cleaned = text
        
        if strategy == MissingValueStrategy.PLACEHOLDER:
            cleaned = cleaned.replace("null", "[データなし]")
            cleaned = cleaned.replace("None", "[データなし]")
            cleaned = cleaned.replace("NULL", "[データなし]")
        
        elif strategy == MissingValueStrategy.INTERPOLATE:
            # 連続する空行を単一の空行に
            import re
            cleaned = re.sub(r'\n{3,}', '\n\n', cleaned)
            
            # null値を周囲の文脈に基づいて補間
            cleaned = cleaned.replace("null", "")
        
        elif strategy == MissingValueStrategy.INFER:
            # AI推論による補間（HolySheep API使用）
            # 実際の実装ではAPIを呼び出して欠損値を推論
            pass
        
        # 共通の前処理
        cleaned = cleaned.strip()
        return cleaned
    
    async def _generate_embeddings(
        self, 
        documents: List[Dict]
    ) -> List[Dict]:
        """Embedding生成（HolySheep API）"""
        headers = {
            "Authorization": f"Bearer {self.api_key}",
            "Content-Type": "application/json"
        }
        
        texts = [doc["content"] for doc in documents]
        
        payload = {
            "model": "text-embedding-3-small",
            "input": texts
        }
        
        async with aiohttp.ClientSession() as session:
            async with session.post(
                f"{self.base_url}/embeddings",
                headers=headers,
                json=payload,
                timeout=aiohttp.ClientTimeout(total=60)
            ) as response:
                if response.status == 200:
                    result = await response.json()
                    return result.get("data", [])
                else:
                    error_text = await response.text()
                    raise Exception(f"Embedding APIエラー: {error_text}")
    
    def _generate_report(
        self, 
        quality_reports: List[Dict]
    ) -> DataQualityReport:
        """データ品質レポートを生成"""
        total = len(quality_reports)
        processed = sum(1 for r in quality_reports if r["status"] == "processed")
        rejected = sum(1 for r in quality_reports if r["status"] == "rejected")
        
        all_qualities = [
            r["quality"] for r in quality_reports 
            if "quality" in r
        ]
        
        avg_missing = np.mean([q["missing_rate"] for q in all_qualities]) if all_qualities else 0
        avg_anomaly = np.mean([q["anomaly_rate"] for q in all_qualities]) if all_qualities else 0
        avg_score = np.mean([q["quality_score"] for q in all_qualities]) if all_qualities else 0
        
        return DataQualityReport(
            total_records=total,
            missing_count=sum(1 for r in quality_reports if r["status"] == "rejected" and "欠損" in r.get("reason", "")),
            anomaly_count=rejected,
            clean_records=processed,
            missing_rate=avg_missing,
            anomaly_rate=avg_anomaly,
            quality_score=avg_score
        )


使用例
async def main():
    from cryptography.fernet import Fernet
    
    # テスト用暗号化ドキュメント
    key = Fernet.generate_key()
    cipher = Fernet(key)
    
    test_docs = [
        cipher.encrypt(b"商品	A についての説明本文...\n\nカテゴリ: 電子機器\n日付: 2025-01-15"),
        cipher.encrypt(b"null"),
        cipher.encrypt(b"価格	B の情報は...\n\n著者: 営業部\nnull\nカテゴリ: >null<"),
        cipher.encrypt(b"これは正常なドキュメントです。\nタイトル: 製品ガイド\n本文: 詳細な説明..."),
    ]
    
    processor = SecureRAGProcessor(api_key="YOUR_HOLYSHEEP_API_KEY")
    
    result = await processor.process_documents_batch(
        encrypted_documents=test_docs,
        decryption_key=key.decode(),
        missing_strategy=MissingValueStrategy.PLACEHOLDER,
        quality_threshold=70.0
    )
    
    print("=== 処理結果 ===")
    print(f"処理済みドキュメント数: {len(result['documents'])}")
    print(f"品質スコア: {result['report'].quality_score:.2f}")
    print(f"欠損率: {result['report'].missing_rate:.2%}")
    print(f"異常率: {result['report'].anomaly_rate:.4%}")

if __name__ == "__main__":
    asyncio.run(main())

個人開発者向け：最小構成のデータ品質チェック

個人開発者が急に立ち上げたプロジェクトにおいても、プロダクションレベルのデータ品質管理は重要です。HolySheep AIのAPIは登録で無料クレジットがもらえるため、コストリスクを最小限に抑えられます。

#!/usr/bin/env python3
"""
個人開発者向け・最小構成のデータ品質チェック
HolySheep AI API 使用
"""

import requests
import json
from typing import Optional

def check_data_quality_and_analyze(
    api_key: str,
    text: str,
    check_missing: bool = True,
    check_anomalies: bool = True
) -> dict:
    """
    データ品質チェック + 分析を1度に実行
    
    Returns:
        dict: 品質チェック結果と分析結果
    """
    # 入力検証
    if not text or not text.strip():
        return {
            "quality_check": {
                "status": "FAIL",
                "reason": "EMPTY_INPUT",
                "score": 0
            },
            "analysis": None,
            "error": "入力が空です"
        }
    
    # 品質スコア計算
    quality_score = 100
    issues = []
    
    # 欠損値チェック
    if check_missing:
        missing_indicators = ["null", "None", "NULL", "undefined", "N/A", "n/a", ""]
        found_missing = any(indicator in text for indicator in missing_indicators)
        if found_missing:
            quality_score -= 30
            issues.append("欠損値 indicator を検出")
    
    # 異常値チェック
    if check_anomalies:
        # 長さ異常
        if len(text) > 10000:
            quality_score -= 20
            issues.append(f"テキストが長すぎます ({len(text)}文字)")
        elif len(text) < 10:
            quality_score -= 15
            issues.append(f"テキストが短すぎます ({len(text)}文字)")
        
        # エンコーディング異常
        encoding_issues = ["\ufffd", "�"]
        for issue in encoding_issues:
            if issue in text:
                quality_score -= 25
                issues.append("エンコーディング異常を検出")
                break
        
        # 特殊文字過多
        special_chars = sum(1 for c in text if ord(c) > 127)
        if special_chars / len(text) > 0.5:
            quality_score -= 10
            issues.append("特殊文字比率が高め")
    
    # 品質判定
    if quality_score >= 80:
        quality_status = "EXCELLENT"
    elif quality_score >= 60:
        quality_status = "GOOD"
    elif quality_score >= 40:
        quality_status = "FAIR"
    else:
        quality_status = "POOR"
    
    quality_result = {
        "status": quality_status,
        "score": max(0, quality_score),
        "issues": issues,
        "char_count": len(text)
    }
    
    # HolySheep AI で分析（品質スコアが閾値以上の場合）
    analysis_result = None
    if quality_score >= 40:
        headers = {
            "Authorization": f"Bearer {api_key}",
            "Content-Type": "application/json"
        }
        
        payload = {
            "model": "deepseek-v3.2",  # $0.42/MTokのコスト効率モデル
            "messages": [
                {
                    "role": "system", 
                    "content": f"データ品質スコア: {quality_score}/100\n問題点: {', '.join(issues) if issues else 'なし'}"
                },
                {
                    "role": "user", 
                    "content": f"以下のデータを分析してください:\n{text[:2000]}"
                }
            ],
            "temperature": 0.3,
            "max_tokens": 300
        }
        
        try:
            response = requests.post(
                "https://api.holysheep.ai/v1/chat/completions",
                headers=headers,
                json=payload,
                timeout=15
            )
            response.raise_for_status()
            result = response.json()
            
            analysis_result = {
                "content": result["choices"][0]["message"]["content"],
                "model": result.get("model", "unknown"),
                "usage": result.get("usage", {})
            }
        except requests.exceptions.RequestException as e:
            return {
                "quality_check": quality_result,
                "analysis": None,
                "error": f"API呼び出し失敗: {str(e)}"
            }
    
    return {
        "quality_check": quality_result,
        "analysis": analysis_result,
        "error": None
    }


def batch_quality_report(
    api_key: str,
    texts: list,
    quality_threshold: int = 60
) -> dict:
    """バッチ処理用の品質レポート"""
    results = []
    stats = {
        "total": len(texts),
        "excellent": 0,
        "good": 0,
        "fair": 0,
        "poor": 0,
        "total_score": 0
    }
    
    for idx, text in enumerate(texts):
        result = check_data_quality_and_analyze(
            api_key=api_key,
            text=text,
            check_missing=True,
            check_anomalies=True
        )
        results.append({
            "index": idx,
            "status": result["quality_check"]["status"],
            "score": result["quality_check"]["score"],
            "issues": result["quality_check"]["issues"]
        })
        
        # 統計更新
        status = result["quality_check"]["status"].lower()
        if status in stats:
            stats[status] += 1
        stats["total_score"] += result["quality_check"]["score"]
    
    # 平均スコア計算
    stats["average_score"] = stats["total_score"] / max(stats["total"], 1)
    stats["pass_rate"] = (stats["excellent"] + stats["good"]) / max(stats["total"], 1) * 100
    
    return {
        "results": results,
        "summary": stats,
        "recommendation": "全データ使用" if stats["pass_rate"] >= 90 else "要確認"
    }


CLI使用例
if __name__ == "__main__":
    import sys
    
    if len(sys.argv) < 2:
        print("使用方法: python quality_check.py ")
        sys.exit(1)
    
    api_key = sys.argv[1]
    
    # テストデータ
    test_texts = [
        "正常な商品名は、価格は1000円です。",
        "null",
        "これは非常に長いテキストです" * 1000,
        "商品情報：A-B-C企業" + "\ufffd" + "の案内",
        "短い",
    ]
    
    report = batch_quality_report(api_key, test_texts)
    
    print("=== 品質レポート ===")
    print(json.dumps(report, indent=2, ensure_ascii=False))

# エラー例
{"status": "error", "error": "復号化失敗: ..."}

原因と解決
from cryptography.fernet import InvalidToken

try:
    decrypted = cipher.decrypt(encrypted_data.encode())
except InvalidToken:
    # 解決方法: キーの不一致またはデータ破損
    # 1. 暗号化・復号化キーが同一か確認
    # 2. 暗号化キーをローテーションせずに再試行
    # 3. データが暗号化済みか確認
    
    # フォールバック処理
    return {
        "status": "decryption_failed",
        "action": "notify_admin",
        "message": "復号化キーの不一致を検出"
    }

# エラー例
{"status": "error", "error_code": "MISSING_VALUE", "error": "必須フィールドが存在しません"}

原因と解決
def safe_extract(data: dict, keys: list, default: str = "[不明]") -> dict:
    """安全なフィールド抽出"""
    result = {}
    for key in keys:
        value = data.get(key)
        if value is None or value == "" or str(value).lower() == "null":
            # デフォルト値を設定して処理を続行
            result[key] = default
            result[f"{key}_was_missing"] = True
        else:
            result[key] = value
    return result

使用例
safe_data = safe_extract(
    encrypted_response,
    keys=["content", "author", "timestamp"],
    default="[データなし]"
)

# 問題: 正常なテキストが異常と判定される
例: 絵文字や特殊文字を含む自然なテキスト

解決: コンテキストを考慮した異常値判定
def smart_anomaly_detection(text: str, context: str = "general") -> dict:
    """コンテキスト適応型異常値検出"""
    
    # コンテキスト別の閾値設定
    thresholds = {
        "chat": {"length_max": 2000, "special_ratio": 0.6},
        "document": {"length_max": 50000, "special_ratio": 0.3},
        "code": {"length_max": 100000, "special_ratio": 0.7},
        "general": {"length_max": 10000, "special_ratio": 0.5}
    }
    
    config = thresholds.get(context, thresholds["general"])
    
    anomalies = []
    
    # 長すぎる場合（コンテキスト考慮）
    if len(text) > config["length_max"]:
        anomalies.append({
            "type": "LENGTH_WARNING",
            "detail": f"長さ{int(len(text)/config['length_max']*100)}%超過（閾値: {config['length_max']}）",
            "severity": "low"  # 長すぎるだけでは重大ではない
        })
    
    # 特殊文字判定（絵文字は正常とみなす）
    normal_special = sum(1 for c in text if c in "!?.,。！？～…「」【】()（）")
    abnormal_special = sum(1 for c in text if ord(c) > 127 and not c.isalnum())
    
    if abnormal_special / max(len(text), 1) > config["special_ratio"]:
        anomalies.append({
            "type": "ENCODING_CONCERN",
            "detail": "エンコーディング異常の可能性がある",
            "severity": "medium",
            "action": "manual_review_recommended"
        })
    
    return {
        "has_critical_anomaly": any(a.get("severity") == "high" for a in anomalies),
        "anomalies": anomalies,
        "recommendation": "proceed" if not anomalies else "review"
    }

# エラー例
{"error": "rate_limit_exceeded", "retry_after": 60}

解決: 指数バックオフによるリトライ
import time
from functools import wraps

def retry_with_backoff(max_retries: int = 5, base_delay: float = 1.0):
    """指数バックオフデコレータ"""
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            for attempt in range(max_retries):
                try:
                    return func(*args, **kwargs)
                except requests.exceptions.RequestException as e:
                    if "rate_limit" in str(e).lower():
                        delay = base_delay * (2 ** attempt)
                        # HolySheepは<50msレイテンシなので短めのバックオフ
                        wait_time = min(delay, 30)
                        time.sleep(wait_time)
                        continue
                    raise
            return {"error": "max_retries_exceeded"}
        return wrapper
    return decorator

@retry_with_backoff(max_retries=3, base_delay=2.0)
def call_holySheep_api(payload: dict, api_key: str):
    headers = {
        "Authorization": f"Bearer {api_key}",
        "Content-Type": "application/json"
    }
    response = requests.post(
        "https://api.holysheep.ai/v1/chat/completions",
        headers=headers,
        json=payload,
        timeout=30
    )
    return response.json()

# 問題: 暗号化されたデータと平文データが混在

解決: 型安全なラッパー
from typing import Union, TypeVar
from dataclasses import dataclass

T = TypeVar('T')

@dataclass
class SecureData:
    """型安全なセキュアデータラッパー"""
    encrypted: Optional[str] = None
    decrypted: Optional[str] = None
    is_encrypted: bool = True
    quality_metadata: dict = None
    
    @classmethod
    def from_encrypted(cls, encrypted_str: str, cipher) -> "SecureData":
        """暗号化された文字列から作成"""
        try:
            decrypted = cipher.decrypt(encrypted_str.encode()).decode()
            return cls(
                encrypted=encrypted_str,
                decrypted=decrypted,
                is_encrypted=True,
                quality_metadata={}
            )
        except:
            # 復号化失敗時
            return cls(
                encrypted=encrypted_str,
                decrypted=None,
                is_encrypted=True,
                quality_metadata={"error": "decryption_failed"}
            )
    
    @classmethod
    def from_plaintext(cls, text: str, cipher) -> "SecureData":
        """平文から暗号化して作成"""
        encrypted = cipher.encrypt(text.encode()).decode()
        return cls(
            encrypted=encrypted,
            decrypted=text,
            is_encrypted=True,
            quality_metadata={}
        )
    
    def get_content(self) -> str:
        """内容を安全に取り出す"""
        if self.decrypted:
            return self.decrypted
        raise ValueError("復号化に失敗したデータは取得できません")

👉 HolySheep AI に登録して無料クレジットを獲得