diff --git a/backend/routers/pipeline.py b/backend/routers/pipeline.py index af2f01f..7b720b2 100644 --- a/backend/routers/pipeline.py +++ b/backend/routers/pipeline.py @@ -924,6 +924,109 @@ async def bulk_resynthesize( } +# ── Admin: Wipe All Output ────────────────────────────────────────────────── + +@router.post("/admin/pipeline/wipe-all-output") +async def wipe_all_output( + db: AsyncSession = Depends(get_session), +): + """Wipe ALL pipeline output while preserving raw input data. + + Deletes: technique_page_versions, related_technique_links, technique_pages, + key_moments, pipeline_events, pipeline_runs, Qdrant vectors, Redis cache. + Resets: all video processing_status to 'not_started', classification_data to null. + Preserves: creators, source_videos (metadata), transcript_segments. + """ + from models import TechniquePageVersion, TechniquePage, RelatedTechniqueLink, PipelineRun + + counts = {} + + # Order matters due to FK constraints: versions → links → pages → moments → events → runs + + # 1. Technique page versions + result = await db.execute(TechniquePageVersion.__table__.delete()) + counts["technique_page_versions"] = result.rowcount + + # 2. Related technique links + result = await db.execute(RelatedTechniqueLink.__table__.delete()) + counts["related_technique_links"] = result.rowcount + + # 3. Key moments (FK to technique_pages, must clear before pages) + result = await db.execute(KeyMoment.__table__.delete()) + counts["key_moments"] = result.rowcount + + # 4. Technique pages + result = await db.execute(TechniquePage.__table__.delete()) + counts["technique_pages"] = result.rowcount + + # 5. Pipeline events + result = await db.execute(PipelineEvent.__table__.delete()) + counts["pipeline_events"] = result.rowcount + + # 6. Pipeline runs + result = await db.execute(PipelineRun.__table__.delete()) + counts["pipeline_runs"] = result.rowcount + + # 7. Reset all video statuses and classification data + from sqlalchemy import update + result = await db.execute( + update(SourceVideo).values( + processing_status=ProcessingStatus.not_started, + classification_data=None, + ) + ) + counts["videos_reset"] = result.rowcount + + await db.commit() + + # 8. Clear Qdrant vectors (best-effort) + try: + settings = get_settings() + from pipeline.qdrant_client import QdrantManager + qdrant = QdrantManager(settings) + qdrant.ensure_collection() + # Delete entire collection and recreate empty + from qdrant_client.models import Distance, VectorParams + qdrant.client.delete_collection(settings.qdrant_collection) + qdrant.client.create_collection( + collection_name=settings.qdrant_collection, + vectors_config=VectorParams( + size=settings.embedding_dimensions, + distance=Distance.COSINE, + ), + ) + counts["qdrant"] = "collection_recreated" + except Exception as exc: + logger.warning("Qdrant wipe failed: %s", exc) + counts["qdrant"] = f"skipped: {exc}" + + # 9. Clear Redis pipeline cache keys (best-effort) + try: + import redis as redis_lib + settings = get_settings() + r = redis_lib.Redis.from_url(settings.redis_url) + cursor = 0 + deleted_keys = 0 + while True: + cursor, keys = r.scan(cursor, match="chrysopedia:*", count=100) + if keys: + r.delete(*keys) + deleted_keys += len(keys) + if cursor == 0: + break + counts["redis_keys_deleted"] = deleted_keys + except Exception as exc: + logger.warning("Redis wipe failed: %s", exc) + counts["redis"] = f"skipped: {exc}" + + logger.info("[WIPE] All pipeline output wiped: %s", counts) + + return { + "status": "wiped", + "deleted": counts, + } + + # ── Admin: Prompt Optimization ────────────────────────────────────────────── @router.post("/admin/pipeline/optimize-prompt/{stage}") diff --git a/frontend/src/api/public-client.ts b/frontend/src/api/public-client.ts index 0f18f42..43ef801 100644 --- a/frontend/src/api/public-client.ts +++ b/frontend/src/api/public-client.ts @@ -690,6 +690,19 @@ export async function bulkResynthesize( }); } +// ── Wipe All Output ──────────────────────────────────────────────────────── + +export interface WipeAllResponse { + status: string; + deleted: Record; +} + +export async function wipeAllOutput(): Promise { + return request(`${BASE}/admin/pipeline/wipe-all-output`, { + method: "POST", + }); +} + // ── Debug Mode ────────────────────────────────────────────────────────────── export interface DebugModeResponse { diff --git a/frontend/src/pages/AdminPipeline.tsx b/frontend/src/pages/AdminPipeline.tsx index e6f2712..6ecf240 100644 --- a/frontend/src/pages/AdminPipeline.tsx +++ b/frontend/src/pages/AdminPipeline.tsx @@ -20,6 +20,7 @@ import { fetchChunkingData, fetchStalePages, bulkResynthesize, + wipeAllOutput, fetchCreators, fetchRecentActivity, type PipelineVideoItem, @@ -987,6 +988,27 @@ export default function AdminPipeline() { } }; + const handleWipeAll = async () => { + if (!confirm("WIPE ALL pipeline output? This deletes all technique pages, key moments, pipeline events, runs, and Qdrant vectors. Creators, videos, and transcripts are preserved. This cannot be undone.")) return; + if (!confirm("Are you sure? This is irreversible.")) return; + try { + const res = await wipeAllOutput(); + setActionMessage({ + id: "__wipe__", + text: `Wiped: ${JSON.stringify(res.deleted)}`, + ok: true, + }); + setStalePagesCount(null); + void load(); + } catch (err) { + setActionMessage({ + id: "__wipe__", + text: err instanceof Error ? err.message : "Wipe failed", + ok: false, + }); + } + }; + // Deep-link: auto-expand and scroll to ?video= on first load useEffect(() => { if (deepLinked.current || loading || videos.length === 0) return; @@ -1188,6 +1210,13 @@ export default function AdminPipeline() { {stalePagesCount} stale pages )} +