取材資料のデジタルアーカイブ構築（AIによる整理・知識グラフ化）

取材・調査報道で集めた大量の資料を、検索・分析できる本番品質のデジタルアーカイブに変えるための設計パターン集。 複数の入力源（開示文書のOCR、Webスクレイピング、SNS、反訳）を統一スキーマにまとめ、AIで要約・分類・エンティティ抽出を行い、人物や組織の関係を知識グラフとして持たせる。

注意：本スキルは資料整理の手引きであり、法的助言ではない。取材資料には個人情報・要配慮情報が含まれることが多い。アーカイブ化・外部AIへの送信・共有の各段階で、個人情報保護法・取材源の保護・各社の取材倫理規程に照らして扱いを判断すること。詳しくは末尾の「個人情報・機密の取り扱い」を必ず読む。

アーカイブの全体構成

複数入力源の統合パターン

┌─────────────────┐    ┌──────────────────┐    ┌────────────────┐
│  OCRパイプライン │    │  Webスクレイピング│    │  SNS収集        │
│ （開示文書・紙） │    │ （記事・公開資料）│    │ （投稿・反訳）  │
└────────┬────────┘    └────────┬─────────┘    └───────┬────────┘
         │                      │                      │
         └──────────────────────┼──────────────────────┘
                                │
                    ┌───────────▼───────────┐
                    │  統一スキーマ          │
                    │  （35項目以上）        │
                    └───────────┬───────────┘
                                │
         ┌──────────────────────┼──────────────────────┐
         │                      │                      │
┌────────▼────────┐  ┌──────────▼──────────┐  ┌───────▼───────┐
│  AIエンリッチ    │  │  エンティティ抽出   │  │  PDF保存       │
│ （要約・分類）   │  │ （知識グラフ）      │  │ （アクセシブル）│
└────────┬────────┘  └──────────┬──────────┘  └───────┬───────┘
         │                      │                      │
         └──────────────────────┼──────────────────────┘
                                │
                    ┌───────────▼───────────┐
                    │  スプレッドシート      │
                    │ （主データベース）     │
                    └───────────┬───────────┘
                                │
                    ┌───────────▼───────────┐
                    │  フロントエンド書き出し│
                    │ （JSON／CSV）          │
                    └───────────────────────┘

ツールの注記：図の「スプレッドシート」は Google スプレッドシート等を想定するが、少人数なら SQLite、共有重視なら社内DBでよい。AIエンリッチの実装例では後述の通り Gemini を使うが、要配慮情報を外部APIに送る前に必ず「個人情報・機密の取り扱い」を確認すること。社外送信が許されない資料はローカルで動くモデルやマスキングを検討する。

統一スキーマの設計

取材資料は出所がバラバラでも、横断検索のために同じ項目構造に正規化する。下のデータクラスは1件の資料を表す。日本語コメントを付した。

from dataclasses import dataclass, field
from datetime import date
from typing import Optional
from enum import Enum

class ContentType(Enum):
    # 資料の種別
    ARTICLE = '記事'
    VIDEO = '動画'
    AUDIO = '音声'
    SOCIAL = 'SNS投稿'
    DOCUMENT = '行政文書・開示文書'   # 開示請求で得た文書など
    NEWSPAPER = '新聞記事'

class ThematicCategory(Enum):
    # テーマ分類（調査テーマに合わせて自由に組み替える）
    ADMIN = '行政・政策'
    POLITICS = '政治・選挙'
    MONEY = '公金・契約・補助金'
    DISASTER = '災害・防災'
    INDUSTRY = '企業・産業'
    SOCIAL_ISSUE = '社会・人権'
    LOCAL = '地域・自治体'

class HistoricalEra(Enum):
    # 時期区分（年表・通史の整理用。調査対象期間に合わせる）
    ERA_1990s = '1990-1999'
    ERA_2000_09 = '2000-2009'
    ERA_2010_19 = '2010-2019'
    ERA_2020_25 = '2020-2025'
    ERA_2026_PRESENT = '2026-現在'

@dataclass
class ArchiveRecord:
    # 基本識別子
    id: str                              # 形式: 出所-00001（例 KAIJI-00001）
    url: str                             # 出典URL（ファイルの場合は保存パス）
    title: str

    # 本文・出所
    author: Optional[str] = None         # 作成者・記者・差出人
    publication_date: Optional[date] = None  # 作成日・公開日
    publication: Optional[str] = None    # 媒体・発行元・所管機関
    content_type: ContentType = ContentType.ARTICLE
    text: str = ''                       # 本文（OCR・反訳済みのテキスト）

    # AIで付与する項目
    summary: Optional[str] = None        # 要約
    pull_quote: Optional[str] = None     # 引用に使える一節
    categories: list[ThematicCategory] = field(default_factory=list)
    key_concepts: list[str] = field(default_factory=list)  # 主要論点・キーワード
    tags: list[str] = field(default_factory=list)
    era: Optional[HistoricalEra] = None
    scope: Optional[str] = None  # 性質（一次資料／解説／事例／背景 など）

    # エンティティ参照（知識グラフへの紐づけ）
    entities_mentioned: list[str] = field(default_factory=list)  # 登場する人物・組織のID
    related_to: list[str] = field(default_factory=list)          # 関連する資料ID
    responds_to: list[str] = field(default_factory=list)         # 反論・続報の対象資料ID

    # アーカイブ管理用メタデータ
    pdf_url: Optional[str] = None
    transcript_url: Optional[str] = None
    verified: bool = False               # 裏取り済みか
    sensitivity: str = 'normal'          # 機微度（normal / confidential / source-protected）
    processing_status: str = 'pending'   # 処理状態
    last_updated: Optional[date] = None

def generate_record_id(source: str, sequence: int) -> str:
    """出所の接頭辞を付けた一意なIDを生成する。"""
    prefixes = {
        '開示文書': 'KAIJI',
        '記者会見': 'KAIKEN',
        '新聞': 'NEWS',
        'twitter': 'TW',
        'x': 'TW',
        'youtube': 'YT',
        'インタビュー': 'INT',
    }
    prefix = prefixes.get(source.lower(), 'MISC')
    return f"{prefix}-{sequence:05d}"

sensitivity 項目は原典には無いが、日本の取材実務では資料ごとに機微度（通常／機密／取材源保護）を持たせ、外部AI送信や共有の可否を機械的に判定できるようにしておくと安全。

AIによる分類（カテゴライズ）

タクソノミー（分類辞書）にもとづく分類

あらかじめ「使ってよいカテゴリ・論点・性質」の辞書を固定し、AIにはその中から選ばせる。これで分類のブレと、AIが勝手な語を作る問題（ハルシネーション）を抑える。

# pip install google-genai
# （旧 `google-generativeai` SDK は2024年に非推奨化された。
#  新しい `google-genai` パッケージが正規の経路。以下は新しい書き方。）
import os
from google import genai
from google.genai import types
import json
from typing import Optional

# 既定は現行の Gemini 2.5 系。2026年の本番運用では
# Gemini 3 系（gemini-3-flash, gemini-3-pro）も利用可能。
# 自分のプロンプトで応答形式を検証してからモデル名を上げること。
DEFAULT_GEMINI_MODEL = 'gemini-2.5-flash'

# クライアントは1つだけ生成。GOOGLE_API_KEY を読む（api_key= で明示も可）。
_client = genai.Client(api_key=os.environ.get('GOOGLE_API_KEY'))

# 分類辞書：調査テーマに合わせて差し替える。
# ここでは行政・調査報道向けの汎用例を示す。
TAXONOMY = {
    "thematic_categories": [   # テーマ
        "行政・政策",
        "政治・選挙",
        "公金・契約・補助金",
        "災害・防災",
        "企業・産業",
        "社会・人権",
        "地域・自治体",
    ],
    "key_concepts": [          # 主要論点（調査の切り口）
        "意思決定過程",
        "随意契約・入札",
        "天下り・人事",
        "公文書の不開示・黒塗り",
        "予算・補助金の流れ",
        "利益相反",
        "前例・先例の踏襲",
        "規制と監督",
        "情報公開と説明責任",
        "個人情報の扱い",
    ],
    "scope_types": [           # 資料の性質
        "一次資料",            # 原本・公文書・録音など
        "解説・分析",
        "通史・経緯",
        "事例",
        "背景・参考",
        "当事者の証言",
    ]
}

class ArchiveCategorizer:
    def __init__(self, model: str = DEFAULT_GEMINI_MODEL, client: genai.Client = None):
        self.model = model
        self.client = client or _client

    def categorize(self, record: ArchiveRecord) -> dict:
        prompt = f"""次の資料を分析し、タクソノミーに従って分類してください。

資料:
タイトル: {record.title}
作成者: {record.author or '不明'}
日付: {record.publication_date or '不明'}
本文（先頭8000字）:
{record.text[:8000]}

タクソノミー:
{json.dumps(TAXONOMY, ensure_ascii=False, indent=2)}

次のJSONで回答してください:
{{
  "categories": ["カテゴリ1", "カテゴリ2"],  // thematic_categories から1〜3個
  "key_concepts": ["論点1", "論点2"],        // key_concepts から0〜5個
  "scope": "性質",                          // scope_types から1個
  "era": "YYYY-YYYY",                       // 年代範囲
  "tags": ["タグ1", "タグ2", "タグ3", "タグ4", "タグ5"],  // 文脈に即したキーワード5個
  "summary": "2〜3文の要約",
  "pull_quote": "本文中の最も重要な一節（原文ママ）"
}}

重要:
- カテゴリ・論点はタクソノミーにある語だけを使う
- タグは短い名詞句にする
- 要約は中心的な主張をとらえる
- pull_quote は本文からの正確な抜粋にする（創作しない）
"""

        # response_mime_type='application/json' を指定すると Gemini は
        # ```json フェンス無しの生JSONを返す。_parse_response() のフェンス
        # 除去は、フェンスを付ける旧モデル向けの保険として残してある。
        response = self.client.models.generate_content(
            model=self.model,
            contents=prompt,
            config=types.GenerateContentConfig(
                response_mime_type='application/json',
            ),
        )
        result = self._parse_response(response.text)

        # タクソノミーに照らして検証（辞書外の語を捨てる）
        result['categories'] = [c for c in result.get('categories', [])
                               if c in TAXONOMY['thematic_categories']]
        result['key_concepts'] = [c for c in result.get('key_concepts', [])
                                  if c in TAXONOMY['key_concepts']]

        return result

    def _parse_response(self, text: str) -> dict:
        """応答からJSONを取り出す。```json フェンスがあれば除去する。

        request 側で response_mime_type='application/json' を指定していれば
        Gemini は素のJSONを返す。この除去処理は旧モデルや設定未適用時の保険。
        """
        if '```json' in text:
            text = text.split('```json')[1].split('```')[0]
        elif '```' in text:
            text = text.split('```')[1].split('```')[0]

        return json.loads(text.strip())

    def validate_response(self, result: dict, text: str) -> bool:
        """AIのハルシネーション（でっち上げ）の兆候を検出する。"""
        # 全タグが同一など、応答が定型的すぎる場合は疑う
        if len(set(result.get('tags', []))) < 3:
            return False

        # pull_quote が本文に実在するか確認（創作を弾く）
        pull_quote = result.get('pull_quote', '')
        if pull_quote and pull_quote.lower() not in text.lower():
            return False

        # 要約が中身の無い定型句でないか確認
        generic_phrases = ['この記事は', 'について述べている', 'を検討している', '本資料は']
        summary = result.get('summary', '')
        if any(phrase in summary for phrase in generic_phrases):
            return False

        return True

AIの分類は下書きとして扱い、重要な資料は人が必ず確認する。特に pull_quote（引用候補）は記事に使う前に原本との一致を再確認すること。

エンティティ抽出と知識グラフ

資料に登場する人物・組織・出来事を「エンティティ」として抜き出し、それらの関係（誰が誰を批判した、どの組織がどの事業を所管する等）を記録すると、人物相関や資金の流れを横断的に追える。

エンティティの型と関係

from dataclasses import dataclass
from typing import Literal

# エンティティの種別
EntityType = Literal['Person', 'Organization', 'Work', 'Concept', 'Event', 'Location']
# 関係の種別（調査報道で使う関係を網羅）
RelationshipType = Literal[
    '言及', '批判', '引用', '論及', '発展させる', '支持',
    '設立した', '主導した', '影響を受けた',
    '所属', '掲載先', '発端', '発生場所',
    '所有する', '所有される',
    '所管する', '契約相手', '補助先',
]

@dataclass
class Entity:
    id: str                    # P-001（人）, O-001（組織）, W-001（著作物）など
    name: str
    type: EntityType
    aliases: list[str]         # 別名・表記ゆれ
    prominence: float          # 0-10：議論の深さにもとづく重要度
    mention_count: int = 0
    first_mentioned_in: str = ''  # 初出の資料ID

@dataclass
class Relationship:
    source_entity_id: str
    target_entity_id: str
    relationship_type: RelationshipType
    source_record_id: str      # この関係を裏づける資料ID（出典）
    confidence: float = 1.0    # 確信度

class EntityRegistry:
    """エンティティの名寄せ（重複排除）と正規化を行う。"""

    # 表記ゆれ・略称を正式名称に統一する辞書（調査対象に合わせて拡充）
    NORMALIZATIONS = {
        '財務省': '財務省',
        '経産省': '経済産業省',
        '経済産業省': '経済産業省',
        '厚労省': '厚生労働省',
        '厚生労働省': '厚生労働省',
        '国交省': '国土交通省',
        '都': '東京都',
        '東京都庁': '東京都',
    }

    def __init__(self):
        self.entities: dict[str, Entity] = {}
        self.name_to_id: dict[str, str] = {}

    def normalize_name(self, name: str) -> str:
        """エンティティ名を正式形に正規化する。"""
        name_lower = name.lower().strip()
        return self.NORMALIZATIONS.get(name.strip(), self.NORMALIZATIONS.get(name_lower, name.strip()))

    def find_or_create(self, name: str, entity_type: EntityType) -> Entity:
        """既存エンティティを探し、無ければ新規作成する。"""
        normalized = self.normalize_name(name)

        # すでに登録済みか確認
        if normalized.lower() in self.name_to_id:
            entity_id = self.name_to_id[normalized.lower()]
            entity = self.entities[entity_id]
            entity.mention_count += 1
            return entity

        # 新規エンティティを作成
        type_prefix = entity_type[0].upper()  # P, O, W, C, E, L
        count = sum(1 for e in self.entities.values() if e.type == entity_type)
        entity_id = f"{type_prefix}-{count + 1:04d}"

        entity = Entity(
            id=entity_id,
            name=normalized,
            type=entity_type,
            aliases=[name] if name != normalized else [],
            prominence=0.0,
            mention_count=1
        )

        self.entities[entity_id] = entity
        self.name_to_id[normalized.lower()] = entity_id

        return entity

AIによるエンティティ抽出

class EntityExtractor:
    def __init__(self, registry: EntityRegistry, model: str = DEFAULT_GEMINI_MODEL,
                 client: genai.Client = None):
        self.registry = registry
        self.model = model
        self.client = client or _client

    def extract(self, record: ArchiveRecord) -> tuple[list[Entity], list[Relationship]]:
        prompt = f"""次の資料から固有のエンティティと関係を抽出してください。

資料:
タイトル: {record.title}
本文: {record.text[:10000]}

エンティティの種別:
- Person（人物）: 政治家・官僚・記者・専門家・当事者など
- Organization（組織）: 行政機関・企業・団体・報道機関など
- Work（著作物）: 記事・報告書・契約書・公文書・調査など
- Concept（概念）: 制度・政策・論点・枠組みなど
- Event（出来事）: 会見・選挙・事件・災害など
- Location（場所）: 関連する地理的場所

関係の種別:
- 言及, 批判, 引用, 論及, 発展させる, 支持
- 設立した, 主導した, 影響を受けた
- 所属, 掲載先, 発端, 発生場所
- 所有する, 所有される
- 所管する, 契約相手, 補助先

次のJSONで回答してください:
{{
  "entities": [
    {{"name": "エンティティ名", "type": "Person|Organization|...", "prominence": 1-10}}
  ],
  "relationships": [
    {{"source": "エンティティ名", "target": "エンティティ名", "type": "関係の種別"}}
  ]
}}

重要:
- prominence（重要度）: 1-3 = 一言触れた程度, 4-6 = 論じられている, 7-10 = 中心的主題
- 単なる通りすがりの言及ではなく、実際に論じられているエンティティだけ抽出する
- 関係は同じ本文に現れるエンティティ同士でのみ結ぶ
"""

        response = self.client.models.generate_content(
            model=self.model,
            contents=prompt,
            config=types.GenerateContentConfig(
                response_mime_type='application/json',
            ),
        )
        data = json.loads(response.text)

        entities = []
        entity_name_to_obj = {}

        # エンティティの処理
        for e in data.get('entities', []):
            entity = self.registry.find_or_create(e['name'], e['type'])
            entity.prominence = max(entity.prominence, e.get('prominence', 5))
            entities.append(entity)
            entity_name_to_obj[e['name'].lower()] = entity

        # 関係の処理
        relationships = []
        for r in data.get('relationships', []):
            source = entity_name_to_obj.get(r['source'].lower())
            target = entity_name_to_obj.get(r['target'].lower())

            if source and target:
                relationships.append(Relationship(
                    source_entity_id=source.id,
                    target_entity_id=target.id,
                    relationship_type=r['type'],
                    source_record_id=record.id
                ))

        return entities, relationships

知識グラフの関係（例「Aが Bに 補助先」）は、必ず source_record_id の原資料に当たって裏取りする。AIが抽出した関係は仮説であって、報道で断定する根拠にはそのままでは使えない。

PDFによる保存版の生成

開示文書やWeb記事は消える・差し替わることがある。受領時点の内容を、メタデータ付きの読みやすいPDFとして固定保存しておく。

from reportlab.lib.pagesizes import A4
from reportlab.lib.styles import getSampleStyleSheet, ParagraphStyle
from reportlab.platypus import SimpleDocTemplate, Paragraph, Spacer, Image
from reportlab.lib.units import mm
from reportlab.pdfbase import pdfmetrics
from reportlab.pdfbase.cidfonts import UnicodeCIDFont
from pathlib import Path

class ArchivePDFGenerator:
    """保存用のアクセシブルなPDFを生成する。"""

    def __init__(self, output_dir: Path):
        self.output_dir = output_dir
        self.output_dir.mkdir(parents=True, exist_ok=True)

        # 日本語フォントを登録（reportlab同梱のCIDフォント。日本語の文字化け防止）
        pdfmetrics.registerFont(UnicodeCIDFont('HeiseiKakuGo-W5'))
        self.styles = getSampleStyleSheet()

        # 独自スタイル（日本語フォントを指定）
        self.styles.add(ParagraphStyle(
            'ArchiveTitle',
            parent=self.styles['Heading1'],
            fontName='HeiseiKakuGo-W5',
            fontSize=16,
            spaceAfter=12
        ))
        self.styles.add(ParagraphStyle(
            'ArchiveMeta',
            parent=self.styles['Normal'],
            fontName='HeiseiKakuGo-W5',
            fontSize=10,
            textColor='#666666',
            spaceAfter=6
        ))
        self.styles.add(ParagraphStyle(
            'ArchiveBody',
            parent=self.styles['Normal'],
            fontName='HeiseiKakuGo-W5',
            fontSize=11,
        ))

    def generate(self, record: ArchiveRecord) -> Path:
        output_path = self.output_dir / f"{record.id}.pdf"

        doc = SimpleDocTemplate(
            str(output_path),
            pagesize=A4,                      # 日本では A4 が標準
            title=record.title,
            author=record.author or '不明',
            subject=f"アーカイブ資料 {record.id}"
        )

        story = []

        # タイトル
        story.append(Paragraph(record.title, self.styles['ArchiveTitle']))

        # メタデータ欄
        meta_lines = [
            f"<b>作成者:</b> {record.author or '不明'}",
            f"<b>日付:</b> {record.publication_date or '不明'}",
            f"<b>出所:</b> {record.publication or '不明'}",
            f"<b>URL:</b> {record.url}",
            f"<b>アーカイブID:</b> {record.id}",
        ]
        for line in meta_lines:
            story.append(Paragraph(line, self.styles['ArchiveMeta']))

        story.append(Spacer(1, 6 * mm))

        # 要約（あれば）
        if record.summary:
            story.append(Paragraph("<b>要約:</b>", self.styles['ArchiveTitle']))
            story.append(Paragraph(record.summary, self.styles['ArchiveBody']))
            story.append(Spacer(1, 6 * mm))

        # 本文
        story.append(Paragraph("<b>全文:</b>", self.styles['ArchiveTitle']))

        # 段落ごとに分割して追加
        paragraphs = record.text.split('\n\n')
        for para in paragraphs:
            if para.strip():
                story.append(Paragraph(para.strip(), self.styles['ArchiveBody']))
                story.append(Spacer(1, 3 * mm))

        # PDFを生成
        doc.build(story)

        return output_path

原典は米国の letter 判・英語フォント前提だったため、日本向けに A4判と日本語CIDフォントの登録を加えた。証拠としての保存では、改ざんしていないことを示すため、受領した原本（生のPDF・スキャン）も別に保管し、生成版と取り違えないこと。

データ品質と検証

アーカイブは欠損や不整合があると検索の信頼性が落ちる。投入前に各資料を機械的に点検する。

from dataclasses import dataclass
from typing import Callable

@dataclass
class ValidationResult:
    field: str
    valid: bool
    message: str
    severity: Literal['error', 'warning', 'info']

class ArchiveValidator:
    """アーカイブ資料の完全性・整合性を検証する。"""

    REQUIRED_FIELDS = ['id', 'url', 'title', 'text']          # 必須
    CRITICAL_FIELDS = ['publication_date', 'author', 'summary']  # 重要（無くても投入は可）
    OPTIONAL_FIELDS = ['categories', 'tags', 'pull_quote']    # 任意

    def validate(self, record: ArchiveRecord) -> list[ValidationResult]:
        results = []

        # 必須項目
        for field in self.REQUIRED_FIELDS:
            value = getattr(record, field, None)
            if not value:
                results.append(ValidationResult(
                    field=field,
                    valid=False,
                    message=f"必須項目 '{field}' が欠落",
                    severity='error'
                ))

        # 重要項目（あるべきだが処理は止めない）
        for field in self.CRITICAL_FIELDS:
            value = getattr(record, field, None)
            if not value:
                results.append(ValidationResult(
                    field=field,
                    valid=False,
                    message=f"重要項目 '{field}' が欠落",
                    severity='warning'
                ))

        # 本文の長さチェック（OCR失敗などを検出）
        if record.text and len(record.text) < 100:
            results.append(ValidationResult(
                field='text',
                valid=False,
                message=f"本文が異常に短い（{len(record.text)}字）",
                severity='warning'
            ))

        # 日付の妥当性チェック
        if record.publication_date:
            try:
                _ = record.publication_date.isoformat()
            except (AttributeError, ValueError):
                results.append(ValidationResult(
                    field='publication_date',
                    valid=False,
                    message="日付の形式が不正",
                    severity='error'
                ))

        # カテゴリの妥当性チェック（辞書外の値を検出）
        for cat in record.categories:
            if cat not in ThematicCategory:
                results.append(ValidationResult(
                    field='categories',
                    valid=False,
                    message=f"未知のカテゴリ: {cat}",
                    severity='warning'
                ))

        return results

    def is_complete(self, record: ArchiveRecord) -> bool:
        """重要項目がすべて埋まっているか（エラーが無いか）を判定する。"""
        results = self.validate(record)
        errors = [r for r in results if r.severity == 'error']
        return len(errors) == 0

統合ワークフロー

収集 → 記録作成 → AI分類 → エンティティ抽出 → PDF保存 → 検証 → 保存、を一連で回すオーケストレータ。

class ArchiveWorkflow:
    """アーカイブ処理パイプライン全体を統括する。"""

    def __init__(self, config: Config):
        self.scraper = ScrapingCascade()                      # 収集（複数手段を順に試す）
        self.categorizer = ArchiveCategorizer()               # AI分類
        self.entity_registry = EntityRegistry()               # エンティティ台帳
        self.entity_extractor = EntityExtractor(self.entity_registry)
        self.pdf_generator = ArchivePDFGenerator(config.PDF_DIR)
        self.sheets_service = SheetsService(config.CREDENTIALS_PATH)  # 保存先（スプレッドシート等）
        self.validator = ArchiveValidator()
        self.progress = ProgressTracker(config.PROGRESS_FILE)  # 進捗（再開用）

    def process_url(self, url: str, record_id: str) -> ArchiveRecord:
        """1件のURLをパイプライン全体に通す。"""

        # 1. 本文を収集
        result = self.scraper.fetch(url)
        if not result:
            raise ValueError(f"収集に失敗: {url}")

        # 2. 初期レコードを作成
        record = ArchiveRecord(
            id=record_id,
            url=url,
            title=result.title,
            text=result.content
        )

        # 3. AI分類
        categories = self.categorizer.categorize(record)
        record.summary = categories.get('summary')
        record.pull_quote = categories.get('pull_quote')
        record.categories = categories.get('categories', [])
        record.key_concepts = categories.get('key_concepts', [])
        record.tags = categories.get('tags', [])
        record.era = categories.get('era')
        record.scope = categories.get('scope')

        # 4. エンティティ抽出
        entities, relationships = self.entity_extractor.extract(record)
        record.entities_mentioned = [e.id for e in entities]

        # 5. PDF生成
        pdf_path = self.pdf_generator.generate(record)
        record.pdf_url = str(pdf_path)

        # 6. 検証
        validation = self.validator.validate(record)
        record.verified = self.validator.is_complete(record)
        record.processing_status = 'completed'

        return record

    def run_batch(self, input_csv: Path):
        """入力CSVのURLをすべて処理する（途中再開に対応）。"""
        for row in read_input(input_csv):
            if self.progress.is_processed(row.id):
                continue   # 処理済みは飛ばす

            try:
                record = self.process_url(row.url, row.id)
                self.sheets_service.append_row(self.worksheet, record_to_row(record))
                self.progress.mark_processed(row.id)
            except Exception as e:
                self.progress.log_error(row.id, str(e))  # 失敗は記録して継続

ScrapingCascade SheetsService ProgressTracker Config などは環境依存の補助クラス（実装は環境に合わせて用意する）。要配慮情報を含む資料を外部のクラウド保存先やAIに送る前に、機微度（sensitivity）で振り分け、社外送信不可のものはこのパイプラインから除外する分岐を入れること。

フロントエンド向けの書き出し

検索サイトや可視化ツールに渡すため、資料・エンティティ・関係を扱いやすい形式で出力する。

import json
from pathlib import Path

def export_for_frontend(records: list[ArchiveRecord], output_dir: Path):
    """アーカイブデータをフロントエンド向け形式で書き出す。"""

    # 本体（アーカイブ全件）のJSON
    archive_data = {
        'metadata': {
            'total_records': len(records),
            'last_updated': datetime.now().isoformat(),
            'schema_version': '2.0'
        },
        'records': [asdict(r) for r in records]
    }

    # ensure_ascii=False で日本語をエスケープせず可読にする
    (output_dir / 'archive-data.json').write_text(
        json.dumps(archive_data, ensure_ascii=False, indent=2, default=str),
        encoding='utf-8'
    )

    # エンティティの書き出し
    entities_data = [asdict(e) for e in entity_registry.entities.values()]
    (output_dir / 'entities.json').write_text(
        json.dumps(entities_data, ensure_ascii=False, indent=2),
        encoding='utf-8'
    )

    # 関係（知識グラフのエッジ）の書き出し
    relationships_data = [asdict(r) for r in all_relationships]
    (output_dir / 'relationships.json').write_text(
        json.dumps(relationships_data, ensure_ascii=False, indent=2),
        encoding='utf-8'
    )

    # 表計算ソフトと相性のよいCSVも出力
    records_df = pd.DataFrame([asdict(r) for r in records])
    records_df.to_csv(output_dir / 'archive_records.csv', index=False,
                      encoding='utf-8-sig')  # Excelで開く場合はBOM付きUTF-8が無難

原典の英語前提に対し、日本語が文字化けしないよう JSON は ensure_ascii=False、CSV は Excel 対応の utf-8-sig を指定した。

個人情報・機密の取り扱い（日本の取材実務）

取材資料のアーカイブは、個人情報・要配慮情報・取材源につながる情報の塊になりやすい。便利さと引き換えに漏えい・特定のリスクを抱える点を常に意識する。

• 外部AI・クラウドへの送信：本文を外部API（Gemini等）に送ると、その内容が事業者の処理を経る。要配慮情報・取材源を特定し得る情報を送ってよいかは、各社の規程・契約（学習利用の有無等）を確認してから判断する。送れない資料は、固有名のマスキング、ローカルで動くモデルの利用、AI処理の対象外とする、のいずれかで対応する。
• 取材源の保護：差出人・連絡先・内部関係者を示す情報は、sensitivity を source-protected にしてアクセス・共有・書き出しの対象から外す運用にする。知識グラフの関係づけは、意図せず情報源を特定し得るため特に注意。
• 個人情報保護法：取材・報道目的の個人情報の取り扱いには適用除外の規定があるが、保管・共有の安全管理は別途求められる。アクセス権限・暗号化・保存期間・廃棄の方針を決めておく。
• 黒塗りの再現に注意：開示文書をOCR・テキスト化すると、黒塗り（不開示）部分が周辺情報から推測しやすくなることがある。テキスト化後のデータも原本と同じ機微度で扱う。
• 保存と廃棄：何を、どこに、誰がアクセスできる形で、いつまで保管するかをプロジェクト単位で決める。共同取材では共有範囲を明文化する。
• 裏取り原則：AIが付けた要約・分類・エンティティ・関係は、すべて下書き。報道に用いる事実は必ず原資料に当たって確認する。

ここに挙げた制度・運用は一般的な整理であり、法的助言ではない。具体的な可否は所属社の法務・デスク、必要に応じて弁護士に確認すること。

---

出典・ライセンス（帰属表示）

本スキルは research-toolkit（by Joe Amditis, MIT License, © 2025 Joe Amditis, https://github.com/jamditis/claude-skills-journalism ）の digital-archive（Jay Rosen Digital Archive プロジェクトのパターンに由来）を 原典の構成（アーカイブ全体構成・統一スキーマ・AI分類・エンティティ抽出と知識グラフ・PDF保存・データ検証・統合ワークフロー・書き出し）に沿って日本語化し、 取材資料／調査報道の資料管理という文脈に再構成したものです。 原典の Jay Rosen 個人アーカイブ固有のタクソノミー（ジャーナリズム理論の概念群）や米国メディア名の正規化例は、日本の調査報道向けの汎用的なテーマ・論点・機関名に置き換え、 PDF生成・文字コードは日本語（A4・CIDフォント・UTF-8）に適応し、日本の個人情報・取材倫理に関する注記を加えました。 コードは機能を保持し、コメントのみ日本語化しています。第三者の著作物本文は転載していません。原作者の公認・推奨を受けたものではありません。 MIT ライセンス全文は配布元のリポジトリを参照してください。