import re
import unicodedata
from typing import List, Optional, Dict
from dataclasses import dataclass
 
@dataclass
class TokenInfo:
    """トークンの詳細情報"""
    surface: str      # 表層形
    base: str = ""    # 原形
    pos: str = ""     # 品詞
    reading: str = "" # 読み
 
class TextPreprocessor:
    """日本語/英語対応の高度なテキスト前処理"""
 
    def __init__(self, language: str = "ja"):
        self.language = language
        self._stopwords_ja = {
            "の", "に", "は", "を", "た", "が", "で", "て", "と", "し",
            "れ", "さ", "ある", "いる", "も", "する", "から", "な",
            "こと", "として", "い", "や", "れる", "など", "なっ",
            "ない", "この", "ため", "その", "あっ", "よう", "また",
            "もの", "という", "あり", "まで", "られ", "なる", "へ",
            "か", "だ", "これ", "によって", "により", "おり", "より",
        }
        self._stopwords_en = {
            "the", "a", "an", "is", "are", "was", "were", "be", "been",
            "being", "have", "has", "had", "do", "does", "did", "will",
            "would", "could", "should", "may", "might", "shall", "can",
            "to", "of", "in", "for", "on", "with", "at", "by", "from",
            "as", "into", "through", "during", "before", "after",
            "above", "below", "between", "out", "off", "over", "under",
            "again", "further", "then", "once", "here", "there", "when",
            "where", "why", "how", "all", "both", "each", "few", "more",
            "most", "other", "some", "such", "no", "nor", "not", "only",
            "own", "same", "so", "than", "too", "very", "just",
            "it", "its", "he", "she", "they", "them", "his", "her",
            "this", "that", "these", "those", "i", "me", "my", "we",
        }
 
    def normalize(self, text: str) -> str:
        """Unicode正規化 + 基本的なクリーニング"""
        # NFKC正規化（全角→半角、異体字統一）
        text = unicodedata.normalize("NFKC", text)
        # URL/メール/ハッシュタグの置換
        text = re.sub(r"https?://\S+", "[URL]", text)
        text = re.sub(r"\S+@\S+", "[EMAIL]", text)
        text = re.sub(r"#(\w+)", r"[HASHTAG:\1]", text)
        text = re.sub(r"@(\w+)", r"[MENTION:\1]", text)
        # 連続する空白の正規化
        text = re.sub(r"\s+", " ", text).strip()
        # HTML タグの除去
        text = re.sub(r"<[^>]+>", "", text)
        # 制御文字の除去
        text = "".join(c for c in text if unicodedata.category(c)[0] != "C" or c in "\n\t ")
        return text
 
    def normalize_neologd(self, text: str) -> str:
        """NEologd 風の正規化（日本語向け）"""
        text = self.normalize(text)
        # 長音記号の正規化
        text = re.sub(r"[〜～]", "ー", text)
        # 繰り返し記号の削減
        text = re.sub(r"([!?！？]){2,}", r"\1", text)
        text = re.sub(r"(ー){2,}", "ー", text)
        text = re.sub(r"(っ){2,}", "っ", text)
        text = re.sub(r"(。){2,}", "。", text)
        # 括弧の正規化
        text = re.sub(r"[（\(]", "(", text)
        text = re.sub(r"[）\)]", ")", text)
        return text
 
    def tokenize_ja(self, text: str, with_pos: bool = False) -> List:
        """日本語形態素解析（MeCab）"""
        import MeCab
        tagger = MeCab.Tagger()
        parsed = tagger.parse(text)
 
        tokens = []
        for line in parsed.split("\n"):
            if line == "EOS" or line == "":
                continue
            parts = line.split("\t")
            if len(parts) < 2:
                continue
            surface = parts[0]
            features = parts[1].split(",") if len(parts) > 1 else []
 
            if with_pos:
                tokens.append(TokenInfo(
                    surface=surface,
                    pos=features[0] if len(features) > 0 else "",
                    base=features[6] if len(features) > 6 else surface,
                    reading=features[7] if len(features) > 7 else "",
                ))
            else:
                tokens.append(surface)
 
        return tokens
 
    def tokenize_en(self, text: str) -> List[str]:
        """英語トークン化"""
        text = text.lower()
        # 基本的なトークン化（句読点を分離）
        text = re.sub(r"([.!?,;:'\"-])", r" \1 ", text)
        text = re.sub(r"\s+", " ", text).strip()
        return text.split()
 
    def tokenize(self, text: str) -> List[str]:
        text = self.normalize(text)
        if self.language == "ja":
            return self.tokenize_ja(text)
        return self.tokenize_en(text)
 
    def remove_stopwords(self, tokens: List[str],
                          custom_stopwords: set = None) -> List[str]:
        stopwords = (custom_stopwords or
                     (self._stopwords_ja if self.language == "ja"
                      else self._stopwords_en))
        return [t for t in tokens if t not in stopwords and len(t) > 1]
 
    def extract_keywords(self, text: str, top_k: int = 10) -> List[Dict]:
        """TF-IDFベースのキーワード抽出"""
        from sklearn.feature_extraction.text import TfidfVectorizer
        import numpy as np
 
        tokens = self.tokenize(text)
        clean_tokens = self.remove_stopwords(tokens)
 
        # 疑似TF-IDF（単文書の場合はTFのみ）
        word_freq = {}
        for token in clean_tokens:
            word_freq[token] = word_freq.get(token, 0) + 1
 
        total = sum(word_freq.values())
        keywords = sorted(
            [{"word": w, "score": c / total} for w, c in word_freq.items()],
            key=lambda x: x["score"], reverse=True
        )[:top_k]
 
        return keywords
 
# 使用例
preprocessor = TextPreprocessor(language="ja")
text = "東京は今日も晴れです。  気温は２５度でした。https://example.com"
normalized = preprocessor.normalize(text)
tokens = preprocessor.tokenize(normalized)
clean_tokens = preprocessor.remove_stopwords(tokens)
print(f"正規化: {normalized}")
print(f"トークン: {tokens}")
print(f"前処理後: {clean_tokens}")

from transformers import AutoTokenizer
 
def compare_tokenizers(text: str):
    """複数のトークナイザの挙動を比較する"""
    tokenizer_names = {
        "BERT (日本語)": "cl-tohoku/bert-base-japanese-v3",
        "GPT-2": "gpt2",
        "T5": "t5-small",
        "Llama2": "meta-llama/Llama-2-7b-hf",
    }
 
    print(f"テキスト: '{text}'")
    print("-" * 60)
 
    for name, model_name in tokenizer_names.items():
        try:
            tokenizer = AutoTokenizer.from_pretrained(model_name)
            tokens = tokenizer.tokenize(text)
            ids = tokenizer.encode(text)
            decoded = tokenizer.decode(ids)
 
            print(f"\n{name}:")
            print(f"  トークン数: {len(tokens)}")
            print(f"  トークン: {tokens[:20]}{'...' if len(tokens) > 20 else ''}")
            print(f"  ID: {ids[:10]}{'...' if len(ids) > 10 else ''}")
            print(f"  復元: {decoded[:100]}")
        except Exception as e:
            print(f"\n{name}: スキップ ({e})")
 
# compare_tokenizers("自然言語処理は人工知能の重要な研究分野です。")
# compare_tokenizers("Natural language processing is important.")

import pandas as pd
import numpy as np
from sklearn.feature_extraction.text import TfidfVectorizer
from sklearn.model_selection import cross_val_score, StratifiedKFold
from sklearn.linear_model import LogisticRegression
from sklearn.svm import LinearSVC
from sklearn.naive_bayes import MultinomialNB
from sklearn.ensemble import VotingClassifier
from sklearn.pipeline import make_pipeline, Pipeline
from sklearn.metrics import classification_report, confusion_matrix
from sklearn.calibration import CalibratedClassifierCV
 
class TextClassificationPipeline:
    """テキスト分類の包括的パイプライン"""
 
    def __init__(self, language: str = "ja"):
        self.language = language
        self.preprocessor = TextPreprocessor(language)
        self.pipeline = None
 
    def build_pipeline(self, model_type: str = "svm") -> Pipeline:
        """分類パイプラインを構築する"""
        tfidf = TfidfVectorizer(
            analyzer="char_wb" if self.language == "ja" else "word",
            ngram_range=(2, 4) if self.language == "ja" else (1, 2),
            max_features=50000,
            sublinear_tf=True,     # TFのlog正規化
            min_df=2,              # 最低2文書に出現
            max_df=0.95,           # 95%以上の文書に出現する語は除外
        )
 
        models = {
            "lr": LogisticRegression(
                max_iter=1000, C=1.0, class_weight="balanced"
            ),
            "svm": CalibratedClassifierCV(
                LinearSVC(C=1.0, class_weight="balanced", max_iter=5000),
                cv=3
            ),
            "nb": MultinomialNB(alpha=0.1),
            "ensemble": VotingClassifier(
                estimators=[
                    ("lr", LogisticRegression(max_iter=1000, C=1.0)),
                    ("svm", CalibratedClassifierCV(LinearSVC(C=1.0), cv=3)),
                    ("nb", MultinomialNB(alpha=0.1)),
                ],
                voting="soft"
            ),
        }
 
        self.pipeline = make_pipeline(tfidf, models[model_type])
        return self.pipeline
 
    def evaluate(self, texts, labels, cv=5):
        """交差検証で評価する"""
        skf = StratifiedKFold(n_splits=cv, shuffle=True, random_state=42)
        scores = cross_val_score(
            self.pipeline, texts, labels,
            cv=skf, scoring="f1_weighted", n_jobs=-1
        )
        print(f"F1 (weighted): {scores.mean():.4f} (+/- {scores.std():.4f})")
        return scores
 
    def train_and_report(self, X_train, y_train, X_test, y_test):
        """学習して詳細レポートを出力する"""
        self.pipeline.fit(X_train, y_train)
        y_pred = self.pipeline.predict(X_test)
        print(classification_report(y_test, y_pred))
        return y_pred
 
# 使用例
texts = [
    "この映画は素晴らしい演技で感動した",
    "ストーリーが退屈で眠くなった",
    "映像美が際立つ傑作だ",
    "期待外れの駄作だった",
    "心温まる感動的な作品",
    "つまらない展開の連続で苦痛だった",
    "演技力に圧倒される名作",
    "時間の無駄だった",
]
labels = [1, 0, 1, 0, 1, 0, 1, 0]
 
pipeline = TextClassificationPipeline(language="ja")
pipeline.build_pipeline("lr")
pipeline.pipeline.fit(texts, labels)
 
test_texts = ["感動的な映画だった", "退屈な映画だった"]
preds = pipeline.pipeline.predict(test_texts)
for t, p in zip(test_texts, preds):
    print(f"  '{t}' -> {'ポジティブ' if p == 1 else 'ネガティブ'}")

from transformers import (
    AutoTokenizer, AutoModelForSequenceClassification,
    TrainingArguments, Trainer, EarlyStoppingCallback
)
from datasets import Dataset, DatasetDict
import torch
import numpy as np
from sklearn.metrics import accuracy_score, f1_score, precision_score, recall_score
 
class BERTClassifier:
    """日本語BERTのファインチューニングパイプライン"""
 
    def __init__(self, model_name: str = "cl-tohoku/bert-base-japanese-v3",
                 num_labels: int = 2, max_length: int = 128):
        self.model_name = model_name
        self.num_labels = num_labels
        self.max_length = max_length
        self.tokenizer = AutoTokenizer.from_pretrained(model_name)
        self.model = AutoModelForSequenceClassification.from_pretrained(
            model_name, num_labels=num_labels
        )
 
    def prepare_dataset(self, texts, labels, test_size=0.2):
        """データセットを準備する"""
        dataset = Dataset.from_dict({"text": texts, "label": labels})
        dataset = dataset.train_test_split(test_size=test_size, seed=42,
                                            stratify_by_column="label")
 
        def tokenize_fn(examples):
            return self.tokenizer(
                examples["text"],
                padding="max_length",
                truncation=True,
                max_length=self.max_length
            )
 
        tokenized = dataset.map(tokenize_fn, batched=True,
                                 remove_columns=["text"])
        tokenized.set_format("torch")
        return tokenized
 
    def compute_metrics(self, eval_pred):
        """評価メトリクスを計算する"""
        logits, labels = eval_pred
        predictions = np.argmax(logits, axis=-1)
        return {
            "accuracy": accuracy_score(labels, predictions),
            "f1": f1_score(labels, predictions, average="weighted"),
            "precision": precision_score(labels, predictions, average="weighted"),
            "recall": recall_score(labels, predictions, average="weighted"),
        }
 
    def train(self, tokenized_dataset, output_dir="./results",
              epochs=5, batch_size=16, lr=2e-5):
        """学習を実行する"""
        training_args = TrainingArguments(
            output_dir=output_dir,
            num_train_epochs=epochs,
            per_device_train_batch_size=batch_size,
            per_device_eval_batch_size=batch_size * 2,
            learning_rate=lr,
            weight_decay=0.01,
            warmup_ratio=0.1,
            evaluation_strategy="epoch",
            save_strategy="epoch",
            load_best_model_at_end=True,
            metric_for_best_model="f1",
            greater_is_better=True,
            logging_steps=50,
            fp16=torch.cuda.is_available(),
            dataloader_num_workers=2,
            report_to="none",
        )
 
        trainer = Trainer(
            model=self.model,
            args=training_args,
            train_dataset=tokenized_dataset["train"],
            eval_dataset=tokenized_dataset["test"],
            compute_metrics=self.compute_metrics,
            callbacks=[EarlyStoppingCallback(early_stopping_patience=3)],
        )
 
        trainer.train()
        return trainer
 
    def predict(self, texts: list) -> list:
        """テキストを分類する"""
        self.model.eval()
        inputs = self.tokenizer(
            texts, padding=True, truncation=True,
            max_length=self.max_length, return_tensors="pt"
        )
        device = next(self.model.parameters()).device
        inputs = {k: v.to(device) for k, v in inputs.items()}
 
        with torch.no_grad():
            outputs = self.model(**inputs)
            predictions = torch.argmax(outputs.logits, dim=-1)
 
        return predictions.cpu().tolist()
 
# 使用例
# classifier = BERTClassifier(num_labels=2)
# dataset = classifier.prepare_dataset(texts, labels)
# trainer = classifier.train(dataset)
# predictions = classifier.predict(["素晴らしい映画だった"])

from transformers import pipeline
import json
 
class LLMClassifier:
    """LLMを使ったゼロショット/フューショット分類"""
 
    def __init__(self, model_name: str = "facebook/bart-large-mnli"):
        self.zero_shot = pipeline(
            "zero-shot-classification",
            model=model_name
        )
 
    def classify_zero_shot(self, text: str, labels: list,
                            multi_label: bool = False) -> dict:
        """ゼロショット分類"""
        result = self.zero_shot(
            text, labels,
            multi_label=multi_label
        )
        return {
            "text": text,
            "predictions": [
                {"label": label, "score": round(score, 4)}
                for label, score in zip(result["labels"], result["scores"])
            ],
            "top_label": result["labels"][0],
            "top_score": round(result["scores"][0], 4),
        }
 
    def classify_batch(self, texts: list, labels: list) -> list:
        """バッチゼロショット分類"""
        results = self.zero_shot(texts, labels)
        if not isinstance(results, list):
            results = [results]
        return [
            {
                "text": text,
                "label": r["labels"][0],
                "score": round(r["scores"][0], 4),
            }
            for text, r in zip(texts, results)
        ]
 
    @staticmethod
    def few_shot_prompt(text: str, examples: list,
                         labels: list) -> str:
        """フューショット学習用のプロンプトを構築する"""
        prompt = "以下のテキストを分類してください。\n\n"
        prompt += f"カテゴリ: {', '.join(labels)}\n\n"
        prompt += "例:\n"
        for ex in examples:
            prompt += f"テキスト: {ex['text']}\n"
            prompt += f"カテゴリ: {ex['label']}\n\n"
        prompt += f"テキスト: {text}\n"
        prompt += "カテゴリ: "
        return prompt
 
# 使用例
# classifier = LLMClassifier()
# result = classifier.classify_zero_shot(
#     "この商品は品質が良く、価格も手頃です",
#     ["ポジティブ", "ネガティブ", "ニュートラル"]
# )
# print(f"分類結果: {result['top_label']} ({result['top_score']})")

import spacy
from transformers import pipeline
from typing import List, Dict, Optional
from dataclasses import dataclass
 
@dataclass
class Entity:
    text: str
    label: str
    start: int
    end: int
    score: float = 1.0
 
class NERPipeline:
    """複数バックエンドに対応するNERパイプライン"""
 
    def __init__(self, backend: str = "spacy", model_name: str = None):
        self.backend = backend
 
        if backend == "spacy":
            model_name = model_name or "ja_core_news_lg"
            self.nlp = spacy.load(model_name)
        elif backend == "transformers":
            model_name = model_name or "dslim/bert-base-NER"
            self.ner_pipeline = pipeline(
                "ner", model=model_name,
                aggregation_strategy="simple"
            )
 
    def extract(self, text: str) -> List[Entity]:
        """固有表現を抽出する"""
        if self.backend == "spacy":
            return self._extract_spacy(text)
        else:
            return self._extract_transformers(text)
 
    def _extract_spacy(self, text: str) -> List[Entity]:
        doc = self.nlp(text)
        return [
            Entity(
                text=ent.text,
                label=ent.label_,
                start=ent.start_char,
                end=ent.end_char,
            )
            for ent in doc.ents
        ]
 
    def _extract_transformers(self, text: str) -> List[Entity]:
        results = self.ner_pipeline(text)
        return [
            Entity(
                text=ent["word"],
                label=ent["entity_group"],
                start=ent["start"],
                end=ent["end"],
                score=round(ent["score"], 4),
            )
            for ent in results
        ]
 
    def extract_batch(self, texts: List[str]) -> List[List[Entity]]:
        """バッチでNERを実行する"""
        return [self.extract(text) for text in texts]
 
    def format_output(self, text: str, entities: List[Entity]) -> str:
        """エンティティをハイライトした文字列を生成する"""
        result = text
        # 末尾からの置換で位置がずれないようにする
        for ent in sorted(entities, key=lambda e: e.start, reverse=True):
            result = (
                result[:ent.start]
                + f"{ent.text}"
                + result[ent.end:]
            )
        return result
 
# 使用例
ner = NERPipeline(backend="transformers", model_name="dslim/bert-base-NER")
text = "Apple CEO Tim Cook announced new products in San Francisco."
entities = ner.extract(text)
for ent in entities:
    print(f"  [{ent.label}] {ent.text} (信頼度: {ent.score})")
print(f"\n  注釈付き: {ner.format_output(text, entities)}")

from transformers import (
    AutoTokenizer, AutoModelForTokenClassification,
    TrainingArguments, Trainer, DataCollatorForTokenClassification
)
from datasets import Dataset
import numpy as np
 
class CustomNERTrainer:
    """カスタムNERモデルの学習パイプライン"""
 
    def __init__(self, model_name: str = "cl-tohoku/bert-base-japanese-v3",
                 label_list: list = None):
        self.model_name = model_name
        self.label_list = label_list or [
            "O", "B-PER", "I-PER", "B-ORG", "I-ORG",
            "B-LOC", "I-LOC", "B-DATE", "I-DATE",
        ]
        self.label2id = {l: i for i, l in enumerate(self.label_list)}
        self.id2label = {i: l for i, l in enumerate(self.label_list)}
 
        self.tokenizer = AutoTokenizer.from_pretrained(model_name)
        self.model = AutoModelForTokenClassification.from_pretrained(
            model_name,
            num_labels=len(self.label_list),
            id2label=self.id2label,
            label2id=self.label2id,
        )
 
    def tokenize_and_align(self, examples):
        """トークン化してラベルをアラインメントする"""
        tokenized = self.tokenizer(
            examples["tokens"],
            truncation=True,
            is_split_into_words=True,
            max_length=128,
        )
 
        labels = []
        for i, label in enumerate(examples["ner_tags"]):
            word_ids = tokenized.word_ids(batch_index=i)
            label_ids = []
            previous_word_idx = None
 
            for word_idx in word_ids:
                if word_idx is None:
                    label_ids.append(-100)  # 特殊トークンは無視
                elif word_idx != previous_word_idx:
                    label_ids.append(label[word_idx])
                else:
                    # サブワードの2番目以降: B- を I- に変換
                    lbl = label[word_idx]
                    if self.label_list[lbl].startswith("B-"):
                        lbl = self.label2id[
                            self.label_list[lbl].replace("B-", "I-")
                        ]
                    label_ids.append(lbl)
                previous_word_idx = word_idx
 
            labels.append(label_ids)
 
        tokenized["labels"] = labels
        return tokenized
 
    def compute_metrics(self, eval_pred):
        """NER用のメトリクスを計算する"""
        predictions, labels = eval_pred
        predictions = np.argmax(predictions, axis=2)
 
        # -100を除外して評価
        true_labels = []
        true_predictions = []
        for pred, label in zip(predictions, labels):
            for p, l in zip(pred, label):
                if l != -100:
                    true_labels.append(self.label_list[l])
                    true_predictions.append(self.label_list[p])
 
        # エンティティレベルのF1
        from seqeval.metrics import f1_score, classification_report
        f1 = f1_score([true_labels], [true_predictions])
        return {"f1": f1}
 
    def train(self, train_dataset, eval_dataset, output_dir="./ner_model"):
        """NERモデルを学習する"""
        data_collator = DataCollatorForTokenClassification(
            self.tokenizer, pad_to_multiple_of=8
        )
 
        training_args = TrainingArguments(
            output_dir=output_dir,
            num_train_epochs=10,
            per_device_train_batch_size=16,
            per_device_eval_batch_size=32,
            learning_rate=3e-5,
            weight_decay=0.01,
            warmup_ratio=0.1,
            evaluation_strategy="epoch",
            save_strategy="epoch",
            load_best_model_at_end=True,
            metric_for_best_model="f1",
        )
 
        trainer = Trainer(
            model=self.model,
            args=training_args,
            train_dataset=train_dataset,
            eval_dataset=eval_dataset,
            data_collator=data_collator,
            compute_metrics=self.compute_metrics,
        )
 
        trainer.train()
        return trainer

from transformers import pipeline
import pandas as pd
from typing import Dict, List
import numpy as np
 
class SentimentAnalyzer:
    """マルチ言語・マルチアスペクト感情分析"""
 
    def __init__(self, model_name: str = None):
        if model_name is None:
            model_name = "nlptown/bert-base-multilingual-uncased-sentiment"
        self.pipe = pipeline("sentiment-analysis", model=model_name,
                              device=-1)
 
    def analyze(self, texts: list) -> pd.DataFrame:
        """テキストリストの感情分析"""
        results = self.pipe(texts, truncation=True, max_length=512)
 
        df = pd.DataFrame({
            "text": texts,
            "label": [r["label"] for r in results],
            "score": [round(r["score"], 4) for r in results],
        })
        return df
 
    def analyze_aspects(self, text: str, aspects: list) -> dict:
        """アスペクトベースの感情分析"""
        results = {}
        for aspect in aspects:
            prompt = f"{aspect}について: {text}"
            result = self.pipe(prompt, truncation=True)[0]
            results[aspect] = {
                "label": result["label"],
                "score": round(result["score"], 4),
            }
        return results
 
    def analyze_with_context(self, text: str,
                               context: str = None) -> Dict:
        """コンテキスト付き感情分析"""
        if context:
            input_text = f"[コンテキスト: {context}] {text}"
        else:
            input_text = text
 
        result = self.pipe(input_text, truncation=True)[0]
        return {
            "text": text,
            "context": context,
            "label": result["label"],
            "score": round(result["score"], 4),
        }
 
    def analyze_trends(self, texts: list,
                        timestamps: list = None) -> pd.DataFrame:
        """時系列での感情トレンド分析"""
        results = self.pipe(texts, truncation=True)
 
        df = pd.DataFrame({
            "text": texts,
            "label": [r["label"] for r in results],
            "score": [r["score"] for r in results],
        })
 
        if timestamps:
            df["timestamp"] = timestamps
            df = df.sort_values("timestamp")
 
        # 感情スコアの移動平均
        df["score_ma"] = df["score"].rolling(window=5, min_periods=1).mean()
 
        return df
 
# 使用例
analyzer = SentimentAnalyzer()
 
reviews = [
    "This product is amazing! Best purchase ever.",
    "Terrible quality, broke after one day.",
    "It's okay, nothing special.",
    "Exceeded my expectations, highly recommended!",
    "Not worth the price at all.",
]
 
df = analyzer.analyze(reviews)
print(df.to_string(index=False))
 
# アスペクトベース
# result = analyzer.analyze_aspects(
#     "料理は美味しかったが、サービスが遅かった",
#     aspects=["料理", "サービス", "雰囲気"]
# )

from sentence_transformers import SentenceTransformer
import numpy as np
from typing import List, Dict, Optional
import json
 
class SemanticSearch:
    """文埋め込みによる高度なセマンティック検索"""
 
    def __init__(self, model_name: str = "all-MiniLM-L6-v2"):
        self.model = SentenceTransformer(model_name)
        self.documents = []
        self.metadata = []
        self.embeddings = None
 
    def index(self, documents: list, metadata: list = None) -> None:
        """ドキュメントをインデックス化"""
        self.documents = documents
        self.metadata = metadata or [{}] * len(documents)
        self.embeddings = self.model.encode(
            documents, normalize_embeddings=True,
            show_progress_bar=True, batch_size=32
        )
 
    def search(self, query: str, top_k: int = 5,
               threshold: float = 0.0) -> list:
        """クエリに最も類似するドキュメントを検索"""
        query_emb = self.model.encode(
            [query], normalize_embeddings=True
        )
        similarities = np.dot(self.embeddings, query_emb.T).flatten()
        top_indices = np.argsort(similarities)[::-1][:top_k]
 
        results = []
        for idx in top_indices:
            sim = float(similarities[idx])
            if sim >= threshold:
                results.append({
                    "document": self.documents[idx],
                    "similarity": round(sim, 4),
                    "index": int(idx),
                    "metadata": self.metadata[idx],
                })
        return results
 
    def find_similar_pairs(self, threshold: float = 0.8) -> list:
        """類似度が高い文書ペアを検出する"""
        sim_matrix = np.dot(self.embeddings, self.embeddings.T)
        pairs = []
        n = len(self.documents)
 
        for i in range(n):
            for j in range(i + 1, n):
                if sim_matrix[i][j] >= threshold:
                    pairs.append({
                        "doc1": self.documents[i],
                        "doc2": self.documents[j],
                        "similarity": round(float(sim_matrix[i][j]), 4),
                    })
 
        return sorted(pairs, key=lambda x: x["similarity"], reverse=True)
 
    def cluster_documents(self, n_clusters: int = 5) -> dict:
        """ドキュメントをクラスタリングする"""
        from sklearn.cluster import KMeans
 
        kmeans = KMeans(n_clusters=n_clusters, random_state=42, n_init=10)
        labels = kmeans.fit_predict(self.embeddings)
 
        clusters = {}
        for i, label in enumerate(labels):
            if label not in clusters:
                clusters[label] = []
            clusters[label].append({
                "document": self.documents[i],
                "index": i,
            })
 
        return clusters
 
# 使用例
search = SemanticSearch()
docs = [
    "Python is a programming language.",
    "Machine learning uses data to learn patterns.",
    "Tokyo is the capital of Japan.",
    "Neural networks are inspired by the brain.",
    "Deep learning requires large datasets.",
    "Japan is an island country in East Asia.",
    "Python supports object-oriented programming.",
    "Gradient descent optimizes neural network weights.",
]
search.index(docs)
 
results = search.search("AI and data science", top_k=3)
for r in results:
    print(f"  [{r['similarity']:.3f}] {r['document']}")
 
# 類似ペア検出
print("\n類似文書ペア:")
pairs = search.find_similar_pairs(threshold=0.5)
for p in pairs[:5]:
    print(f"  [{p['similarity']:.3f}] '{p['doc1']}' <-> '{p['doc2']}'")

from sentence_transformers import SentenceTransformer
import numpy as np
 
class SimpleRAG:
    """シンプルなRAGパイプライン"""
 
    def __init__(self, embedding_model: str = "all-MiniLM-L6-v2"):
        self.encoder = SentenceTransformer(embedding_model)
        self.documents = []
        self.embeddings = None
 
    def add_documents(self, documents: list):
        """ドキュメントを追加する"""
        self.documents.extend(documents)
        self.embeddings = self.encoder.encode(
            self.documents, normalize_embeddings=True
        )
 
    def retrieve(self, query: str, top_k: int = 3) -> list:
        """関連ドキュメントを検索する"""
        query_emb = self.encoder.encode(
            [query], normalize_embeddings=True
        )
        similarities = np.dot(self.embeddings, query_emb.T).flatten()
        top_indices = np.argsort(similarities)[::-1][:top_k]
 
        return [
            {
                "text": self.documents[idx],
                "score": float(similarities[idx]),
            }
            for idx in top_indices
        ]
 
    def build_prompt(self, query: str, contexts: list) -> str:
        """コンテキスト付きプロンプトを構築する"""
        context_text = "\n\n".join(
            f"[文書{i+1}] {ctx['text']}"
            for i, ctx in enumerate(contexts)
        )
 
        prompt = f"""以下の参考文書に基づいて質問に回答してください。
参考文書に情報がない場合は「情報がありません」と回答してください。
 
## 参考文書
{context_text}
 
## 質問
{query}
 
## 回答
"""
        return prompt
 
    def query(self, question: str, top_k: int = 3) -> dict:
        """質問に対して検索→回答生成を行う"""
        # 1. 検索
        contexts = self.retrieve(question, top_k=top_k)
 
        # 2. プロンプト構築
        prompt = self.build_prompt(question, contexts)
 
        # 3. LLMで生成（ここではプロンプトを返す）
        return {
            "question": question,
            "contexts": contexts,
            "prompt": prompt,
            # "answer": llm.generate(prompt)  # LLM呼び出し
        }
 
# 使用例
# rag = SimpleRAG()
# rag.add_documents([
#     "Pythonは1991年にGuido van Rossumによって作られた。",
#     "Pythonはインタプリタ型の高水準プログラミング言語。",
#     "Pythonはデータサイエンスや機械学習で広く使われている。",
# ])
# result = rag.query("Pythonは誰が作った？")
# print(result["prompt"])

# BAD: BERTの512トークン制限を無視
inputs = tokenizer(long_text, return_tensors="pt")  # 切り詰められる!
 
# GOOD: 長文対応戦略
def handle_long_text(text, tokenizer, model, max_length=512, stride=128):
    """オーバーラップチャンク分割で長文を処理"""
    inputs = tokenizer(
        text, max_length=max_length, truncation=True,
        stride=stride, return_overflowing_tokens=True,
        return_tensors="pt", padding=True
    )
 
    all_logits = []
    for i in range(inputs["input_ids"].shape[0]):
        chunk_inputs = {
            k: v[i:i+1] for k, v in inputs.items()
            if k != "overflow_to_sample_mapping"
        }
        with torch.no_grad():
            outputs = model(**chunk_inputs)
            all_logits.append(outputs.logits)
 
    # チャンクの結果を統合（平均）
    avg_logits = torch.stack(all_logits).mean(dim=0)
    return avg_logits

# BAD: 学習時と推論時で前処理が異なる
# 学習時: 小文字化 + 記号除去
# 推論時: そのまま入力 → 性能低下
 
# GOOD: 前処理をパイプラインに組み込む
from sklearn.pipeline import Pipeline
from sklearn.feature_extraction.text import TfidfVectorizer
 
pipeline = Pipeline([
    ("tfidf", TfidfVectorizer(
        preprocessor=lambda x: TextPreprocessor("en").normalize(x),
        analyzer="char_wb",
        ngram_range=(2, 4)
    )),
    ("classifier", LogisticRegression()),
])
# 学習時も推論時も同じ前処理が自動適用

# BAD: 英語用BPEトークナイザを日本語に適用
tokenizer = AutoTokenizer.from_pretrained("gpt2")
tokens = tokenizer.tokenize("東京は日本の首都です")
# → 1文字ずつバイト分割され、大量のトークンを消費
 
# GOOD: 日本語対応モデルを使う
tokenizer = AutoTokenizer.from_pretrained("cl-tohoku/bert-base-japanese-v3")
tokens = tokenizer.tokenize("東京は日本の首都です")
# → 形態素に近い単位でトークン化される
 
# 多言語の場合: 多言語対応モデルを使用
tokenizer = AutoTokenizer.from_pretrained(
    "sentence-transformers/paraphrase-multilingual-MiniLM-L12-v2"
)

# 演習1: 基本実装のテンプレート
class Exercise1:
    """基本的な実装パターンの演習"""
 
    def __init__(self):
        self.data = []
 
    def validate_input(self, value):
        """入力値の検証"""
        if value is None:
            raise ValueError("入力値がNoneです")
        return True
 
    def process(self, value):
        """データ処理のメインロジック"""
        self.validate_input(value)
        self.data.append(value)
        return self.data
 
    def get_results(self):
        """処理結果の取得"""
        return {
            'count': len(self.data),
            'data': self.data
        }
 
# テスト
def test_exercise1():
    ex = Exercise1()
    assert ex.process(1) == [1]
    assert ex.process(2) == [1, 2]
    assert ex.get_results()['count'] == 2
 
    try:
        ex.process(None)
        assert False, "例外が発生するべき"
    except ValueError:
        pass
 
    print("全テスト合格!")
 
test_exercise1()

# 演習2: 応用パターン
from typing import List, Dict, Optional
from datetime import datetime
 
class AdvancedExercise:
    """応用パターンの演習"""
 
    def __init__(self, max_size: int = 100):
        self._items: List[Dict] = []
        self._max_size = max_size
        self._created_at = datetime.now()
 
    def add(self, key: str, value: any) -> bool:
        """アイテムの追加（サイズ制限付き）"""
        if len(self._items) >= self._max_size:
            return False
        self._items.append({
            'key': key,
            'value': value,
            'timestamp': datetime.now().isoformat()
        })
        return True
 
    def find(self, key: str) -> Optional[Dict]:
        """キーによる検索"""
        for item in reversed(self._items):
            if item['key'] == key:
                return item
        return None
 
    def remove(self, key: str) -> bool:
        """キーによる削除"""
        for i, item in enumerate(self._items):
            if item['key'] == key:
                self._items.pop(i)
                return True
        return False
 
    def stats(self) -> Dict:
        """統計情報"""
        return {
            'total_items': len(self._items),
            'max_size': self._max_size,
            'usage_percent': len(self._items) / self._max_size * 100,
            'uptime': str(datetime.now() - self._created_at)
        }
 
# テスト
def test_advanced():
    ex = AdvancedExercise(max_size=3)
    assert ex.add("a", 1) == True
    assert ex.add("b", 2) == True
    assert ex.add("c", 3) == True
    assert ex.add("d", 4) == False  # サイズ制限
    assert ex.find("b")['value'] == 2
    assert ex.remove("b") == True
    assert ex.find("b") is None
    stats = ex.stats()
    assert stats['total_items'] == 2
    print("応用テスト全合格!")
 
test_advanced()

# 演習3: パフォーマンス最適化
import time
from functools import lru_cache
 
# 最適化前（O(n^2)）
def slow_search(data: list, target: int) -> int:
    """非効率な検索"""
    for i in range(len(data)):
        for j in range(i + 1, len(data)):
            if data[i] + data[j] == target:
                return (i, j)
    return (-1, -1)
 
# 最適化後（O(n)）
def fast_search(data: list, target: int) -> tuple:
    """ハッシュマップを使った効率的な検索"""
    seen = {}
    for i, num in enumerate(data):
        complement = target - num
        if complement in seen:
            return (seen[complement], i)
        seen[num] = i
    return (-1, -1)
 
# ベンチマーク
def benchmark():
    import random
    data = list(range(5000))
    random.shuffle(data)
    target = data[100] + data[4000]
 
    start = time.time()
    result1 = slow_search(data, target)
    slow_time = time.time() - start
 
    start = time.time()
    result2 = fast_search(data, target)
    fast_time = time.time() - start
 
    print(f"非効率版: {slow_time:.4f}秒")
    print(f"効率版:   {fast_time:.6f}秒")
    print(f"高速化率: {slow_time/fast_time:.0f}倍")
 
benchmark()