DataOps-platform/app/api/data_development/routes.py

"""HTTP orchestration boundary for data-research ingestion jobs."""

from __future__ import annotations

import logging

from flask import g, jsonify, request

from app import db
from app.api.data_development import bp
from app.core.data_research.errors import DataResearchError
from app.models.result import failed, success


logger = logging.getLogger(__name__)


def get_ingestion_service():
    from app.core.data_research.ingestion import IngestionService
    from app.core.data_research.repository import SqlAlchemyIngestionJobRepository

    return IngestionService(
        SqlAlchemyIngestionJobRepository(db.session),
        commit=db.session.commit,
        rollback=db.session.rollback,
    )


def _identity():
    return getattr(g, "current_user", {}) or {}


def _record(record):
    return {
        "uid": str(record.uid),
        "source_uid": str(record.source_uid),
        "artifact_uid": str(record.artifact_uid) if record.artifact_uid else None,
        "job_type": record.job_type,
        "parser_version": record.parser_version,
        "status": record.status,
        "parameters": dict(record.parameters or {}),
        "statistics": dict(record.statistics or {}),
        "last_error": record.last_error,
        "actor_uid": record.actor_uid,
        "created_at": record.created_at.isoformat() if record.created_at else None,
        "updated_at": record.updated_at.isoformat() if record.updated_at else None,
        "started_at": record.started_at.isoformat() if record.started_at else None,
        "finished_at": record.finished_at.isoformat() if record.finished_at else None,
    }


def _error(error):
    if isinstance(error, DataResearchError):
        return (
            jsonify(
                failed(
                    str(error),
                    code=error.http_status,
                    error={"code": error.code},
                )
            ),
            error.http_status,
        )
    logger.exception("data-research ingestion request failed")
    return (
        jsonify(
            failed(
                "数据采集任务处理失败",
                code=500,
                error={"code": "DATA_RESEARCH_ERROR"},
            )
        ),
        500,
    )


@bp.route("/ingestion-jobs", methods=["POST"])
def create_ingestion_job():
    payload = request.get_json(silent=True) or {}
    try:
        record, created = get_ingestion_service().create_job(
            payload,
            actor_uid=_identity().get("id") or _identity().get("sub"),
        )
        return jsonify(success(_record(record))), 201 if created else 200
    except Exception as error:
        return _error(error)


@bp.route("/ingestion-jobs", methods=["GET"])
def list_ingestion_jobs():
    filters = {
        name: request.args.get(name)
        for name in ("status", "source_uid")
        if request.args.get(name)
    }
    try:
        records = get_ingestion_service().list_jobs(filters)
        return jsonify(
            success({"records": [_record(item) for item in records], "total": len(records)})
        ), 200
    except Exception as error:
        return _error(error)


@bp.route("/ingestion-jobs/<job_uid>", methods=["GET"])
def get_ingestion_job(job_uid):
    try:
        return jsonify(success(_record(get_ingestion_service().get_job(job_uid)))), 200
    except Exception as error:
        return _error(error)


@bp.route("/ingestion-jobs/<job_uid>/retry", methods=["POST"])
def retry_ingestion_job(job_uid):
    try:
        return jsonify(success(_record(get_ingestion_service().retry(job_uid)))), 200
    except Exception as error:
        return _error(error)


@bp.route("/ingestion-jobs/<job_uid>/cancel", methods=["POST"])
def cancel_ingestion_job(job_uid):
    try:
        service = get_ingestion_service()
        record = service.get_job(job_uid)
        identity = _identity()
        permissions = set(identity.get("permissions") or [])
        actor_uid = identity.get("id") or identity.get("sub")
        if record.actor_uid != actor_uid and "ingestion:admin" not in permissions:
            return jsonify(failed("权限不足", code=403)), 403
        return jsonify(success(_record(service.cancel(job_uid)))), 200
    except Exception as error:
        return _error(error)