from typing import Any from spec.jsonschema.job_inputs import JOB_TYPE_TO_CREATE_JOB_INPUTS_SCHEMA_MAP from spec.jsonschema.job_outputs import JOB_TYPE_TO_CREATE_JOB_OUTPUTS_SCHEMA_MAP from video.constants import job_io_fields, job_statuses from video.constants.job_types import JOB_TYPES MANAGE_JOB_POST_SCHEMA: dict[str, Any] = { "type": "object", "properties": { "id": {"type": "integer", "minimum": 1}, }, } ORCHARD_USER_ID_REGEX = r"\A(?:alw|oa):[1-9]{1}\d*\Z" POSITIVE_INTEGER_REGEX = r"\A[1-9]\d*\Z" VALID_STATUS_TRANSITIONS = { job_statuses.SETUP: [ job_statuses.CANCELLED, job_statuses.SUBMITTED, ], job_statuses.SUBMITTED: [ job_statuses.CANCELLED, job_statuses.PROGRESSING, job_statuses.ERROR, ], job_statuses.PROGRESSING: [ job_statuses.PROGRESSING, job_statuses.CANCELLED, job_statuses.ERROR, job_statuses.COMPLETE, ], job_statuses.CANCELLED: [ job_statuses.COMPLETE, job_statuses.ERROR, job_statuses.PROGRESSING, ], job_statuses.ERROR: [ job_statuses.ERROR, job_statuses.PROGRESSING, ], job_statuses.COMPLETE: [ job_statuses.COMPLETE, job_statuses.ERROR, job_statuses.PROGRESSING, ], } def build_response_schema_file_upload_info() -> dict[str, Any]: return { "type": "object", "additionalProperties": False, "properties": { job_io_fields.FILENAME: {"type": "string"}, job_io_fields.FILESIZE: {"type": "number"}, }, "required": [job_io_fields.FILENAME, job_io_fields.FILESIZE], } def build_response_schema_jobs(job_type: str) -> dict[str, Any]: job_inputs = JOB_TYPE_TO_CREATE_JOB_INPUTS_SCHEMA_MAP[job_type] job_outputs = JOB_TYPE_TO_CREATE_JOB_OUTPUTS_SCHEMA_MAP[job_type] job_outputs["properties"][job_io_fields.ERROR_UNKNOWN] = {"type": "string"} return { "type": "object", "additionalProperties": False, "properties": { "id": { "type": "integer", }, "parent_id": { "type": ["integer", "null"], }, "type": { "type": "string", "enum": [job_type], }, "status": { "type": "string", "enum": job_statuses.JOB_STATUSES, }, "inputs": job_inputs, "outputs": job_outputs, "context": { "type": "object", "additionalProperties": False, "properties": { "id": { "type": "integer", "minimum": 1, }, "correlation_id": { "type": "string", }, "product_id": { "type": ["integer", "null"], "minimum": 1, }, "upc": { "type": ["integer", "null"], "minimum": 100000000000, "maximum": 9999999999999, }, "vendor_id": { "type": ["integer", "null"], "minimum": 1, }, "subaccount_id": { "type": ["integer", "null"], "minimum": 1, }, "user_id": { "type": ["string", "null"], "pattern": ORCHARD_USER_ID_REGEX, }, "datetime": { "type": "string", "format": "date-time", }, }, "required": [ "id", "correlation_id", "datetime", ], }, }, "required": [ "id", "parent_id", "type", "status", "inputs", "outputs", ], } def build_request_schema_create_job(job_type: str) -> dict[str, Any]: job_inputs = JOB_TYPE_TO_CREATE_JOB_INPUTS_SCHEMA_MAP[job_type] return { "type": "object", "additionalProperties": False, "properties": { "status": { "type": "string", "enum": job_statuses.JOB_STATUSES, }, "parent_id": { "type": ["integer", "null"], "minimum": 1, }, "type": { "type": "string", "enum": JOB_TYPES, }, "inputs": job_inputs, "context": { "type": "object", "additionalProperties": False, "properties": { "product_id": { "type": "integer", "minimum": 1, }, "upc": { "type": "integer", "minimum": 100000000000, "maximum": 9999999999999, }, }, }, }, "required": [ "type", ], } def build_request_schema_setup_ingestion_reencode_workflow() -> dict[str, Any]: return { "type": "object", "additionalProperties": False, "properties": { job_io_fields.WORKFLOW_INGEST_JOB_ID: {"type": "integer", "minimum": 1}, # Overrides correct the decisions that can misfire: # the crop rectangle (cropdetect) and the clip boundaries — the # trim, which validate_analysis otherwise derives by combining # blackdetect and silencedetect. The raw black/silent intervals are # not exposed directly; the clip is the abstraction over them. # validate_analysis re-derives the dependent outputs (output # resolution, audio adjustment) and, when the clip is not # overridden, the trim. Deterministic measurements (frame rate, # durations, volume) are not judgment calls, so they are not # overridable. "overrides": { "type": "object", "additionalProperties": False, "minProperties": 1, "properties": { job_io_fields.VIDEO_INNER_TOP_LEFT_CORNER_X_PIXELS: { "type": "integer", "minimum": 0, }, job_io_fields.VIDEO_INNER_TOP_LEFT_CORNER_Y_PIXELS: { "type": "integer", "minimum": 0, }, job_io_fields.VIDEO_INNER_WIDTH_PIXELS: { "type": "integer", "minimum": 1, }, job_io_fields.VIDEO_INNER_HEIGHT_PIXELS: { "type": "integer", "minimum": 1, }, job_io_fields.INPUT_CLIP_START_TIMECODE: { "type": ["string", "null"], }, job_io_fields.INPUT_CLIP_END_TIMECODE: { "type": ["string", "null"], }, }, }, "context": { "type": "object", "additionalProperties": False, "properties": { "product_id": {"type": "integer", "minimum": 1}, "upc": { "type": "integer", "minimum": 100000000000, "maximum": 9999999999999, }, }, "required": ["product_id"], }, }, "required": [ job_io_fields.WORKFLOW_INGEST_JOB_ID, "overrides", "context", ], } def build_request_schema_change_job_status( job_type: str, current_job_status: str ) -> dict[str, Any]: job_inputs = JOB_TYPE_TO_CREATE_JOB_INPUTS_SCHEMA_MAP[job_type] job_outputs = JOB_TYPE_TO_CREATE_JOB_OUTPUTS_SCHEMA_MAP[job_type] job_outputs["properties"][job_io_fields.ERROR_UNKNOWN] = {"type": "string"} response: dict[str, Any] = { "type": "object", "additionalProperties": False, "properties": { "id": { "type": "integer", "minimum": 1, }, "inputs": job_inputs, "outputs": job_outputs, }, "required": [ "id", ], } valid_status_transitions = VALID_STATUS_TRANSITIONS[current_job_status] if valid_status_transitions: response["properties"]["status"] = { "type": "string", "enum": valid_status_transitions, } response["required"].append("status") return response def build_request_schema_create_job_inputs(job_type: str) -> dict[str, Any]: return { "type": "object", "additionalProperties": False, "properties": { "id": { "type": "integer", "minimum": 1, }, "status": {"type": "string", "enum": job_statuses.JOB_STATUSES}, "inputs": JOB_TYPE_TO_CREATE_JOB_INPUTS_SCHEMA_MAP[job_type], }, "required": [ "id", "inputs", ], } def build_request_schema_create_job_outputs(job_type: str) -> dict[str, Any]: job_outputs = JOB_TYPE_TO_CREATE_JOB_OUTPUTS_SCHEMA_MAP[job_type] job_outputs["properties"][job_io_fields.ERROR_UNKNOWN] = {"type": "string"} return { "type": "object", "additionalProperties": False, "properties": { "id": { "type": "integer", "minimum": 1, }, "status": {"type": "string", "enum": job_statuses.JOB_STATUSES}, "outputs": job_outputs, }, "required": [ "id", "outputs", ], } def build_query_param_schema_get_jobs() -> dict[str, Any]: return { "type": "object", "additionalProperties": False, "properties": { "job_id": { "type": "string", "pattern": POSITIVE_INTEGER_REGEX, }, "job_parent_id": { "type": "string", "pattern": POSITIVE_INTEGER_REGEX, }, }, }