"""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/", 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//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//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)