File size: 3,127 Bytes
345855e | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 | import asyncio
from dataclasses import dataclass
# -------------------------------------------------
# CONTEXT NORMALIZER
# -------------------------------------------------
@dataclass
class Context:
input_file: any = None
url_input: str = None
params: dict = None
def normalize_context(ctx):
if isinstance(ctx, dict):
return Context(
input_file=ctx.get("input_file"),
url_input=ctx.get("url_input"),
params=ctx.get("params", {}) or {}
)
return ctx
# -------------------------------------------------
# PIPELINE ORCHESTRATOR
# -------------------------------------------------
async def run(context):
"""
Autonomous pipeline:
1. Transcribe
2. Extract highlights
3. Compute viral score
4. Generate strategy
"""
context = normalize_context(context)
try:
# -------------------------------------------------
# STEP 1 — TRANSCRIPTION
# -------------------------------------------------
from publisher.tasks.transcribe import run as transcribe_task
transcript = await transcribe_task(context)
if transcript.get("status") != "success":
return {
"status": "error",
"stage": "transcribe",
"detail": transcript
}
# -------------------------------------------------
# STEP 2 — HIGHLIGHTS
# -------------------------------------------------
from publisher.tasks.highlights import run as highlights_task
highlights = await highlights_task(context)
# -------------------------------------------------
# STEP 3 — VIRAL SCORE
# -------------------------------------------------
from publisher.tasks.viral_score import run as viral_task
viral = await viral_task(context)
# -------------------------------------------------
# STEP 4 — STRATEGY GENERATION
# -------------------------------------------------
from publisher.tasks.strategy import run as strategy_task
strategy = await strategy_task(context)
# -------------------------------------------------
# AGGREGATED OUTPUT
# -------------------------------------------------
segments = transcript.get("segments", [])
full_text = " ".join([s["text"] for s in segments])
return {
"status": "success",
"pipeline": "autonomous-v11",
"summary": {
"segments": len(segments),
"highlights": highlights.get("count", 0),
"viral_score": viral.get("viral_score", 0),
},
"transcript": {
"text": full_text,
"segments": segments
},
"highlights": highlights.get("highlights", []),
"viral": viral,
"strategy": strategy
}
except Exception as e:
return {
"status": "error",
"message": str(e),
"stage": "autonomous_pipeline_failed"
} |