Fine-tuning/backend/app/services/dataset_service.py

import asyncio
import json
import uuid
from datetime import datetime, timezone
from pathlib import Path
from typing import Any


from fastapi import UploadFile

from app.config import get_settings
from app.core.background_tasks import background_task_manager
from app.core.db import async_session, DatasetRecord, DatasetDownloadTask
from app.core.logging import logger
from app.schemas.dataset import DatasetDownloadRequest, DatasetDownloadResponse
from sqlalchemy import select

settings = get_settings()

# Known metadata filenames that are NOT training data
META_FILENAMES = frozenset({
    "configuration.json", "configuration.yaml", "README.md",
    ".mdl", ".msc", ".mv", "model_index.json", "generation_config.json",
    "special_tokens_map.json", "tokenizer_config.json",
    "added_tokens.json", "vocab.json", "merges.txt",
    "config.json", "preprocessor_config.json",
    "dataset_infos.json", "dataset_info.json",
    "state.json", "card_data.json",
})

# File size threshold: files smaller than this (bytes) are likely metadata
META_SIZE_THRESHOLD = 500


def _is_training_data_file(path: Path) -> bool:
    """判断文件是否可能是训练数据文件（而非配置/元数据）。"""
    if path.name in META_FILENAMES:
        return False
    if path.suffix in (".jsonl", ".parquet", ".csv"):
        # 小文件可能是元数据（如 ModelScope CLI 生成的 data.jsonl 只有几十字节）
        if path.stat().st_size < META_SIZE_THRESHOLD:
            return False
        return True
    if path.suffix == ".json":
        # 小 JSON 文件通常是配置
        if path.stat().st_size < META_SIZE_THRESHOLD:
            return False
        # 尝试读取首行判断格式
        try:
            first_line = path.read_text(encoding="utf-8", errors="ignore").splitlines()[0].strip()
            obj = json.loads(first_line)
            # 如果有 input/output/conversation/instruction 等字段，则是训练数据
            if isinstance(obj, dict):
                data_keys = {"input", "output", "conversations", "instruction", "prompt",
                             "text", "completion", "source", "target", "query", "response"}
                if data_keys & set(obj.keys()):
                    return True
                # 如果只有 framework/task/model_type 等字段，则是元数据
                meta_keys = {"framework", "task", "license", "base_model", "model_type",
                             "language", "domains", "tags", "authors"}
                if meta_keys & set(obj.keys()):
                    return False
            return True  # 大 JSON 文件默认是数据
        except Exception:
            return False
    # 无后缀文件：尝试读取判断是否为 JSON/JSONL
    if not path.suffix:
        try:
            first_line = path.read_text(encoding="utf-8", errors="ignore").splitlines()[0].strip()
            json.loads(first_line)
            return True
        except Exception:
            return False
    return False


async def download_dataset(req: DatasetDownloadRequest) -> DatasetDownloadResponse:
    """启动数据集下载后台任务，立即返回 task_id。"""
    task_id = str(uuid.uuid4())

    # 写 DB
    record = DatasetDownloadTask(
        id=task_id,
        dataset_id=req.dataset_id,
        use_modelscope=1 if req.use_modelscope else 0,
        status="pending",
    )
    async with async_session() as session:
        session.add(record)
        await session.commit()

    # 注册并启动
    background_task_manager.register_task(task_id, "dataset_download", {"dataset_id": req.dataset_id})
    background_task_manager.run(
        task_id, "dataset_download", _execute_dataset_download(task_id, req)
    )

    logger.info(f"Dataset download task started: {req.dataset_id} (task_id={task_id})")
    return DatasetDownloadResponse(
        dataset_id=req.dataset_id, status="pending", path=task_id
    )


async def _execute_dataset_download(task_id: str, req: DatasetDownloadRequest) -> dict:
    """后台执行数据集下载。"""
    try:
        if req.use_modelscope:
            ds_dir, jsonl_path, record_count = await asyncio.to_thread(
                _download_modelscope_dataset, req.dataset_id
            )
        else:
            from datasets import load_dataset

            ds = load_dataset(req.dataset_id)
            ds_dir = settings.processed_dir / f"hf_{req.dataset_id.replace('/', '_')}"
            ds_dir.mkdir(parents=True, exist_ok=True)
            if "train" in ds:
                split = ds["train"]
            else:
                split = ds[list(ds.keys())[0]]
            output_path = ds_dir / "data.jsonl"
            with open(output_path, "w", encoding="utf-8") as f:
                for item in split:
                    f.write(json.dumps(item, ensure_ascii=False) + "\n")
            jsonl_path = output_path
            record_count = len(split) if hasattr(split, "__len__") else 0

        db_record = DatasetRecord(
            id=str(uuid.uuid4()),
            name=req.dataset_id,
            format="jsonl",
            record_count=record_count,
            file_path=str(jsonl_path),
            created_at=datetime.utcnow(),
        )
        async with async_session() as session:
            session.add(db_record)
            await session.commit()

        await _update_dataset_download_status(task_id, "completed", path=str(jsonl_path), record_count=record_count)

        logger.info(f"Dataset downloaded: {req.dataset_id} ({record_count} records)")
        return {"path": str(jsonl_path), "record_count": record_count}

    except Exception as e:
        logger.error(f"Dataset download failed: {e}")
        await _update_dataset_download_status(task_id, "failed", error=str(e))
        return {"error": str(e)}


async def _update_dataset_download_status(task_id: str, status: str, path: str = None, error: str = None, record_count: int = 0):
    async with async_session() as session:
        result = await session.execute(select(DatasetDownloadTask).where(DatasetDownloadTask.id == task_id))
        record = result.scalar_one_or_none()
        if record:
            record.status = status
            if path:
                record.path = path
            if error:
                record.error = error
            if record_count:
                record.record_count = record_count
            if status in ("completed", "failed"):
                record.finished_at = datetime.utcnow()
            await session.commit()

    background_task_manager.update_task(
        task_id, status=status, path=path, error=error, record_count=record_count,
    )


async def get_dataset_download_status(task_id: str) -> dict[str, Any]:
    async with async_session() as session:
        result = await session.execute(select(DatasetDownloadTask).where(DatasetDownloadTask.id == task_id))
        record = result.scalar_one_or_none()
        if record:
            return {
                "task_id": record.id,
                "dataset_id": record.dataset_id,
                "status": record.status,
                "use_modelscope": bool(record.use_modelscope),
                "path": record.path,
                "error": record.error,
                "record_count": record.record_count,
                "created_at": record.created_at.isoformat() if record.created_at else "",
            }
    mem = background_task_manager.get_task(task_id)
    if mem:
        return {
            "task_id": task_id,
            "dataset_id": mem.get("dataset_id", ""),
            "status": mem["status"],
            "error": mem.get("error"),
            "record_count": mem.get("record_count", 0),
        }
    return {"task_id": task_id, "status": "not_found"}


async def list_dataset_downloads() -> list[dict[str, Any]]:
    async with async_session() as session:
        result = await session.execute(
            select(DatasetDownloadTask).order_by(DatasetDownloadTask.created_at.desc())
        )
        records = result.scalars().all()
    return [
        {
            "task_id": r.id,
            "dataset_id": r.dataset_id,
            "status": r.status,
            "use_modelscope": bool(r.use_modelscope),
            "path": r.path,
            "error": r.error,
            "record_count": r.record_count,
            "created_at": r.created_at.isoformat() if r.created_at else "",
        }
        for r in records
    ]


async def cancel_dataset_download(task_id: str) -> dict[str, Any]:
    background_task_manager.cancel_task(task_id)
    async with async_session() as session:
        result = await session.execute(select(DatasetDownloadTask).where(DatasetDownloadTask.id == task_id))
        record = result.scalar_one_or_none()
        if record and record.status in ("pending", "downloading"):
            record.status = "cancelled"
            record.error = "Cancelled by user"
            record.finished_at = datetime.utcnow()
            await session.commit()
    return {"task_id": task_id, "status": "cancelled"}


async def recover_stale_downloads() -> None:
    async with async_session() as session:
        result = await session.execute(
            select(DatasetDownloadTask).where(
                DatasetDownloadTask.status.in_(["pending", "downloading"])
            )
        )
        records = result.scalars().all()
        for record in records:
            record.status = "failed"
            record.error = "Server restarted, task interrupted"
            record.finished_at = datetime.utcnow()
        if records:
            await session.commit()
            logger.info(f"Recovered {len(records)} stale dataset download tasks")


def _download_modelscope_dataset(dataset_id: str) -> tuple[Path, Path, int]:
    """用 modelscope CLI 下载数据集，完全绕过 datasets 库，避免版本兼容问题。"""
    import subprocess

    ds_dir = settings.processed_dir / f"ms_{dataset_id.replace('/', '_')}"
    ds_dir.mkdir(parents=True, exist_ok=True)

    # 使用 CLI 方式下载，避免 snapshot_download API 的路径问题
    cmd = ["modelscope", "download", "--dataset", dataset_id, "--local_dir", str(ds_dir)]
    logger.info(f"Running: {' '.join(cmd)}")
    proc = subprocess.run(cmd, capture_output=True, text=True, timeout=600)
    if proc.returncode != 0:
        logger.error(f"ModelScope CLI download failed (code={proc.returncode}): {proc.stderr[:500]}")
        raise RuntimeError(f"ModelScope download failed: {proc.stderr[:500]}")

    # 扫描下载目录中的所有文件
    all_files = [p for p in ds_dir.rglob("*") if p.is_file()]
    logger.info(f"ModelScope CLI downloaded {len(all_files)} files to {ds_dir}")

    # 识别训练数据文件
    data_files = [f for f in all_files if _is_training_data_file(f)]

    if not data_files:
        fallback = [f for f in all_files if f.suffix in (".json", ".jsonl") and f.name not in META_FILENAMES and f.name != "README.md"]
        logger.warning(f"No training data files found in {dataset_id}. "
                       f"Available JSON files: {[f.name for f in fallback]}")
        if fallback:
            data_files = fallback
        else:
            # 如果还是没有，列出所有文件供排查
            logger.error(f"All downloaded files: {[str(f.relative_to(ds_dir)) for f in all_files]}")
            raise ValueError(f"No JSON/JSONL data files found in dataset {dataset_id}. "
                             f"Available files: {[f.name for f in all_files]}")

    # 按文件大小排序，取最大的文件作为训练数据（真正的数据集通常是最大的）
    target = sorted(data_files, key=lambda f: f.stat().st_size, reverse=True)[0]

    logger.info(f"Selected data file: {target} (size={target.stat().st_size})")

    # 读取并统一转为 JSONL
    jsonl_path = ds_dir / "data.jsonl"
    record_count = 0
    content = target.read_text(encoding="utf-8")

    if target.suffix == ".jsonl" or not target.suffix:
        # JSONL 或无后缀文件：逐行解析
        records = []
        for line in content.splitlines():
            line = line.strip()
            if not line:
                continue
            try:
                records.append(json.loads(line))
            except json.JSONDecodeError:
                records = json.loads(content)
                if not isinstance(records, list):
                    records = [records]
                break
    elif target.suffix == ".json":
        # JSON 文件：先尝试 JSON 数组，失败再逐行解析（可能是 JSONL 格式）
        try:
            records = json.loads(content)
            if not isinstance(records, list):
                records = [records]
        except json.JSONDecodeError:
            records = []
            for line in content.splitlines():
                line = line.strip()
                if not line:
                    continue
                try:
                    records.append(json.loads(line))
                except json.JSONDecodeError:
                    continue

    with open(jsonl_path, "w", encoding="utf-8") as f:
        for item in records:
            f.write(json.dumps(item, ensure_ascii=False) + "\n")
            record_count += 1

    return ds_dir, jsonl_path, record_count


async def upload_dataset(file: UploadFile) -> dict[str, Any]:
    """保存上传文件并写入数据库。"""
    upload_dir = settings.uploads_dir
    upload_dir.mkdir(parents=True, exist_ok=True)

    safe_name = file.filename or "unknown"
    file_path = upload_dir / safe_name
    if file_path.exists():
        file_path = upload_dir / f"{uuid.uuid4().hex}_{safe_name}"

    content = await file.read()
    file_path.write_bytes(content)

    fmt = _detect_format(safe_name)
    record_count = _count_records(file_path, fmt)

    record_id = str(uuid.uuid4())
    record = DatasetRecord(
        id=record_id,
        name=safe_name,
        format=fmt,
        record_count=record_count,
        file_path=str(file_path),
        created_at=datetime.utcnow(),
    )
    async with async_session() as session:
        session.add(record)
        await session.commit()

    logger.info(f"Uploaded dataset: {safe_name} ({record_count} records, format={fmt})")
    return {
        "id": record_id,
        "name": safe_name,
        "format": fmt,
        "record_count": record_count,
        "file_path": str(file_path),
        "created_at": record.created_at.isoformat(),
    }


def _format_value(value) -> str:
    """将复杂值格式化为可读字符串。"""
    if isinstance(value, (dict, list)):
        return json.dumps(value, ensure_ascii=False, indent=2)
    return str(value)


def _is_sharegpt_format(records: list[dict]) -> bool:
    """检测是否为 ShareGPT 格式。"""
    if not records:
        return False
    first = records[0]
    if "conversations" in first and isinstance(first["conversations"], list):
        if len(first["conversations"]) > 0 and isinstance(first["conversations"][0], dict):
            conv = first["conversations"][0]
            return "from" in conv and "value" in conv
    return False


def _flatten_sharegpt(records: list[dict]) -> tuple[list[dict], list[str]]:
    """将 ShareGPT 格式展平为 input/output 列。"""
    flat_rows = []
    for row in records:
        conversations = row.get("conversations", [])
        for i in range(0, len(conversations) - 1, 2):
            user_turn = conversations[i]
            assistant_turn = conversations[i + 1] if i + 1 < len(conversations) else None

            if user_turn.get("from") in ("human", "user"):
                input_text = str(user_turn.get("value", ""))
                output_text = str(assistant_turn.get("value", "")) if assistant_turn else ""
            else:
                input_text = str(assistant_turn.get("value", "")) if assistant_turn else ""
                output_text = str(user_turn.get("value", ""))

            if len(input_text) > 500:
                input_text = input_text[:500] + "..."
            if len(output_text) > 500:
                output_text = output_text[:500] + "..."

            flat_rows.append({"input": input_text, "output": output_text})

    return flat_rows, ["input", "output"]


async def preview_dataset(dataset_id: str, rows: int = 10) -> dict[str, Any]:
    """预览数据集前 N 行。"""
    async with async_session() as session:
        from sqlalchemy import select

        result = await session.execute(select(DatasetRecord).where(DatasetRecord.id == dataset_id))
        record = result.scalar_one_or_none()
        if not record:
            return {"total_records": 0, "preview_rows": [], "columns": []}

    file_path = Path(record.file_path)
    if not file_path.exists():
        return {"total_records": 0, "preview_rows": [], "columns": []}

    fmt = record.format
    preview_data = _read_records(file_path, fmt, rows)

    # 检测是否为 ShareGPT 格式，如果是则展平为 input/output 列
    if _is_sharegpt_format(preview_data):
        preview_data, columns = _flatten_sharegpt(preview_data)
    else:
        columns = list(preview_data[0].keys()) if preview_data else []

    return {
        "total_records": record.record_count,
        "preview_rows": [
            {
                "row_index": i,
                "data": {k: _format_value(v) for k, v in row.items()},
            }
            for i, row in enumerate(preview_data)
        ],
        "columns": columns,
    }


async def validate_dataset(dataset_id: str) -> dict[str, Any]:
    """校验数据集格式和 Schema。"""
    async with async_session() as session:
        from sqlalchemy import select

        result = await session.execute(select(DatasetRecord).where(DatasetRecord.id == dataset_id))
        record = result.scalar_one_or_none()
        if not record:
            return {"is_valid": False, "errors": ["Dataset not found"], "warnings": []}

    file_path = Path(record.file_path)
    if not file_path.exists():
        return {"is_valid": False, "errors": ["File not found"], "warnings": []}

    errors = []
    warnings = []

    fmt = record.format
    if fmt not in ("jsonl", "csv", "json", "parquet"):
        errors.append(f"Unsupported format: {fmt}")

    try:
        preview = _read_records(file_path, fmt, 5)
        if not preview:
            warnings.append("Dataset appears to be empty")
        else:
            first = preview[0]
            has_sft_fields = any(k in first for k in ("instruction", "prompt", "text", "input", "output", "completion"))
            if not has_sft_fields:
                warnings.append(f"No common SFT fields found. Keys: {list(first.keys())}")
    except Exception as e:
        errors.append(f"Failed to read file: {str(e)}")

    return {"is_valid": len(errors) == 0, "errors": errors, "warnings": warnings}


async def list_datasets() -> list[dict[str, Any]]:
    """列出所有已上传数据集。"""
    async with async_session() as session:
        from sqlalchemy import select

        result = await session.execute(select(DatasetRecord).order_by(DatasetRecord.created_at.desc()))
        records = result.scalars().all()

    return [
        {
            "id": r.id,
            "name": r.name,
            "format": r.format,
            "record_count": r.record_count,
            "file_path": r.file_path,
            "created_at": r.created_at.isoformat(),
        }
        for r in records
    ]


async def delete_dataset(dataset_id: str) -> dict[str, Any]:
    """删除数据集。"""
    async with async_session() as session:
        from sqlalchemy import select

        result = await session.execute(select(DatasetRecord).where(DatasetRecord.id == dataset_id))
        record = result.scalar_one_or_none()
        if record:
            file_path = Path(record.file_path)
            if file_path.exists():
                file_path.unlink()
            await session.delete(record)
            await session.commit()
            logger.info(f"Deleted dataset: {record.name}")

    return {"status": "deleted"}


def _detect_format(filename: str) -> str:
    ext = Path(filename).suffix.lower().lstrip(".")
    if ext in ("jsonl", "csv", "parquet", "json"):
        return ext
    return "unknown"


def _count_records(file_path: Path, fmt: str) -> int:
    try:
        if fmt == "jsonl":
            return sum(1 for line in open(file_path, encoding="utf-8") if line.strip())
        elif fmt == "json":
            with open(file_path, encoding="utf-8") as f:
                data = json.load(f)
                return len(data) if isinstance(data, list) else 1
        elif fmt == "csv":
            import csv
            with open(file_path, encoding="utf-8") as f:
                return sum(1 for _ in csv.reader(f)) - 1
        elif fmt == "parquet":
            import pandas as pd
            return len(pd.read_parquet(file_path))
    except Exception:
        pass
    return 0


def _read_records(file_path: Path, fmt: str, n: int) -> list[dict]:
    if fmt == "jsonl":
        records = []
        with open(file_path, encoding="utf-8") as f:
            for i, line in enumerate(f):
                if i >= n:
                    break
                line = line.strip()
                if line:
                    records.append(json.loads(line))
        return records
    elif fmt == "json":
        with open(file_path, encoding="utf-8") as f:
            data = json.load(f)
            return data[:n] if isinstance(data, list) else [data]
    elif fmt == "csv":
        import csv
        with open(file_path, encoding="utf-8") as f:
            reader = csv.DictReader(f)
            return [dict(row) for i, row in enumerate(reader) if i < n]
    elif fmt == "parquet":
        import pandas as pd
        df = pd.read_parquet(file_path)
        return df.head(n).to_dict(orient="records")
    return []