hunjing/backend/app/services/imports.py

"""资料导入服务（api-spec §8）。

5a：结构解析（消息数组 / 段落 / 字段）落 ``import_items``，不调模型；媒体资料
（``voice_sample``）仅存元数据与授权记录（spec §18 仅预留，不落盘不解析）。
候选抽取（档案 / 记忆 / 人格）与应用留 5b / 5c。
"""

import hashlib

from fastapi import UploadFile
from sqlalchemy.ext.asyncio import AsyncSession

from app.core.config import settings
from app.core.errors import AppError
from app.core.time import utc_now
from app.db.session import AsyncSessionLocal
from app.models.imports import CharacterImportJob, ImportItem
from app.models.memory import Memory
from app.modules.imports import parse_import
from app.repositories import candidates as candidates_repo
from app.repositories import imports as imports_repo
from app.repositories.json_utils import dumps, loads_dict
from app.schemas.candidates import (
    BatchReviewRequest,
    BatchReviewResult,
    MemoryCandidateResponse,
    ProfileCandidateResponse,
)
from app.schemas.common import Page
from app.schemas.imports import (
    CreateImportRequest,
    ImportItemResponse,
    ImportJobResponse,
)
from app.services.audit import record_audit_event
from app.services.characters import get_owned_character
from app.services.serializers import (
    import_item_to_response,
    import_job_to_response,
    memory_candidate_to_response,
    profile_candidate_to_response,
)

_TEXT_MIME_FORMAT: dict[str, str] = {
    "text/plain": "text",
    "text/markdown": "markdown",
    "application/json": "json",
    "text/csv": "csv",
    "application/yaml": "yaml",
    "text/yaml": "yaml",
}
_AUDIO_FORMAT: dict[str, str] = {
    "audio/wav": "wav",
    "audio/mpeg": "mp3",
    "audio/mp4": "m4a",
    "audio/x-m4a": "m4a",
    "audio/flac": "flac",
}


async def create_import_from_json(
    character_id: str,
    payload: CreateImportRequest,
    owner_scope: str = settings.default_owner_scope,
) -> ImportJobResponse:
    """JSON 提交导入（粘贴文本 / 结构化消息 / 手动设定）。"""
    async with AsyncSessionLocal() as session:
        character = await get_owned_character(session, character_id, owner_scope=owner_scope)
        job = CharacterImportJob(
            character_id=character.id,
            material_type=payload.material_type,
            source_type="pasted_text",
            title=payload.title,
            status="pending",
            format=payload.format,
            source_description=payload.source_note,
            raw_content=payload.content,
            checksum=_checksum(payload.content),
        )
        await imports_repo.add_job(session, job)
        await _parse_into_items(session, job, payload.content)
        await _audit(session, character.id, "import.created", job.id, payload.material_type)
        await session.commit()
        await session.refresh(job)
        return import_job_to_response(job)


async def create_import_from_upload(
    character_id: str,
    file: UploadFile,
    *,
    title: str | None = None,
    source_description: str | None = None,
    speaker_label: str | None = None,
    consent_status: str = "unknown",
    usage_scope: str = "reference_only",
    owner_scope: str = settings.default_owner_scope,
) -> ImportJobResponse:
    """文件上传导入（文本类解析为 items；音频类仅存元数据，不落盘）。"""
    content_bytes = await file.read()
    mime = file.content_type or "application/octet-stream"
    filename = file.filename or "upload"

    async with AsyncSessionLocal() as session:
        character = await get_owned_character(session, character_id, owner_scope=owner_scope)
        if mime in _AUDIO_FORMAT:
            job = CharacterImportJob(
                character_id=character.id,
                material_type="voice_sample",
                source_type="media_upload",
                title=title,
                status="uploaded",
                format=_AUDIO_FORMAT[mime],
                media_type="audio",
                mime_type=mime,
                source_description=source_description,
                speaker_label=speaker_label,
                consent_status=consent_status,
                usage_scope=usage_scope,
                checksum=_checksum_bytes(content_bytes),
            )
            await imports_repo.add_job(session, job)
        else:
            try:
                content = content_bytes.decode("utf-8")
            except UnicodeDecodeError as exc:
                raise AppError("unsupported_media_type", "仅支持 UTF-8 文本文件。") from exc
            job = CharacterImportJob(
                character_id=character.id,
                material_type=_infer_material_type(filename, mime),
                source_type="uploaded_file",
                title=title or filename,
                status="pending",
                format=_TEXT_MIME_FORMAT.get(mime, "text"),
                mime_type=mime,
                source_description=source_description,
                raw_content=content,
                checksum=_checksum_bytes(content_bytes),
            )
            await imports_repo.add_job(session, job)
            await _parse_into_items(session, job, content)
        await _audit(session, character.id, "import.created", job.id, job.material_type)
        await session.commit()
        await session.refresh(job)
        return import_job_to_response(job)


async def get_import(
    character_id: str,
    import_id: str,
    owner_scope: str = settings.default_owner_scope,
) -> ImportJobResponse:
    async with AsyncSessionLocal() as session:
        character = await get_owned_character(session, character_id, owner_scope=owner_scope)
        job = await imports_repo.get_job(session, import_id)
        if job is None or job.character_id != character.id:
            raise AppError("not_found", "导入任务不存在。")
        return import_job_to_response(job)


async def list_jobs_page(
    character_id: str,
    *,
    page: int = 1,
    page_size: int = 20,
    owner_scope: str = settings.default_owner_scope,
) -> Page[ImportJobResponse]:
    async with AsyncSessionLocal() as session:
        character = await get_owned_character(session, character_id, owner_scope=owner_scope)
        safe_page = max(page, 1)
        safe_size = min(max(page_size, 1), 100)
        items, total = await imports_repo.list_jobs(
            session, character.id, page=safe_page, page_size=safe_size
        )
        return Page(
            items=[import_job_to_response(item) for item in items],
            page=safe_page,
            page_size=safe_size,
            total=total,
        )


async def list_import_items(
    character_id: str,
    import_id: str,
    *,
    page: int = 1,
    page_size: int = 20,
    owner_scope: str = settings.default_owner_scope,
) -> Page[ImportItemResponse]:
    async with AsyncSessionLocal() as session:
        character = await get_owned_character(session, character_id, owner_scope=owner_scope)
        job = await imports_repo.get_job(session, import_id)
        if job is None or job.character_id != character.id:
            raise AppError("not_found", "导入任务不存在。")
        safe_page = max(page, 1)
        safe_size = min(max(page_size, 1), 100)
        items, total = await imports_repo.list_items(
            session, import_id, page=safe_page, page_size=safe_size
        )
        return Page(
            items=[import_item_to_response(item) for item in items],
            page=safe_page,
            page_size=safe_size,
            total=total,
        )


async def _parse_into_items(session: AsyncSession, job: CharacterImportJob, content: str) -> None:
    """结构解析为 import_items 并回填 job 状态；纯结构，不调模型。"""
    job.status = "parsing"
    await session.flush()
    drafts = parse_import(job.material_type, job.format, content=content)
    items = [
        ImportItem(
            import_job_id=job.id,
            character_id=job.character_id,
            item_type=draft.item_type,
            role=draft.role,
            sequence=idx,
            content=draft.content,
            metadata_json=dumps(draft.metadata),
        )
        for idx, draft in enumerate(drafts)
    ]
    await imports_repo.add_items(session, items)
    job.item_count = len(items)
    job.status = "parsed" if items else "uploaded"


async def _audit(
    session: AsyncSession,
    character_id: str,
    action: str,
    resource_id: str,
    summary: str,
) -> None:
    await record_audit_event(
        session,
        action=action,
        character_id=character_id,
        actor="user",
        decision="auto",
        input_summary=summary,
        resource_type="import_job",
        resource_id=resource_id,
    )


def _checksum(text: str) -> str:
    return hashlib.sha256(text.encode("utf-8")).hexdigest()


def _checksum_bytes(data: bytes) -> str:
    return hashlib.sha256(data).hexdigest()


def _infer_material_type(filename: str, mime: str) -> str:
    name = filename.lower()
    if "chat" in name or "history" in name:
        return "chat_history"
    if mime == "application/json":
        return "other"
    return "background_story"


# —— 候选抽取触发与列表（5b，异步后台）——

_EXTRACTABLE_STATUSES = frozenset({"parsed", "uploaded", "reviewing"})


async def _get_owned_job(
    session: AsyncSession, character_id: str, import_id: str
) -> CharacterImportJob:
    job = await imports_repo.get_job(session, import_id)
    if job is None or job.character_id != character_id:
        raise AppError("not_found", "导入任务不存在。")
    return job


async def _reload_job(import_id: str) -> ImportJobResponse:
    async with AsyncSessionLocal() as session:
        job = await imports_repo.get_job(session, import_id)
        return import_job_to_response(job)


async def trigger_profile_extraction(
    character_id: str,
    import_id: str,
    owner_scope: str = settings.default_owner_scope,
) -> ImportJobResponse:
    return await _trigger_extraction(
        character_id, import_id, "extracting_profile", "profile", owner_scope=owner_scope
    )


async def trigger_memory_extraction(
    character_id: str,
    import_id: str,
    owner_scope: str = settings.default_owner_scope,
) -> ImportJobResponse:
    return await _trigger_extraction(
        character_id, import_id, "extracting_memory", "memory", owner_scope=owner_scope
    )


async def trigger_personality_extraction(
    character_id: str,
    import_id: str,
    owner_scope: str = settings.default_owner_scope,
) -> ImportJobResponse:
    return await _trigger_extraction(
        character_id,
        import_id,
        "generating_personality_proposals",
        "personality",
        owner_scope=owner_scope,
    )


async def _trigger_extraction(
    character_id: str,
    import_id: str,
    status: str,
    kind: str,
    owner_scope: str = settings.default_owner_scope,
) -> ImportJobResponse:
    """校验 job → 置 extracting_* → 启动后台抽取 → 返回 job 当前状态。"""
    async with AsyncSessionLocal() as session:
        character = await get_owned_character(session, character_id, owner_scope=owner_scope)
        job = await _get_owned_job(session, character.id, import_id)
        if job.status not in _EXTRACTABLE_STATUSES:
            raise AppError("conflict", f"任务状态 {job.status} 不可触发抽取。")
        job.status = status
        await session.commit()

    from app.application.imports import (
        start_memory_extraction,
        start_personality_extraction,
        start_profile_extraction,
    )

    starters = {
        "profile": start_profile_extraction,
        "memory": start_memory_extraction,
        "personality": start_personality_extraction,
    }
    starters[kind](import_id)
    return await _reload_job(import_id)


async def list_profile_candidates(
    character_id: str,
    import_id: str,
    *,
    status: str | None = None,
    owner_scope: str = settings.default_owner_scope,
) -> list[ProfileCandidateResponse]:
    async with AsyncSessionLocal() as session:
        character = await get_owned_character(session, character_id, owner_scope=owner_scope)
        await _get_owned_job(session, character.id, import_id)
        items = await candidates_repo.list_profile_candidates(
            session, character.id, import_job_id=import_id, status=status
        )
        return [profile_candidate_to_response(c) for c in items]


async def list_memory_candidates(
    character_id: str,
    import_id: str,
    *,
    status: str | None = None,
    owner_scope: str = settings.default_owner_scope,
) -> list[MemoryCandidateResponse]:
    async with AsyncSessionLocal() as session:
        character = await get_owned_character(session, character_id, owner_scope=owner_scope)
        await _get_owned_job(session, character.id, import_id)
        items = await candidates_repo.list_memory_candidates(
            session, character.id, import_job_id=import_id, status=status
        )
        return [memory_candidate_to_response(c) for c in items]


# —— 候选审核应用（5c）——


async def batch_review_profile_candidates(
    character_id: str,
    payload: BatchReviewRequest,
    owner_scope: str = settings.default_owner_scope,
) -> BatchReviewResult:
    """批量审核档案候选：approve 写入人物档案，reject 标记驳回，非 pending 跳过。"""
    async with AsyncSessionLocal() as session:
        character = await get_owned_character(session, character_id, owner_scope=owner_scope)
        profile = loads_dict(character.profile_json)
        applied = rejected = skipped = 0
        for decision in payload.decisions:
            candidate = await candidates_repo.get_profile_candidate(session, decision.candidate_id)
            if (
                candidate is None
                or candidate.character_id != character.id
                or candidate.status != "pending"
            ):
                skipped += 1
                continue
            if decision.decision == "approve":
                profile[candidate.field] = candidate.value
                if candidate.field == "display_name":
                    character.display_name = candidate.value
                candidate.status = "applied"
                applied += 1
            elif decision.decision == "reject":
                candidate.status = "rejected"
                rejected += 1
            else:
                skipped += 1
                continue
            candidate.reviewed_at = utc_now()
        character.profile_json = dumps(profile)
        await _audit(
            session,
            character.id,
            "import.profile_candidate_reviewed",
            character.id,
            f"applied={applied} rejected={rejected}",
        )
        await session.commit()
        return BatchReviewResult(applied=applied, rejected=rejected, skipped=skipped)


async def batch_review_memory_candidates(
    character_id: str,
    payload: BatchReviewRequest,
    owner_scope: str = settings.default_owner_scope,
) -> BatchReviewResult:
    """批量审核记忆候选：approve 创建记忆，reject 标记驳回。"""
    async with AsyncSessionLocal() as session:
        character = await get_owned_character(session, character_id, owner_scope=owner_scope)
        applied = rejected = skipped = 0
        for decision in payload.decisions:
            candidate = await candidates_repo.get_memory_candidate(session, decision.candidate_id)
            if (
                candidate is None
                or candidate.character_id != character.id
                or candidate.status != "pending"
            ):
                skipped += 1
                continue
            if decision.decision == "approve":
                session.add(
                    Memory(
                        character_id=character.id,
                        type=candidate.type,
                        title=candidate.title,
                        content=candidate.content,
                        source="user_explicit",
                        confidence=candidate.confidence,
                        importance=candidate.importance,
                        sensitivity=candidate.sensitivity,
                        evidence_ids_json=candidate.evidence_ids_json,
                        status="active",
                    )
                )
                candidate.status = "applied"
                applied += 1
            elif decision.decision == "reject":
                candidate.status = "rejected"
                rejected += 1
            else:
                skipped += 1
                continue
            candidate.reviewed_at = utc_now()
        await _audit(
            session,
            character.id,
            "import.memory_candidate_reviewed",
            character.id,
            f"applied={applied} rejected={rejected}",
        )
        await session.commit()
        return BatchReviewResult(applied=applied, rejected=rejected, skipped=skipped)