1239 lines
42 KiB
Python
1239 lines
42 KiB
Python
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.logging import get_logger
|
|
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 (
|
|
sync_datasource_job,
|
|
)
|
|
from app.services.data_jobs import (
|
|
JOB_STATUS_CANCELLING,
|
|
JOB_STATUS_QUEUED,
|
|
JOB_STATUS_RUNNING,
|
|
JOB_TYPE_CLEAR_CACHE,
|
|
JOB_TYPE_CLEAR_DATA,
|
|
JOB_TYPE_COLLECT,
|
|
enqueue_datasource_job,
|
|
get_active_datasource_job,
|
|
request_cancel_datasource_task,
|
|
)
|
|
from app.services.business_logs import emit_business_log
|
|
|
|
router = APIRouter()
|
|
logger = get_logger(__name__, service="api")
|
|
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.in_((JOB_STATUS_QUEUED, JOB_STATUS_RUNNING, JOB_STATUS_CANCELLING)))
|
|
.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_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 and running_task.task_type == JOB_TYPE_COLLECT,
|
|
"is_task_active": running_task is not None,
|
|
"task_status": running_task.status if running_task else None,
|
|
"task_id": display_task.id if display_task else None,
|
|
"task_type": display_task.task_type 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,
|
|
actor_id: int | None = None,
|
|
trigger_kind: str = "batch",
|
|
) -> dict:
|
|
await emit_business_log(
|
|
logger,
|
|
event=f"collector.trigger.{trigger_kind}.start",
|
|
message="Datasource batch trigger started",
|
|
category="collector",
|
|
service="api",
|
|
module=__name__,
|
|
user_id=actor_id,
|
|
context={
|
|
"trigger_kind": trigger_kind,
|
|
"force": force,
|
|
"requested_count": len(datasources),
|
|
},
|
|
)
|
|
if not datasources:
|
|
result = {
|
|
"status": "noop",
|
|
"message": "No matching data sources to trigger",
|
|
"force": force,
|
|
"triggered": [],
|
|
"skipped": [],
|
|
"failed": [],
|
|
}
|
|
await emit_business_log(
|
|
logger,
|
|
event=f"collector.trigger.{trigger_kind}.completed",
|
|
message="Datasource batch trigger completed with no matching sources",
|
|
category="collector",
|
|
service="api",
|
|
module=__name__,
|
|
user_id=actor_id,
|
|
context={"trigger_kind": trigger_kind, "force": force, "status": "noop", "triggered_count": 0},
|
|
)
|
|
return result
|
|
|
|
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
|
|
await request_cancel_datasource_task(
|
|
db,
|
|
running_task,
|
|
reason="superseded_by_forced_collection",
|
|
)
|
|
|
|
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
|
|
|
|
task = await enqueue_datasource_job(
|
|
db,
|
|
datasource,
|
|
JOB_TYPE_COLLECT,
|
|
payload={"force": force, "trigger": "batch"},
|
|
)
|
|
|
|
triggered_sources.append(
|
|
{
|
|
"id": datasource.id,
|
|
"source": datasource.source,
|
|
"name": datasource.name,
|
|
"task_id": task.id,
|
|
}
|
|
)
|
|
|
|
result = {
|
|
"status": "queued" if triggered_sources else "partial",
|
|
"message": f"Queued {len(triggered_sources)} data source jobs",
|
|
"force": force,
|
|
"triggered": triggered_sources,
|
|
"skipped": skipped_sources,
|
|
"failed": failed_sources,
|
|
}
|
|
await emit_business_log(
|
|
logger,
|
|
event=f"collector.trigger.{trigger_kind}.completed",
|
|
message="Datasource batch trigger completed",
|
|
category="collector",
|
|
service="api",
|
|
module=__name__,
|
|
user_id=actor_id,
|
|
context={
|
|
"trigger_kind": trigger_kind,
|
|
"force": force,
|
|
"status": result["status"],
|
|
"requested_count": len(datasources),
|
|
"triggered_count": len(triggered_sources),
|
|
"skipped_count": len(skipped_sources),
|
|
"failed_count": len(failed_sources),
|
|
"triggered_sources": [item["source"] for item in triggered_sources],
|
|
"skipped_reasons": [item["reason"] for item in skipped_sources],
|
|
},
|
|
)
|
|
return result
|
|
|
|
|
|
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,
|
|
actor_id=current_user.id,
|
|
trigger_kind="all",
|
|
)
|
|
|
|
|
|
@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,
|
|
actor_id=current_user.id,
|
|
trigger_kind="batch",
|
|
)
|
|
|
|
|
|
@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_active_datasource_job(db, datasource.id, task_types=(JOB_TYPE_COLLECT,))
|
|
if running_task is not None and not force:
|
|
await emit_business_log(
|
|
logger,
|
|
event="collector.trigger.single.skipped_already_running",
|
|
message="Datasource trigger skipped because a task is already running",
|
|
category="collector",
|
|
level="warning",
|
|
service="api",
|
|
module=__name__,
|
|
user_id=current_user.id,
|
|
context={
|
|
"collector_name": datasource.source,
|
|
"datasource_id": datasource.id,
|
|
"task_id": running_task.id,
|
|
"status": "skipped",
|
|
},
|
|
)
|
|
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:
|
|
await request_cancel_datasource_task(
|
|
db,
|
|
running_task,
|
|
reason="superseded_by_forced_collection",
|
|
)
|
|
|
|
task = await enqueue_datasource_job(
|
|
db,
|
|
datasource,
|
|
JOB_TYPE_COLLECT,
|
|
payload={"force": force, "trigger": "single"},
|
|
)
|
|
await emit_business_log(
|
|
logger,
|
|
event="collector.trigger.single.completed",
|
|
message="Datasource trigger queued",
|
|
category="collector",
|
|
service="api",
|
|
module=__name__,
|
|
user_id=current_user.id,
|
|
context={
|
|
"collector_name": datasource.source,
|
|
"datasource_id": datasource.id,
|
|
"task_id": task.id,
|
|
"force": force,
|
|
"status": "queued",
|
|
},
|
|
)
|
|
|
|
return {
|
|
"status": "queued",
|
|
"source_id": datasource.id,
|
|
"task_id": task.id,
|
|
"collector_name": datasource.source,
|
|
"force": force,
|
|
"message": f"Collector '{datasource.source}' has been queued",
|
|
}
|
|
|
|
|
|
@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")
|
|
|
|
active_task = await get_active_datasource_job(db, datasource.id)
|
|
if active_task is not None:
|
|
raise HTTPException(
|
|
status_code=409,
|
|
detail={
|
|
"reason": "datasource_job_in_progress",
|
|
"message": "当前数据源已有任务在执行，请等待完成或先取消任务。",
|
|
"task_id": active_task.id,
|
|
"task_type": active_task.task_type,
|
|
"status": active_task.status,
|
|
},
|
|
)
|
|
task = await enqueue_datasource_job(
|
|
db,
|
|
datasource,
|
|
JOB_TYPE_CLEAR_DATA,
|
|
payload={"source": datasource.source},
|
|
)
|
|
|
|
return {
|
|
"status": "queued",
|
|
"message": f"Queued data clearing for data source '{datasource.name}'",
|
|
"task_id": task.id,
|
|
"deleted_count": None,
|
|
}
|
|
|
|
|
|
@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")
|
|
|
|
task = await enqueue_datasource_job(
|
|
db,
|
|
datasource,
|
|
JOB_TYPE_CLEAR_CACHE,
|
|
payload={"source": datasource.source},
|
|
dedupe_key=f"clear_cache:{datasource.source}",
|
|
)
|
|
|
|
return {
|
|
"status": "queued",
|
|
"message": f"Queued cache clearing for data source '{datasource.name}'",
|
|
"task_id": task.id,
|
|
"deleted_count": None,
|
|
}
|
|
|
|
|
|
@router.post("/{source_id}/tasks/{task_id}/cancel")
|
|
async def cancel_datasource_task(
|
|
source_id: str,
|
|
task_id: int,
|
|
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")
|
|
|
|
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")
|
|
|
|
task = await request_cancel_datasource_task(db, task)
|
|
return {
|
|
"status": "cancelled" if task.completed_at else "cancelling",
|
|
"task_id": task.id,
|
|
"task_type": task.task_type,
|
|
"phase": task.phase,
|
|
"requested_cancel_at": to_iso8601_utc(task.requested_cancel_at),
|
|
}
|
|
|
|
|
|
@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_active_datasource_job(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 in {JOB_STATUS_QUEUED, JOB_STATUS_RUNNING, JOB_STATUS_CANCELLING}
|
|
and task.task_type == JOB_TYPE_COLLECT,
|
|
"is_task_active": task.status in {JOB_STATUS_QUEUED, JOB_STATUS_RUNNING, JOB_STATUS_CANCELLING},
|
|
"task_id": task.id,
|
|
"task_type": task.task_type,
|
|
"task_status": task.status,
|
|
"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,
|
|
"requested_cancel_at": to_iso8601_utc(task.requested_cancel_at),
|
|
"error_message": task.error_message,
|
|
}
|