sprint1: agent queue worker, proxy support, auto-mark-on-navigate-away, test fixes

This commit is contained in:
Gan, Jimmy
2026-07-08 22:05:05 +08:00
parent a154bc7978
commit ffa47a1335
8 changed files with 429 additions and 28 deletions
+166
View File
@@ -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())
+1
View File
@@ -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)
)
""")
+5 -1
View File
@@ -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):
+31 -1
View File
@@ -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")
+6 -1
View File
@@ -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",
+112
View File
@@ -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"
+78 -11
View File
@@ -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 @@
</script>
<div style="background:var(--bg-primary);color:var(--text-primary);height:100vh;overflow:hidden;display:flex;flex-direction:column">
<!-- Token Setup Modal -->
{#if showTokenSetup}
<div class="fixed inset-0 z-50 flex items-center justify-center" style="background:rgba(0,0,0,0.5)">
<div class="bg-[var(--bg-secondary)] p-6 rounded-lg max-w-md w-full mx-4"
style="border:1px solid var(--border)">
<h2 class="text-xl font-bold mb-2">API Token Required</h2>
<p class="text-sm mb-4" style="color:var(--text-secondary)">
This app requires an API token to save notes and manage videos.
Please enter the token configured on the server.
</p>
<input type="text" bind:value={apiTokenInput}
placeholder="Enter API token..." class="w-full px-3 py-2 rounded"
style="background:var(--bg-primary);border:1px solid var(--border);color:var(--text-primary)"
onkeydown={(e) => e.key === 'Enter' && setupToken()} />
<div class="flex space-x-2 mt-4">
<button onclick={setupToken}
class="px-4 py-2 rounded text-white font-bold"
style="background:var(--accent)">
Submit
</button>
</div>
</div>
</div>
{/if}
<header style="background:var(--bg-secondary);border-bottom:1px solid var(--border);" class="flex items-center justify-between px-5 py-2.5">
<div class="flex items-center space-x-3">
<span class="text-lg font-black tracking-wider" style="color:var(--accent)">T YouTube</span>
+30 -14
View File
@@ -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