diff --git a/atlas-terminal/README.md b/atlas-terminal/README.md index 6ea1181..a7c0a64 100644 --- a/atlas-terminal/README.md +++ b/atlas-terminal/README.md @@ -15,6 +15,7 @@ ATLAS Terminal brings market overview, quant research, valuation, technical anal - Technical analysis with candlesticks, moving averages, Bollinger Bands, RSI, MACD, and Fibonacci levels - Cross-market monitoring through macro, smart-money, yield/FX, economic calendar, earnings, news, filings, and portfolio pages - Portfolio tooling for OCR import, cross-asset correlation, and a UK CGT planning calculator +- Video transcript ingestion for YouTube, direct media URLs, and local uploads with searchable storage and optional Korean translation - Institutional report generation with printable PDF-style layouts ## Core Product Principle @@ -67,6 +68,7 @@ You can also click the screenshot below to open the recorded walkthrough: - `/valuation` DCF and scenario analysis - `/technical` chart-driven technical analysis - `/macro` macro and smart-money dashboard +- `/transcripts` video and audio transcript workbench - `/filings` SEC, DART, and EDINET workflows - `/report` institutional report generator - `/portfolio` portfolio tracking and OCR import @@ -177,3 +179,29 @@ Credential API: ATLAS Terminal started as an attempt to build a personal Bloomberg-lite for retail investing workflows: high information density, clean narrative structure, and a hard separation between AI-generated language and deterministic financial computation. It is currently optimized as a desktop-first personal research environment rather than a SaaS product. + +## Video Transcript Prerequisite + +The `/transcripts` workflow uses `faster-whisper` plus system `ffmpeg` for local speech-to-text and media conversion. On macOS, install `ffmpeg` before running transcript jobs: + +```bash +brew install ffmpeg +``` + +If Homebrew is not available, the repo also includes an `imageio-ffmpeg` fallback so local media extraction can still run in lightweight environments. + +`faster-whisper` downloads its first model automatically on demand. The default is `base`, and you can override it with `WHISPER_MODEL_SIZE=small` or `WHISPER_MODEL_SIZE=medium`. + +## Transcript Workflow + +Updated May 10, 2026. + +The new `/transcripts` route adds a dedicated ingestion and review flow for long-form media research: + +- Submit a YouTube URL, direct media URL, or local file upload +- Prefer subtitle extraction first, then fall back to local Whisper transcription +- Persist transcript text, summary, keywords, topics, and intent in the app database +- Search saved transcripts with full-text search across completed jobs +- Translate completed transcripts into Korean on demand when a Gemini key is configured in Settings + +This makes it easier to turn interviews, news clips, and earnings-related video into structured research notes inside ATLAS without leaving the terminal workflow. diff --git a/atlas-terminal/apps/web/src/app/components/sidebar.tsx b/atlas-terminal/apps/web/src/app/components/sidebar.tsx index c6fdb83..8dad20f 100644 --- a/atlas-terminal/apps/web/src/app/components/sidebar.tsx +++ b/atlas-terminal/apps/web/src/app/components/sidebar.tsx @@ -1,7 +1,7 @@ "use client"; import Link from "next/link"; import { usePathname } from "next/navigation"; -import { BarChart3, Briefcase, CalendarDays, CalendarRange, FileSearch, FileText, Globe, Landmark, LineChart, Microscope, Newspaper, Search, Settings, Sparkles, Target, TrendingUp } from "lucide-react"; +import { BarChart3, Briefcase, CalendarDays, CalendarRange, FileSearch, FileText, FileVideo, Globe, Landmark, LineChart, Microscope, Newspaper, Search, Settings, Sparkles, Target, TrendingUp } from "lucide-react"; import { useMemo, useState } from "react"; import { flags } from "../lib/flags"; import { searchTickerSuggestions, type TickerSuggestion } from "../lib/ticker-alias"; @@ -17,6 +17,7 @@ const NAV_ITEMS = [ { href: "/earnings", label: "Earnings", icon: CalendarRange }, ...(flags.calendar ? [{ href: "/calendar", label: "Calendar", icon: CalendarDays }] : []), { href: "/news", label: "News", icon: Newspaper }, + { href: "/transcripts", label: "Transcripts", icon: FileVideo }, { href: "/screener", label: "Screener", icon: Target }, { href: "/portfolio", label: "Portfolio", icon: Briefcase }, { href: "/filings", label: "Filings", icon: FileSearch }, diff --git a/atlas-terminal/apps/web/src/app/components/transcripts/TranscriptDetailPanel.tsx b/atlas-terminal/apps/web/src/app/components/transcripts/TranscriptDetailPanel.tsx new file mode 100644 index 0000000..99ff789 --- /dev/null +++ b/atlas-terminal/apps/web/src/app/components/transcripts/TranscriptDetailPanel.tsx @@ -0,0 +1,220 @@ +"use client"; + +import { Check, Copy, Languages, Search } from "lucide-react"; +import { useDeferredValue, useMemo, useState } from "react"; +import { Card } from "../ui/Card"; +import { ErrorBanner } from "../ui/ErrorBanner"; +import { LoadingPulse } from "../ui/LoadingPulse"; +import type { VideoJobDetailResponse, VideoTranslation } from "../../lib/video-transcript-types"; + +interface TranscriptDetailPanelProps { + detail: VideoJobDetailResponse | null; + loading: boolean; + translation: VideoTranslation | null; + translating: boolean; + onTranslate: (jobId: string) => Promise; +} + +function escapeRegExp(value: string) { + return value.replace(/[.*+?^${}()|[\]\\]/g, "\\$&"); +} + +function sentimentTone(sentiment: string | null | undefined) { + if (sentiment === "positive") return "border-fin-positive/40 bg-fin-positive/10 text-fin-positive"; + if (sentiment === "negative") return "border-fin-negative/40 bg-fin-negative/10 text-fin-negative"; + return "border-border bg-surface-sunken text-text-secondary"; +} + +function highlightText(text: string, query: string) { + if (!query.trim()) return text; + const matcher = new RegExp(`(${escapeRegExp(query)})`, "ig"); + const lowered = query.toLowerCase(); + return text.split(matcher).map((part, index) => ( + part.toLowerCase() === lowered ? ( + + {part} + + ) : ( + {part} + ) + )); +} + +export function TranscriptDetailPanel({ + detail, + loading, + translation, + translating, + onTranslate, +}: TranscriptDetailPanelProps) { + const [query, setQuery] = useState(""); + const [copied, setCopied] = useState(false); + const [showTranslation, setShowTranslation] = useState(false); + const deferredQuery = useDeferredValue(query); + const transcriptText = showTranslation ? (translation?.text || "") : (detail?.transcript?.text || ""); + const highlightedTranscript = useMemo( + () => highlightText(transcriptText, deferredQuery), + [deferredQuery, transcriptText], + ); + + async function handleCopy() { + if (!transcriptText) return; + await navigator.clipboard.writeText(transcriptText); + setCopied(true); + window.setTimeout(() => setCopied(false), 1600); + } + + if (loading && !detail) { + return ; + } + + if (!detail) { + return ( + +
+ Select a transcript job from the left column or search your saved transcripts above. +
+ + ); + } + + const { job, transcript } = detail; + const displaySummary = showTranslation ? (translation?.summary || transcript?.summary) : transcript?.summary; + const displayKeywords = showTranslation ? (translation?.keywords || transcript?.keywords || []) : (transcript?.keywords || []); + const displayTopics = showTranslation ? (translation?.topics || transcript?.topics || []) : (transcript?.topics || []); + const displayIntent = showTranslation ? (translation?.intent || transcript?.intent) : transcript?.intent; + + async function handleTranslateToggle() { + if (!transcript) return; + if (showTranslation) { + setShowTranslation(false); + return; + } + if (!translation) { + await onTranslate(job.job_id); + } + setShowTranslation(true); + } + + return ( +
+ void handleTranslateToggle()} + disabled={translating} + className="inline-flex items-center gap-2 rounded-md border border-border px-3 py-2 text-sm font-semibold text-brand-navy transition-colors hover:bg-surface-sunken disabled:cursor-not-allowed disabled:opacity-70" + > + + {showTranslation ? "원문 보기" : translating ? "번역 중..." : "한국어 번역"} + + ) : undefined} + > +
+
+ + {job.status === "failed" ? "Failed" : `Sentiment: ${transcript?.sentiment || "pending"}`} + + + {job.status.toUpperCase()} · {job.progress}% + +
+ + + + + +
+
+
Summary
+

+ {displaySummary || "Summary will appear here once transcript extraction and analysis complete."} +

+
+ +
+
+
Keywords
+
+ {displayKeywords.map((keyword) => ( + + #{keyword} + + ))} + {displayKeywords.length === 0 && ( + Keywords pending. + )} +
+
+ +
+
Topics
+
+ {displayTopics.map((topic) => ( + + {topic} + + ))} + {displayTopics.length === 0 && ( + Topics pending. + )} +
+
+ +
+
Intent
+

+ {displayIntent || "Intent analysis pending."} +

+
+
+
+
+ + + void handleCopy()} + className="inline-flex items-center gap-2 rounded-md border border-border px-3 py-2 text-sm font-semibold text-brand-navy transition-colors hover:bg-surface-sunken" + > + {copied ? : } + {copied ? "Copied" : "Copy"} + + )} + > +
+ + setQuery(event.target.value)} + placeholder="Search inside this transcript..." + className="w-full bg-transparent text-sm text-text-primary outline-none" + /> +
+ + {transcript ? ( +
+ {highlightedTranscript} +
+ ) : ( +
+ Transcript text will appear here once extraction completes. +
+ )} + +
+ ); +} diff --git a/atlas-terminal/apps/web/src/app/components/transcripts/TranscriptJobList.tsx b/atlas-terminal/apps/web/src/app/components/transcripts/TranscriptJobList.tsx new file mode 100644 index 0000000..22eb5ea --- /dev/null +++ b/atlas-terminal/apps/web/src/app/components/transcripts/TranscriptJobList.tsx @@ -0,0 +1,112 @@ +"use client"; + +import { AlertTriangle, CheckCircle2, Clock3, LoaderCircle, Trash2 } from "lucide-react"; +import { Card } from "../ui/Card"; +import { LoadingPulse } from "../ui/LoadingPulse"; +import type { VideoJob, VideoJobStatus } from "../../lib/video-transcript-types"; + +interface TranscriptJobListProps { + jobs: VideoJob[]; + loading: boolean; + selectedJobId: string | null; + onSelectJob: (jobId: string) => void; + onDeleteJob: (jobId: string) => Promise; +} + +const STATUS_LABELS: Record = { + queued: "Queued", + fetching: "Fetching", + transcribing: "Transcribing", + analyzing: "Analyzing", + completed: "Completed", + failed: "Failed", +}; + +const STATUS_TONES: Record = { + queued: "border-border bg-surface-sunken text-text-secondary", + fetching: "border-brand-gold/40 bg-brand-gold/10 text-brand-navy", + transcribing: "border-brand-blue/40 bg-brand-blue/10 text-brand-blue", + analyzing: "border-brand-gold/40 bg-brand-gold/10 text-brand-navy", + completed: "border-fin-positive/40 bg-fin-positive/10 text-fin-positive", + failed: "border-fin-negative/40 bg-fin-negative/10 text-fin-negative", +}; + +function StatusIcon({ status }: { status: VideoJobStatus }) { + if (status === "completed") return ; + if (status === "failed") return ; + if (status === "queued") return ; + return ; +} + +export function TranscriptJobList({ + jobs, + loading, + selectedJobId, + onSelectJob, + onDeleteJob, +}: TranscriptJobListProps) { + return ( + + {loading && jobs.length === 0 ? ( + + ) : jobs.length === 0 ? ( +
+ No transcript jobs yet. Submit a video, podcast clip, or subtitle file to get started. +
+ ) : ( +
+ {jobs.map((job) => { + const active = selectedJobId === job.job_id; + return ( +
+
+ + +
+ +
+ + + {STATUS_LABELS[job.status]} + + {job.progress}% + {job.language && {job.language.toUpperCase()}} + {job.duration_sec != null && {job.duration_sec}s} +
+ +
+
+
+ +
+ {new Date(job.created_at).toLocaleString()} +
+
+ ); + })} +
+ )} + + ); +} diff --git a/atlas-terminal/apps/web/src/app/components/transcripts/TranscriptSearchBar.tsx b/atlas-terminal/apps/web/src/app/components/transcripts/TranscriptSearchBar.tsx new file mode 100644 index 0000000..d8cbbff --- /dev/null +++ b/atlas-terminal/apps/web/src/app/components/transcripts/TranscriptSearchBar.tsx @@ -0,0 +1,75 @@ +"use client"; + +import { LoaderCircle, Search } from "lucide-react"; +import type { VideoSearchHit } from "../../lib/video-transcript-types"; + +interface TranscriptSearchBarProps { + query: string; + searching: boolean; + results: VideoSearchHit[]; + onQueryChange: (value: string) => void; + onSelectHit: (jobId: string) => void; +} + +function renderSnippet(snippet: string) { + return snippet.split(/(\[[^\]]+\])/g).map((part, index) => ( + part.startsWith("[") && part.endsWith("]") ? ( + + {part.slice(1, -1)} + + ) : ( + {part} + ) + )); +} + +export function TranscriptSearchBar({ + query, + searching, + results, + onQueryChange, + onSelectHit, +}: TranscriptSearchBarProps) { + const showResults = query.trim().length >= 2; + + return ( +
+
+ + onQueryChange(event.target.value)} + placeholder="Search saved transcripts for earnings, margins, AI, guidance..." + className="w-full bg-transparent text-sm text-text-primary outline-none" + /> + {searching && } +
+ + {showResults && ( +
+ {results.length > 0 ? ( + results.map((hit) => ( + + )) + ) : !searching ? ( +
+ No saved transcript matched this query yet. +
+ ) : null} +
+ )} +
+ ); +} diff --git a/atlas-terminal/apps/web/src/app/components/transcripts/TranscriptUploadForm.tsx b/atlas-terminal/apps/web/src/app/components/transcripts/TranscriptUploadForm.tsx new file mode 100644 index 0000000..ddc170a --- /dev/null +++ b/atlas-terminal/apps/web/src/app/components/transcripts/TranscriptUploadForm.tsx @@ -0,0 +1,140 @@ +"use client"; + +import { Link2, PlayCircle, UploadCloud, type LucideIcon } from "lucide-react"; +import { useState } from "react"; +import { Card } from "../ui/Card"; +import type { VideoSourceType } from "../../lib/video-transcript-types"; + +interface TranscriptUploadFormProps { + submitting: boolean; + onSubmitSource: (url: string, sourceType: VideoSourceType, language?: string) => Promise; + onUploadMedia: (file: File, language?: string) => Promise; +} + +type InputMode = "youtube" | "url" | "upload"; + +const TABS: Array<{ id: InputMode; label: string; icon: LucideIcon }> = [ + { id: "youtube", label: "YouTube", icon: PlayCircle }, + { id: "url", label: "Web URL", icon: Link2 }, + { id: "upload", label: "Upload", icon: UploadCloud }, +]; + +export function TranscriptUploadForm({ submitting, onSubmitSource, onUploadMedia }: TranscriptUploadFormProps) { + const [mode, setMode] = useState("youtube"); + const [sourceValue, setSourceValue] = useState(""); + const [language, setLanguage] = useState(""); + const [selectedFile, setSelectedFile] = useState(null); + const [localError, setLocalError] = useState(null); + + async function handleSubmit() { + setLocalError(null); + try { + if (mode === "upload") { + if (!selectedFile) { + setLocalError("Choose an mp4, audio file, or subtitle file first."); + return; + } + await onUploadMedia(selectedFile, language); + setSelectedFile(null); + return; + } + + if (!sourceValue.trim()) { + setLocalError(mode === "youtube" ? "Paste a YouTube URL to continue." : "Paste a media URL or local file path to continue."); + return; + } + + await onSubmitSource(sourceValue.trim(), mode === "youtube" ? "youtube" : "url", language); + setSourceValue(""); + } catch { + // Parent hook already surfaces the actionable error banner. + } + } + + return ( + +
+
+ {TABS.map(({ id, label, icon: Icon }) => { + const active = mode === id; + return ( + + ); + })} +
+ + {mode === "upload" ? ( + + ) : ( + + )} + + + + {localError && ( +
+ {localError} +
+ )} + + +
+ + ); +} diff --git a/atlas-terminal/apps/web/src/app/lib/use-video-transcript.ts b/atlas-terminal/apps/web/src/app/lib/use-video-transcript.ts new file mode 100644 index 0000000..12ed2f1 --- /dev/null +++ b/atlas-terminal/apps/web/src/app/lib/use-video-transcript.ts @@ -0,0 +1,249 @@ +"use client"; + +import { useCallback, useEffect, useState } from "react"; +import type { VideoJob, VideoJobDetailResponse, VideoSearchHit, VideoSourceType, VideoTranslation } from "./video-transcript-types"; + +const PENDING_STATUSES = new Set(["queued", "fetching", "transcribing", "analyzing"]); + +function getGeminiHeader(): HeadersInit { + if (typeof window === "undefined") return {}; + const apiKey = (localStorage.getItem("atlas_gemini_key") || "").trim(); + return apiKey ? { "x-gemini-api-key": apiKey } : {}; +} + +function sortJobs(jobs: VideoJob[]): VideoJob[] { + return [...jobs].sort((left, right) => right.created_at.localeCompare(left.created_at)); +} + +export function useVideoTranscript() { + const [jobs, setJobs] = useState([]); + const [selectedJobId, setSelectedJobId] = useState(null); + const [detail, setDetail] = useState(null); + const [searchResults, setSearchResults] = useState([]); + const [translationsByJob, setTranslationsByJob] = useState>({}); + const [loadingJobs, setLoadingJobs] = useState(true); + const [loadingDetail, setLoadingDetail] = useState(false); + const [submitting, setSubmitting] = useState(false); + const [searching, setSearching] = useState(false); + const [translating, setTranslating] = useState(false); + const [error, setError] = useState(null); + + const request = useCallback(async (path: string, init?: RequestInit, json = true) => { + const response = await fetch(`/api/video${path}`, { + ...init, + headers: { + ...(json ? { "Content-Type": "application/json" } : {}), + ...getGeminiHeader(), + ...(init?.headers ?? {}), + }, + }); + + if (!response.ok) { + const payload = await response.json().catch(() => null); + throw new Error(payload?.detail || `Request failed with status ${response.status}.`); + } + + return response.json(); + }, []); + + const refreshJobs = useCallback(async (showSpinner = true) => { + if (showSpinner) setLoadingJobs(true); + try { + const rows = (await request("/jobs")) as VideoJob[]; + const sorted = sortJobs(rows); + setJobs(sorted); + setSelectedJobId((current) => { + if (current && sorted.some((job) => job.job_id === current)) { + return current; + } + return sorted[0]?.job_id ?? null; + }); + if (sorted.length === 0) { + setDetail(null); + } + setError(null); + } catch (requestError) { + setError(requestError instanceof Error ? requestError.message : "Failed to load transcript jobs."); + } finally { + if (showSpinner) setLoadingJobs(false); + } + }, [request]); + + const loadJobDetail = useCallback(async (jobId: string, showSpinner = true) => { + if (showSpinner) setLoadingDetail(true); + try { + const payload = (await request(`/jobs/${jobId}`)) as VideoJobDetailResponse; + setDetail(payload); + setJobs((current) => { + const merged = current.some((job) => job.job_id === payload.job.job_id) + ? current.map((job) => (job.job_id === payload.job.job_id ? payload.job : job)) + : [payload.job, ...current]; + return sortJobs(merged); + }); + setError(null); + } catch (requestError) { + setError(requestError instanceof Error ? requestError.message : "Failed to load transcript detail."); + } finally { + if (showSpinner) setLoadingDetail(false); + } + }, [request]); + + useEffect(() => { + void refreshJobs(); + }, [refreshJobs]); + + useEffect(() => { + if (!selectedJobId) return; + void loadJobDetail(selectedJobId); + }, [selectedJobId, loadJobDetail]); + + useEffect(() => { + if (!selectedJobId || !detail || detail.job.job_id !== selectedJobId) return undefined; + if (!PENDING_STATUSES.has(detail.job.status)) return undefined; + + const timer = window.setInterval(() => { + void loadJobDetail(selectedJobId, false); + void refreshJobs(false); + }, 5000); + + return () => window.clearInterval(timer); + }, [detail, loadJobDetail, refreshJobs, selectedJobId]); + + const submitSource = useCallback(async (url: string, sourceType: VideoSourceType, language?: string) => { + setSubmitting(true); + try { + const job = (await request("/submit", { + method: "POST", + body: JSON.stringify({ + url, + source_type: sourceType, + language: language?.trim() || null, + }), + })) as VideoJob; + setJobs((current) => sortJobs([job, ...current.filter((item) => item.job_id !== job.job_id)])); + setSelectedJobId(job.job_id); + setDetail({ job, transcript: null }); + setError(null); + return job; + } catch (requestError) { + const message = requestError instanceof Error ? requestError.message : "Failed to submit transcript job."; + setError(message); + throw requestError; + } finally { + setSubmitting(false); + } + }, [request]); + + const uploadMedia = useCallback(async (file: File, language?: string) => { + setSubmitting(true); + try { + const formData = new FormData(); + formData.append("file", file); + if (language?.trim()) { + formData.append("language", language.trim()); + } + const job = (await request("/upload", { + method: "POST", + body: formData, + }, false)) as VideoJob; + setJobs((current) => sortJobs([job, ...current.filter((item) => item.job_id !== job.job_id)])); + setSelectedJobId(job.job_id); + setDetail({ job, transcript: null }); + setError(null); + return job; + } catch (requestError) { + const message = requestError instanceof Error ? requestError.message : "Failed to upload media."; + setError(message); + throw requestError; + } finally { + setSubmitting(false); + } + }, [request]); + + const deleteTranscriptJob = useCallback(async (jobId: string) => { + try { + await request(`/jobs/${jobId}`, { method: "DELETE" }); + setJobs((current) => { + const nextJobs = current.filter((job) => job.job_id !== jobId); + if (selectedJobId === jobId) { + setSelectedJobId(nextJobs[0]?.job_id ?? null); + if (nextJobs.length === 0) { + setDetail(null); + } + } + return nextJobs; + }); + setTranslationsByJob((current) => { + const next = { ...current }; + delete next[jobId]; + return next; + }); + setError(null); + } catch (requestError) { + setError(requestError instanceof Error ? requestError.message : "Failed to delete transcript job."); + } + }, [request, selectedJobId]); + + const searchVideos = useCallback(async (query: string) => { + if (!query.trim()) { + setSearchResults([]); + return; + } + setSearching(true); + try { + const rows = (await request(`/search?q=${encodeURIComponent(query.trim())}`)) as VideoSearchHit[]; + setSearchResults(rows); + setError(null); + } catch (requestError) { + setError(requestError instanceof Error ? requestError.message : "Failed to search transcripts."); + } finally { + setSearching(false); + } + }, [request]); + + const clearSearch = useCallback(() => { + setSearchResults([]); + }, []); + + const translateJobToKorean = useCallback(async (jobId: string) => { + setTranslating(true); + try { + const payload = (await request(`/jobs/${jobId}/translate?target_language=ko`, { + method: "POST", + })) as VideoTranslation; + setTranslationsByJob((current) => ({ + ...current, + [jobId]: payload, + })); + setError(null); + return payload; + } catch (requestError) { + setError(requestError instanceof Error ? requestError.message : "Failed to translate transcript."); + throw requestError; + } finally { + setTranslating(false); + } + }, [request]); + + return { + jobs, + selectedJobId, + detail, + searchResults, + translation: detail ? translationsByJob[detail.job.job_id] ?? null : null, + loadingJobs, + loadingDetail, + submitting, + searching, + translating, + error, + selectJob: setSelectedJobId, + refreshJobs, + submitSource, + uploadMedia, + deleteTranscriptJob, + searchVideos, + clearSearch, + translateJobToKorean, + }; +} diff --git a/atlas-terminal/apps/web/src/app/lib/video-transcript-types.ts b/atlas-terminal/apps/web/src/app/lib/video-transcript-types.ts new file mode 100644 index 0000000..1a06302 --- /dev/null +++ b/atlas-terminal/apps/web/src/app/lib/video-transcript-types.ts @@ -0,0 +1,48 @@ +export type VideoSourceType = "youtube" | "url" | "local"; +export type VideoJobStatus = "queued" | "fetching" | "transcribing" | "analyzing" | "completed" | "failed"; + +export interface VideoJob { + job_id: string; + status: VideoJobStatus; + source_url: string; + source_type: VideoSourceType; + progress: number; + error: string | null; + title: string | null; + duration_sec: number | null; + language: string | null; + created_at: string; + completed_at: string | null; +} + +export interface VideoTranscript { + job_id: string; + text: string; + summary: string | null; + keywords: string[]; + topics: string[]; + sentiment: "positive" | "neutral" | "negative" | null; + intent: string | null; +} + +export interface VideoSearchHit { + job_id: string; + title: string | null; + snippet: string; + rank: number; +} + +export interface VideoJobDetailResponse { + job: VideoJob; + transcript: VideoTranscript | null; +} + +export interface VideoTranslation { + job_id: string; + target_language: string; + summary: string | null; + keywords: string[]; + topics: string[]; + intent: string | null; + text: string; +} diff --git a/atlas-terminal/apps/web/src/app/transcripts/page.tsx b/atlas-terminal/apps/web/src/app/transcripts/page.tsx new file mode 100644 index 0000000..43e3b4f --- /dev/null +++ b/atlas-terminal/apps/web/src/app/transcripts/page.tsx @@ -0,0 +1,100 @@ +"use client"; + +import { startTransition, useDeferredValue, useEffect, useState } from "react"; +import { TranscriptDetailPanel } from "../components/transcripts/TranscriptDetailPanel"; +import { TranscriptJobList } from "../components/transcripts/TranscriptJobList"; +import { TranscriptSearchBar } from "../components/transcripts/TranscriptSearchBar"; +import { TranscriptUploadForm } from "../components/transcripts/TranscriptUploadForm"; +import { ErrorBanner } from "../components/ui/ErrorBanner"; +import { SectionHeading } from "../components/ui/SectionHeading"; +import { useVideoTranscript } from "../lib/use-video-transcript"; + +export default function TranscriptsPage() { + const { + jobs, + selectedJobId, + detail, + searchResults, + translation, + loadingJobs, + loadingDetail, + submitting, + searching, + translating, + error, + selectJob, + submitSource, + uploadMedia, + deleteTranscriptJob, + searchVideos, + clearSearch, + translateJobToKorean, + } = useVideoTranscript(); + const [searchQuery, setSearchQuery] = useState(""); + const deferredQuery = useDeferredValue(searchQuery); + + useEffect(() => { + if (deferredQuery.trim().length < 2) { + clearSearch(); + return undefined; + } + + const timer = window.setTimeout(() => { + void searchVideos(deferredQuery.trim()); + }, 250); + + return () => window.clearTimeout(timer); + }, [clearSearch, deferredQuery, searchVideos]); + + return ( +
+
+
+ Transcripts +

+ Extract captions or speech from YouTube clips, local media, and direct video URLs, then store the results for analysis and full-text search. +

+
+
+ + + + { + startTransition(() => { + selectJob(jobId); + }); + }} + /> + +
+
+ + +
+ + +
+
+ ); +} diff --git a/atlas-terminal/requirements.txt b/atlas-terminal/requirements.txt index 4a83c43..f9efd0c 100644 --- a/atlas-terminal/requirements.txt +++ b/atlas-terminal/requirements.txt @@ -9,6 +9,7 @@ requests>=2.31.0 pandas>=2.0.0 lxml>=4.9.0 python-dotenv>=1.0.0 +python-multipart>=0.0.9 yfinance>=0.2.40 yahooquery>=2.2.0 sec-edgar-downloader>=5.0.0 @@ -24,3 +25,8 @@ dart-fss>=0.4.0 numpy>=1.20.0 scipy>=1.10.0 dbnomics>=1.2.0 +faster-whisper>=1.0.3 +yt-dlp>=2024.10.7 +srt>=3.5.3 +ffmpeg-python>=0.2.0 +imageio-ffmpeg>=0.6.0 diff --git a/atlas-terminal/server/db/database.py b/atlas-terminal/server/db/database.py index 3ccd720..e0b3c6d 100644 --- a/atlas-terminal/server/db/database.py +++ b/atlas-terminal/server/db/database.py @@ -133,6 +133,57 @@ async def init_db() -> None: """ ) + await db.executescript( + """ + CREATE TABLE IF NOT EXISTS video_jobs ( + job_id TEXT PRIMARY KEY, + source_url TEXT NOT NULL, + source_type TEXT NOT NULL, + status TEXT NOT NULL DEFAULT 'queued', + progress INTEGER NOT NULL DEFAULT 0, + error TEXT, + title TEXT, + duration_sec INTEGER, + language TEXT, + transcript_text TEXT, + summary TEXT, + keywords_json TEXT, + topics_json TEXT, + sentiment TEXT, + intent TEXT, + created_at TEXT NOT NULL, + completed_at TEXT + ); + + CREATE INDEX IF NOT EXISTS idx_video_jobs_created_at ON video_jobs(created_at DESC); + + CREATE VIRTUAL TABLE IF NOT EXISTS video_jobs_fts USING fts5( + job_id UNINDEXED, + title, + transcript_text, + content='video_jobs', + content_rowid='rowid' + ); + + CREATE TRIGGER IF NOT EXISTS video_jobs_ai AFTER INSERT ON video_jobs BEGIN + INSERT INTO video_jobs_fts(rowid, job_id, title, transcript_text) + VALUES (new.rowid, new.job_id, new.title, new.transcript_text); + END; + + CREATE TRIGGER IF NOT EXISTS video_jobs_ad AFTER DELETE ON video_jobs BEGIN + INSERT INTO video_jobs_fts(video_jobs_fts, rowid, job_id, title, transcript_text) + VALUES ('delete', old.rowid, old.job_id, old.title, old.transcript_text); + END; + + CREATE TRIGGER IF NOT EXISTS video_jobs_au AFTER UPDATE ON video_jobs BEGIN + INSERT INTO video_jobs_fts(video_jobs_fts, rowid, job_id, title, transcript_text) + VALUES ('delete', old.rowid, old.job_id, old.title, old.transcript_text); + INSERT INTO video_jobs_fts(rowid, job_id, title, transcript_text) + VALUES (new.rowid, new.job_id, new.title, new.transcript_text); + END; + """ + ) + await db.commit() diff --git a/atlas-terminal/server/db/pg_database.py b/atlas-terminal/server/db/pg_database.py index e9063af..f569521 100644 --- a/atlas-terminal/server/db/pg_database.py +++ b/atlas-terminal/server/db/pg_database.py @@ -126,6 +126,35 @@ async def init_pg_tables() -> None: await conn.execute(""" CREATE INDEX IF NOT EXISTS idx_credential_access_log_at ON credential_access_log(at); """) + await conn.execute(""" + CREATE TABLE IF NOT EXISTS video_jobs ( + job_id TEXT PRIMARY KEY, + source_url TEXT NOT NULL, + source_type TEXT NOT NULL, + status TEXT NOT NULL DEFAULT 'queued', + progress INTEGER NOT NULL DEFAULT 0, + error TEXT, + title TEXT, + duration_sec INTEGER, + language TEXT, + transcript_text TEXT, + summary TEXT, + keywords_json JSONB DEFAULT '[]'::jsonb, + topics_json JSONB DEFAULT '[]'::jsonb, + sentiment TEXT, + intent TEXT, + created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), + completed_at TIMESTAMPTZ + ); + """) + await conn.execute(""" + CREATE INDEX IF NOT EXISTS idx_video_jobs_created_at ON video_jobs(created_at DESC); + """) + await conn.execute(""" + CREATE INDEX IF NOT EXISTS idx_video_jobs_search + ON video_jobs + USING GIN (to_tsvector('simple', coalesce(title, '') || ' ' || coalesce(transcript_text, ''))); + """) logger.info("PostgreSQL tables initialized.") diff --git a/atlas-terminal/server/db/pg_video_repo.py b/atlas-terminal/server/db/pg_video_repo.py new file mode 100644 index 0000000..c905871 --- /dev/null +++ b/atlas-terminal/server/db/pg_video_repo.py @@ -0,0 +1,179 @@ +"""PostgreSQL repository for persisted video transcript jobs.""" + +from __future__ import annotations + +from datetime import datetime +from typing import Any + +from server.db.pg_database import get_pg_pool + + +def _coerce_list(value: Any) -> list[str]: + if isinstance(value, list): + return [str(item) for item in value if str(item).strip()] + if isinstance(value, str) and value.strip(): + return [value] + return [] + + +def _coerce_datetime(value: Any) -> Any: + if isinstance(value, str) and value.strip(): + return datetime.fromisoformat(value.replace("Z", "+00:00")) + return value + + +def _normalise_row(row: Any) -> dict[str, Any]: + payload = dict(row) + payload["keywords"] = _coerce_list(payload.pop("keywords_json", [])) + payload["topics"] = _coerce_list(payload.pop("topics_json", [])) + if payload.get("created_at") is not None: + payload["created_at"] = payload["created_at"].isoformat() + if payload.get("completed_at") is not None: + payload["completed_at"] = payload["completed_at"].isoformat() + return payload + + +def _normalise_updates(fields: dict[str, Any]) -> dict[str, Any]: + alias_map = {"keywords": "keywords_json", "topics": "topics_json"} + allowed = { + "source_url", + "source_type", + "status", + "progress", + "error", + "title", + "duration_sec", + "language", + "transcript_text", + "summary", + "keywords_json", + "topics_json", + "sentiment", + "intent", + "completed_at", + } + normalised: dict[str, Any] = {} + for key, value in fields.items(): + mapped_key = alias_map.get(key, key) + if mapped_key not in allowed: + continue + if mapped_key in {"keywords_json", "topics_json"}: + normalised[mapped_key] = _coerce_list(value) + elif mapped_key in {"completed_at"}: + normalised[mapped_key] = _coerce_datetime(value) + else: + normalised[mapped_key] = value + return normalised + + +async def pg_add_video_job(job_id: str, url: str, source_type: str) -> dict[str, Any] | None: + pool = await get_pg_pool() + if not pool: + return None + async with pool.acquire() as conn: + row = await conn.fetchrow( + """ + INSERT INTO video_jobs (job_id, source_url, source_type, status, progress) + VALUES ($1, $2, $3, 'queued', 0) + RETURNING * + """, + job_id, + url, + source_type, + ) + return _normalise_row(row) if row else None + + +async def pg_update_video_job(job_id: str, **fields: Any) -> dict[str, Any] | None: + pool = await get_pg_pool() + if not pool: + return None + updates = _normalise_updates(fields) + async with pool.acquire() as conn: + if updates: + set_clause = ", ".join(f"{column} = ${index + 2}" for index, column in enumerate(updates)) + values = [job_id] + list(updates.values()) + row = await conn.fetchrow( + f"UPDATE video_jobs SET {set_clause} WHERE job_id = $1 RETURNING *", # noqa: S608 + *values, + ) + return _normalise_row(row) if row else None + row = await conn.fetchrow("SELECT * FROM video_jobs WHERE job_id = $1", job_id) + return _normalise_row(row) if row else None + + +async def pg_get_video_job(job_id: str) -> dict[str, Any] | None: + pool = await get_pg_pool() + if not pool: + return None + async with pool.acquire() as conn: + row = await conn.fetchrow("SELECT * FROM video_jobs WHERE job_id = $1", job_id) + return _normalise_row(row) if row else None + + +async def pg_list_video_jobs(limit: int = 50) -> list[dict[str, Any]]: + pool = await get_pg_pool() + if not pool: + return [] + async with pool.acquire() as conn: + rows = await conn.fetch( + """ + SELECT + job_id, source_url, source_type, status, progress, error, title, + duration_sec, language, created_at, completed_at + FROM video_jobs + ORDER BY created_at DESC + LIMIT $1 + """, + limit, + ) + return [_normalise_row(row) for row in rows] + + +async def pg_search_videos(query: str, limit: int = 20) -> list[dict[str, Any]]: + pool = await get_pg_pool() + if not pool: + return [] + async with pool.acquire() as conn: + rows = await conn.fetch( + """ + SELECT + job_id, + title, + ts_headline( + 'simple', + coalesce(transcript_text, ''), + plainto_tsquery('simple', $1), + 'StartSel=[,StopSel=],MaxFragments=1,MaxWords=18,MinWords=6' + ) AS snippet, + ts_rank_cd( + to_tsvector('simple', coalesce(title, '') || ' ' || coalesce(transcript_text, '')), + plainto_tsquery('simple', $1) + ) AS rank + FROM video_jobs + WHERE to_tsvector('simple', coalesce(title, '') || ' ' || coalesce(transcript_text, '')) + @@ plainto_tsquery('simple', $1) + ORDER BY rank DESC, created_at DESC + LIMIT $2 + """, + query, + limit, + ) + return [ + { + "job_id": row["job_id"], + "title": row["title"], + "snippet": row["snippet"] or "", + "rank": float(row["rank"] or 0.0), + } + for row in rows + ] + + +async def pg_delete_video_job(job_id: str) -> bool: + pool = await get_pg_pool() + if not pool: + return False + async with pool.acquire() as conn: + result = await conn.execute("DELETE FROM video_jobs WHERE job_id = $1", job_id) + return result == "DELETE 1" diff --git a/atlas-terminal/server/db/unified_repo.py b/atlas-terminal/server/db/unified_repo.py index 76b41f7..fafc3f6 100644 --- a/atlas-terminal/server/db/unified_repo.py +++ b/atlas-terminal/server/db/unified_repo.py @@ -70,6 +70,48 @@ class UnifiedRepo: from server.db.cache import cache_manager await cache_manager.set(key, value, ttl) + async def add_video_job(self, job_id: str, url: str, source_type: str) -> Optional[dict]: + if _use_postgres(): + from server.db.pg_video_repo import pg_add_video_job + return await pg_add_video_job(job_id, url, source_type) + from server.db.video_repo import add_video_job + return await add_video_job(job_id, url, source_type) + + async def update_video_job(self, job_id: str, **fields: Any) -> Optional[dict]: + if _use_postgres(): + from server.db.pg_video_repo import pg_update_video_job + return await pg_update_video_job(job_id, **fields) + from server.db.video_repo import update_video_job + return await update_video_job(job_id, **fields) + + async def get_video_job(self, job_id: str) -> Optional[dict]: + if _use_postgres(): + from server.db.pg_video_repo import pg_get_video_job + return await pg_get_video_job(job_id) + from server.db.video_repo import get_video_job + return await get_video_job(job_id) + + async def list_video_jobs(self, limit: int = 50) -> list[dict]: + if _use_postgres(): + from server.db.pg_video_repo import pg_list_video_jobs + return await pg_list_video_jobs(limit=limit) + from server.db.video_repo import list_video_jobs + return await list_video_jobs(limit=limit) + + async def search_videos(self, query: str, limit: int = 20) -> list[dict]: + if _use_postgres(): + from server.db.pg_video_repo import pg_search_videos + return await pg_search_videos(query, limit=limit) + from server.db.video_repo import search_videos + return await search_videos(query, limit=limit) + + async def delete_video_job(self, job_id: str) -> bool: + if _use_postgres(): + from server.db.pg_video_repo import pg_delete_video_job + return await pg_delete_video_job(job_id) + from server.db.video_repo import delete_video_job + return await delete_video_job(job_id) + async def init_db(self) -> None: """Initialize the appropriate database.""" if _use_postgres(): diff --git a/atlas-terminal/server/db/video_repo.py b/atlas-terminal/server/db/video_repo.py new file mode 100644 index 0000000..169bc4c --- /dev/null +++ b/atlas-terminal/server/db/video_repo.py @@ -0,0 +1,176 @@ +"""SQLite repository for persisted video transcript jobs.""" + +from __future__ import annotations + +import json +import re +from datetime import datetime, timezone +from typing import Any + +import aiosqlite + +from server.db.database import get_db + +_SEARCH_TOKEN_RE = re.compile(r"[A-Za-z0-9_]+") + + +def _now_iso() -> str: + return datetime.now(timezone.utc).isoformat() + + +def _coerce_json_list(value: Any) -> list[str]: + if isinstance(value, list): + return [str(item) for item in value if str(item).strip()] + if isinstance(value, str) and value.strip(): + try: + parsed = json.loads(value) + except json.JSONDecodeError: + return [value] + if isinstance(parsed, list): + return [str(item) for item in parsed if str(item).strip()] + return [] + + +def _row_to_dict(row: aiosqlite.Row | None) -> dict[str, Any] | None: + if row is None: + return None + data = dict(row) + data["keywords"] = _coerce_json_list(data.pop("keywords_json", [])) + data["topics"] = _coerce_json_list(data.pop("topics_json", [])) + return data + + +def _normalise_updates(fields: dict[str, Any]) -> dict[str, Any]: + alias_map = { + "keywords": "keywords_json", + "topics": "topics_json", + } + allowed = { + "source_url", + "source_type", + "status", + "progress", + "error", + "title", + "duration_sec", + "language", + "transcript_text", + "summary", + "keywords_json", + "topics_json", + "sentiment", + "intent", + "completed_at", + } + normalised: dict[str, Any] = {} + for key, value in fields.items(): + mapped_key = alias_map.get(key, key) + if mapped_key not in allowed: + continue + if mapped_key in {"keywords_json", "topics_json"}: + normalised[mapped_key] = json.dumps(_coerce_json_list(value), ensure_ascii=False) + else: + normalised[mapped_key] = value + return normalised + + +def _build_fts_query(query: str) -> str: + tokens = [token.lower() for token in _SEARCH_TOKEN_RE.findall(query)] + if not tokens: + return "" + return " AND ".join(f"{token}*" for token in tokens) + + +async def add_video_job(job_id: str, url: str, source_type: str) -> dict[str, Any] | None: + db = await get_db() + created_at = _now_iso() + await db.execute( + """ + INSERT INTO video_jobs ( + job_id, source_url, source_type, status, progress, created_at + ) VALUES (?, ?, ?, 'queued', 0, ?) + """, + (job_id, url, source_type, created_at), + ) + await db.commit() + return await get_video_job(job_id) + + +async def update_video_job(job_id: str, **fields: Any) -> dict[str, Any] | None: + updates = _normalise_updates(fields) + db = await get_db() + if updates: + set_clause = ", ".join(f"{column} = ?" for column in updates) + values = list(updates.values()) + [job_id] + await db.execute( + f"UPDATE video_jobs SET {set_clause} WHERE job_id = ?", # noqa: S608 + values, + ) + await db.commit() + return await get_video_job(job_id) + + +async def get_video_job(job_id: str) -> dict[str, Any] | None: + db = await get_db() + cursor = await db.execute("SELECT * FROM video_jobs WHERE job_id = ?", (job_id,)) + row = await cursor.fetchone() + return _row_to_dict(row) + + +async def list_video_jobs(limit: int = 50) -> list[dict[str, Any]]: + db = await get_db() + cursor = await db.execute( + """ + SELECT + job_id, source_url, source_type, status, progress, error, title, + duration_sec, language, created_at, completed_at + FROM video_jobs + ORDER BY created_at DESC + LIMIT ? + """, + (limit,), + ) + rows = await cursor.fetchall() + return [row for row in (_row_to_dict(item) for item in rows) if row is not None] + + +async def search_videos(query: str, limit: int = 20) -> list[dict[str, Any]]: + fts_query = _build_fts_query(query) + if not fts_query: + return [] + db = await get_db() + cursor = await db.execute( + """ + SELECT + video_jobs.job_id, + video_jobs.title, + COALESCE( + snippet(video_jobs_fts, 2, '[', ']', '...', 16), + substr(video_jobs.transcript_text, 1, 180) + ) AS snippet, + (1.0 / (1.0 + bm25(video_jobs_fts))) AS rank + FROM video_jobs_fts + JOIN video_jobs ON video_jobs.rowid = video_jobs_fts.rowid + WHERE video_jobs_fts MATCH ? + ORDER BY bm25(video_jobs_fts), video_jobs.created_at DESC + LIMIT ? + """, + (fts_query, limit), + ) + rows = await cursor.fetchall() + return [ + { + "job_id": row["job_id"], + "title": row["title"], + "snippet": row["snippet"] or "", + "rank": float(row["rank"] or 0.0), + } + for row in rows + ] + + +async def delete_video_job(job_id: str) -> bool: + db = await get_db() + cursor = await db.execute("DELETE FROM video_jobs WHERE job_id = ?", (job_id,)) + await db.commit() + return cursor.rowcount > 0 diff --git a/atlas-terminal/server/main.py b/atlas-terminal/server/main.py index 8289948..e5aea26 100644 --- a/atlas-terminal/server/main.py +++ b/atlas-terminal/server/main.py @@ -81,7 +81,7 @@ app.add_middleware( ) # --- Mount routers --- -from server.routers import edgar, analysis, valuation, market_data, news, crypto, fx, portfolio, technical, financials, estimates, earnings, insider, screener, markets, fmp, macro, dart, edinet, research, chat, copilot, credentials, calendar, tax # noqa: E402 +from server.routers import edgar, analysis, valuation, market_data, news, crypto, fx, portfolio, technical, financials, estimates, earnings, insider, screener, markets, fmp, macro, dart, edinet, research, chat, copilot, credentials, calendar, tax, video_transcript # noqa: E402 app.include_router(edgar.router, prefix="/api/edgar", tags=["SEC EDGAR"]) app.include_router(analysis.router, prefix="/api/analysis", tags=["AI Analysis"]) @@ -107,6 +107,7 @@ app.include_router(chat.router, prefix="/api/chat", tags=["Chat"]) app.include_router(copilot.router, prefix="/api/copilot", tags=["Copilot"]) app.include_router(credentials.router, prefix="/api/credentials", tags=["Credentials"]) app.include_router(calendar.router, prefix="/api/calendar", tags=["Calendar"]) +app.include_router(video_transcript.router, prefix="/api/video", tags=["Video Transcript"]) app.include_router(tax.router, prefix="/api/tax", tags=["Tax"]) diff --git a/atlas-terminal/server/models/schemas.py b/atlas-terminal/server/models/schemas.py index 93fcf4b..5f71134 100644 --- a/atlas-terminal/server/models/schemas.py +++ b/atlas-terminal/server/models/schemas.py @@ -367,3 +367,56 @@ class HealthCheckResponse(BaseModel): """API health check.""" status: str = "ok" version: str = "1.0.0" + + + +class VideoSubmitRequest(BaseModel): + """Request to start transcript extraction for a video or audio source.""" + url: str = Field(..., description="YouTube URL, web URL, or local file path") + source_type: Literal["youtube", "url", "local"] = "url" + language: Optional[str] = Field(default=None, description="Optional source language hint for Whisper") + + +class VideoJob(BaseModel): + """Video transcript job metadata.""" + job_id: str + status: Literal["queued", "fetching", "transcribing", "analyzing", "completed", "failed"] + source_url: str + source_type: Literal["youtube", "url", "local"] + progress: int = 0 + error: Optional[str] = None + title: Optional[str] = None + duration_sec: Optional[int] = None + language: Optional[str] = None + created_at: str + completed_at: Optional[str] = None + + +class VideoTranscript(BaseModel): + """Stored transcript text plus analysis output.""" + job_id: str + text: str + summary: Optional[str] = None + keywords: List[str] = Field(default_factory=list) + topics: List[str] = Field(default_factory=list) + sentiment: Optional[Literal["positive", "neutral", "negative"]] = None + intent: Optional[str] = None + + +class VideoSearchHit(BaseModel): + """One transcript full-text search match.""" + job_id: str + title: Optional[str] = None + snippet: str + rank: float + + +class VideoTranslation(BaseModel): + """On-demand translated transcript payload.""" + job_id: str + target_language: str = "ko" + summary: Optional[str] = None + keywords: List[str] = Field(default_factory=list) + topics: List[str] = Field(default_factory=list) + intent: Optional[str] = None + text: str = "" diff --git a/atlas-terminal/server/routers/video_transcript.py b/atlas-terminal/server/routers/video_transcript.py new file mode 100644 index 0000000..711c2fb --- /dev/null +++ b/atlas-terminal/server/routers/video_transcript.py @@ -0,0 +1,113 @@ +"""Video transcript endpoints.""" + +from __future__ import annotations + +from fastapi import APIRouter, File, Form, Header, HTTPException, Query, UploadFile + +from server.models.schemas import VideoJob, VideoSearchHit, VideoSubmitRequest, VideoTranscript, VideoTranslation +from server.services.video_transcript_service import ( + delete_job, + get_job_detail, + list_jobs, + search_jobs, + submit_job, + submit_upload, + translate_job_content, +) + +router = APIRouter() + + +def _resolve_api_key(header_value: str | None) -> str | None: + return (header_value or "").strip() or None + + +@router.post("/submit", response_model=VideoJob, summary="Submit a video or audio transcript job") +async def submit_video_job( + body: VideoSubmitRequest, + x_gemini_api_key: str | None = Header(default=None), +) -> VideoJob: + job_id = await submit_job( + body.url, + body.source_type, + language=body.language, + api_key=_resolve_api_key(x_gemini_api_key), + ) + detail = await get_job_detail(job_id) + if detail is None: + raise HTTPException(status_code=500, detail="Failed to create transcript job.") + return VideoJob(**detail["job"]) + + +@router.post("/upload", response_model=VideoJob, summary="Upload local media and start transcript extraction") +async def upload_video_file( + file: UploadFile = File(...), + language: str | None = Form(default=None), + x_gemini_api_key: str | None = Header(default=None), +) -> VideoJob: + job_id = await submit_upload( + file, + language=language, + api_key=_resolve_api_key(x_gemini_api_key), + ) + detail = await get_job_detail(job_id) + if detail is None: + raise HTTPException(status_code=500, detail="Failed to create transcript job.") + return VideoJob(**detail["job"]) + + +@router.get("/jobs", response_model=list[VideoJob], summary="List recent transcript jobs") +async def video_jobs(limit: int = Query(50, ge=1, le=200)) -> list[VideoJob]: + rows = await list_jobs(limit=limit) + return [VideoJob(**row) for row in rows] + + +@router.get("/jobs/{job_id}", summary="Get transcript job details") +async def video_job_detail(job_id: str) -> dict[str, VideoJob | VideoTranscript | None]: + payload = await get_job_detail(job_id) + if payload is None: + raise HTTPException(status_code=404, detail="Transcript job not found.") + transcript = payload["transcript"] + return { + "job": VideoJob(**payload["job"]), + "transcript": None if transcript is None else VideoTranscript(**transcript), + } + + +@router.post("/jobs/{job_id}/translate", response_model=VideoTranslation, summary="Translate a stored transcript on demand") +async def translate_video_job( + job_id: str, + x_gemini_api_key: str | None = Header(default=None), + target_language: str = Query("ko", pattern="^[a-z]{2}$"), +) -> VideoTranslation: + try: + payload = await translate_job_content( + job_id, + target_language=target_language, + api_key=_resolve_api_key(x_gemini_api_key), + ) + except ValueError as exc: + raise HTTPException(status_code=400, detail=str(exc)) from exc + except RuntimeError as exc: + raise HTTPException(status_code=400, detail=str(exc)) from exc + + if payload is None: + raise HTTPException(status_code=404, detail="Transcript job not found.") + return VideoTranslation(**payload) + + +@router.get("/search", response_model=list[VideoSearchHit], summary="Full-text search saved transcripts") +async def search_transcripts( + q: str = Query(..., min_length=1, description="Query text"), + limit: int = Query(20, ge=1, le=100), +) -> list[VideoSearchHit]: + rows = await search_jobs(q, limit=limit) + return [VideoSearchHit(**row) for row in rows] + + +@router.delete("/jobs/{job_id}", summary="Delete a transcript job") +async def delete_video_job(job_id: str) -> dict[str, object]: + deleted = await delete_job(job_id) + if not deleted: + raise HTTPException(status_code=404, detail="Transcript job not found.") + return {"job_id": job_id, "deleted": True} diff --git a/atlas-terminal/server/services/gemini_service.py b/atlas-terminal/server/services/gemini_service.py index 12d9d37..896247d 100644 --- a/atlas-terminal/server/services/gemini_service.py +++ b/atlas-terminal/server/services/gemini_service.py @@ -40,7 +40,12 @@ def get_gemini_model(api_key: str) -> Any: return genai.GenerativeModel(GEMINI_MODEL) -async def generate_text(prompt: str, temperature: float = 0.3, max_tokens: int = 1200) -> str: +async def generate_text( + prompt: str, + temperature: float = 0.3, + max_tokens: int = 1200, + api_key: str | None = None, +) -> str: """Async convenience wrapper used by lightweight best-effort AI features. It reads a server-side Gemini key from the environment. Browser-local keys @@ -48,12 +53,12 @@ async def generate_text(prompt: str, temperature: float = 0.3, max_tokens: int = secrets implicitly from localStorage. """ - api_key = (os.getenv("GOOGLE_API_KEY") or os.getenv("GEMINI_API_KEY") or "").strip() - if not api_key: + resolved_api_key = (api_key or os.getenv("GOOGLE_API_KEY") or os.getenv("GEMINI_API_KEY") or "").strip() + if not resolved_api_key: raise RuntimeError("GOOGLE_API_KEY or GEMINI_API_KEY is not configured") def _run() -> str: - model = get_gemini_model(api_key) + model = get_gemini_model(resolved_api_key) response = _generate_with_retry( model, prompt, diff --git a/atlas-terminal/server/services/video_transcript_service.py b/atlas-terminal/server/services/video_transcript_service.py new file mode 100644 index 0000000..f09a0f6 --- /dev/null +++ b/atlas-terminal/server/services/video_transcript_service.py @@ -0,0 +1,929 @@ +"""Video transcript ingestion, transcription, and analysis service.""" + +from __future__ import annotations + +import asyncio +import json +import logging +import os +import re +import shutil +import tempfile +from contextlib import suppress +from dataclasses import dataclass, field +from datetime import datetime, timezone +from pathlib import Path +from threading import Lock +from typing import Any, Literal +from urllib.parse import urlparse +from uuid import uuid4 + +import httpx +from fastapi import UploadFile + +from server.db.unified_repo import repo +from server.services.gemini_service import generate_text +from server.services.text_chunker import _split_into_chunks, smart_chunk + +try: # pragma: no cover - optional dependency is exercised in integration use. + import ffmpeg +except ImportError: # pragma: no cover - tests run without multimedia stack installed. + ffmpeg = None + +try: # pragma: no cover - optional dependency is exercised in integration use. + import imageio_ffmpeg +except ImportError: # pragma: no cover - tests run without fallback binary installed. + imageio_ffmpeg = None + +try: # pragma: no cover - optional dependency is exercised in integration use. + from faster_whisper import WhisperModel +except ImportError: # pragma: no cover - tests run without Whisper installed. + WhisperModel = None + +try: # pragma: no cover - optional dependency is exercised in integration use. + import srt +except ImportError: # pragma: no cover - tests run without subtitle stack installed. + srt = None + +try: # pragma: no cover - optional dependency is exercised in integration use. + from yt_dlp import YoutubeDL +except ImportError: # pragma: no cover - tests run without downloader installed. + YoutubeDL = None + +logger = logging.getLogger(__name__) + +_YOUTUBE_CACHE_TTL = 30 * 24 * 60 * 60 +_REMOTE_MAX_BYTES = 500 * 1024 * 1024 +_PENDING_STATUSES = {"queued", "fetching", "transcribing", "analyzing"} +_WHISPER_MODEL_LOCK = Lock() +_WHISPER_MODEL: Any | None = None +_TASKS: dict[str, asyncio.Task[None]] = {} + +_TEXT_STOPWORDS = { + "about", "after", "again", "also", "been", "being", "because", "between", "could", "every", + "from", "have", "into", "just", "like", "more", "most", "only", "other", "over", "should", + "some", "such", "than", "that", "their", "there", "these", "they", "this", "those", "through", + "very", "were", "what", "when", "where", "which", "while", "with", "would", "your", "ourselves", + "ours", "you", "them", "then", "well", "will", "call", "speaker", "question", "answer", "video", +} +_POSITIVE_WORDS = {"growth", "strong", "improve", "upside", "record", "bullish", "accelerate", "opportunity"} +_NEGATIVE_WORDS = {"risk", "weak", "decline", "pressure", "downside", "loss", "uncertain", "headwind"} +_YOUTUBE_HOSTS = { + "youtube.com", + "www.youtube.com", + "m.youtube.com", + "youtu.be", + "www.youtu.be", +} +_VIDEO_EXTENSIONS = {".mp4", ".mov", ".mkv", ".avi", ".webm", ".m4v"} +_AUDIO_EXTENSIONS = {".mp3", ".wav", ".m4a", ".aac", ".ogg", ".flac", ".webm"} +_CAPTION_EXTENSIONS = {".srt", ".vtt"} + + +@dataclass +class SourceMaterial: + """Resolved source asset for one transcript job.""" + + title: str | None = None + duration_sec: int | None = None + language: str | None = None + transcript_text: str | None = None + media_path: Path | None = None + cleanup_paths: list[Path] = field(default_factory=list) + + +def _now_iso() -> str: + return datetime.now(timezone.utc).isoformat() + + +def _resolved_gemini_api_key(api_key: str | None = None) -> str | None: + resolved = (api_key or os.getenv("GOOGLE_API_KEY") or os.getenv("GEMINI_API_KEY") or "").strip() + return resolved or None + + +def _cache_key(source_url: str) -> str: + return f"video-transcript:{source_url.strip()}" + + +def _display_name_from_source(source_url: str) -> str: + parsed = urlparse(source_url) + if parsed.scheme: + if parsed.path: + tail = Path(parsed.path).name + return tail or parsed.netloc or source_url + return parsed.netloc or source_url + return Path(source_url).expanduser().name or source_url + + +def _detect_source_type(url_or_path: str, source_type: Literal["youtube", "url", "local"]) -> Literal["youtube", "url", "local"]: + candidate = (url_or_path or "").strip() + if source_type == "youtube": + return "youtube" + if source_type == "local": + return "local" + parsed = urlparse(candidate) + if parsed.scheme in {"http", "https"} and parsed.netloc.lower() in _YOUTUBE_HOSTS: + return "youtube" + if parsed.scheme in {"http", "https"}: + return "url" + if Path(candidate).expanduser().exists(): + return "local" + return source_type + + +def _is_caption_file(path: Path) -> bool: + return path.suffix.lower() in _CAPTION_EXTENSIONS + + +def _requires_audio_extraction(path: Path) -> bool: + return path.suffix.lower() in _VIDEO_EXTENSIONS + + +def _is_supported_remote_content(content_type: str, suffix: str) -> bool: + if suffix in _CAPTION_EXTENSIONS | _VIDEO_EXTENSIONS | _AUDIO_EXTENSIONS: + return True + if content_type.startswith("video/") or content_type.startswith("audio/"): + return True + return content_type in {"text/plain", "text/vtt", "application/x-subrip", "application/octet-stream"} + + +def _infer_suffix(source_url: str, content_type: str) -> str: + parsed = urlparse(source_url) + suffix = Path(parsed.path).suffix.lower() + if suffix: + return suffix + mapping = { + "video/mp4": ".mp4", + "video/quicktime": ".mov", + "audio/mpeg": ".mp3", + "audio/mp4": ".m4a", + "audio/x-m4a": ".m4a", + "text/plain": ".txt", + "text/vtt": ".vtt", + "application/x-subrip": ".srt", + } + return mapping.get(content_type, ".bin") + + +def _cleanup_paths(paths: list[Path]) -> None: + seen: set[str] = set() + for path in paths: + raw = str(path) + if not raw or raw in seen: + continue + seen.add(raw) + if path.is_dir(): + shutil.rmtree(path, ignore_errors=True) + else: + with suppress(FileNotFoundError): + path.unlink() + + +def _resolve_ffmpeg_binary() -> str: + binary = shutil.which("ffmpeg") + if binary: + return binary + if imageio_ffmpeg is not None: + return imageio_ffmpeg.get_ffmpeg_exe() + raise RuntimeError( + "ffmpeg is not installed. Install Homebrew ffmpeg or add `imageio-ffmpeg` to the Python environment." + ) + + +def _persist_upload(upload: UploadFile, target_path: Path) -> None: + target_path.parent.mkdir(parents=True, exist_ok=True) + upload.file.seek(0) + with target_path.open("wb") as handle: + shutil.copyfileobj(upload.file, handle) + + +def _pick_caption_file(directory: Path) -> Path | None: + candidates = [path for path in directory.iterdir() if path.suffix.lower() in _CAPTION_EXTENSIONS] + if not candidates: + return None + + def _score(path: Path) -> tuple[int, str]: + name = path.name.lower() + if ".ko" in name: + return (0, name) + if ".en" in name: + return (1, name) + return (2, name) + + return sorted(candidates, key=_score)[0] + + +def _language_from_filename(path: Path) -> str | None: + name = path.name.lower() + if ".ko" in name: + return "ko" + if ".en" in name: + return "en" + return None + + +def _parse_vtt_text(vtt_content: str) -> str: + lines: list[str] = [] + for raw_line in vtt_content.splitlines(): + line = raw_line.strip() + if not line or line == "WEBVTT" or "-->" in line: + continue + if line.isdigit() or line.startswith(("NOTE", "Kind:", "Language:")): + continue + lines.append(line) + return "\n".join(lines).strip() + + +def _parse_caption_text(content: str, suffix: str) -> str: + if suffix == ".vtt": + return _parse_vtt_text(content) + return _parse_srt_text(content) + + +def _extract_json_object(text: str) -> dict[str, Any] | None: + if not text or not text.strip(): + return None + cleaned = re.sub(r"^```(?:json)?\s*", "", text.strip()) + cleaned = re.sub(r"\s*```$", "", cleaned).strip() + try: + parsed = json.loads(cleaned) + except json.JSONDecodeError: + match = re.search(r"\{.*\}", cleaned, flags=re.DOTALL) + if not match: + return None + try: + parsed = json.loads(match.group(0)) + except json.JSONDecodeError: + return None + return parsed if isinstance(parsed, dict) else None + + +def _top_terms(text: str, limit: int = 6) -> list[str]: + counter: dict[str, int] = {} + for token in re.findall(r"[A-Za-z][A-Za-z\-]{3,}", text.lower()): + if token in _TEXT_STOPWORDS: + continue + counter[token] = counter.get(token, 0) + 1 + ranked = sorted(counter.items(), key=lambda item: (-item[1], item[0])) + return [token for token, _count in ranked[:limit]] + + +def _fallback_analysis(text: str) -> dict[str, Any]: + excerpt = smart_chunk(text.strip(), max_chars=720) + positives = sum(text.lower().count(word) for word in _POSITIVE_WORDS) + negatives = sum(text.lower().count(word) for word in _NEGATIVE_WORDS) + if positives > negatives: + sentiment = "positive" + elif negatives > positives: + sentiment = "negative" + else: + sentiment = "neutral" + keywords = _top_terms(text, limit=6) + return { + "summary": excerpt or "Transcript extracted, but AI summary is unavailable.", + "keywords": keywords, + "topics": keywords[:4], + "sentiment": sentiment, + "intent": "Review transcript manually or add a Gemini API key for richer analysis.", + } + + +def _language_display_name(target_language: str) -> str: + return { + "ko": "Korean", + "ja": "Japanese", + "en": "English", + }.get(target_language.lower(), target_language.upper()) + + +def _normalise_analysis(payload: dict[str, Any], fallback: dict[str, Any]) -> dict[str, Any]: + keywords = payload.get("keywords") + topics = payload.get("topics") + sentiment = str(payload.get("sentiment") or fallback["sentiment"]).strip().lower() + if sentiment not in {"positive", "neutral", "negative"}: + sentiment = fallback["sentiment"] + return { + "summary": str(payload.get("summary") or fallback["summary"]).strip(), + "keywords": [str(item).strip() for item in (keywords if isinstance(keywords, list) else fallback["keywords"]) if str(item).strip()], + "topics": [str(item).strip() for item in (topics if isinstance(topics, list) else fallback["topics"]) if str(item).strip()], + "sentiment": sentiment, + "intent": str(payload.get("intent") or fallback["intent"]).strip(), + } + + +def _get_whisper_model() -> Any: + global _WHISPER_MODEL + if _WHISPER_MODEL is not None: + return _WHISPER_MODEL + if WhisperModel is None: + raise RuntimeError("faster-whisper is not installed. Run `pip install -r requirements.txt`.") + with _WHISPER_MODEL_LOCK: + if _WHISPER_MODEL is None: + _WHISPER_MODEL = WhisperModel( + os.getenv("WHISPER_MODEL_SIZE", "base"), + device="cpu", + compute_type=os.getenv("WHISPER_COMPUTE_TYPE", "int8"), + ) + return _WHISPER_MODEL + + +async def submit_job( + url_or_path: str, + source_type: Literal["youtube", "url", "local"], + *, + language: str | None = None, + api_key: str | None = None, + source_label: str | None = None, + cleanup_source: bool = False, +) -> str: + """Create a new transcript job and start asynchronous processing.""" + + resolved_source_type = _detect_source_type(url_or_path, source_type) + source_value = url_or_path.strip() if resolved_source_type != "local" else str(Path(url_or_path).expanduser()) + job_id = uuid4().hex + await repo.add_video_job(job_id, source_value, resolved_source_type) + if source_label: + await repo.update_video_job(job_id, title=source_label) + + task = asyncio.create_task( + _process_job( + job_id, + language=language, + api_key=(api_key or "").strip() or None, + source_label=source_label, + cleanup_source=cleanup_source, + ) + ) + _TASKS[job_id] = task + task.add_done_callback(lambda _: _TASKS.pop(job_id, None)) + return job_id + + +async def submit_upload(upload: UploadFile, *, language: str | None = None, api_key: str | None = None) -> str: + """Persist a multipart upload locally and enqueue it as a transcript job.""" + + temp_dir = Path(tempfile.mkdtemp(prefix="atlas-video-upload-")) + suffix = Path(upload.filename or "upload.bin").suffix + target_path = temp_dir / f"upload{suffix}" + await asyncio.to_thread(_persist_upload, upload, target_path) + await upload.close() + return await submit_job( + str(target_path), + "local", + language=language, + api_key=api_key, + source_label=upload.filename or target_path.name, + cleanup_source=True, + ) + + +async def list_jobs(limit: int = 50) -> list[dict[str, Any]]: + return await repo.list_video_jobs(limit=limit) + + +async def get_job_detail(job_id: str) -> dict[str, Any] | None: + row = await repo.get_video_job(job_id) + if row is None: + return None + return { + "job": { + "job_id": row["job_id"], + "status": row["status"], + "source_url": row["source_url"], + "source_type": row["source_type"], + "progress": row.get("progress", 0), + "error": row.get("error"), + "title": row.get("title"), + "duration_sec": row.get("duration_sec"), + "language": row.get("language"), + "created_at": row["created_at"], + "completed_at": row.get("completed_at"), + }, + "transcript": None if not row.get("transcript_text") else { + "job_id": row["job_id"], + "text": row.get("transcript_text", ""), + "summary": row.get("summary"), + "keywords": row.get("keywords", []), + "topics": row.get("topics", []), + "sentiment": row.get("sentiment"), + "intent": row.get("intent"), + }, + } + + +async def search_jobs(query: str, limit: int = 20) -> list[dict[str, Any]]: + return await repo.search_videos(query, limit=limit) + + +async def delete_job(job_id: str) -> bool: + task = _TASKS.pop(job_id, None) + if task is not None and not task.done(): + task.cancel() + with suppress(asyncio.CancelledError, Exception): + await task + return await repo.delete_video_job(job_id) + + +async def translate_job_content( + job_id: str, + *, + target_language: str = "ko", + api_key: str | None = None, +) -> dict[str, Any] | None: + """Translate a stored transcript payload on demand.""" + + row = await repo.get_video_job(job_id) + if row is None: + return None + + transcript_text = (row.get("transcript_text") or "").strip() + if not transcript_text: + raise ValueError("Transcript text is not available yet.") + + resolved_api_key = _resolved_gemini_api_key(api_key) + if not resolved_api_key: + raise RuntimeError("Set a Gemini API key in Settings to use Korean translation.") + + target_label = _language_display_name(target_language) + translated_text = await _translate_text_chunks( + transcript_text, + target_language=target_language, + api_key=resolved_api_key, + ) + + source_payload = { + "summary": row.get("summary") or "", + "keywords": row.get("keywords", []), + "topics": row.get("topics", []), + "intent": row.get("intent") or "", + } + meta_prompt = ( + f"Translate the following analyst metadata into natural {target_label}.\n" + "Return ONLY valid JSON with this exact shape:\n" + '{"summary":"...","keywords":["..."],"topics":["..."],"intent":"..."}\n' + "- Preserve company names, proper nouns, and figures.\n" + "- Translate keywords and topics into concise Korean finance/media terms.\n" + "- Do not add markdown fences or commentary.\n\n" + f"SOURCE_JSON:\n{json.dumps(source_payload, ensure_ascii=False)}" + ) + translated_meta = _extract_json_object( + await generate_text(meta_prompt, temperature=0.2, max_tokens=1200, api_key=resolved_api_key) + ) or {} + + translated_summary = str(translated_meta.get("summary") or "").strip() + if not translated_summary: + translated_summary = await _translate_text_chunks( + source_payload["summary"] or smart_chunk(transcript_text, max_chars=1500), + target_language=target_language, + api_key=resolved_api_key, + max_chars=4_500, + min_chunk=1_200, + ) + + return { + "job_id": job_id, + "target_language": target_language, + "summary": translated_summary, + "keywords": [str(item).strip() for item in (translated_meta.get("keywords") or source_payload["keywords"]) if str(item).strip()], + "topics": [str(item).strip() for item in (translated_meta.get("topics") or source_payload["topics"]) if str(item).strip()], + "intent": str(translated_meta.get("intent") or source_payload["intent"]).strip(), + "text": translated_text, + } + + +async def _process_job( + job_id: str, + *, + language: str | None = None, + api_key: str | None = None, + source_label: str | None = None, + cleanup_source: bool = False, +) -> None: + """Resolve source media, extract transcript text, run analysis, and persist results.""" + + row = await repo.get_video_job(job_id) + if row is None: + return + + cleanup_paths: list[Path] = [] + try: + source_url = row["source_url"] + source_type = _detect_source_type(source_url, row["source_type"]) + await repo.update_video_job(job_id, source_type=source_type, status="fetching", progress=15, error=None) + + if source_type == "youtube": + cached = await repo.cache_get(_cache_key(source_url)) + if isinstance(cached, dict) and cached.get("transcript_text"): + await repo.update_video_job( + job_id, + title=source_label or cached.get("title"), + duration_sec=cached.get("duration_sec"), + language=cached.get("language"), + transcript_text=cached.get("transcript_text"), + summary=cached.get("summary"), + keywords=cached.get("keywords", []), + topics=cached.get("topics", []), + sentiment=cached.get("sentiment"), + intent=cached.get("intent"), + status="completed", + progress=100, + completed_at=_now_iso(), + error=None, + ) + return + material = await _fetch_youtube_source(source_url, preferred_language=language) + elif source_type == "local": + material = await _prepare_local_source(source_url, owned_source=cleanup_source) + else: + material = await _fetch_remote_source(source_url) + + cleanup_paths.extend(material.cleanup_paths) + title = source_label or material.title or row.get("title") or _display_name_from_source(source_url) + detected_language = material.language or language + duration_sec = material.duration_sec + + await repo.update_video_job( + job_id, + title=title, + duration_sec=duration_sec, + language=detected_language, + status="transcribing", + progress=45, + ) + + transcript_text = (material.transcript_text or "").strip() + if not transcript_text: + if material.media_path is None: + raise RuntimeError("No subtitle track or downloadable media was found for this source.") + transcribe_path = material.media_path + if _requires_audio_extraction(material.media_path): + transcribe_path = await asyncio.to_thread(_extract_audio_track, material.media_path) + cleanup_paths.append(transcribe_path) + whisper_result = await asyncio.to_thread(_whisper_transcribe, str(transcribe_path), language) + transcript_text = (whisper_result.get("text") or "").strip() + detected_language = whisper_result.get("language") or detected_language + duration_value = whisper_result.get("duration_sec") or duration_sec + duration_sec = int(round(duration_value)) if duration_value else duration_sec + + if not transcript_text: + raise RuntimeError("Transcript extraction finished, but no text was produced.") + + await repo.update_video_job( + job_id, + title=title, + duration_sec=duration_sec, + language=detected_language, + transcript_text=transcript_text, + progress=65, + ) + + await repo.update_video_job(job_id, status="analyzing", progress=80) + analysis = await _analyze_text(transcript_text, api_key=api_key, title=title) + completed_at = _now_iso() + await repo.update_video_job( + job_id, + status="completed", + progress=100, + completed_at=completed_at, + error=None, + title=title, + duration_sec=duration_sec, + language=detected_language, + transcript_text=transcript_text, + summary=analysis["summary"], + keywords=analysis["keywords"], + topics=analysis["topics"], + sentiment=analysis["sentiment"], + intent=analysis["intent"], + ) + + if source_type == "youtube": + await repo.cache_set( + _cache_key(source_url), + { + "title": title, + "duration_sec": duration_sec, + "language": detected_language, + "transcript_text": transcript_text, + "summary": analysis["summary"], + "keywords": analysis["keywords"], + "topics": analysis["topics"], + "sentiment": analysis["sentiment"], + "intent": analysis["intent"], + "completed_at": completed_at, + }, + ttl=_YOUTUBE_CACHE_TTL, + ) + except asyncio.CancelledError: + logger.info("Video transcript job %s cancelled.", job_id) + raise + except Exception as exc: + logger.exception("Video transcript job %s failed", job_id) + await repo.update_video_job( + job_id, + status="failed", + progress=100, + error=str(exc), + completed_at=_now_iso(), + ) + finally: + await asyncio.to_thread(_cleanup_paths, cleanup_paths) + + +async def _prepare_local_source(source_path: str, *, owned_source: bool = False) -> SourceMaterial: + path = Path(source_path).expanduser() + if not path.exists(): + raise RuntimeError(f"Local file not found: {path}") + cleanup_paths = [path.parent] if owned_source else [] + if _is_caption_file(path): + content = await asyncio.to_thread(path.read_text, encoding="utf-8", errors="ignore") + return SourceMaterial( + title=path.name, + language=_language_from_filename(path), + transcript_text=_parse_caption_text(content, path.suffix.lower()), + cleanup_paths=cleanup_paths, + ) + return SourceMaterial( + title=path.name, + media_path=path, + cleanup_paths=cleanup_paths, + ) + + +async def _fetch_remote_source(source_url: str) -> SourceMaterial: + if _detect_source_type(source_url, "url") == "youtube": + return await _fetch_youtube_source(source_url) + + temp_dir = Path(tempfile.mkdtemp(prefix="atlas-video-remote-")) + content_type = "" + target_path = temp_dir / "download.bin" + try: + async with httpx.AsyncClient(follow_redirects=True, timeout=httpx.Timeout(60.0, connect=20.0)) as client: + async with client.stream("GET", source_url) as response: + response.raise_for_status() + content_type = response.headers.get("content-type", "").split(";")[0].strip().lower() + content_length = int(response.headers.get("content-length") or 0) + if content_length and content_length > _REMOTE_MAX_BYTES: + raise RuntimeError("Remote media is larger than the 500MB safety limit.") + suffix = _infer_suffix(source_url, content_type) + if not _is_supported_remote_content(content_type, suffix): + raise RuntimeError(f"Unsupported remote content type: {content_type or suffix}") + target_path = temp_dir / f"download{suffix}" + total_bytes = 0 + with target_path.open("wb") as handle: + async for chunk in response.aiter_bytes(): + total_bytes += len(chunk) + if total_bytes > _REMOTE_MAX_BYTES: + raise RuntimeError("Remote media exceeded the 500MB safety limit while downloading.") + handle.write(chunk) + except Exception: + await asyncio.to_thread(_cleanup_paths, [temp_dir]) + raise + + if _is_caption_file(target_path): + content = await asyncio.to_thread(target_path.read_text, encoding="utf-8", errors="ignore") + return SourceMaterial( + title=_display_name_from_source(source_url), + language=_language_from_filename(target_path), + transcript_text=_parse_caption_text(content, target_path.suffix.lower()), + cleanup_paths=[temp_dir], + ) + + return SourceMaterial( + title=_display_name_from_source(source_url), + media_path=target_path, + cleanup_paths=[temp_dir], + ) + + +async def _fetch_youtube_source(source_url: str, preferred_language: str | None = None) -> SourceMaterial: + temp_dir = Path(tempfile.mkdtemp(prefix="atlas-video-youtube-")) + try: + info: dict[str, Any] = {} + try: + info = await asyncio.to_thread(_download_youtube_subtitles, source_url, temp_dir, preferred_language) + except Exception: + logger.warning("YouTube subtitle download failed for %s; falling back to audio transcription.", source_url, exc_info=True) + title = str(info.get("title") or _display_name_from_source(source_url)) + duration_sec = int(info.get("duration")) if info.get("duration") else None + caption_path = _pick_caption_file(temp_dir) + if caption_path is not None: + content = await asyncio.to_thread(caption_path.read_text, encoding="utf-8", errors="ignore") + return SourceMaterial( + title=title, + duration_sec=duration_sec, + language=_language_from_filename(caption_path), + transcript_text=_parse_caption_text(content, caption_path.suffix.lower()), + cleanup_paths=[temp_dir], + ) + + media_path, audio_info = await asyncio.to_thread(_download_youtube_audio, source_url, temp_dir) + return SourceMaterial( + title=str(audio_info.get("title") or title), + duration_sec=int(audio_info.get("duration")) if audio_info.get("duration") else duration_sec, + media_path=media_path, + cleanup_paths=[temp_dir], + ) + except Exception: + await asyncio.to_thread(_cleanup_paths, [temp_dir]) + raise + + +def _normalise_subtitle_languages(preferred_language: str | None = None) -> list[str]: + ordered: list[str] = [] + for value in [preferred_language, "en", "ko", "en-US", "ko-KR"]: + candidate = (value or "").strip() + if candidate and candidate not in ordered: + ordered.append(candidate) + return ordered + + +def _download_youtube_subtitles(source_url: str, output_dir: Path, preferred_language: str | None = None) -> dict[str, Any]: + if YoutubeDL is None: + raise RuntimeError("yt-dlp is not installed. Run `pip install -r requirements.txt`.") + last_error: Exception | None = None + for language in _normalise_subtitle_languages(preferred_language): + options = { + "quiet": True, + "no_warnings": True, + "noprogress": True, + "noplaylist": True, + "skip_download": True, + "writesubtitles": True, + "writeautomaticsub": True, + "subtitleslangs": [language], + "subtitlesformat": "srt/vtt/best", + "outtmpl": str(output_dir / "%(id)s.%(ext)s"), + } + try: + with YoutubeDL(options) as ydl: + info = ydl.extract_info(source_url, download=True) + if _pick_caption_file(output_dir) is not None: + return info + except Exception as exc: + last_error = exc + logger.warning("Subtitle attempt failed for language %s on %s", language, source_url, exc_info=True) + + if last_error is not None: + raise last_error + return {} + + +def _download_youtube_audio(source_url: str, output_dir: Path) -> tuple[Path, dict[str, Any]]: + if YoutubeDL is None: + raise RuntimeError("yt-dlp is not installed. Run `pip install -r requirements.txt`.") + options = { + "quiet": True, + "no_warnings": True, + "noprogress": True, + "noplaylist": True, + "format": "bestaudio[ext=m4a]/bestaudio/best", + "outtmpl": str(output_dir / "%(id)s.%(ext)s"), + } + with YoutubeDL(options) as ydl: + info = ydl.extract_info(source_url, download=True) + return Path(ydl.prepare_filename(info)), info + + +def _extract_audio_track(media_path: Path) -> Path: + if ffmpeg is None: + raise RuntimeError("ffmpeg-python is not installed. Run `pip install -r requirements.txt`.") + output_path = media_path.parent / f"{media_path.stem}.atlas.wav" + ffmpeg_binary = _resolve_ffmpeg_binary() + try: + stream = ffmpeg.input(str(media_path)) + stream = ffmpeg.output(stream, str(output_path), acodec="pcm_s16le", ac=1, ar="16000", format="wav") + ffmpeg.run(stream.overwrite_output(), cmd=ffmpeg_binary, capture_stdout=True, capture_stderr=True) + except ffmpeg.Error as exc: # type: ignore[attr-defined] + stderr = exc.stderr.decode("utf-8", errors="ignore") if getattr(exc, "stderr", None) else str(exc) + raise RuntimeError(f"ffmpeg failed to extract audio: {stderr.strip()}") from exc + return output_path + + +def _whisper_transcribe(audio_path: str, language: str | None = None) -> dict[str, Any]: + """Transcribe audio or video with a lazily loaded faster-whisper model.""" + + model = _get_whisper_model() + segments, info = model.transcribe(audio_path, language=language, vad_filter=True, beam_size=5) + parsed_segments: list[dict[str, Any]] = [] + for segment in segments: + text = (segment.text or "").strip() + if not text: + continue + parsed_segments.append( + { + "start": round(float(segment.start), 2), + "end": round(float(segment.end), 2), + "text": text, + } + ) + full_text = " ".join(segment["text"] for segment in parsed_segments).strip() + duration_sec = getattr(info, "duration", None) or (parsed_segments[-1]["end"] if parsed_segments else 0) + return { + "text": full_text, + "language": getattr(info, "language", None) or language, + "duration_sec": duration_sec, + "segments": parsed_segments, + } + + +def _parse_srt_text(srt_content: str) -> str: + """Strip timing metadata and join subtitle text into a clean transcript.""" + + if not srt_content.strip(): + return "" + if srt is None: + cleaned = re.sub(r"\d+\s+\d{2}:\d{2}:\d{2},\d{3}\s+-->\s+\d{2}:\d{2}:\d{2},\d{3}", "", srt_content) + cleaned = re.sub(r"^\d+\s*$", "", cleaned, flags=re.MULTILINE) + return "\n".join(line.strip() for line in cleaned.splitlines() if line.strip()) + + subtitles = [] + for item in srt.parse(srt_content): + text = " ".join(line.strip() for line in item.content.splitlines() if line.strip()).strip() + if text: + subtitles.append(text) + return "\n".join(subtitles) + + +async def _analyze_text(text: str, *, api_key: str | None = None, title: str | None = None) -> dict[str, Any]: + """Generate a summary, topics, and sentiment for a transcript.""" + + fallback = _fallback_analysis(text) + chunk_candidates = _split_into_chunks(text, max_chars=10_000, min_chunk=3_000) + chunks = [smart_chunk(chunk, max_chars=10_000) for chunk in chunk_candidates if chunk.strip()] or [smart_chunk(text, max_chars=10_000)] + + try: + partial_summaries: list[str] = [] + for index, chunk in enumerate(chunks, start=1): + prompt = ( + "You are preparing analyst notes from one transcript chunk.\n" + f"Chunk {index} of {len(chunks)} for: {title or 'Untitled media'}\n\n" + "Return 4-6 concise bullet points covering factual takeaways, recurring themes, risks, and speaker tone.\n" + "Do not invent facts. Keep names, products, and figures exact when present.\n\n" + f"TRANSCRIPT CHUNK:\n{chunk}" + ) + partial_summaries.append( + await generate_text(prompt, temperature=0.2, max_tokens=700, api_key=api_key) + ) + + summary_blob = "\n\n".join(f"Chunk {idx} notes:\n{summary.strip()}" for idx, summary in enumerate(partial_summaries, start=1)) + head_sample = smart_chunk(text[:10_000], max_chars=2_000) + tail_sample = smart_chunk(text[-10_000:], max_chars=2_000) + final_prompt = ( + "You are an expert media and transcript analyst.\n" + "Use the chunk notes and transcript samples below.\n" + "Return ONLY valid JSON with this exact shape:\n" + '{"summary":"...","keywords":["..."],"topics":["..."],"sentiment":"positive|neutral|negative","intent":"..."}\n' + "Rules:\n" + "- summary: 3-5 sentences, specific and factual\n" + "- keywords: 5-8 concise phrases\n" + "- topics: 3-5 broader themes\n" + "- sentiment: choose only positive, neutral, or negative\n" + "- intent: one concise sentence describing what the speaker or content is trying to achieve\n" + "- Do not add markdown fences or commentary.\n\n" + f"TITLE: {title or 'Untitled media'}\n\n" + f"CHUNK NOTES:\n{summary_blob}\n\n" + f"HEAD SAMPLE:\n{head_sample}\n\n" + f"TAIL SAMPLE:\n{tail_sample}" + ) + parsed = _extract_json_object( + await generate_text(final_prompt, temperature=0.2, max_tokens=1200, api_key=api_key) + ) + if parsed is None: + return fallback + return _normalise_analysis(parsed, fallback) + except Exception: + logger.warning("Falling back to heuristic transcript analysis.", exc_info=True) + return fallback + + +async def _translate_text_chunks( + text: str, + *, + target_language: str, + api_key: str, + max_chars: int = 7_000, + min_chunk: int = 2_000, +) -> str: + """Translate long text in chunks while preserving structure.""" + + source = text.strip() + if not source: + return "" + + target_label = _language_display_name(target_language) + chunks = _split_into_chunks(source, max_chars=max_chars, min_chunk=min_chunk) or [source] + translated_chunks: list[str] = [] + for index, chunk in enumerate(chunks, start=1): + prompt = ( + f"Translate the following transcript chunk into natural {target_label}.\n" + "- Preserve meaning, numbers, names, and paragraph breaks.\n" + "- Keep speaker turns and emphasis where obvious.\n" + "- Return ONLY the translated text.\n" + f"- This is chunk {index} of {len(chunks)}.\n\n" + f"TRANSCRIPT CHUNK:\n{chunk}" + ) + translated_chunks.append( + (await generate_text(prompt, temperature=0.2, max_tokens=2200, api_key=api_key)).strip() + ) + return "\n\n".join(chunk for chunk in translated_chunks if chunk) diff --git a/atlas-terminal/tests/test_video_transcript_service.py b/atlas-terminal/tests/test_video_transcript_service.py new file mode 100644 index 0000000..73bdce3 --- /dev/null +++ b/atlas-terminal/tests/test_video_transcript_service.py @@ -0,0 +1,165 @@ +"""Tests for the video transcript service and repository integration.""" + +from __future__ import annotations + +import asyncio + +from server.db.unified_repo import repo +from server.services import video_transcript_service as service + + +def test_parse_srt_text_strips_timestamps() -> None: + raw_srt = """1 +00:00:00,000 --> 00:00:01,500 +Welcome to the call. + +2 +00:00:01,500 --> 00:00:03,000 +Revenue grew 12%. +""" + + parsed = service._parse_srt_text(raw_srt) + + assert "00:00:00,000" not in parsed + assert parsed == "Welcome to the call.\nRevenue grew 12%." + + +def test_analyze_text_runs_chunk_map_reduce(monkeypatch) -> None: + calls: list[str] = [] + + async def fake_generate_text(prompt: str, **_: object) -> str: + calls.append(prompt) + if len(calls) < 3: + return f"- chunk note {len(calls)}" + return ( + '{"summary":"Management highlighted revenue growth.",' + '"keywords":["revenue growth","guidance"],' + '"topics":["Earnings"],' + '"sentiment":"positive",' + '"intent":"Reassure investors about momentum."}' + ) + + monkeypatch.setattr(service, "generate_text", fake_generate_text) + monkeypatch.setattr(service, "_split_into_chunks", lambda text, max_chars=10_000, min_chunk=3_000: ["chunk one", "chunk two"]) + + result = asyncio.run(service._analyze_text("long transcript body", api_key="test-key", title="Demo call")) + + assert result["summary"] == "Management highlighted revenue growth." + assert result["keywords"] == ["revenue growth", "guidance"] + assert len(calls) == 3 + + +def test_submit_job_processes_statuses_to_completion(monkeypatch) -> None: + async def run_test() -> None: + await repo.init_db() + statuses: list[str] = [] + original_update = service.repo.update_video_job + + async def recording_update(job_id: str, **fields: object): + if "status" in fields: + statuses.append(str(fields["status"])) + return await original_update(job_id, **fields) + + async def fake_prepare_local_source(source_path: str, *, owned_source: bool = False) -> service.SourceMaterial: + return service.SourceMaterial( + title="demo.mp4", + duration_sec=96, + language="en", + transcript_text="Revenue grew strongly and management stayed confident about AI demand.", + ) + + async def fake_analyze_text(text: str, *, api_key: str | None = None, title: str | None = None): + return { + "summary": "Revenue and AI demand were the main focus.", + "keywords": ["revenue", "ai demand"], + "topics": ["Earnings"], + "sentiment": "positive", + "intent": "Reassure investors about execution.", + } + + monkeypatch.setattr(service.repo, "update_video_job", recording_update) + monkeypatch.setattr(service, "_prepare_local_source", fake_prepare_local_source) + monkeypatch.setattr(service, "_analyze_text", fake_analyze_text) + + job_id = await service.submit_job("/tmp/demo.mp4", "local", api_key="test-key", source_label="demo.mp4") + await service._TASKS[job_id] + row = await repo.get_video_job(job_id) + + assert row is not None + assert row["status"] == "completed" + assert row["progress"] == 100 + assert row["summary"] == "Revenue and AI demand were the main focus." + assert statuses == ["fetching", "transcribing", "analyzing", "completed"] + + asyncio.run(run_test()) + + +def test_video_fts_search_returns_snippet() -> None: + async def run_test() -> None: + await repo.init_db() + await repo.add_video_job("job-1", "https://example.com/earnings", "url") + await repo.update_video_job( + "job-1", + title="Q1 Earnings Call", + transcript_text="Management discussed AI revenue, guidance, and margin expansion in detail.", + summary="AI revenue and guidance dominated the discussion.", + ) + await repo.add_video_job("job-2", "https://example.com/interview", "url") + await repo.update_video_job( + "job-2", + title="CEO Interview", + transcript_text="Consumer demand remained steady and product cadence was unchanged.", + ) + + hits = await repo.search_videos("guidance", limit=10) + + assert hits + assert hits[0]["job_id"] == "job-1" + assert "guidance" in hits[0]["snippet"].lower() + assert hits[0]["rank"] > 0 + + asyncio.run(run_test()) + + +def test_translate_job_content_translates_meta_and_chunks(monkeypatch) -> None: + async def run_test() -> None: + await repo.init_db() + await repo.add_video_job("job-ko", "https://example.com/demo", "url") + await repo.update_video_job( + "job-ko", + transcript_text="first chunk\n\nsecond chunk", + summary="Original summary.", + keywords=["guidance", "margin"], + topics=["Earnings"], + intent="Reassure investors.", + status="completed", + progress=100, + ) + + prompts: list[str] = [] + + async def fake_generate_text(prompt: str, **_: object) -> str: + prompts.append(prompt) + if "SOURCE_JSON" in prompt: + return ( + '{"summary":"번역된 요약","keywords":["가이던스","마진"],' + '"topics":["실적"],"intent":"투자자를 안심시키려는 목적입니다."}' + ) + if "chunk 1 of 2" in prompt: + return "첫 번째 청크" + if "chunk 2 of 2" in prompt: + return "두 번째 청크" + return "대체 요약" + + monkeypatch.setattr(service, "generate_text", fake_generate_text) + monkeypatch.setattr(service, "_split_into_chunks", lambda text, max_chars=7000, min_chunk=2000: ["first chunk", "second chunk"]) + + translated = await service.translate_job_content("job-ko", target_language="ko", api_key="test-key") + + assert translated is not None + assert translated["summary"] == "번역된 요약" + assert translated["keywords"] == ["가이던스", "마진"] + assert translated["text"] == "첫 번째 청크\n\n두 번째 청크" + assert len(prompts) == 3 + + asyncio.run(run_test())