import asyncio
from datetime import datetime, timedelta, timezone
from typing import Optional

from fastapi import APIRouter, Depends, HTTPException, Query
from pydantic import BaseModel, Field
from sqlalchemy import func, or_, select, text
from sqlalchemy.ext.asyncio import AsyncSession

from app.core.cache import cache
from app.core.time import to_iso8601_utc
from app.core.security import get_current_user
from app.core.data_sources import get_data_sources_config
from app.core.datasource_defaults import DEFAULT_DATASOURCES
from app.db.session import get_db
from app.models.collected_data import CollectedData
from app.models.data_snapshot import DataSnapshot
from app.models.datasource import DataSource
from app.models.datasource_config import DataSourceConfig
from app.models.task import CollectionTask
from app.models.user import User
from app.models.vessel import AISRawObservation
from app.services.vessel_ais_aggregation import VESSEL_AIS_SCHEMA
from app.services.scheduler import (
    cancel_running_collector_now,
    get_latest_task_id_for_datasource,
    run_collector_now,
    sync_datasource_job,
)
from app.services.earth_layer_cache import invalidate_earth_layer_cache_for_source

router = APIRouter()
STALE_RUNNING_TASK_TIMEOUT_MINUTES = 90

PRODUCT_SOURCE_KEYWORDS: tuple[tuple[str, tuple[str, ...]], ...] = (
    ("vessels", ("vessel", "ais")),
    ("cables", ("cable", "landing", "telegeography", "arcgis", "fao")),
    ("satellites", ("tle", "satellite", "spacetrack", "celestrak")),
    ("bgp", ("bgp", "asn", "prefix_geo", "opengeofeed", "nro")),
    ("compute", ("top500", "gpu", "supercomputer", "compute")),
    ("ai", ("huggingface", "epoch_ai")),
    ("media", ("news", "tv", "live_stream")),
)


class DatasourceBatchTriggerRequest(BaseModel):
    source_ids: list[int] = Field(default_factory=list)
    force: bool = False
    module: Optional[str] = None
    product: Optional[str] = None
    is_active: Optional[bool] = None
    priority: Optional[str] = None
    run_status: Optional[str] = None
    collected: Optional[bool] = None
    credential_status: Optional[str] = None
    q: Optional[str] = None


def format_frequency_label(minutes: int) -> str:
    if minutes % 1440 == 0:
        return f"{minutes // 1440}d"
    if minutes % 60 == 0:
        return f"{minutes // 60}h"
    return f"{minutes}m"


def datasource_metadata(source: str) -> dict:
    info = DEFAULT_DATASOURCES.get(source, {})
    return {
        "display_name": info.get("display_name") or info.get("name") or source,
        "is_free": bool(info.get("is_free", True)),
        "requires_credentials": bool(info.get("requires_credentials", False)),
        "credential_provider": info.get("credential_provider"),
        "credential_status": info.get("credential_status", "none"),
    }


def datasource_product_key(datasource: DataSource) -> str:
    haystack = " ".join(
        [
            datasource.source or "",
            datasource.name or "",
            datasource.collector_class or "",
        ]
    ).lower()
    for product, keywords in PRODUCT_SOURCE_KEYWORDS:
        if any(keyword in haystack for keyword in keywords):
            return product
    return "other"


def is_due_for_collection(datasource: DataSource, now: datetime) -> bool:
    if datasource.last_run_at is None:
        return True
    return datasource.last_run_at + timedelta(minutes=datasource.frequency_minutes) <= now


def _task_rank_column(order_column):
    return func.row_number().over(
        partition_by=CollectionTask.datasource_id,
        order_by=(order_column.desc().nullslast(), CollectionTask.id.desc()),
    ).label("row_num")


async def _load_latest_running_tasks(
    db: AsyncSession,
    datasource_ids: list[int],
) -> dict[int, CollectionTask]:
    if not datasource_ids:
        return {}

    ranked_tasks = (
        select(
            CollectionTask.id.label("task_id"),
            _task_rank_column(CollectionTask.started_at),
        )
        .where(CollectionTask.datasource_id.in_(datasource_ids))
        .where(CollectionTask.status == "running")
        .subquery()
    )
    result = await db.execute(
        select(CollectionTask)
        .join(ranked_tasks, CollectionTask.id == ranked_tasks.c.task_id)
        .where(ranked_tasks.c.row_num == 1)
    )
    return {task.datasource_id: task for task in result.scalars().all()}


async def _load_latest_task_ids(
    db: AsyncSession,
    datasource_ids: list[int],
) -> dict[int, int]:
    if not datasource_ids:
        return {}

    ranked_tasks = (
        select(
            CollectionTask.id.label("task_id"),
            CollectionTask.datasource_id.label("datasource_id"),
            func.row_number().over(
                partition_by=CollectionTask.datasource_id,
                order_by=CollectionTask.id.desc(),
            ).label("row_num"),
        )
        .where(CollectionTask.datasource_id.in_(datasource_ids))
        .subquery()
    )
    result = await db.execute(
        select(ranked_tasks.c.datasource_id, ranked_tasks.c.task_id)
        .where(ranked_tasks.c.row_num == 1)
    )
    return {datasource_id: task_id for datasource_id, task_id in result.all()}


async def _load_latest_tasks(
    db: AsyncSession,
    datasource_ids: list[int],
) -> dict[int, CollectionTask]:
    if not datasource_ids:
        return {}

    ranked_tasks = (
        select(
            CollectionTask.id.label("task_id"),
            CollectionTask.datasource_id.label("datasource_id"),
            func.row_number().over(
                partition_by=CollectionTask.datasource_id,
                order_by=(CollectionTask.created_at.desc().nullslast(), CollectionTask.id.desc()),
            ).label("row_num"),
        )
        .where(CollectionTask.datasource_id.in_(datasource_ids))
        .subquery()
    )
    result = await db.execute(
        select(CollectionTask)
        .join(ranked_tasks, CollectionTask.id == ranked_tasks.c.task_id)
        .where(ranked_tasks.c.row_num == 1)
    )
    return {task.datasource_id: task for task in result.scalars().all()}


async def _load_collected_record_counts(
    db: AsyncSession,
    sources: list[str],
) -> dict[str, int]:
    if not sources:
        return {}

    result = await db.execute(
        select(CollectedData.source, func.count(CollectedData.id))
        .where(CollectedData.source.in_(sources))
        .where(CollectedData.is_current.is_(True))
        .group_by(CollectedData.source)
    )
    counts = {source: int(count or 0) for source, count in result.all()}

    vessel_sources = [
        source
        for source in sources
        if datasource_metadata(source)["credential_provider"] in {"aisstream", "barentswatch"}
        or "vessel" in source
        or "ais" in source
    ]
    if vessel_sources:
        raw_result = await db.execute(
            select(AISRawObservation.source, func.count(AISRawObservation.id))
            .where(AISRawObservation.target_schema == VESSEL_AIS_SCHEMA)
            .where(AISRawObservation.source.in_(vessel_sources))
            .group_by(AISRawObservation.source)
        )
        for source, count in raw_result.all():
            counts[source] = max(counts.get(source, 0), int(count or 0))

    return counts


async def _load_datasource_endpoint_overrides(
    db: AsyncSession,
    sources: list[str],
) -> dict[str, str]:
    if not sources:
        return {}

    result = await db.execute(
        select(DataSourceConfig.name, DataSourceConfig.endpoint)
        .where(DataSourceConfig.name.in_(sources))
        .where(DataSourceConfig.is_active.is_(True))
        .where(DataSourceConfig.endpoint.isnot(None))
    )
    return {
        name: endpoint
        for name, endpoint in result.all()
        if endpoint
    }


async def _load_datasource_list_context(
    db: AsyncSession,
    datasources: list[DataSource],
    *,
    include_endpoint: bool = True,
) -> tuple[dict[int, CollectionTask], dict[int, CollectionTask], dict[str, str]]:
    datasource_ids = [datasource.id for datasource in datasources]
    sources = [datasource.source for datasource in datasources]

    running_tasks = await _load_latest_running_tasks(db, datasource_ids)
    datasource_by_id = {datasource.id: datasource for datasource in datasources}
    now = datetime.now(timezone.utc)

    stale_datasource_ids: list[int] = []
    for datasource_id, task in running_tasks.items():
        started_at = task.started_at
        if started_at is None:
            continue
        if started_at.tzinfo is None:
            started_at = started_at.replace(tzinfo=timezone.utc)
        if now - started_at > timedelta(minutes=STALE_RUNNING_TASK_TIMEOUT_MINUTES):
            datasource = datasource_by_id.get(datasource_id)
            if datasource is not None:
                await fail_and_rollback_stale_running_task(db, datasource, task)
                stale_datasource_ids.append(datasource_id)

    if stale_datasource_ids:
        running_tasks = await _load_latest_running_tasks(db, datasource_ids)

    latest_tasks = await _load_latest_tasks(db, datasource_ids)
    endpoint_overrides = await _load_datasource_endpoint_overrides(db, sources) if include_endpoint else {}
    return running_tasks, latest_tasks, endpoint_overrides


def serialize_datasource_row(
    datasource: DataSource,
    *,
    running_tasks: dict[int, CollectionTask],
    latest_tasks: dict[int, CollectionTask],
    record_counts: dict[str, int],
    endpoint_overrides: dict[str, str],
    config,
    include_endpoint: bool,
) -> dict:
    running_task = running_tasks.get(datasource.id)
    latest_task = latest_tasks.get(datasource.id)
    display_task = running_task or latest_task
    endpoint = None
    if include_endpoint:
        endpoint = endpoint_overrides.get(datasource.source) or config.get_yaml_url(datasource.source)
    last_run_at = datasource.last_run_at or (latest_task.completed_at if latest_task else None)
    last_status = datasource.last_status or (latest_task.status if latest_task else None)
    collected_records = record_counts.get(datasource.source, 0)

    row = {
        "id": datasource.id,
        "source": datasource.source,
        "name": datasource.name,
        **datasource_metadata(datasource.source),
        "product": datasource_product_key(datasource),
        "module": datasource.module,
        "priority": datasource.priority,
        "frequency": format_frequency_label(datasource.frequency_minutes),
        "frequency_minutes": datasource.frequency_minutes,
        "is_active": datasource.is_active,
        "collector_class": datasource.collector_class,
        "last_run": to_iso8601_utc(last_run_at),
        "last_run_at": to_iso8601_utc(last_run_at),
        "last_status": last_status,
        "is_running": running_task is not None,
        "task_id": display_task.id if display_task else None,
        "progress": display_task.progress if display_task else None,
        "phase": display_task.phase if display_task else None,
        "phase_progress": display_task.phase_progress if display_task else None,
        "phase_message": display_task.phase_message if display_task else None,
        "phase_current": display_task.phase_current if display_task else None,
        "phase_total": display_task.phase_total if display_task else None,
        "phase_unit": display_task.phase_unit if display_task else None,
        "records_processed": display_task.records_processed if display_task else None,
        "total_records": display_task.total_records if display_task else None,
        "error_message": display_task.error_message if display_task else None,
        "collected_records": collected_records,
        "has_collected_data": collected_records > 0,
    }
    if include_endpoint:
        row["endpoint"] = endpoint
    return row


def _apply_datasource_query_filters(
    query,
    *,
    module: Optional[str] = None,
    is_active: Optional[bool] = None,
    priority: Optional[str] = None,
    run_status: Optional[str] = None,
    q: Optional[str] = None,
) -> object:
    if module:
        query = query.where(DataSource.module == module)
    if is_active is not None:
        query = query.where(DataSource.is_active == is_active)
    if priority:
        query = query.where(DataSource.priority == priority)
    if q:
        like_value = f"%{q.strip()}%"
        query = query.where(
            or_(
                DataSource.name.ilike(like_value),
                DataSource.source.ilike(like_value),
                DataSource.collector_class.ilike(like_value),
            )
        )
    return query


def _filter_datasources_in_memory(
    datasources: list[DataSource],
    *,
    running_tasks: dict[int, CollectionTask],
    latest_tasks: dict[int, CollectionTask] | None = None,
    record_counts: dict[str, int],
    product: Optional[str] = None,
    run_status: Optional[str] = None,
    collected: Optional[bool] = None,
    credential_status: Optional[str] = None,
) -> list[DataSource]:
    latest_tasks = latest_tasks or {}
    filtered: list[DataSource] = []
    for datasource in datasources:
        record_count = record_counts.get(datasource.source, 0)
        latest_task = latest_tasks.get(datasource.id)
        effective_status = (
            "running"
            if datasource.id in running_tasks
            else latest_task.status
            if latest_task is not None
            else datasource.last_status
        )
        if product and datasource_product_key(datasource) != product:
            continue
        if collected is not None and (record_count > 0) != collected:
            continue
        if credential_status:
            metadata = datasource_metadata(datasource.source)
            if metadata["credential_status"] != credential_status:
                continue
        if run_status == "running" and datasource.id not in running_tasks:
            continue
        if run_status == "not_run" and effective_status is not None:
            continue
        if run_status not in {None, "running", "not_run", "collected", "uncollected"} and effective_status != run_status:
            continue
        if run_status == "collected" and record_count <= 0:
            continue
        if run_status == "uncollected" and record_count > 0:
            continue
        filtered.append(datasource)
    return filtered


async def _trigger_datasource_batch(
    db: AsyncSession,
    datasources: list[DataSource],
    *,
    force: bool,
) -> dict:
    if not datasources:
        return {
            "status": "noop",
            "message": "No matching data sources to trigger",
            "force": force,
            "triggered": [],
            "skipped": [],
            "failed": [],
        }

    previous_task_ids: dict[int, Optional[int]] = {}
    triggered_sources: list[dict] = []
    skipped_sources: list[dict] = []
    failed_sources: list[dict] = []
    now = datetime.now(timezone.utc)
    running_tasks = await _load_latest_running_tasks(
        db,
        [datasource.id for datasource in datasources],
    )

    for datasource in datasources:
        if not datasource.is_active:
            skipped_sources.append(
                {
                    "id": datasource.id,
                    "source": datasource.source,
                    "name": datasource.name,
                    "reason": "disabled",
                }
            )
            continue

        running_task = running_tasks.get(datasource.id)
        if running_task is not None:
            if not force:
                skipped_sources.append(
                    {
                        "id": datasource.id,
                        "source": datasource.source,
                        "name": datasource.name,
                        "reason": "already_running",
                        "task_id": running_task.id,
                    }
                )
                continue
            cancelled = await cancel_running_collector_now(datasource.source)
            if not cancelled:
                await rollback_orphaned_running_task(db, datasource, running_task)

        if not force and not is_due_for_collection(datasource, now):
            skipped_sources.append(
                {
                    "id": datasource.id,
                    "source": datasource.source,
                    "name": datasource.name,
                    "reason": "within_frequency_window",
                    "last_run_at": to_iso8601_utc(datasource.last_run_at),
                    "next_run_at": to_iso8601_utc(
                        datasource.last_run_at + timedelta(minutes=datasource.frequency_minutes)
                    ),
                }
            )
            continue

        previous_task_ids[datasource.id] = None
        success = run_collector_now(datasource.source)
        if not success:
            failed_sources.append(
                {
                    "id": datasource.id,
                    "source": datasource.source,
                    "name": datasource.name,
                    "reason": "trigger_failed",
                }
            )
            continue

        triggered_sources.append(
            {
                "id": datasource.id,
                "source": datasource.source,
                "name": datasource.name,
                "task_id": None,
            }
        )

    latest_task_ids = await _load_latest_task_ids(
        db,
        [datasource.id for datasource in datasources],
    )
    for datasource_id in previous_task_ids:
        previous_task_ids[datasource_id] = latest_task_ids.get(datasource_id)

    for _ in range(20):
        await asyncio.sleep(0.1)
        pending = [item for item in triggered_sources if item["task_id"] is None]
        if not pending:
            break
        latest_task_ids = await _load_latest_task_ids(
            db,
            [item["id"] for item in pending],
        )
        for item in pending:
            task_id = latest_task_ids.get(item["id"])
            if task_id is not None and task_id != previous_task_ids.get(item["id"]):
                item["task_id"] = task_id

    return {
        "status": "triggered" if triggered_sources else "partial",
        "message": f"Triggered {len(triggered_sources)} data sources",
        "force": force,
        "triggered": triggered_sources,
        "skipped": skipped_sources,
        "failed": failed_sources,
    }


async def get_datasource_record(db: AsyncSession, source_id: str) -> Optional[DataSource]:
    datasource = None
    try:
        datasource = await db.get(DataSource, int(source_id))
    except ValueError:
        pass

    if datasource is not None:
        return datasource

    result = await db.execute(
        select(DataSource).where(
            (DataSource.source == source_id) | (DataSource.collector_class == source_id)
        )
    )
    return result.scalar_one_or_none()


async def get_running_task(db: AsyncSession, datasource_id: int) -> Optional[CollectionTask]:
    result = await db.execute(
        select(CollectionTask)
        .where(CollectionTask.datasource_id == datasource_id)
        .where(CollectionTask.status == "running")
        .order_by(CollectionTask.started_at.desc())
        .limit(1)
    )
    task = result.scalar_one_or_none()
    if not task:
        return None

    started_at = task.started_at
    if started_at is None:
        return task

    now = datetime.now(timezone.utc)
    if started_at.tzinfo is None:
        started_at = started_at.replace(tzinfo=timezone.utc)

    if now - started_at <= timedelta(minutes=STALE_RUNNING_TASK_TIMEOUT_MINUTES):
        return task

    datasource = await db.get(DataSource, datasource_id)
    if datasource is not None:
        await fail_and_rollback_stale_running_task(db, datasource, task)
    else:
        existing_error = (task.error_message or "").strip()
        stale_reason = (
            f"Marked failed automatically after stale running timeout "
            f"({STALE_RUNNING_TASK_TIMEOUT_MINUTES}m)"
        )
        task.status = "failed"
        task.phase = "failed"
        task.completed_at = now
        task.error_message = f"{existing_error}\n{stale_reason}".strip() if existing_error else stale_reason
        await db.commit()
    return None


async def rollback_orphaned_running_task(
    db: AsyncSession,
    datasource: DataSource,
    running_task: CollectionTask,
) -> None:
    snapshot_result = await db.execute(
        select(DataSnapshot)
        .where(
            DataSnapshot.datasource_id == datasource.id,
            DataSnapshot.task_id == running_task.id,
        )
        .order_by(DataSnapshot.id.desc())
        .limit(1)
    )
    snapshot = snapshot_result.scalar_one_or_none()

    await db.execute(CollectedData.__table__.delete().where(CollectedData.task_id == running_task.id))

    await db.execute(
        text(
            """
            UPDATE collected_data
            SET is_current = FALSE
            WHERE source = :source
            """
        ),
        {"source": datasource.source},
    )

    if snapshot is not None:
        snapshot.status = "cancelled"
        snapshot.is_current = False
        snapshot.completed_at = datetime.now(timezone.utc)
        summary = dict(snapshot.summary or {})
        summary["rollback"] = True
        summary["rollback_reason"] = "orphaned_running_task_after_backend_restart"
        snapshot.summary = summary

        if snapshot.parent_snapshot_id is not None:
            parent_snapshot = await db.get(DataSnapshot, snapshot.parent_snapshot_id)
            if parent_snapshot:
                parent_snapshot.is_current = True
            await db.execute(
                text(
                    """
                    UPDATE collected_data
                    SET is_current = TRUE
                    WHERE snapshot_id = :snapshot_id
                    """
                ),
                {"snapshot_id": snapshot.parent_snapshot_id},
            )

    running_task.status = "cancelled"
    running_task.phase = "cancelled"
    running_task.completed_at = datetime.now(timezone.utc)
    existing_error = (running_task.error_message or "").strip()
    cancel_reason = "Cancelled after backend restart because the running task handle was lost; incomplete writes rolled back"
    running_task.error_message = f"{existing_error}\n{cancel_reason}".strip() if existing_error else cancel_reason
    datasource.last_status = "cancelled"
    datasource.last_run_at = datetime.now(timezone.utc)
    await db.commit()


async def fail_and_rollback_stale_running_task(
    db: AsyncSession,
    datasource: DataSource,
    running_task: CollectionTask,
) -> None:
    snapshot_result = await db.execute(
        select(DataSnapshot)
        .where(
            DataSnapshot.datasource_id == datasource.id,
            DataSnapshot.task_id == running_task.id,
        )
        .order_by(DataSnapshot.id.desc())
        .limit(1)
    )
    snapshot = snapshot_result.scalar_one_or_none()

    await db.execute(CollectedData.__table__.delete().where(CollectedData.task_id == running_task.id))

    await db.execute(
        text(
            """
            UPDATE collected_data
            SET is_current = FALSE
            WHERE source = :source
            """
        ),
        {"source": datasource.source},
    )

    if snapshot is not None:
        snapshot.status = "failed"
        snapshot.is_current = False
        snapshot.completed_at = datetime.now(timezone.utc)
        summary = dict(snapshot.summary or {})
        summary["rollback"] = True
        summary["rollback_reason"] = "stale_running_task_timeout"
        snapshot.summary = summary

        if snapshot.parent_snapshot_id is not None:
            parent_snapshot = await db.get(DataSnapshot, snapshot.parent_snapshot_id)
            if parent_snapshot:
                parent_snapshot.is_current = True
            await db.execute(
                text(
                    """
                    UPDATE collected_data
                    SET is_current = TRUE
                    WHERE snapshot_id = :snapshot_id
                    """
                ),
                {"snapshot_id": snapshot.parent_snapshot_id},
            )

    existing_error = (running_task.error_message or "").strip()
    stale_reason = (
        f"Marked failed automatically after stale running timeout "
        f"({STALE_RUNNING_TASK_TIMEOUT_MINUTES}m); incomplete writes rolled back"
    )
    running_task.status = "failed"
    running_task.phase = "failed"
    running_task.completed_at = datetime.now(timezone.utc)
    running_task.error_message = f"{existing_error}\n{stale_reason}".strip() if existing_error else stale_reason
    datasource.last_status = "failed"
    datasource.last_run_at = datetime.now(timezone.utc)
    await db.commit()


@router.get("")
async def list_datasources(
    module: Optional[str] = None,
    is_active: Optional[bool] = None,
    priority: Optional[str] = None,
    product: Optional[str] = None,
    run_status: Optional[str] = None,
    collected: Optional[bool] = None,
    credential_status: Optional[str] = None,
    q: Optional[str] = None,
    include_endpoint: bool = True,
    current_user: User = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    query = select(DataSource).order_by(DataSource.module, DataSource.id)
    query = _apply_datasource_query_filters(
        query,
        module=module,
        is_active=is_active,
        priority=priority,
        run_status=run_status,
        q=q,
    )

    result = await db.execute(query)
    datasources = result.scalars().all()

    collector_list = []
    config = get_data_sources_config()
    running_tasks, latest_tasks, endpoint_overrides = await _load_datasource_list_context(
        db,
        datasources,
        include_endpoint=include_endpoint,
    )
    record_counts = await _load_collected_record_counts(db, [datasource.source for datasource in datasources])
    datasources = _filter_datasources_in_memory(
        datasources,
        running_tasks=running_tasks,
        latest_tasks=latest_tasks,
        record_counts=record_counts,
        product=product,
        run_status=run_status,
        collected=collected,
        credential_status=credential_status,
    )
    for datasource in datasources:
        collector_list.append(
            serialize_datasource_row(
                datasource,
                running_tasks=running_tasks,
                latest_tasks=latest_tasks,
                record_counts=record_counts,
                endpoint_overrides=endpoint_overrides,
                config=config,
                include_endpoint=include_endpoint,
            )
        )

    return {"total": len(collector_list), "data": collector_list}


@router.post("/trigger-all")
async def trigger_all_datasources(
    force: bool = Query(False),
    current_user: User = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    result = await db.execute(
        select(DataSource)
        .where(DataSource.is_active.is_(True))
        .order_by(DataSource.module, DataSource.id)
    )
    datasources = result.scalars().all()
    return await _trigger_datasource_batch(db, datasources, force=force)


@router.post("/trigger-batch")
async def trigger_datasource_batch(
    payload: DatasourceBatchTriggerRequest,
    current_user: User = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    query = select(DataSource).order_by(DataSource.module, DataSource.id)
    if payload.source_ids:
        query = query.where(DataSource.id.in_(payload.source_ids))
    else:
        query = _apply_datasource_query_filters(
            query,
            module=payload.module,
            is_active=payload.is_active,
            priority=payload.priority,
            run_status=payload.run_status,
            q=payload.q,
        )

    result = await db.execute(query)
    datasources = result.scalars().all()
    running_tasks, latest_tasks, _ = await _load_datasource_list_context(db, datasources)
    record_counts = await _load_collected_record_counts(db, [datasource.source for datasource in datasources])
    datasources = _filter_datasources_in_memory(
        datasources,
        running_tasks=running_tasks,
        latest_tasks=latest_tasks,
        record_counts=record_counts,
        product=None if payload.source_ids else payload.product,
        run_status=None if payload.source_ids else payload.run_status,
        collected=None if payload.source_ids else payload.collected,
        credential_status=None if payload.source_ids else payload.credential_status,
    )
    return await _trigger_datasource_batch(db, datasources, force=payload.force)


@router.get("/snapshots")
async def list_datasource_snapshots(
    source_id: Optional[str] = None,
    current_only: Optional[bool] = None,
    limit: int = Query(default=100, ge=1, le=500),
    current_user: User = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    query = (
        select(DataSnapshot, DataSource.name, DataSource.module)
        .outerjoin(DataSource, DataSource.id == DataSnapshot.datasource_id)
        .order_by(DataSnapshot.created_at.desc().nullslast(), DataSnapshot.id.desc())
        .limit(limit)
    )
    if source_id:
        query = query.where(DataSnapshot.source == source_id)
    if current_only is not None:
        query = query.where(DataSnapshot.is_current.is_(current_only))

    result = await db.execute(query)
    rows = []
    for snapshot, datasource_name, datasource_module in result.all():
        rows.append(
            {
                "id": snapshot.id,
                "datasource_id": snapshot.datasource_id,
                "datasource_name": datasource_name,
                "module": datasource_module,
                "task_id": snapshot.task_id,
                "source": snapshot.source,
                "snapshot_key": snapshot.snapshot_key,
                "reference_date": to_iso8601_utc(snapshot.reference_date),
                "started_at": to_iso8601_utc(snapshot.started_at),
                "completed_at": to_iso8601_utc(snapshot.completed_at),
                "record_count": snapshot.record_count,
                "status": snapshot.status,
                "is_current": snapshot.is_current,
                "parent_snapshot_id": snapshot.parent_snapshot_id,
                "summary": snapshot.summary or {},
                "created_at": to_iso8601_utc(snapshot.created_at),
            }
        )
    return {"total": len(rows), "data": rows}


@router.get("/{source_id}")
async def get_datasource(
    source_id: str,
    current_user: User = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    datasource = await get_datasource_record(db, source_id)
    if not datasource:
        raise HTTPException(status_code=404, detail="Data source not found")

    config = get_data_sources_config()
    endpoint = await config.get_url(datasource.source, db)

    return {
        "id": datasource.id,
        "name": datasource.name,
        **datasource_metadata(datasource.source),
        "module": datasource.module,
        "priority": datasource.priority,
        "frequency": format_frequency_label(datasource.frequency_minutes),
        "frequency_minutes": datasource.frequency_minutes,
        "collector_class": datasource.collector_class,
        "source": datasource.source,
        "endpoint": endpoint,
        "is_active": datasource.is_active,
    }


@router.get("/{source_id}/row")
async def get_datasource_row(
    source_id: str,
    include_endpoint: bool = True,
    current_user: User = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    datasource = await get_datasource_record(db, source_id)
    if not datasource:
        raise HTTPException(status_code=404, detail="Data source not found")

    config = get_data_sources_config()
    running_tasks, latest_tasks, endpoint_overrides = await _load_datasource_list_context(
        db,
        [datasource],
        include_endpoint=include_endpoint,
    )
    record_counts = await _load_collected_record_counts(db, [datasource.source])
    return {
        "data": serialize_datasource_row(
            datasource,
            running_tasks=running_tasks,
            latest_tasks=latest_tasks,
            record_counts=record_counts,
            endpoint_overrides=endpoint_overrides,
            config=config,
            include_endpoint=include_endpoint,
        )
    }


@router.post("/{source_id}/enable")
async def enable_datasource(
    source_id: str,
    current_user: User = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    datasource = await get_datasource_record(db, source_id)
    if not datasource:
        raise HTTPException(status_code=404, detail="Data source not found")

    datasource.is_active = True
    await db.commit()
    await sync_datasource_job(datasource.id)
    return {"status": "enabled", "source_id": datasource.id}


@router.post("/{source_id}/disable")
async def disable_datasource(
    source_id: str,
    current_user: User = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    datasource = await get_datasource_record(db, source_id)
    if not datasource:
        raise HTTPException(status_code=404, detail="Data source not found")

    datasource.is_active = False
    await db.commit()
    await sync_datasource_job(datasource.id)
    return {"status": "disabled", "source_id": datasource.id}


@router.get("/{source_id}/stats")
async def get_datasource_stats(
    source_id: str,
    current_user: User = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    datasource = await get_datasource_record(db, source_id)
    if not datasource:
        raise HTTPException(status_code=404, detail="Data source not found")

    result = await db.execute(
        select(func.count(CollectedData.id)).where(CollectedData.source == datasource.source)
    )
    total = result.scalar() or 0

    return {
        "source_id": datasource.id,
        "collector_name": datasource.collector_class,
        "name": datasource.name,
        "total_records": total,
    }


@router.post("/{source_id}/trigger")
async def trigger_datasource(
    source_id: str,
    force: bool = Query(False),
    current_user: User = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    datasource = await get_datasource_record(db, source_id)
    if not datasource:
        raise HTTPException(status_code=404, detail="Data source not found")

    if not datasource.is_active:
        raise HTTPException(status_code=400, detail="Data source is disabled")

    running_task = await get_running_task(db, datasource.id)
    if running_task is not None and not force:
        raise HTTPException(
            status_code=409,
            detail={
                "reason": "running_task_in_progress",
                "message": "当前采集任务尚未完成，重新触发会丢失本次未完成进度。是否强制重新采集？",
                "task_id": running_task.id,
                "phase": running_task.phase,
                "phase_progress": running_task.phase_progress,
                "phase_message": running_task.phase_message,
                "phase_current": running_task.phase_current,
                "phase_total": running_task.phase_total,
                "phase_unit": running_task.phase_unit,
                "progress": running_task.progress,
                "records_processed": running_task.records_processed,
                "total_records": running_task.total_records,
            },
        )

    if running_task is not None and force:
        cancelled = await cancel_running_collector_now(datasource.source)
        if not cancelled:
            await rollback_orphaned_running_task(db, datasource, running_task)

    previous_task_id = await get_latest_task_id_for_datasource(datasource.id)
    success = run_collector_now(datasource.source)
    if not success:
        raise HTTPException(status_code=500, detail=f"Failed to trigger collector '{datasource.source}'")

    task_id = None
    for _ in range(20):
        await asyncio.sleep(0.1)
        task_id = await get_latest_task_id_for_datasource(datasource.id)
        if task_id is not None and task_id != previous_task_id:
            break
    if task_id == previous_task_id:
        task_id = None

    return {
        "status": "triggered",
        "source_id": datasource.id,
        "task_id": task_id,
        "collector_name": datasource.source,
        "force": force,
        "message": f"Collector '{datasource.source}' has been triggered",
    }


@router.delete("/{source_id}/data")
async def clear_datasource_data(
    source_id: str,
    current_user: User = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    datasource = await get_datasource_record(db, source_id)
    if not datasource:
        raise HTTPException(status_code=404, detail="Data source not found")

    result = await db.execute(
        select(func.count(CollectedData.id)).where(CollectedData.source == datasource.source)
    )
    count = result.scalar() or 0

    if count == 0:
        return {"status": "success", "message": "No data to clear", "deleted_count": 0}

    delete_query = CollectedData.__table__.delete().where(CollectedData.source == datasource.source)
    await db.execute(delete_query)
    await db.commit()

    return {
        "status": "success",
        "message": f"Cleared {count} records for data source '{datasource.name}'",
        "deleted_count": count,
    }


@router.delete("/{source_id}/cache")
async def clear_datasource_cache(
    source_id: str,
    current_user: User = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    datasource = await get_datasource_record(db, source_id)
    if not datasource:
        raise HTTPException(status_code=404, detail="Data source not found")

    earth_deleted_count = invalidate_earth_layer_cache_for_source(datasource.source)
    dashboard_deleted_count = int(cache.delete("dashboard:stats")) + int(cache.delete("dashboard:summary"))
    deleted_count = earth_deleted_count + dashboard_deleted_count

    return {
        "status": "success",
        "message": f"Cleared {deleted_count} cache keys for data source '{datasource.name}'",
        "deleted_count": deleted_count,
        "earth_layer_deleted_count": earth_deleted_count,
        "dashboard_deleted_count": dashboard_deleted_count,
    }


@router.get("/{source_id}/task-status")
async def get_task_status(
    source_id: str,
    task_id: Optional[int] = None,
    current_user: User = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    datasource = await get_datasource_record(db, source_id)
    if not datasource:
        raise HTTPException(status_code=404, detail="Data source not found")

    if task_id is not None:
        task = await db.get(CollectionTask, task_id)
        if not task or task.datasource_id != datasource.id:
            raise HTTPException(status_code=404, detail="Task not found")
    else:
        task = await get_running_task(db, datasource.id)
        if task is None:
            result = await db.execute(
                select(CollectionTask)
                .where(CollectionTask.datasource_id == datasource.id)
                .order_by(CollectionTask.created_at.desc().nullslast(), CollectionTask.id.desc())
                .limit(1)
            )
            task = result.scalar_one_or_none()

    if not task:
        return {
            "is_running": False,
            "task_id": None,
            "progress": None,
            "phase": None,
            "phase_progress": None,
            "phase_message": None,
            "phase_current": None,
            "phase_total": None,
            "phase_unit": None,
            "status": "idle",
        }

    return {
        "is_running": task.status == "running",
        "task_id": task.id,
        "progress": task.progress,
        "phase": task.phase,
        "phase_progress": task.phase_progress,
        "phase_message": task.phase_message,
        "phase_current": task.phase_current,
        "phase_total": task.phase_total,
        "phase_unit": task.phase_unit,
        "records_processed": task.records_processed,
        "total_records": task.total_records,
        "status": task.status,
        "error_message": task.error_message,
    }