The hard nudity detection task analyzes an MP4 video for explicit nudity involving genitals. Submit a task with `POST /streaming/ai/tasks`, then poll `GET /streaming/ai/tasks/{task_id}` until the status is `SUCCESS` or `FAILURE`.
An [API token](/account-settings/api-tokens) is required.
```bash theme={null}
export GCORE_API_KEY="{YOUR_API_KEY}"
```
## Create a hard nudity detection task
Send `POST /streaming/ai/tasks` with `task_name: "content-moderation"` and `category: "hard_nudity"`. The response returns a `task_id` to poll for results.
```python theme={null}
from gcore import Gcore
client = Gcore()
task = client.streaming.ai_tasks.create(
task_name="content-moderation",
category="hard_nudity",
url="https://example.com/video.mp4",
)
print(f"Task ID: {task.task_id}")
```
```go theme={null}
package main
import (
"context"
"fmt"
gcore "github.com/G-Core/gcore-go"
"github.com/G-Core/gcore-go/streaming"
)
func main() {
client := gcore.NewClient()
ctx := context.Background()
task, err := client.Streaming.AITasks.New(ctx, streaming.AITaskNewParams{
OfAIContentModerationHardNudityTaskData: &streaming.AIContentmoderationHardnudityParam{
TaskName: streaming.AIContentmoderationHardnudityTaskNameContentModeration,
Category: streaming.AIContentmoderationHardnudityCategoryHardNudity,
URL: "https://example.com/video.mp4",
},
})
if err != nil {
panic(err)
}
fmt.Printf("Task ID: %s\n", task.TaskID)
}
```
```bash theme={null}
curl -X POST "https://api.gcore.com/streaming/ai/tasks" \
-H "Authorization: APIKey $GCORE_API_KEY" \
-H "Content-Type: application/json" \
-d '{
"task_name": "content-moderation",
"category": "hard_nudity",
"url": "https://example.com/video.mp4"
}'
```
The API returns:
```json theme={null}
{"task_id": "18a5b96f-c17d-413e-b8ed-f5bf1fdc158c"}
```
## Poll for detection result
Use `GET /streaming/ai/tasks/{task_id}` to check task status. Poll until `status` is `"SUCCESS"` or `"FAILURE"`.
```python theme={null}
import os
import time
from gcore import Gcore
task_id = os.environ["TASK_ID"]
client = Gcore()
while True:
task = client.streaming.ai_tasks.get(task_id)
print(f"Status: {task.status} Progress: {task.progress}%")
if task.status in ("SUCCESS", "FAILURE"):
break
time.sleep(5)
if task.status == "SUCCESS":
print(f"Nudity detected: {task.result.nudity_detected}")
for frame in task.result.frames:
print(f" Frame {frame.frame_number}: {frame.label} (confidence {frame.confidence:.0%})")
```
```go theme={null}
package main
import (
"context"
"fmt"
"os"
"time"
gcore "github.com/G-Core/gcore-go"
)
func main() {
taskID := os.Getenv("TASK_ID")
client := gcore.NewClient()
ctx := context.Background()
for {
task, err := client.Streaming.AITasks.Get(ctx, taskID)
if err != nil {
panic(err)
}
fmt.Printf("Status: %s Progress: %d%%\n", task.Status, task.Progress)
if task.Status == "SUCCESS" || task.Status == "FAILURE" {
if task.Status == "SUCCESS" {
result := task.Result.AsAITaskGetResponseResultAIResultsContentmoderationHardnudity()
fmt.Printf("Nudity detected: %v\n", result.NudityDetected)
for _, frame := range result.Frames {
fmt.Printf(" Frame %d: %s (confidence %.0f%%)\n",
frame.FrameNumber, frame.Label, frame.Confidence*100)
}
}
break
}
time.Sleep(5 * time.Second)
}
}
```
```bash theme={null}
curl "https://api.gcore.com/streaming/ai/tasks/$TASK_ID" \
-H "Authorization: APIKey $GCORE_API_KEY"
```
The API returns:
```json theme={null}
{
"task_id": "18a5b96f-c17d-413e-b8ed-f5bf1fdc158c",
"status": "SUCCESS",
"progress": 100,
"result": {
"nudity_detected": false,
"detection_results": [],
"frames": []
}
}
```
**Interpreting the result**
On success, the `result` object contains:
* `nudity_detected: false` — no hard nudity content found in the video.
* `nudity_detected: true` — hard nudity detected. The `detection_results` array lists the detected object categories (e.g. `"FEMALE_BREAST_EXPOSED"`, `"MALE_GENITALIA_EXPOSED"`). The `frames` array lists each flagged frame with `frame_number`, `label`, and `confidence` (decimal 0.0–1.0, e.g. `0.75` means 75% confidence).
## List AI tasks
To retrieve all AI tasks, send `GET /streaming/ai/tasks`. Use the optional `search` query parameter to filter by ID, URL, task type, or status.
```python theme={null}
from gcore import Gcore
client = Gcore()
for task in client.streaming.ai_tasks.list():
print(f"{task.task_id} {task.status} {task.progress}%")
```
```go theme={null}
package main
import (
"context"
"fmt"
gcore "github.com/G-Core/gcore-go"
"github.com/G-Core/gcore-go/streaming"
)
func main() {
client := gcore.NewClient()
ctx := context.Background()
page, err := client.Streaming.AITasks.List(ctx, streaming.AITaskListParams{})
if err != nil {
panic(err)
}
for _, task := range page.Results {
fmt.Printf("%s %s %d%%\n", task.TaskID, task.Status, task.Progress)
}
}
```
```bash theme={null}
curl "https://api.gcore.com/streaming/ai/tasks" \
-H "Authorization: APIKey $GCORE_API_KEY"
```
## Cancel an AI task
To cancel a running task, send `POST /streaming/ai/tasks/{task_id}/cancel`.
```python theme={null}
import os
from gcore import Gcore
task_id = os.environ["TASK_ID"]
client = Gcore()
client.streaming.ai_tasks.cancel(task_id)
print(f"Task {task_id} canceled")
```
```go theme={null}
package main
import (
"context"
"fmt"
"os"
gcore "github.com/G-Core/gcore-go"
)
func main() {
taskID := os.Getenv("TASK_ID")
client := gcore.NewClient()
ctx := context.Background()
_, err := client.Streaming.AITasks.Cancel(ctx, taskID)
if err != nil {
panic(err)
}
fmt.Printf("Task %s canceled\n", taskID)
}
```
```bash theme={null}
curl -X POST "https://api.gcore.com/streaming/ai/tasks/$TASK_ID/cancel" \
-H "Authorization: APIKey $GCORE_API_KEY"
```
