From ffa47a13353c08814747c4dbff466762ec6e5949 Mon Sep 17 00:00:00 2001 From: "Gan, Jimmy" Date: Wed, 8 Jul 2026 22:05:05 +0800 Subject: [PATCH] sprint1: agent queue worker, proxy support, auto-mark-on-navigate-away, test fixes --- backend/agent_worker.py | 166 ++++++++++++++++++++ backend/db.py | 1 + backend/fetcher.py | 6 +- backend/main.py | 32 +++- backend/summarizer.py | 7 +- backend/tests/test_agent_worker.py | 112 +++++++++++++ frontend/src/routes/YoutubeDashboard.svelte | 89 +++++++++-- run_tests.sh | 44 ++++-- 8 files changed, 429 insertions(+), 28 deletions(-) create mode 100644 backend/agent_worker.py create mode 100644 backend/tests/test_agent_worker.py diff --git a/backend/agent_worker.py b/backend/agent_worker.py new file mode 100644 index 0000000..97782d8 --- /dev/null +++ b/backend/agent_worker.py @@ -0,0 +1,166 @@ +""" +Agent Queue Worker — polls agent_queue for pending items and +processes them against the local Qwen27B API (:8080). +""" +import asyncio +import json +import logging +import os +import aiosqlite +import httpx + +import db + +log = logging.getLogger(__name__) + +QWEN_BASE_URL = os.environ.get("QWEN_BASE_URL", "http://localhost:8080") +QWEN_MODEL = os.environ.get("QWEN_MODEL", "current") +POLL_INTERVAL = int(os.environ.get("AGENT_POLL_INTERVAL", "30")) +MAX_CONCURRENT = int(os.environ.get("AGENT_MAX_CONCURRENT", "3")) +SYSTEM_PROMPT = os.environ.get( + "AGENT_SYSTEM_PROMPT", + "You are a helpful AI assistant. The user has watched a YouTube video and " + "left you instructions. Read the video info below and respond to their " + "prompt. Be concise and actionable." +) + +semaphore = asyncio.Semaphore(MAX_CONCURRENT) + + +async def _call_qwen(prompt: str, video_title: str, video_context: str) -> str: + """Call the local Qwen27B API with video context + user prompt.""" + messages = [ + {"role": "system", "content": SYSTEM_PROMPT}, + { + "role": "user", + "content": ( + f"## Video Title\n{video_title}\n\n" + f"## Video Context\n{video_context}\n\n" + f"## User's Request\n{prompt}" + ), + }, + ] + + http_proxy = os.environ.get("HTTP_PROXY") or os.environ.get("http_proxy") + https_proxy = os.environ.get("HTTPS_PROXY") or os.environ.get("https_proxy") + proxies = http_proxy or https_proxy + + async with httpx.AsyncClient(proxy=proxies, timeout=180) as client: + resp = await client.post( + f"{QWEN_BASE_URL}/v1/chat/completions", + json={ + "model": QWEN_MODEL, + "messages": messages, + "max_tokens": 2048, + "temperature": 0.3, + }, + ) + resp.raise_for_status() + data = resp.json() + return data["choices"][0]["message"]["content"] + + +async def _process_item(item: dict) -> None: + """Process a single agent_queue item.""" + video_id = item["video_id"] + user_prompt = item["user_prompt"] + + async with aiosqlite.connect(db.DB_PATH) as conn: + conn.row_factory = aiosqlite.Row + + # Mark as processing + await conn.execute( + "UPDATE agent_queue SET status = 'processing' WHERE id = ?", + (item["id"],), + ) + await conn.commit() + + # Get video info + cursor = await conn.execute( + "SELECT title, channel_name, summary, transcript FROM videos WHERE video_id = ?", + (video_id,), + ) + video = await cursor.fetchone() + + if not video: + async with aiosqlite.connect(db.DB_PATH) as conn: + await conn.execute( + "UPDATE agent_queue SET status = 'failed', result = 'Video not found in DB' WHERE id = ?", + (item["id"],), + ) + await conn.commit() + return + + # Build video context from whatever we have + context_parts = [] + if video["channel_name"]: + context_parts.append(f"Channel: {video['channel_name']}") + if video["summary"] and not video["summary"].startswith("📺 Short clip"): + context_parts.append(f"Summary:\n{video['summary']}") + if video["transcript"] and video["transcript"] != "No transcript available": + context_parts.append(f"Transcript (first 4000 chars):\n{video['transcript'][:4000]}") + video_context = "\n\n".join(context_parts) if context_parts else "No additional context available." + + try: + result = await _call_qwen(user_prompt, video["title"], video_context) + status = "done" + except Exception as e: + log.warning(f"Qwen API call failed for item {item['id']}: {e}") + result = f"Error: {e}" + status = "failed" + + async with aiosqlite.connect(db.DB_PATH) as conn: + await conn.execute( + "UPDATE agent_queue SET status = ?, result = ?, processed_at = CURRENT_TIMESTAMP WHERE id = ?", + (status, result, item["id"]), + ) + await conn.commit() + + +async def process_queue_loop() -> None: + """Main polling loop — runs forever.""" + log.info( + f"Agent worker started: Qwen at {QWEN_BASE_URL}, " + f"poll every {POLL_INTERVAL}s, max {MAX_CONCURRENT} concurrent" + ) + + while True: + try: + async with aiosqlite.connect(db.DB_PATH) as conn: + conn.row_factory = aiosqlite.Row + cursor = await conn.execute( + "SELECT id, video_id, user_prompt FROM agent_queue WHERE status = 'pending' ORDER BY created_at ASC LIMIT ?", + (MAX_CONCURRENT,), + ) + items = await cursor.fetchall() + + if items: + log.info(f"Agent queue: processing {len(items)} item(s)") + tasks = [_process_item(dict(item)) for item in items] + await asyncio.gather(*tasks) + + except Exception as e: + log.error(f"Agent worker loop error: {e}") + + await asyncio.sleep(POLL_INTERVAL) + + +async def start_worker(): + """Entry point — called from main.py startup.""" + task = asyncio.create_task(process_queue_loop()) + log.info("Agent queue worker scheduled") + return task + + +if __name__ == "__main__": + logging.basicConfig( + level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s" + ) + + async def main(): + from db import init_db + await init_db() + log.info("Agent worker running in foreground. Ctrl+C to stop.") + await process_queue_loop() + + asyncio.run(main()) diff --git a/backend/db.py b/backend/db.py index 76098db..7f81d55 100644 --- a/backend/db.py +++ b/backend/db.py @@ -39,6 +39,7 @@ async def init_db(): status TEXT DEFAULT 'pending', created_at TEXT DEFAULT CURRENT_TIMESTAMP, processed_at TEXT, + result TEXT, FOREIGN KEY (video_id) REFERENCES videos(video_id) ) """) diff --git a/backend/fetcher.py b/backend/fetcher.py index d07ee93..c606e26 100644 --- a/backend/fetcher.py +++ b/backend/fetcher.py @@ -99,7 +99,11 @@ async def fetch_subscriptions(): # Step 3: Fetch RSS from slice channels (free, parallel) rss_videos = {} - async with httpx.AsyncClient(timeout=20) as client: + async with httpx.AsyncClient( + proxy=os.environ.get("HTTP_PROXY") or os.environ.get("http_proxy") or + os.environ.get("HTTPS_PROXY") or os.environ.get("https_proxy") or None, + timeout=20 + ) as client: tasks = [_fetch_rss(client, cid) for cid in slice_ids] results = await asyncio.gather(*tasks) for cid, entries in zip(slice_ids, results): diff --git a/backend/main.py b/backend/main.py index 3e82548..ad4af08 100644 --- a/backend/main.py +++ b/backend/main.py @@ -1,7 +1,9 @@ import os import json import logging -from fastapi import FastAPI, BackgroundTasks, Depends, HTTPException, Query +from fastapi import FastAPI, BackgroundTasks, Depends, HTTPException, Query, Response +from fastapi import Request +from fastapi.responses import JSONResponse from fastapi.middleware.cors import CORSMiddleware from fastapi.staticfiles import StaticFiles from pydantic import BaseModel @@ -11,6 +13,7 @@ import aiosqlite from db import init_db, get_db from fetcher import fetch_subscriptions from summarizer import summarize_video, summarize_all_pending +from agent_worker import start_worker logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s') log = logging.getLogger(__name__) @@ -23,8 +26,25 @@ app.add_middleware( allow_credentials=True, allow_methods=["*"], allow_headers=["*"], + ) +# API token validation for write operations +API_TOKEN=os.environ.get('API_TOKEN', '') + +async def validate_api_token(request: Request, call_next): + path = request.url.path + method = request.method + # Only check auth for write operations on /api/* + if path.startswith("/api/") and method in ("POST", "PATCH", "DELETE") and API_TOKEN: + auth_header = request.headers.get("Authorization", "") + if not auth_header.startswith("Bearer ") or auth_header[7:] != API_TOKEN: + return JSONResponse(status_code=401, content={"detail": "Unauthorized"}) + return await call_next(request) + +app.middleware("http")(validate_api_token) + + class StatusUpdate(BaseModel): status: str @@ -45,11 +65,21 @@ class AgentQueueUpdate(BaseModel): @app.on_event("startup") async def startup(): await init_db() + if os.environ.get("AGENT_ENABLED", "").lower() == "true": + await start_worker() + log.info("Agent queue worker enabled") + else: + log.info("Agent queue worker disabled (set AGENT_ENABLED=true to enable)") @app.get("/api/health") async def health(): return {"status": "ok"} +@app.get("/api/config") +async def api_config(): + """Return API config without exposing the token.""" + return {"auth_enabled": bool(API_TOKEN)} + @app.get("/auto_categories.json") async def auto_categories(): path = os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "frontend", "src", "auto_categories.json") diff --git a/backend/summarizer.py b/backend/summarizer.py index 2cd8bb9..cbdd0eb 100644 --- a/backend/summarizer.py +++ b/backend/summarizer.py @@ -9,6 +9,11 @@ from db import DB_PATH log = logging.getLogger(__name__) +def proxy_from_env(): + """Return proxy URL from HTTP_PROXY/HTTPS_PROXY env vars, or None.""" + return os.environ.get("HTTP_PROXY") or os.environ.get("http_proxy") or \ + os.environ.get("HTTPS_PROXY") or os.environ.get("https_proxy") or None + # Fallback pattern to retrieve the correct bytecatcode credentials API_KEY = os.environ.get("DEEPSEEK_API_KEY", os.environ.get("OPENAI_API_KEY", "")) BASE_URL = os.environ.get("DEEPSEEK_BASE_URL", "https://api.deepseek.com") @@ -108,7 +113,7 @@ async def summarize_video(video_id: str) -> str | None: ] } - async with httpx.AsyncClient(timeout=120) as client: + async with httpx.AsyncClient(proxy=proxy_from_env(), timeout=120) as client: try: resp = await client.post( f"{BASE_URL}/v1/chat/completions", diff --git a/backend/tests/test_agent_worker.py b/backend/tests/test_agent_worker.py new file mode 100644 index 0000000..3de3186 --- /dev/null +++ b/backend/tests/test_agent_worker.py @@ -0,0 +1,112 @@ +"""Tests for agent_worker.py — mocks Qwen API.""" +import pytest +import os +import json +import aiosqlite +import httpx +from unittest.mock import patch, AsyncMock, MagicMock + + +@pytest.mark.asyncio +async def test_agent_worker_polls_and_processes(test_db, monkeypatch): + """Worker picks up a pending item, calls Qwen, marks it done.""" + from agent_worker import _process_item + + # Seed a pending queue item + video + await test_db.execute(""" + INSERT INTO videos (video_id, title, channel_name, url, summary, status) + VALUES ('ag_vid', 'Test Video', 'Test Channel', 'https://youtube.com/watch?v=ag_vid', + '**Summary**\n- Point 1\n- Point 2\nVerdict.', 'pending') + """) + await test_db.execute(""" + INSERT INTO agent_queue (video_id, user_prompt, status) + VALUES ('ag_vid', 'Save this as a skill', 'pending') + """) + await test_db.commit() + + # Get the item + cursor = await test_db.execute( + "SELECT id, video_id, user_prompt FROM agent_queue WHERE status = 'pending'" + ) + item = dict(await cursor.fetchone()) + + # Mock Qwen API + mock_response = MagicMock(spec=httpx.Response) + mock_response.status_code = 200 + mock_response.json.return_value = { + "choices": [{"message": {"content": "I've saved the skill. Done."}}] + } + + with patch("httpx.AsyncClient.post", new_callable=AsyncMock, return_value=mock_response): + await _process_item(item) + + # Verify + cursor = await test_db.execute( + "SELECT status, result FROM agent_queue WHERE id = ?", (item["id"],) + ) + row = await cursor.fetchone() + assert row["status"] == "done" + assert "I've saved the skill" in row["result"] + + +@pytest.mark.asyncio +async def test_agent_worker_handles_missing_video(test_db): + """When video doesn't exist, marks as failed gracefully.""" + from agent_worker import _process_item + + await test_db.execute(""" + INSERT INTO agent_queue (video_id, user_prompt, status) + VALUES ('missing_vid', 'Do something', 'pending') + """) + await test_db.commit() + + cursor = await test_db.execute( + "SELECT id, video_id, user_prompt FROM agent_queue WHERE status = 'pending'" + ) + item = dict(await cursor.fetchone()) + + await _process_item(item) + + cursor = await test_db.execute( + "SELECT status, result FROM agent_queue WHERE id = ?", (item["id"],) + ) + row = await cursor.fetchone() + assert row["status"] == "failed" + assert "not found" in row["result"].lower() + + +@pytest.mark.asyncio +async def test_agent_worker_qwen_error_marks_failed(test_db): + """Qwen API failure should mark as failed, not crash.""" + from agent_worker import _process_item + + await test_db.execute(""" + INSERT INTO videos (video_id, title, channel_name, url, status) + VALUES ('err_vid', 'Error Video', 'Test', 'https://youtube.com/watch?v=err_vid', 'pending') + """) + await test_db.execute(""" + INSERT INTO agent_queue (video_id, user_prompt, status) + VALUES ('err_vid', 'Do it', 'pending') + """) + await test_db.commit() + + cursor = await test_db.execute( + "SELECT id, video_id, user_prompt FROM agent_queue WHERE status = 'pending'" + ) + item = dict(await cursor.fetchone()) + + # Mock HTTP error + mock_response = MagicMock(spec=httpx.Response) + mock_response.status_code = 503 + mock_response.raise_for_status.side_effect = httpx.HTTPStatusError( + "503 Service Unavailable", request=MagicMock(), response=mock_response + ) + + with patch("httpx.AsyncClient.post", new_callable=AsyncMock, return_value=mock_response): + await _process_item(item) + + cursor = await test_db.execute( + "SELECT status FROM agent_queue WHERE id = ?", (item["id"],) + ) + row = await cursor.fetchone() + assert row["status"] == "failed" diff --git a/frontend/src/routes/YoutubeDashboard.svelte b/frontend/src/routes/YoutubeDashboard.svelte index 82ec240..e86877d 100644 --- a/frontend/src/routes/YoutubeDashboard.svelte +++ b/frontend/src/routes/YoutubeDashboard.svelte @@ -9,6 +9,9 @@ let syncMessage = $state(''); let pendingCount = $state(0); let theme = $state(localStorage.getItem('theme') || 'dark'); + let apiToken = $state(localStorage.getItem('t-yt-token') || ''); + let apiTokenInput = $state(''); + let showTokenSetup = $state(false); let summaryPanel = $state(null); let notes = $state({}); let activeNote = $state(null); @@ -17,6 +20,7 @@ let channelFilter = $state(''); let allChannels = $state([]); let editingCategories = $state(false); + let prevSelectedId = $state(null); // Agent queue state let agentItems = $state([]); @@ -93,7 +97,7 @@ isSyncing = true; syncMessage = 'Syncing...'; try { - const res = await fetch('/api/videos/fetch', { method: 'POST' }); + const res = await fetch('/api/videos/fetch', { method: 'POST', headers: getAuthHeaders() }); if (res.ok) { syncMessage = 'Sync started...'; setTimeout(() => { loadVideos(); isSyncing = false; syncMessage = ''; }, 6000); @@ -105,7 +109,7 @@ } async function updateStatus(videoId, newStatus, silent = false) { - await fetch(`/api/videos/${videoId}/status`, { method: 'POST', headers: { 'Content-Type': 'application/json' }, body: JSON.stringify({ status: newStatus }) }); + await fetch(`/api/videos/${videoId}/status`, { method: 'POST', headers: getAuthHeaders({ 'Content-Type': 'application/json' }), body: JSON.stringify({ status: newStatus }) }); if (!silent) { videos = videos.filter(v => v.video_id !== videoId); if (selectedVideo?.video_id === videoId) selectedVideo = videos[0] || null; @@ -115,7 +119,7 @@ async function generateSummary(videoId) { if (selectedVideo?.video_id === videoId) selectedVideo.summary = "Generating..."; - await fetch(`/api/videos/${videoId}/summarize`, { method: 'POST' }); + await fetch(`/api/videos/${videoId}/summarize`, { method: 'POST', headers: getAuthHeaders() }); setTimeout(() => loadVideos(), 3000); } @@ -144,7 +148,7 @@ for (const k of missing) payload[k] = local[k]; await fetch('/api/notes/migrate', { method: 'POST', - headers: { 'Content-Type': 'application/json' }, + headers: getAuthHeaders({ 'Content-Type': 'application/json' }), body: JSON.stringify({ notes: payload }) }); await loadNotes(); @@ -160,7 +164,7 @@ // Save to server await fetch(`/api/notes/${vid}`, { method: 'POST', - headers: { 'Content-Type': 'application/json' }, + headers: getAuthHeaders({ 'Content-Type': 'application/json' } ), body: JSON.stringify({ note_text: current, queue_to_agent: queueToAgent }) }); // Update local state @@ -181,21 +185,58 @@ } async function removeAgentItem(id) { - await fetch(`/api/agent-queue/${id}`, { method: 'DELETE' }); + await fetch(`/api/agent-queue/${id}`, { method: 'DELETE', headers: getAuthHeaders() }); agentItems = agentItems.filter(i => i.id !== id); } + async function initAuth() { + // Check if auth is enabled on the backend + try { + const res = await fetch('/api/config'); + if (res.ok) { + const config = await res.json(); + if (config.auth_enabled && !apiToken) { + showTokenSetup = true; + return false; + } + } + } catch(e) {} + showTokenSetup = false; + return true; + } + + async function setupToken() { + if (apiTokenInput.trim()) { + apiToken = apiTokenInput.trim(); + localStorage.setItem('t-yt-token', apiToken); + showTokenSetup = false; + } + } + + function getAuthHeaders(extraHeaders = {}) { + const headers = { ...extraHeaders }; + if (apiToken) { + headers['Authorization'] = 'Bearer ' + apiToken; + } + return headers; + } + // ---- Lifecycle ---- $effect(() => { if (activeTab) loadVideos(); }); $effect(() => { + // Mark previous video as read when navigating away if (selectedVideo && summaryPanel) { tick().then(() => summaryPanel.scrollTop = 0); - const vid = selectedVideo.video_id; - const hasSummary = selectedVideo.summary && !selectedVideo.summary.includes('Error') && !selectedVideo.summary.includes('📺 Short clip'); - if (hasSummary && activeTab === 'pending') { - const timer = setTimeout(() => updateStatus(vid, 'read', true), 3000); - return () => clearTimeout(timer); + } + const prevId = prevSelectedId; + prevSelectedId = selectedVideo?.video_id || null; + + if (prevId && prevId !== selectedVideo?.video_id && activeTab === 'pending') { + // Look up the previous video to check it had a valid summary + const prevVideo = videos.find(v => v.video_id === prevId); + if (prevVideo && prevVideo.summary && !prevVideo.summary.includes('Error') && !prevVideo.summary.includes('📺 Short clip')) { + updateStatus(prevId, 'read', true); } } }); @@ -205,6 +246,7 @@ onMount(async () => { document.documentElement.className = theme; + await initAuth(); await loadNotes(); // Migrate localStorage notes to server (non-blocking) migrateLocalNotes(); @@ -323,6 +365,31 @@
+ + {#if showTokenSetup} +
+
+

API Token Required

+

+ This app requires an API token to save notes and manage videos. + Please enter the token configured on the server. +

+ e.key === 'Enter' && setupToken()} /> +
+ +
+
+
+ {/if} +
T YouTube diff --git a/run_tests.sh b/run_tests.sh index 508db9e..c7bf22a 100755 --- a/run_tests.sh +++ b/run_tests.sh @@ -4,15 +4,20 @@ set -e DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)" VIDEO_ID="jNQXAC9IVRw" -# 1. Auto-detect and configure SOCKS5 proxy if active on 7890 +# 1. Auto-detect and configure proxy +PROXY_DETECTED=false if nc -z 127.0.0.1 7890 >/dev/null 2>&1; then - echo "=== Auto-detected local proxy on port 7890. Enabling proxy... ===" + echo "=== Auto-detected local proxy on port 7890. Enabling SOCKS5 proxy... ===" export http_proxy=socks5://127.0.0.1:7890 export https_proxy=socks5://127.0.0.1:7890 export all_proxy=socks5://127.0.0.1:7890 export HTTP_PROXY=socks5://127.0.0.1:7890 export HTTPS_PROXY=socks5://127.0.0.1:7890 export ALL_PROXY=socks5://127.0.0.1:7890 + PROXY_DETECTED=true +elif [ -n "$HTTP_PROXY" ] || [ -n "$http_proxy" ]; then + echo "=== Using existing HTTP_PROXY env var: ${HTTP_PROXY:-$http_proxy} ===" + PROXY_DETECTED=true fi # 2. Set up virtual environment and install requirements if not set up @@ -25,9 +30,11 @@ fi echo "=== Step 1: Running Mock Unit Tests ===" "$DIR/backend/.venv/bin/python3" -m pytest "$DIR/backend/tests/" -echo "" -echo "=== Step 2: Seeding DB for Live Integration Test ===" -"$DIR/backend/.venv/bin/python3" -c "import asyncio, aiosqlite, sys; sys.path.insert(0, '$DIR/backend'); import db; +# Only run live integration tests if a proxy is available (needed from China) +if [ "$PROXY_DETECTED" = true ]; then + echo "" + echo "=== Step 2: Seeding DB for Live Integration Test ===" + "$DIR/backend/.venv/bin/python3" -c "import asyncio, aiosqlite, sys; sys.path.insert(0, '$DIR/backend'); import db; async def run(): await db.init_db() async with aiosqlite.connect(db.DB_PATH) as conn: @@ -35,14 +42,15 @@ async def run(): await conn.commit() asyncio.run(run())" -echo "=== Step 3: Running Live Ingestion & AI Summarization ===" -set +e -"$DIR/backend/.venv/bin/python3" "$DIR/backend/summarizer.py" "$VIDEO_ID" -set -e + echo "=== Step 3: Running Live Ingestion & AI Summarization ===" + set +e + "$DIR/backend/.venv/bin/python3" "$DIR/backend/summarizer.py" "$VIDEO_ID" + SUMMARIZER_EXIT=$? + set -e -echo "" -echo "=== Step 4: Verifying SQLite Database Storage ===" -"$DIR/backend/.venv/bin/python3" -c "import asyncio, aiosqlite, sys; sys.path.insert(0, '$DIR/backend'); import db; + echo "" + echo "=== Step 4: Verifying SQLite Database Storage ===" + "$DIR/backend/.venv/bin/python3" -c "import asyncio, aiosqlite, sys; sys.path.insert(0, '$DIR/backend'); import db; async def run(): async with aiosqlite.connect(db.DB_PATH) as conn: conn.row_factory = aiosqlite.Row @@ -57,9 +65,17 @@ async def run(): print() print('=============================================') print('WARNING: LIVE INTEGRATION TEST COULD NOT COMPLETE') - print('YouTube blocked this request or cookies.txt was missing.') + print('YouTube blocked this request or DeepSeek is unreachable.') print('Please ensure your proxy is active on port 7890 or export') - print('your youtube cookies to config/cookies.txt.') + print('HTTP_PROXY before running.') print('This warning is expected and did not break the build.') print('=============================================') asyncio.run(run())" +else + echo "" + echo "=== Skipping Live Integration Test (no proxy detected) ===" + echo "From China, YouTube API and DeepSeek require a proxy." + echo "Export HTTP_PROXY or SOCKS5 on port 7890 to enable." + echo "" + echo "=== Mock tests passed — live test skipped (OK) ===" +fi