AI Content Pipeline

from youtube_transcript_api import YouTubeTranscriptApi
import re

def ingest_youtube(video_id: str) -> dict:
    """Fetch and normalise a YouTube transcript."""
    try:
        transcript_list = YouTubeTranscriptApi.get_transcript(
            video_id,
            languages=["en", "en-US", "en-GB"],
        )
    except Exception as e:
        raise ValueError(f"Transcript unavailable for {video_id}: {e}")

    # Merge all segments into a single text block
    full_text = " ".join(seg["text"] for seg in transcript_list)

    # Basic normalisation
    full_text = re.sub(r"\[.*?\]", "", full_text)   # Remove [Music], [Applause] etc.
    full_text = re.sub(r"\s+", " ", full_text).strip()

    return {
        "source_type": "youtube",
        "source_id":   video_id,
        "url":         f"https://youtube.com/watch?v={video_id}",
        "raw_text":    full_text,
        "word_count":  len(full_text.split()),
        "char_count":  len(full_text),
    }

from dataclasses import dataclass, field
from datetime import datetime

@dataclass
class Source:
    source_type:  str          # "youtube" | "web" | "pdf" | "rss"
    source_id:    str          # URL or video ID
    url:          str
    raw_text:     str
    word_count:   int
    char_count:   int
    ingested_at:  str = field(default_factory=lambda: datetime.utcnow().isoformat())
    title:        str = ""
    author:       str = ""
    published_at: str = ""
    tags:         list[str] = field(default_factory=list)

You are a transcript editor. Clean the raw transcript below.

Tasks:
1. Add punctuation and capitalisation where missing
2. Remove filler words (um, uh, you know, like, sort of, basically)
3. Remove false starts and repeated phrases (e.g. "I think — I think what we need...")
4. Correct obvious transcription errors (e.g. "ai" when context means "AI")
5. Preserve all factual content, specific claims, and technical terms exactly

Do NOT:
- Add information not in the original
- Change the meaning of any statement
- Add your own commentary

Return the cleaned transcript as plain text. No preamble or metadata.

RAW TRANSCRIPT:
{raw_text}

You are a research analyst specialising in AI and technology.
Read the source material and extract every distinct insight, claim, statistic, or notable quote.

For each extracted item, assess:
- type: "insight" | "claim" | "statistic" | "quote" | "prediction"
- novelty: 1-5 (5 = something I haven't seen discussed elsewhere)
- importance: 1-5 (5 = genuinely changes how people should think about this)
- source_quote: the verbatim phrase or sentence from the source

Focus on:
✓ Specific claims with concrete numbers or evidence
✓ Counterintuitive or surprising findings
✓ Practical implications for AI practitioners
✓ Predictions with reasoning behind them
✓ Direct quotes from named experts

Ignore:
✗ Vague generalisations ("AI is changing everything")
✗ Common knowledge anyone in the field already knows
✗ Marketing claims without evidence

Return ONLY JSON:
{
  "source_summary": "2-3 sentence summary of what this source is about",
  "speaker":        "Name of main speaker/author if identifiable, else null",
  "insights": [
    {
      "id":           1,
      "type":         "insight",
      "text":         "The insight in one clear sentence",
      "source_quote": "Exact phrase from source",
      "novelty":      4,
      "importance":   5,
      "topic_tags":   ["llm-training", "cost"]
    }
  ]
}

SOURCE MATERIAL:
{cleaned_text}

import json

def score_insight(insight: dict) -> float:
    """Composite score: novelty weighted 60%, importance 40%."""
    return (insight["novelty"] * 0.6) + (insight["importance"] * 0.4)

def filter_insights(extraction: dict, min_score: float = 3.0) -> list[dict]:
    insights = extraction.get("insights", [])
    scored = [(i, score_insight(i)) for i in insights]
    filtered = [(i, s) for i, s in scored if s >= min_score]
    filtered.sort(key=lambda x: x[1], reverse=True)
    return [i for i, _ in filtered]

You are a content strategist for an AI-focused publication.
Create an article outline using the insights below.

Target: {word_count} words, audience: {audience}
Angle: {angle}

The outline must:
1. Have a specific, compelling headline (not generic — make a clear argument)
2. Open with the most surprising or counterintuitive insight
3. Have 4–6 H2 sections, each anchored to 1–3 specific insights
4. End with a practical takeaway the reader can act on

Return JSON:
{
  "headline": "...",
  "subheadline": "One sentence hook under the headline",
  "sections": [
    {
      "heading":  "Section heading",
      "angle":    "What this section argues",
      "insights": [2, 5, 8],
      "word_target": 300
    }
  ]
}

Insights (use insight IDs to reference them):
{insights_json}

You are a technology journalist writing for practitioners who read to learn, not to feel good about AI.

Write the full article following the outline below.

Voice:
- Direct and specific — no filler ("In today's rapidly evolving landscape...")
- Active voice throughout
- Technical depth without jargon for its own sake
- Let the data and insights speak; your job is to frame them clearly

Rules:
- Use ONLY the provided insights as your factual foundation
- When you use an insight, add the insight ID inline: (insight_3)
- Do not add outside knowledge, statistics, or examples not in the insights
- Quotes from the source should be presented as quotes with the speaker's name

Outline: {outline_json}
Insights: {insights_json}

import re, asyncio
from dataclasses import dataclass

BANNED_OPENERS = [
    "in today's rapidly", "in the ever-evolving", "artificial intelligence is",
    "as we stand on the cusp", "the landscape of", "it's no secret that",
    "in recent years,", "needless to say,",
]

@dataclass
class GateResult:
    gate:    str
    passed:  bool
    message: str


def check_length(draft: str, target_words: int) -> GateResult:
    count = len(draft.split())
    within_range = abs(count - target_words) / target_words < 0.20
    return GateResult("length", within_range,
                      f"{count} words (target: {target_words})")


def check_citations(draft: str, insight_ids: set[int]) -> GateResult:
    refs = set(int(m) for m in re.findall(r"\(insight_(\d+)\)", draft))
    orphans = refs - insight_ids
    return GateResult("citations", not orphans,
                      f"Orphaned refs: {orphans}" if orphans else "All citations valid")


def check_filler(draft: str) -> GateResult:
    opening = draft[:300].lower()
    hits = [p for p in BANNED_OPENERS if p in opening]
    return GateResult("filler", not hits,
                      f"Found: {hits}" if hits else "No filler detected")


async def check_groundedness(draft: str, insights_json: str) -> GateResult:
    verdict_raw = await llm_async(
        system=(
            "You are a fact-checker. Review the article draft against the provided insights. "
            "Identify any factual claims in the draft NOT supported by the insights. "
            "Return JSON: {\"ungrounded\": [\"claim 1\", \"claim 2\"], \"passed\": true|false}"
        ),
        user=f"DRAFT:\n{draft}\n\nINSIGHTS:\n{insights_json}",
    )
    verdict = json.loads(verdict_raw)
    return GateResult("groundedness", verdict["passed"],
                      str(verdict.get("ungrounded", [])))


async def run_quality_gates(draft: str, target_words: int,
                             insight_ids: set[int], insights_json: str) -> list[GateResult]:
    # Run automated checks + async groundedness check in parallel
    auto_results = [
        check_length(draft, target_words),
        check_citations(draft, insight_ids),
        check_filler(draft),
    ]
    ground_result = await check_groundedness(draft, insights_json)
    return auto_results + [ground_result]

import json, time
from pathlib import Path

REVIEW_QUEUE = Path("./review_queue")
REVIEW_QUEUE.mkdir(exist_ok=True)

def submit_for_review(state: dict) -> str:
    """Save draft to review queue. Returns the review ID."""
    review_id = f"review_{int(time.time())}_{state['run_id'][:8]}"
    review_path = REVIEW_QUEUE / f"{review_id}.json"
    review_path.write_text(json.dumps({
        "review_id":    review_id,
        "status":       "pending",
        "submitted_at": time.time(),
        "headline":     state["outline"]["headline"],
        "draft":        state["draft"],
        "gate_results": [g.__dict__ for g in state["gate_results"]],
        "insights":     state["filtered_insights"],
        "source_url":   state["source"]["url"],
    }, indent=2))
    return review_id


def get_review_status(review_id: str) -> dict:
    """Check if a review has been completed."""
    review_path = REVIEW_QUEUE / f"{review_id}.json"
    if not review_path.exists():
        raise FileNotFoundError(f"Review {review_id} not found")
    return json.loads(review_path.read_text())


def approve_review(review_id: str, editor_notes: str = "") -> dict:
    """Mark a review as approved. Call this from the review UI or CLI."""
    data = get_review_status(review_id)
    data["status"]       = "approved"
    data["approved_at"]  = time.time()
    data["editor_notes"] = editor_notes
    path = REVIEW_QUEUE / f"{review_id}.json"
    path.write_text(json.dumps(data, indent=2))
    return data

You are an SEO specialist. Generate metadata for the article below.

Return ONLY JSON:
{
  "title":            "Title tag (max 60 chars, include primary keyword)",
  "meta_description": "Meta description (150-160 chars, include keyword, ends with action hook)",
  "primary_keyword":  "Main keyword phrase to target",
  "secondary_keywords": ["keyword 2", "keyword 3", "keyword 4"],
  "slug":             "url-friendly-slug",
  "og_title":         "Open Graph title (can be more engaging than title tag)",
  "og_description":   "OG description (1-2 sentences, compelling for social sharing)",
  "schema_type":      "Article",
  "featured_image_prompt": "Prompt for DALL-E or Midjourney to generate a relevant header image"
}

ARTICLE:
{approved_draft}