mirror of
https://github.com/safishamsi/graphify.git
synced 2026-09-23 22:15:46 +00:00
Add video/audio corpus support with yt-dlp download and Whisper transcription
Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
Safi
co-authored by
Claude Sonnet 4.6
Claude Sonnet 4.6
parent
e3adc0417a
commit
cf5d721bee
@@ -2,6 +2,14 @@
|
||||
|
||||
Full release notes with details on each version: [GitHub Releases](https://github.com/safishamsi/graphify/releases)
|
||||
|
||||
## 0.3.29 (2026-04-10)
|
||||
|
||||
- Add: video and audio corpus support — drop `.mp4`, `.mp3`, `.wav`, `.mov`, `.webm`, `.m4a`, `.ogg`, `.mkv`, `.avi`, `.m4v` files into any corpus and graphify transcribes them with faster-whisper before extraction
|
||||
- Add: YouTube and URL video download — pass a YouTube link (or any video URL) to `/graphify add <url>` and yt-dlp downloads audio-only, which is then transcribed and added to the corpus automatically
|
||||
- Add: domain-aware Whisper prompts — god nodes from non-video files are used to build a one-sentence domain hint for Whisper via a cheap Haiku call, improving transcript accuracy on technical content
|
||||
- Add: `graphify-out/transcripts/` cache — transcripts are cached by filename so re-runs skip already-transcribed files; URLs cached by hash
|
||||
- Requires: `pip install 'graphifyy[video]'` for faster-whisper + yt-dlp
|
||||
|
||||
## 0.3.28 (2026-04-10)
|
||||
|
||||
- Fix: hook installers (Claude Code, Codex, Gemini CLI) now always remove and reinstall the hook on re-run — users upgrading from old versions no longer get stuck with a broken hook format (#182)
|
||||
|
||||
@@ -9,7 +9,7 @@
|
||||
|
||||
**An AI coding assistant skill.** Type `/graphify` in Claude Code, Codex, OpenCode, Cursor, Gemini CLI, GitHub Copilot CLI, Aider, OpenClaw, Factory Droid, or Trae - it reads your files, builds a knowledge graph, and gives you back structure you didn't know was there. Understand a codebase faster. Find the "why" behind architectural decisions.
|
||||
|
||||
Fully multimodal. Drop in code, PDFs, markdown, screenshots, diagrams, whiteboard photos, even images in other languages - graphify uses Claude vision to extract concepts and relationships from all of it and connects them into one graph. 20 languages supported via tree-sitter AST (Python, JS, TS, Go, Rust, Java, C, C++, Ruby, C#, Kotlin, Scala, PHP, Swift, Lua, Zig, PowerShell, Elixir, Objective-C, Julia).
|
||||
Fully multimodal. Drop in code, PDFs, markdown, screenshots, diagrams, whiteboard photos, images in other languages, or video and audio files - graphify extracts concepts and relationships from all of it and connects them into one graph. Videos are transcribed with Whisper using a domain-aware prompt derived from your corpus. 20 languages supported via tree-sitter AST (Python, JS, TS, Go, Rust, Java, C, C++, Ruby, C#, Kotlin, Scala, PHP, Swift, Lua, Zig, PowerShell, Elixir, Objective-C, Julia).
|
||||
|
||||
> Andrej Karpathy keeps a `/raw` folder where he drops papers, tweets, screenshots, and notes. graphify is the answer to that problem - 71.5x fewer tokens per query vs reading the raw files, persistent across sessions, honest about what it found vs guessed.
|
||||
|
||||
|
||||
+7
-1
@@ -13,6 +13,7 @@ class FileType(str, Enum):
|
||||
DOCUMENT = "document"
|
||||
PAPER = "paper"
|
||||
IMAGE = "image"
|
||||
VIDEO = "video"
|
||||
|
||||
|
||||
_MANIFEST_PATH = "graphify-out/manifest.json"
|
||||
@@ -22,6 +23,7 @@ DOC_EXTENSIONS = {'.md', '.txt', '.rst'}
|
||||
PAPER_EXTENSIONS = {'.pdf'}
|
||||
IMAGE_EXTENSIONS = {'.png', '.jpg', '.jpeg', '.gif', '.webp', '.svg'}
|
||||
OFFICE_EXTENSIONS = {'.docx', '.xlsx'}
|
||||
VIDEO_EXTENSIONS = {'.mp4', '.mov', '.webm', '.mkv', '.avi', '.m4v', '.mp3', '.wav', '.m4a', '.ogg'}
|
||||
|
||||
CORPUS_WARN_THRESHOLD = 50_000 # words - below this, warn "you may not need a graph"
|
||||
CORPUS_UPPER_THRESHOLD = 500_000 # words - above this, warn about token cost
|
||||
@@ -95,6 +97,8 @@ def classify_file(path: Path) -> FileType | None:
|
||||
return FileType.DOCUMENT
|
||||
if ext in OFFICE_EXTENSIONS:
|
||||
return FileType.DOCUMENT
|
||||
if ext in VIDEO_EXTENSIONS:
|
||||
return FileType.VIDEO
|
||||
return None
|
||||
|
||||
|
||||
@@ -318,6 +322,7 @@ def detect(root: Path, *, follow_symlinks: bool = False) -> dict:
|
||||
FileType.DOCUMENT: [],
|
||||
FileType.PAPER: [],
|
||||
FileType.IMAGE: [],
|
||||
FileType.VIDEO: [],
|
||||
}
|
||||
total_words = 0
|
||||
|
||||
@@ -388,7 +393,8 @@ def detect(root: Path, *, follow_symlinks: bool = False) -> dict:
|
||||
skipped_sensitive.append(str(p) + " [office conversion failed - pip install graphifyy[office]]")
|
||||
continue
|
||||
files[ftype].append(str(p))
|
||||
total_words += count_words(p)
|
||||
if ftype != FileType.VIDEO:
|
||||
total_words += count_words(p)
|
||||
|
||||
total_files = sum(len(v) for v in files.values())
|
||||
needs_graph = total_words >= CORPUS_WARN_THRESHOLD
|
||||
|
||||
@@ -207,6 +207,12 @@ def ingest(url: str, target_dir: Path, author: str | None = None, contributor: s
|
||||
print(f"Downloaded image: {out.name}")
|
||||
return out
|
||||
|
||||
if url_type == "youtube":
|
||||
from graphify.transcribe import download_audio
|
||||
out = download_audio(url, target_dir)
|
||||
print(f"Downloaded audio: {out.name}")
|
||||
return out
|
||||
|
||||
if url_type == "tweet":
|
||||
content, filename = _fetch_tweet(url, author, contributor)
|
||||
elif url_type == "arxiv":
|
||||
|
||||
+48
-1
@@ -98,13 +98,60 @@ Corpus: X files · ~Y words
|
||||
docs: N files (.md .txt ...)
|
||||
papers: N files (.pdf ...)
|
||||
images: N files
|
||||
video: N files (.mp4 .mp3 ...)
|
||||
```
|
||||
|
||||
Omit any category with 0 files from the summary.
|
||||
|
||||
Then act on it:
|
||||
- If `total_files` is 0: stop with "No supported files found in [path]."
|
||||
- If `skipped_sensitive` is non-empty: mention file count skipped, not the file names.
|
||||
- If `total_words` > 2,000,000 OR `total_files` > 200: show the warning and the top 5 subdirectories by file count, then ask which subfolder to run on. Wait for the user's answer before proceeding.
|
||||
- Otherwise: proceed directly to Step 3 - no need to ask anything.
|
||||
- Otherwise: proceed directly to Step 2.5 if video files were detected, or Step 3 if not.
|
||||
|
||||
### Step 2.5 - Transcribe video / audio files (only if video files detected)
|
||||
|
||||
Skip this step entirely if `detect` returned zero `video` files.
|
||||
|
||||
Video and audio files cannot be read directly. Transcribe them to text first, then treat the transcripts as doc files in Step 3.
|
||||
|
||||
**Strategy:** Run non-video semantic extraction first (Step 3B) to get god nodes, use those to build a domain hint for Whisper, then transcribe. This keeps the prompt relevant without guessing the corpus topic from filenames.
|
||||
|
||||
**However**, if the corpus has *only* video files and no other docs/code, skip the god-node step and transcribe with the generic fallback prompt immediately.
|
||||
|
||||
**Transcription command:**
|
||||
|
||||
```bash
|
||||
$(cat graphify-out/.graphify_python) -c "
|
||||
import json
|
||||
from pathlib import Path
|
||||
from graphify.transcribe import build_whisper_prompt, transcribe_all
|
||||
|
||||
detect = json.loads(Path('graphify-out/.graphify_detect.json').read_text())
|
||||
video_files = detect.get('files', {}).get('video', [])
|
||||
|
||||
# Try to load god nodes from a previous partial run or pass [] if not yet available
|
||||
try:
|
||||
analysis = json.loads(Path('graphify-out/.graphify_analysis.json').read_text())
|
||||
god_nodes = analysis.get('god_nodes', [])
|
||||
except Exception:
|
||||
god_nodes = []
|
||||
|
||||
prompt = build_whisper_prompt(god_nodes)
|
||||
print(f'Whisper prompt: {prompt}')
|
||||
|
||||
transcript_paths = transcribe_all(video_files, initial_prompt=prompt)
|
||||
print(json.dumps(transcript_paths))
|
||||
" > graphify-out/.graphify_transcripts.json
|
||||
```
|
||||
|
||||
After transcription:
|
||||
- Read the transcript paths from `graphify-out/.graphify_transcripts.json`
|
||||
- Add them to the docs list before dispatching semantic subagents in Step 3B
|
||||
- Print how many transcripts were created: `Transcribed N video file(s) -> treating as docs`
|
||||
- If transcription fails for a file, print a warning and continue with the rest
|
||||
|
||||
**Whisper model:** Default is `base`. If the user passed `--whisper-model <name>`, set `GRAPHIFY_WHISPER_MODEL=<name>` in the environment before running the command above.
|
||||
|
||||
### Step 3 - Extract entities and relationships
|
||||
|
||||
|
||||
+48
-1
@@ -98,13 +98,60 @@ Corpus: X files · ~Y words
|
||||
docs: N files (.md .txt ...)
|
||||
papers: N files (.pdf ...)
|
||||
images: N files
|
||||
video: N files (.mp4 .mp3 ...)
|
||||
```
|
||||
|
||||
Omit any category with 0 files from the summary.
|
||||
|
||||
Then act on it:
|
||||
- If `total_files` is 0: stop with "No supported files found in [path]."
|
||||
- If `skipped_sensitive` is non-empty: mention file count skipped, not the file names.
|
||||
- If `total_words` > 2,000,000 OR `total_files` > 200: show the warning and the top 5 subdirectories by file count, then ask which subfolder to run on. Wait for the user's answer before proceeding.
|
||||
- Otherwise: proceed directly to Step 3 - no need to ask anything.
|
||||
- Otherwise: proceed directly to Step 2.5 if video files were detected, or Step 3 if not.
|
||||
|
||||
### Step 2.5 - Transcribe video / audio files (only if video files detected)
|
||||
|
||||
Skip this step entirely if `detect` returned zero `video` files.
|
||||
|
||||
Video and audio files cannot be read directly. Transcribe them to text first, then treat the transcripts as doc files in Step 3.
|
||||
|
||||
**Strategy:** Run non-video semantic extraction first (Step 3B) to get god nodes, use those to build a domain hint for Whisper, then transcribe. This keeps the prompt relevant without guessing the corpus topic from filenames.
|
||||
|
||||
**However**, if the corpus has *only* video files and no other docs/code, skip the god-node step and transcribe with the generic fallback prompt immediately.
|
||||
|
||||
**Transcription command:**
|
||||
|
||||
```bash
|
||||
$(cat graphify-out/.graphify_python) -c "
|
||||
import json
|
||||
from pathlib import Path
|
||||
from graphify.transcribe import build_whisper_prompt, transcribe_all
|
||||
|
||||
detect = json.loads(Path('graphify-out/.graphify_detect.json').read_text())
|
||||
video_files = detect.get('files', {}).get('video', [])
|
||||
|
||||
# Try to load god nodes from a previous partial run or pass [] if not yet available
|
||||
try:
|
||||
analysis = json.loads(Path('graphify-out/.graphify_analysis.json').read_text())
|
||||
god_nodes = analysis.get('god_nodes', [])
|
||||
except Exception:
|
||||
god_nodes = []
|
||||
|
||||
prompt = build_whisper_prompt(god_nodes)
|
||||
print(f'Whisper prompt: {prompt}')
|
||||
|
||||
transcript_paths = transcribe_all(video_files, initial_prompt=prompt)
|
||||
print(json.dumps(transcript_paths))
|
||||
" > graphify-out/.graphify_transcripts.json
|
||||
```
|
||||
|
||||
After transcription:
|
||||
- Read the transcript paths from `graphify-out/.graphify_transcripts.json`
|
||||
- Add them to the docs list before dispatching semantic subagents in Step 3B
|
||||
- Print how many transcripts were created: `Transcribed N video file(s) -> treating as docs`
|
||||
- If transcription fails for a file, print a warning and continue with the rest
|
||||
|
||||
**Whisper model:** Default is `base`. If the user passed `--whisper-model <name>`, set `GRAPHIFY_WHISPER_MODEL=<name>` in the environment before running the command above.
|
||||
|
||||
### Step 3 - Extract entities and relationships
|
||||
|
||||
|
||||
+47
-1
@@ -97,13 +97,59 @@ Corpus: X files · ~Y words
|
||||
docs: N files (.md .txt ...)
|
||||
papers: N files (.pdf ...)
|
||||
images: N files
|
||||
video: N files (.mp4 .mp3 ...)
|
||||
```
|
||||
|
||||
Omit any category with 0 files from the summary.
|
||||
|
||||
Then act on it:
|
||||
- If `total_files` is 0: stop with "No supported files found in [path]."
|
||||
- If `skipped_sensitive` is non-empty: mention file count skipped, not the file names.
|
||||
- If `total_words` > 2,000,000 OR `total_files` > 200: show the warning and the top 5 subdirectories by file count, then ask which subfolder to run on. Wait for the user's answer before proceeding.
|
||||
- Otherwise: proceed directly to Step 3 - no need to ask anything.
|
||||
- Otherwise: proceed directly to Step 2.5 if video files were detected, or Step 3 if not.
|
||||
|
||||
### Step 2.5 - Transcribe video / audio files (only if video files detected)
|
||||
|
||||
Skip this step entirely if `detect` returned zero `video` files.
|
||||
|
||||
Video and audio files cannot be read directly. Transcribe them to text first, then treat the transcripts as doc files in Step 3.
|
||||
|
||||
**Strategy:** Run non-video semantic extraction first (Step 3B) to get god nodes, use those to build a domain hint for Whisper, then transcribe. This keeps the prompt relevant without guessing the corpus topic from filenames.
|
||||
|
||||
**However**, if the corpus has *only* video files and no other docs/code, skip the god-node step and transcribe with the generic fallback prompt immediately.
|
||||
|
||||
**Transcription command:**
|
||||
|
||||
```bash
|
||||
$(cat graphify-out/.graphify_python) -c "
|
||||
import json
|
||||
from pathlib import Path
|
||||
from graphify.transcribe import build_whisper_prompt, transcribe_all
|
||||
|
||||
detect = json.loads(Path('graphify-out/.graphify_detect.json').read_text())
|
||||
video_files = detect.get('files', {}).get('video', [])
|
||||
|
||||
try:
|
||||
analysis = json.loads(Path('graphify-out/.graphify_analysis.json').read_text())
|
||||
god_nodes = analysis.get('god_nodes', [])
|
||||
except Exception:
|
||||
god_nodes = []
|
||||
|
||||
prompt = build_whisper_prompt(god_nodes)
|
||||
print(f'Whisper prompt: {prompt}')
|
||||
|
||||
transcript_paths = transcribe_all(video_files, initial_prompt=prompt)
|
||||
print(json.dumps(transcript_paths))
|
||||
" > graphify-out/.graphify_transcripts.json
|
||||
```
|
||||
|
||||
After transcription:
|
||||
- Read the transcript paths from `graphify-out/.graphify_transcripts.json`
|
||||
- Add them to the docs list before dispatching semantic subagents in Step 3B
|
||||
- Print how many transcripts were created: `Transcribed N video file(s) -> treating as docs`
|
||||
- If transcription fails for a file, print a warning and continue with the rest
|
||||
|
||||
**Whisper model:** Default is `base`. If the user passed `--whisper-model <name>`, set `GRAPHIFY_WHISPER_MODEL=<name>` in the environment before running the command above.
|
||||
|
||||
### Step 3 - Extract entities and relationships
|
||||
|
||||
|
||||
@@ -100,13 +100,60 @@ Corpus: X files · ~Y words
|
||||
docs: N files (.md .txt ...)
|
||||
papers: N files (.pdf ...)
|
||||
images: N files
|
||||
video: N files (.mp4 .mp3 ...)
|
||||
```
|
||||
|
||||
Omit any category with 0 files from the summary.
|
||||
|
||||
Then act on it:
|
||||
- If `total_files` is 0: stop with "No supported files found in [path]."
|
||||
- If `skipped_sensitive` is non-empty: mention file count skipped, not the file names.
|
||||
- If `total_words` > 2,000,000 OR `total_files` > 200: show the warning and the top 5 subdirectories by file count, then ask which subfolder to run on. Wait for the user's answer before proceeding.
|
||||
- Otherwise: proceed directly to Step 3 - no need to ask anything.
|
||||
- Otherwise: proceed directly to Step 2.5 if video files were detected, or Step 3 if not.
|
||||
|
||||
### Step 2.5 - Transcribe video / audio files (only if video files detected)
|
||||
|
||||
Skip this step entirely if `detect` returned zero `video` files.
|
||||
|
||||
Video and audio files cannot be read directly. Transcribe them to text first, then treat the transcripts as doc files in Step 3.
|
||||
|
||||
**Strategy:** Run non-video semantic extraction first (Step 3B) to get god nodes, use those to build a domain hint for Whisper, then transcribe. This keeps the prompt relevant without guessing the corpus topic from filenames.
|
||||
|
||||
**However**, if the corpus has *only* video files and no other docs/code, skip the god-node step and transcribe with the generic fallback prompt immediately.
|
||||
|
||||
**Transcription command:**
|
||||
|
||||
```bash
|
||||
$(cat graphify-out/.graphify_python) -c "
|
||||
import json
|
||||
from pathlib import Path
|
||||
from graphify.transcribe import build_whisper_prompt, transcribe_all
|
||||
|
||||
detect = json.loads(Path('graphify-out/.graphify_detect.json').read_text())
|
||||
video_files = detect.get('files', {}).get('video', [])
|
||||
|
||||
# Try to load god nodes from a previous partial run or pass [] if not yet available
|
||||
try:
|
||||
analysis = json.loads(Path('graphify-out/.graphify_analysis.json').read_text())
|
||||
god_nodes = analysis.get('god_nodes', [])
|
||||
except Exception:
|
||||
god_nodes = []
|
||||
|
||||
prompt = build_whisper_prompt(god_nodes)
|
||||
print(f'Whisper prompt: {prompt}')
|
||||
|
||||
transcript_paths = transcribe_all(video_files, initial_prompt=prompt)
|
||||
print(json.dumps(transcript_paths))
|
||||
" > graphify-out/.graphify_transcripts.json
|
||||
```
|
||||
|
||||
After transcription:
|
||||
- Read the transcript paths from `graphify-out/.graphify_transcripts.json`
|
||||
- Add them to the docs list before dispatching semantic subagents in Step 3B
|
||||
- Print how many transcripts were created: `Transcribed N video file(s) -> treating as docs`
|
||||
- If transcription fails for a file, print a warning and continue with the rest
|
||||
|
||||
**Whisper model:** Default is `base`. If the user passed `--whisper-model <name>`, set `GRAPHIFY_WHISPER_MODEL=<name>` in the environment before running the command above.
|
||||
|
||||
### Step 3 - Extract entities and relationships
|
||||
|
||||
|
||||
+48
-1
@@ -98,13 +98,60 @@ Corpus: X files · ~Y words
|
||||
docs: N files (.md .txt ...)
|
||||
papers: N files (.pdf ...)
|
||||
images: N files
|
||||
video: N files (.mp4 .mp3 ...)
|
||||
```
|
||||
|
||||
Omit any category with 0 files from the summary.
|
||||
|
||||
Then act on it:
|
||||
- If `total_files` is 0: stop with "No supported files found in [path]."
|
||||
- If `skipped_sensitive` is non-empty: mention file count skipped, not the file names.
|
||||
- If `total_words` > 2,000,000 OR `total_files` > 200: show the warning and the top 5 subdirectories by file count, then ask which subfolder to run on. Wait for the user's answer before proceeding.
|
||||
- Otherwise: proceed directly to Step 3 - no need to ask anything.
|
||||
- Otherwise: proceed directly to Step 2.5 if video files were detected, or Step 3 if not.
|
||||
|
||||
### Step 2.5 - Transcribe video / audio files (only if video files detected)
|
||||
|
||||
Skip this step entirely if `detect` returned zero `video` files.
|
||||
|
||||
Video and audio files cannot be read directly. Transcribe them to text first, then treat the transcripts as doc files in Step 3.
|
||||
|
||||
**Strategy:** Run non-video semantic extraction first (Step 3B) to get god nodes, use those to build a domain hint for Whisper, then transcribe. This keeps the prompt relevant without guessing the corpus topic from filenames.
|
||||
|
||||
**However**, if the corpus has *only* video files and no other docs/code, skip the god-node step and transcribe with the generic fallback prompt immediately.
|
||||
|
||||
**Transcription command:**
|
||||
|
||||
```bash
|
||||
$(cat graphify-out/.graphify_python) -c "
|
||||
import json
|
||||
from pathlib import Path
|
||||
from graphify.transcribe import build_whisper_prompt, transcribe_all
|
||||
|
||||
detect = json.loads(Path('graphify-out/.graphify_detect.json').read_text())
|
||||
video_files = detect.get('files', {}).get('video', [])
|
||||
|
||||
# Try to load god nodes from a previous partial run or pass [] if not yet available
|
||||
try:
|
||||
analysis = json.loads(Path('graphify-out/.graphify_analysis.json').read_text())
|
||||
god_nodes = analysis.get('god_nodes', [])
|
||||
except Exception:
|
||||
god_nodes = []
|
||||
|
||||
prompt = build_whisper_prompt(god_nodes)
|
||||
print(f'Whisper prompt: {prompt}')
|
||||
|
||||
transcript_paths = transcribe_all(video_files, initial_prompt=prompt)
|
||||
print(json.dumps(transcript_paths))
|
||||
" > graphify-out/.graphify_transcripts.json
|
||||
```
|
||||
|
||||
After transcription:
|
||||
- Read the transcript paths from `graphify-out/.graphify_transcripts.json`
|
||||
- Add them to the docs list before dispatching semantic subagents in Step 3B
|
||||
- Print how many transcripts were created: `Transcribed N video file(s) -> treating as docs`
|
||||
- If transcription fails for a file, print a warning and continue with the rest
|
||||
|
||||
**Whisper model:** Default is `base`. If the user passed `--whisper-model <name>`, set `GRAPHIFY_WHISPER_MODEL=<name>` in the environment before running the command above.
|
||||
|
||||
### Step 3 - Extract entities and relationships
|
||||
|
||||
|
||||
@@ -98,13 +98,60 @@ Corpus: X files · ~Y words
|
||||
docs: N files (.md .txt ...)
|
||||
papers: N files (.pdf ...)
|
||||
images: N files
|
||||
video: N files (.mp4 .mp3 ...)
|
||||
```
|
||||
|
||||
Omit any category with 0 files from the summary.
|
||||
|
||||
Then act on it:
|
||||
- If `total_files` is 0: stop with "No supported files found in [path]."
|
||||
- If `skipped_sensitive` is non-empty: mention file count skipped, not the file names.
|
||||
- If `total_words` > 2,000,000 OR `total_files` > 200: show the warning and the top 5 subdirectories by file count, then ask which subfolder to run on. Wait for the user's answer before proceeding.
|
||||
- Otherwise: proceed directly to Step 3 - no need to ask anything.
|
||||
- Otherwise: proceed directly to Step 2.5 if video files were detected, or Step 3 if not.
|
||||
|
||||
### Step 2.5 - Transcribe video / audio files (only if video files detected)
|
||||
|
||||
Skip this step entirely if `detect` returned zero `video` files.
|
||||
|
||||
Video and audio files cannot be read directly. Transcribe them to text first, then treat the transcripts as doc files in Step 3.
|
||||
|
||||
**Strategy:** Run non-video semantic extraction first (Step 3B) to get god nodes, use those to build a domain hint for Whisper, then transcribe. This keeps the prompt relevant without guessing the corpus topic from filenames.
|
||||
|
||||
**However**, if the corpus has *only* video files and no other docs/code, skip the god-node step and transcribe with the generic fallback prompt immediately.
|
||||
|
||||
**Transcription command:**
|
||||
|
||||
```bash
|
||||
$(cat graphify-out/.graphify_python) -c "
|
||||
import json
|
||||
from pathlib import Path
|
||||
from graphify.transcribe import build_whisper_prompt, transcribe_all
|
||||
|
||||
detect = json.loads(Path('graphify-out/.graphify_detect.json').read_text())
|
||||
video_files = detect.get('files', {}).get('video', [])
|
||||
|
||||
# Try to load god nodes from a previous partial run or pass [] if not yet available
|
||||
try:
|
||||
analysis = json.loads(Path('graphify-out/.graphify_analysis.json').read_text())
|
||||
god_nodes = analysis.get('god_nodes', [])
|
||||
except Exception:
|
||||
god_nodes = []
|
||||
|
||||
prompt = build_whisper_prompt(god_nodes)
|
||||
print(f'Whisper prompt: {prompt}')
|
||||
|
||||
transcript_paths = transcribe_all(video_files, initial_prompt=prompt)
|
||||
print(json.dumps(transcript_paths))
|
||||
" > graphify-out/.graphify_transcripts.json
|
||||
```
|
||||
|
||||
After transcription:
|
||||
- Read the transcript paths from `graphify-out/.graphify_transcripts.json`
|
||||
- Add them to the docs list before dispatching semantic subagents in Step 3B
|
||||
- Print how many transcripts were created: `Transcribed N video file(s) -> treating as docs`
|
||||
- If transcription fails for a file, print a warning and continue with the rest
|
||||
|
||||
**Whisper model:** Default is `base`. If the user passed `--whisper-model <name>`, set `GRAPHIFY_WHISPER_MODEL=<name>` in the environment before running the command above.
|
||||
|
||||
### Step 3 - Extract entities and relationships
|
||||
|
||||
|
||||
+48
-1
@@ -97,13 +97,60 @@ Corpus: X files · ~Y words
|
||||
docs: N files (.md .txt ...)
|
||||
papers: N files (.pdf ...)
|
||||
images: N files
|
||||
video: N files (.mp4 .mp3 ...)
|
||||
```
|
||||
|
||||
Omit any category with 0 files from the summary.
|
||||
|
||||
Then act on it:
|
||||
- If `total_files` is 0: stop with "No supported files found in [path]."
|
||||
- If `skipped_sensitive` is non-empty: mention file count skipped, not the file names.
|
||||
- If `total_words` > 2,000,000 OR `total_files` > 200: show the warning and the top 5 subdirectories by file count, then ask which subfolder to run on. Wait for the user's answer before proceeding.
|
||||
- Otherwise: proceed directly to Step 3 - no need to ask anything.
|
||||
- Otherwise: proceed directly to Step 2.5 if video files were detected, or Step 3 if not.
|
||||
|
||||
### Step 2.5 - Transcribe video / audio files (only if video files detected)
|
||||
|
||||
Skip this step entirely if `detect` returned zero `video` files.
|
||||
|
||||
Video and audio files cannot be read directly. Transcribe them to text first, then treat the transcripts as doc files in Step 3.
|
||||
|
||||
**Strategy:** Run non-video semantic extraction first (Step 3B) to get god nodes, use those to build a domain hint for Whisper, then transcribe. This keeps the prompt relevant without guessing the corpus topic from filenames.
|
||||
|
||||
**However**, if the corpus has *only* video files and no other docs/code, skip the god-node step and transcribe with the generic fallback prompt immediately.
|
||||
|
||||
**Transcription command:**
|
||||
|
||||
```bash
|
||||
$(cat graphify-out/.graphify_python) -c "
|
||||
import json
|
||||
from pathlib import Path
|
||||
from graphify.transcribe import build_whisper_prompt, transcribe_all
|
||||
|
||||
detect = json.loads(Path('graphify-out/.graphify_detect.json').read_text())
|
||||
video_files = detect.get('files', {}).get('video', [])
|
||||
|
||||
# Try to load god nodes from a previous partial run or pass [] if not yet available
|
||||
try:
|
||||
analysis = json.loads(Path('graphify-out/.graphify_analysis.json').read_text())
|
||||
god_nodes = analysis.get('god_nodes', [])
|
||||
except Exception:
|
||||
god_nodes = []
|
||||
|
||||
prompt = build_whisper_prompt(god_nodes)
|
||||
print(f'Whisper prompt: {prompt}')
|
||||
|
||||
transcript_paths = transcribe_all(video_files, initial_prompt=prompt)
|
||||
print(json.dumps(transcript_paths))
|
||||
" > graphify-out/.graphify_transcripts.json
|
||||
```
|
||||
|
||||
After transcription:
|
||||
- Read the transcript paths from `graphify-out/.graphify_transcripts.json`
|
||||
- Add them to the docs list before dispatching semantic subagents in Step 3B
|
||||
- Print how many transcripts were created: `Transcribed N video file(s) -> treating as docs`
|
||||
- If transcription fails for a file, print a warning and continue with the rest
|
||||
|
||||
**Whisper model:** Default is `base`. If the user passed `--whisper-model <name>`, set `GRAPHIFY_WHISPER_MODEL=<name>` in the environment before running the command above.
|
||||
|
||||
### Step 3 - Extract entities and relationships
|
||||
|
||||
|
||||
@@ -90,13 +90,59 @@ Corpus: X files · ~Y words
|
||||
docs: N files (.md .txt ...)
|
||||
papers: N files (.pdf ...)
|
||||
images: N files
|
||||
video: N files (.mp4 .mp3 ...)
|
||||
```
|
||||
|
||||
Omit any category with 0 files from the summary.
|
||||
|
||||
Then act on it:
|
||||
- If `total_files` is 0: stop with "No supported files found in [path]."
|
||||
- If `skipped_sensitive` is non-empty: mention file count skipped, not the file names.
|
||||
- If `total_words` > 2,000,000 OR `total_files` > 200: show the warning and the top 5 subdirectories by file count, then ask which subfolder to run on. Wait for the user's answer before proceeding.
|
||||
- Otherwise: proceed directly to Step 3 - no need to ask anything.
|
||||
- Otherwise: proceed directly to Step 2.5 if video files were detected, or Step 3 if not.
|
||||
|
||||
### Step 2.5 - Transcribe video / audio files (only if video files detected)
|
||||
|
||||
Skip this step entirely if `detect` returned zero `video` files.
|
||||
|
||||
Video and audio files cannot be read directly. Transcribe them to text first, then treat the transcripts as doc files in Step 3.
|
||||
|
||||
**Strategy:** Run non-video semantic extraction first (Step 3B) to get god nodes, use those to build a domain hint for Whisper, then transcribe. This keeps the prompt relevant without guessing the corpus topic from filenames.
|
||||
|
||||
**However**, if the corpus has *only* video files and no other docs/code, skip the god-node step and transcribe with the generic fallback prompt immediately.
|
||||
|
||||
**Transcription command (PowerShell):**
|
||||
|
||||
```powershell
|
||||
& (Get-Content graphify-out\.graphify_python) -c "
|
||||
import json
|
||||
from pathlib import Path
|
||||
from graphify.transcribe import build_whisper_prompt, transcribe_all
|
||||
|
||||
detect = json.loads(Path('graphify-out/.graphify_detect.json').read_text())
|
||||
video_files = detect.get('files', {}).get('video', [])
|
||||
|
||||
try:
|
||||
analysis = json.loads(Path('graphify-out/.graphify_analysis.json').read_text())
|
||||
god_nodes = analysis.get('god_nodes', [])
|
||||
except Exception:
|
||||
god_nodes = []
|
||||
|
||||
prompt = build_whisper_prompt(god_nodes)
|
||||
print(f'Whisper prompt: {prompt}')
|
||||
|
||||
transcript_paths = transcribe_all(video_files, initial_prompt=prompt)
|
||||
print(json.dumps(transcript_paths))
|
||||
" | Out-File -FilePath graphify-out\.graphify_transcripts.json -Encoding utf8
|
||||
```
|
||||
|
||||
After transcription:
|
||||
- Read the transcript paths from `graphify-out\.graphify_transcripts.json`
|
||||
- Add them to the docs list before dispatching semantic subagents in Step 3B
|
||||
- Print how many transcripts were created: `Transcribed N video file(s) -> treating as docs`
|
||||
- If transcription fails for a file, print a warning and continue with the rest
|
||||
|
||||
**Whisper model:** Default is `base`. If the user passed `--whisper-model <name>`, set `$env:GRAPHIFY_WHISPER_MODEL = "<name>"` before running the command above.
|
||||
|
||||
### Step 3 - Extract entities and relationships
|
||||
|
||||
|
||||
+51
-2
@@ -16,6 +16,7 @@ Turn any folder of files into a navigable knowledge graph with community detecti
|
||||
/graphify <path> --mode deep # thorough extraction, richer INFERRED edges
|
||||
/graphify <path> --update # incremental - re-extract only new/changed files
|
||||
/graphify <path> --directed # build directed graph (preserves edge direction: source→target)
|
||||
/graphify <path> --whisper-model medium # use a larger Whisper model for better transcription accuracy
|
||||
/graphify <path> --cluster-only # rerun clustering on existing graph
|
||||
/graphify <path> --no-viz # skip visualization, just report + JSON
|
||||
/graphify <path> --html # (HTML is generated by default - this flag is a no-op)
|
||||
@@ -101,13 +102,60 @@ Corpus: X files · ~Y words
|
||||
docs: N files (.md .txt ...)
|
||||
papers: N files (.pdf ...)
|
||||
images: N files
|
||||
video: N files (.mp4 .mp3 ...)
|
||||
```
|
||||
|
||||
Omit any category with 0 files from the summary.
|
||||
|
||||
Then act on it:
|
||||
- If `total_files` is 0: stop with "No supported files found in [path]."
|
||||
- If `skipped_sensitive` is non-empty: mention file count skipped, not the file names.
|
||||
- If `total_words` > 2,000,000 OR `total_files` > 200: show the warning and the top 5 subdirectories by file count, then ask which subfolder to run on. Wait for the user's answer before proceeding.
|
||||
- Otherwise: proceed directly to Step 3 - no need to ask anything.
|
||||
- Otherwise: proceed directly to Step 2.5 if video files were detected, or Step 3 if not.
|
||||
|
||||
### Step 2.5 - Transcribe video / audio files (only if video files detected)
|
||||
|
||||
Skip this step entirely if `detect` returned zero `video` files.
|
||||
|
||||
Video and audio files cannot be read directly. Transcribe them to text first, then treat the transcripts as doc files in Step 3.
|
||||
|
||||
**Strategy:** Run non-video semantic extraction first (Step 3B) to get god nodes, use those to build a domain hint for Whisper, then transcribe. This keeps the prompt relevant without guessing the corpus topic from filenames.
|
||||
|
||||
**However**, if the corpus has *only* video files and no other docs/code, skip the god-node step and transcribe with the generic fallback prompt immediately.
|
||||
|
||||
**Transcription command:**
|
||||
|
||||
```bash
|
||||
$(cat graphify-out/.graphify_python) -c "
|
||||
import json
|
||||
from pathlib import Path
|
||||
from graphify.transcribe import build_whisper_prompt, transcribe_all
|
||||
|
||||
detect = json.loads(Path('graphify-out/.graphify_detect.json').read_text())
|
||||
video_files = detect.get('files', {}).get('video', [])
|
||||
|
||||
# Try to load god nodes from a previous partial run or pass [] if not yet available
|
||||
try:
|
||||
analysis = json.loads(Path('graphify-out/.graphify_analysis.json').read_text())
|
||||
god_nodes = analysis.get('god_nodes', [])
|
||||
except Exception:
|
||||
god_nodes = []
|
||||
|
||||
prompt = build_whisper_prompt(god_nodes)
|
||||
print(f'Whisper prompt: {prompt}')
|
||||
|
||||
transcript_paths = transcribe_all(video_files, initial_prompt=prompt)
|
||||
print(json.dumps(transcript_paths))
|
||||
" > graphify-out/.graphify_transcripts.json
|
||||
```
|
||||
|
||||
After transcription:
|
||||
- Read the transcript paths from `graphify-out/.graphify_transcripts.json`
|
||||
- Add them to the docs list before dispatching semantic subagents in Step 3B
|
||||
- Print how many transcripts were created: `Transcribed N video file(s) -> treating as docs`
|
||||
- If transcription fails for a file, print a warning and continue with the rest
|
||||
|
||||
**Whisper model:** Default is `base`. If the user passed `--whisper-model <name>`, set `GRAPHIFY_WHISPER_MODEL=<name>` in the environment before running the command above.
|
||||
|
||||
### Step 3 - Extract entities and relationships
|
||||
|
||||
@@ -1152,8 +1200,9 @@ except RuntimeError as e:
|
||||
Replace `URL` with the actual URL, `AUTHOR` with the user's name if provided, `CONTRIBUTOR` likewise. If the command exits with an error, tell the user what went wrong - do not silently continue. After a successful save, automatically run the `--update` pipeline on `./raw` to merge the new file into the existing graph.
|
||||
|
||||
Supported URL types (auto-detected):
|
||||
- YouTube / any video URL → audio downloaded via yt-dlp, transcribed to `.txt` on next run (requires `pip install 'graphifyy[video]'`)
|
||||
- Twitter/X → fetched via oEmbed, saved as `.md` with tweet text and author
|
||||
- arXiv → abstract + metadata saved as `.md`
|
||||
- arXiv → abstract + metadata saved as `.md`
|
||||
- PDF → downloaded as `.pdf`
|
||||
- Images (.png/.jpg/.webp) → downloaded, Claude vision extracts on next run
|
||||
- Any webpage → converted to markdown via html2text
|
||||
|
||||
@@ -0,0 +1,202 @@
|
||||
# Video transcription using faster-whisper
|
||||
# Converts video/audio files to text transcripts for graph extraction
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
from pathlib import Path
|
||||
|
||||
|
||||
VIDEO_EXTENSIONS = {'.mp4', '.mov', '.webm', '.mkv', '.avi', '.m4v', '.mp3', '.wav', '.m4a', '.ogg'}
|
||||
URL_PREFIXES = ('http://', 'https://', 'www.')
|
||||
|
||||
_DEFAULT_MODEL = "base"
|
||||
_TRANSCRIPTS_DIR = "graphify-out/transcripts"
|
||||
_FALLBACK_PROMPT = "Use proper punctuation and paragraph breaks."
|
||||
|
||||
|
||||
def _model_name() -> str:
|
||||
return os.environ.get("GRAPHIFY_WHISPER_MODEL", _DEFAULT_MODEL)
|
||||
|
||||
|
||||
def _get_whisper():
|
||||
try:
|
||||
from faster_whisper import WhisperModel
|
||||
return WhisperModel
|
||||
except ImportError as exc:
|
||||
raise ImportError(
|
||||
"Video transcription requires faster-whisper. "
|
||||
"Run: pip install 'graphifyy[video]'"
|
||||
) from exc
|
||||
|
||||
|
||||
def _get_yt_dlp():
|
||||
try:
|
||||
import yt_dlp
|
||||
return yt_dlp
|
||||
except ImportError as exc:
|
||||
raise ImportError(
|
||||
"YouTube/URL download requires yt-dlp. "
|
||||
"Run: pip install 'graphifyy[video]'"
|
||||
) from exc
|
||||
|
||||
|
||||
def is_url(path: str) -> bool:
|
||||
"""Return True if the string looks like a URL rather than a file path."""
|
||||
return any(path.startswith(p) for p in URL_PREFIXES)
|
||||
|
||||
|
||||
def download_audio(url: str, output_dir: Path) -> Path:
|
||||
"""Download audio-only stream from a URL using yt-dlp.
|
||||
|
||||
Returns the path to the downloaded audio file (.m4a or .opus).
|
||||
Uses cached file if already downloaded.
|
||||
"""
|
||||
yt_dlp = _get_yt_dlp()
|
||||
output_dir.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
# yt-dlp uses %(title)s which can be long/weird — use a stable name based on URL hash
|
||||
import hashlib
|
||||
url_hash = hashlib.sha1(url.encode()).hexdigest()[:12]
|
||||
out_template = str(output_dir / f"yt_{url_hash}.%(ext)s")
|
||||
|
||||
# Check for already-downloaded file
|
||||
for ext in ('.m4a', '.opus', '.mp3', '.ogg', '.wav', '.webm'):
|
||||
candidate = output_dir / f"yt_{url_hash}{ext}"
|
||||
if candidate.exists():
|
||||
print(f" cached audio: {candidate.name}")
|
||||
return candidate
|
||||
|
||||
ydl_opts = {
|
||||
'format': 'bestaudio[ext=m4a]/bestaudio/best',
|
||||
'outtmpl': out_template,
|
||||
'quiet': True,
|
||||
'no_warnings': True,
|
||||
'noplaylist': True,
|
||||
'postprocessors': [], # no ffmpeg needed — use native audio
|
||||
}
|
||||
|
||||
print(f" downloading audio: {url[:80]} ...", flush=True)
|
||||
with yt_dlp.YoutubeDL(ydl_opts) as ydl:
|
||||
info = ydl.extract_info(url, download=True)
|
||||
ext = info.get('ext', 'm4a')
|
||||
downloaded = output_dir / f"yt_{url_hash}.{ext}"
|
||||
if not downloaded.exists():
|
||||
# yt-dlp may have picked a different extension
|
||||
for p in output_dir.glob(f"yt_{url_hash}.*"):
|
||||
downloaded = p
|
||||
break
|
||||
return downloaded
|
||||
|
||||
|
||||
def build_whisper_prompt(god_nodes: list[dict]) -> str:
|
||||
"""Build a domain hint for Whisper from god nodes extracted from the corpus.
|
||||
|
||||
Takes the top god nodes (most connected concepts) already extracted from
|
||||
non-video files and asks the LLM to summarise them into a one-sentence
|
||||
speech-to-text hint. Falls back to a generic prompt if no nodes available.
|
||||
"""
|
||||
if not god_nodes:
|
||||
return _FALLBACK_PROMPT
|
||||
|
||||
# Use env override if set
|
||||
override = os.environ.get("GRAPHIFY_WHISPER_PROMPT")
|
||||
if override:
|
||||
return override
|
||||
|
||||
labels = [n.get("label", "") for n in god_nodes[:10] if n.get("label")]
|
||||
if not labels:
|
||||
return _FALLBACK_PROMPT
|
||||
|
||||
try:
|
||||
import anthropic
|
||||
client = anthropic.Anthropic()
|
||||
msg = client.messages.create(
|
||||
model="claude-haiku-4-5-20251001",
|
||||
max_tokens=60,
|
||||
messages=[{
|
||||
"role": "user",
|
||||
"content": (
|
||||
f"These are the key concepts from a document corpus: {', '.join(labels)}. "
|
||||
"Write a single short sentence (under 20 words) that describes the domain "
|
||||
"for a speech-to-text model. Start with 'Technical' or the domain name. "
|
||||
"No explanation, just the sentence."
|
||||
),
|
||||
}],
|
||||
)
|
||||
prompt = msg.content[0].text.strip().strip('"')
|
||||
return prompt + " Use proper punctuation and paragraph breaks."
|
||||
except Exception:
|
||||
# If LLM call fails for any reason, fall back gracefully
|
||||
topics = ", ".join(labels[:5])
|
||||
return f"Technical discussion about {topics}. Use proper punctuation and paragraph breaks."
|
||||
|
||||
|
||||
def transcribe(
|
||||
video_path: Path | str,
|
||||
output_dir: Path | None = None,
|
||||
initial_prompt: str | None = None,
|
||||
force: bool = False,
|
||||
) -> Path:
|
||||
"""Transcribe a video/audio file or URL to a .txt transcript.
|
||||
|
||||
If video_path is a URL, audio is downloaded first via yt-dlp.
|
||||
Returns the path to the saved transcript file.
|
||||
Uses cached transcript if it exists unless force=True.
|
||||
|
||||
initial_prompt: domain hint for Whisper (built from corpus god nodes).
|
||||
force: re-transcribe even if transcript already exists.
|
||||
"""
|
||||
out_dir = Path(output_dir) if output_dir else Path(_TRANSCRIPTS_DIR)
|
||||
out_dir.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
if is_url(str(video_path)):
|
||||
audio_path = download_audio(str(video_path), out_dir / "downloads")
|
||||
else:
|
||||
audio_path = Path(video_path)
|
||||
|
||||
transcript_path = out_dir / (audio_path.stem + ".txt")
|
||||
if transcript_path.exists() and not force:
|
||||
return transcript_path
|
||||
|
||||
WhisperModel = _get_whisper()
|
||||
model_name = _model_name()
|
||||
prompt = initial_prompt or _FALLBACK_PROMPT
|
||||
|
||||
print(f" transcribing {audio_path.name} (model={model_name}) ...", flush=True)
|
||||
model = WhisperModel(model_name, device="cpu", compute_type="int8")
|
||||
segments, info = model.transcribe(
|
||||
str(audio_path),
|
||||
beam_size=5,
|
||||
initial_prompt=prompt,
|
||||
)
|
||||
|
||||
lines = [segment.text.strip() for segment in segments if segment.text.strip()]
|
||||
transcript = "\n".join(lines)
|
||||
|
||||
transcript_path.write_text(transcript, encoding="utf-8")
|
||||
lang = info.language if hasattr(info, "language") else "unknown"
|
||||
print(f" transcript saved -> {transcript_path} (lang={lang}, {len(lines)} segments)")
|
||||
return transcript_path
|
||||
|
||||
|
||||
def transcribe_all(
|
||||
video_files: list[str],
|
||||
output_dir: Path | None = None,
|
||||
initial_prompt: str | None = None,
|
||||
) -> list[str]:
|
||||
"""Transcribe a list of video/audio files or URLs, return paths to transcript .txt files.
|
||||
|
||||
Already-transcribed files are returned from cache instantly.
|
||||
initial_prompt is shared across all files — built once from corpus god nodes.
|
||||
"""
|
||||
if not video_files:
|
||||
return []
|
||||
|
||||
transcript_paths = []
|
||||
for vf in video_files:
|
||||
try:
|
||||
t = transcribe(vf, output_dir, initial_prompt=initial_prompt)
|
||||
transcript_paths.append(str(t))
|
||||
except Exception as exc:
|
||||
print(f" warning: could not transcribe {vf}: {exc}")
|
||||
return transcript_paths
|
||||
+3
-2
@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
|
||||
|
||||
[project]
|
||||
name = "graphifyy"
|
||||
version = "0.3.28"
|
||||
version = "0.3.29"
|
||||
description = "AI coding assistant skill (Claude Code, Codex, OpenCode, Cursor, OpenClaw, Factory Droid, Trae) - turn any folder of code, docs, papers, or images into a queryable knowledge graph"
|
||||
readme = "README.md"
|
||||
license = { file = "LICENSE" }
|
||||
@@ -47,7 +47,8 @@ pdf = ["pypdf", "html2text"]
|
||||
watch = ["watchdog"]
|
||||
leiden = ["graspologic"]
|
||||
office = ["python-docx", "openpyxl"]
|
||||
all = ["mcp", "neo4j", "pypdf", "html2text", "watchdog", "graspologic", "python-docx", "openpyxl"]
|
||||
video = ["faster-whisper", "yt-dlp"]
|
||||
all = ["mcp", "neo4j", "pypdf", "html2text", "watchdog", "graspologic", "python-docx", "openpyxl", "faster-whisper", "yt-dlp"]
|
||||
|
||||
[project.scripts]
|
||||
graphify = "graphify.__main__:main"
|
||||
|
||||
@@ -199,3 +199,40 @@ def test_detect_handles_circular_symlinks(tmp_path):
|
||||
|
||||
result = detect(tmp_path, follow_symlinks=True)
|
||||
assert any("main.py" in f for f in result["files"]["code"])
|
||||
|
||||
|
||||
def test_classify_video_extensions():
|
||||
"""Video and audio file extensions should classify as VIDEO."""
|
||||
from graphify.detect import FileType
|
||||
assert classify_file(Path("lecture.mp4")) == FileType.VIDEO
|
||||
assert classify_file(Path("podcast.mp3")) == FileType.VIDEO
|
||||
assert classify_file(Path("talk.mov")) == FileType.VIDEO
|
||||
assert classify_file(Path("recording.wav")) == FileType.VIDEO
|
||||
assert classify_file(Path("webinar.webm")) == FileType.VIDEO
|
||||
assert classify_file(Path("audio.m4a")) == FileType.VIDEO
|
||||
|
||||
|
||||
def test_detect_includes_video_key(tmp_path):
|
||||
"""detect() result always includes a 'video' key even with no video files."""
|
||||
(tmp_path / "main.py").write_text("x = 1")
|
||||
result = detect(tmp_path)
|
||||
assert "video" in result["files"]
|
||||
|
||||
|
||||
def test_detect_finds_video_files(tmp_path):
|
||||
"""detect() correctly counts video files and does not add them to word count."""
|
||||
(tmp_path / "lecture.mp4").write_bytes(b"fake video data")
|
||||
(tmp_path / "notes.md").write_text("# Notes\nSome content here.")
|
||||
result = detect(tmp_path)
|
||||
assert len(result["files"]["video"]) == 1
|
||||
assert any("lecture.mp4" in f for f in result["files"]["video"])
|
||||
# total_words should not include video files (they have no readable text)
|
||||
assert result["total_words"] >= 0 # won't crash
|
||||
|
||||
|
||||
def test_detect_video_not_in_words(tmp_path):
|
||||
"""Video files do not contribute to total_words."""
|
||||
(tmp_path / "clip.mp4").write_bytes(b"\x00" * 100)
|
||||
result = detect(tmp_path)
|
||||
# Only video file present — total_words should be 0
|
||||
assert result["total_words"] == 0
|
||||
|
||||
@@ -0,0 +1,168 @@
|
||||
"""Tests for graphify.transcribe — video/audio transcription support."""
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import os
|
||||
from pathlib import Path
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
import pytest
|
||||
|
||||
from graphify.transcribe import (
|
||||
VIDEO_EXTENSIONS,
|
||||
build_whisper_prompt,
|
||||
transcribe,
|
||||
transcribe_all,
|
||||
)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# VIDEO_EXTENSIONS
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def test_video_extensions_set():
|
||||
assert ".mp4" in VIDEO_EXTENSIONS
|
||||
assert ".mp3" in VIDEO_EXTENSIONS
|
||||
assert ".wav" in VIDEO_EXTENSIONS
|
||||
assert ".mov" in VIDEO_EXTENSIONS
|
||||
assert ".py" not in VIDEO_EXTENSIONS
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# build_whisper_prompt
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def test_build_whisper_prompt_no_nodes():
|
||||
"""Empty god_nodes returns fallback prompt."""
|
||||
prompt = build_whisper_prompt([])
|
||||
assert "punctuation" in prompt.lower() or len(prompt) > 0
|
||||
|
||||
|
||||
def test_build_whisper_prompt_env_override(monkeypatch):
|
||||
"""GRAPHIFY_WHISPER_PROMPT env var short-circuits LLM call."""
|
||||
monkeypatch.setenv("GRAPHIFY_WHISPER_PROMPT", "Custom domain hint.")
|
||||
prompt = build_whisper_prompt([{"label": "Python"}, {"label": "FastAPI"}])
|
||||
assert prompt == "Custom domain hint."
|
||||
|
||||
|
||||
def test_build_whisper_prompt_llm_success():
|
||||
"""Successful LLM call returns generated prompt with punctuation suffix."""
|
||||
god_nodes = [{"label": "neural networks"}, {"label": "transformers"}, {"label": "attention"}]
|
||||
|
||||
fake_response = MagicMock()
|
||||
fake_response.content = [MagicMock(text="Machine learning and deep learning research")]
|
||||
|
||||
with patch.dict(os.environ, {}, clear=False):
|
||||
os.environ.pop("GRAPHIFY_WHISPER_PROMPT", None)
|
||||
with patch("anthropic.Anthropic") as MockClient:
|
||||
MockClient.return_value.messages.create.return_value = fake_response
|
||||
prompt = build_whisper_prompt(god_nodes)
|
||||
|
||||
assert "Machine learning" in prompt
|
||||
assert "punctuation" in prompt.lower()
|
||||
|
||||
|
||||
def test_build_whisper_prompt_llm_failure_fallback():
|
||||
"""If LLM call raises, falls back to topic-based prompt."""
|
||||
god_nodes = [{"label": "kubernetes"}, {"label": "docker"}, {"label": "helm"}]
|
||||
|
||||
with patch.dict(os.environ, {}, clear=False):
|
||||
os.environ.pop("GRAPHIFY_WHISPER_PROMPT", None)
|
||||
with patch("anthropic.Anthropic", side_effect=Exception("API error")):
|
||||
prompt = build_whisper_prompt(god_nodes)
|
||||
|
||||
assert "kubernetes" in prompt.lower() or "docker" in prompt.lower()
|
||||
assert "punctuation" in prompt.lower()
|
||||
|
||||
|
||||
def test_build_whisper_prompt_nodes_without_labels():
|
||||
"""Nodes missing 'label' keys are safely skipped."""
|
||||
god_nodes = [{"id": "1"}, {"id": "2", "label": ""}]
|
||||
prompt = build_whisper_prompt(god_nodes)
|
||||
assert len(prompt) > 0
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# transcribe
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def test_transcribe_uses_cache(tmp_path):
|
||||
"""If transcript already exists, transcribe() returns cached path without running Whisper."""
|
||||
video = tmp_path / "lecture.mp4"
|
||||
video.write_bytes(b"fake")
|
||||
out_dir = tmp_path / "transcripts"
|
||||
out_dir.mkdir()
|
||||
cached = out_dir / "lecture.txt"
|
||||
cached.write_text("Cached transcript content.")
|
||||
|
||||
result = transcribe(video, output_dir=out_dir)
|
||||
assert result == cached
|
||||
|
||||
|
||||
def test_transcribe_force_reruns(tmp_path):
|
||||
"""force=True re-transcribes even when cache exists."""
|
||||
video = tmp_path / "talk.mp4"
|
||||
video.write_bytes(b"fake")
|
||||
out_dir = tmp_path / "transcripts"
|
||||
out_dir.mkdir()
|
||||
(out_dir / "talk.txt").write_text("Old transcript.")
|
||||
|
||||
fake_segment = MagicMock()
|
||||
fake_segment.text = "New transcript segment."
|
||||
fake_info = MagicMock()
|
||||
fake_info.language = "en"
|
||||
|
||||
fake_model = MagicMock()
|
||||
fake_model.transcribe.return_value = ([fake_segment], fake_info)
|
||||
|
||||
with patch("graphify.transcribe._get_whisper", return_value=lambda *a, **kw: fake_model):
|
||||
result = transcribe(video, output_dir=out_dir, force=True)
|
||||
|
||||
assert result.read_text() == "New transcript segment."
|
||||
|
||||
|
||||
def test_transcribe_missing_faster_whisper(tmp_path):
|
||||
"""ImportError propagates when faster_whisper is not installed."""
|
||||
video = tmp_path / "clip.mp4"
|
||||
video.write_bytes(b"fake")
|
||||
|
||||
with patch("graphify.transcribe._get_whisper", side_effect=ImportError("faster-whisper not installed")):
|
||||
with pytest.raises(ImportError):
|
||||
transcribe(video, output_dir=tmp_path / "out")
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# transcribe_all
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def test_transcribe_all_empty():
|
||||
"""Empty input returns empty list without error."""
|
||||
assert transcribe_all([]) == []
|
||||
|
||||
|
||||
def test_transcribe_all_uses_cache(tmp_path):
|
||||
"""transcribe_all() returns cached paths for already-transcribed files."""
|
||||
video = tmp_path / "lecture.mp4"
|
||||
video.write_bytes(b"fake")
|
||||
out_dir = tmp_path / "transcripts"
|
||||
out_dir.mkdir()
|
||||
cached = out_dir / "lecture.txt"
|
||||
cached.write_text("Cached.")
|
||||
|
||||
results = transcribe_all([str(video)], output_dir=out_dir)
|
||||
assert len(results) == 1
|
||||
assert str(cached) in results[0]
|
||||
|
||||
|
||||
def test_transcribe_all_skips_failed(tmp_path):
|
||||
"""transcribe_all() warns and skips files that fail to transcribe."""
|
||||
video = tmp_path / "broken.mp4"
|
||||
video.write_bytes(b"fake")
|
||||
|
||||
def raise_import(*args, **kwargs):
|
||||
raise ImportError("faster_whisper not installed")
|
||||
|
||||
with patch("graphify.transcribe.transcribe", side_effect=RuntimeError("boom")):
|
||||
results = transcribe_all([str(video)], output_dir=tmp_path / "out")
|
||||
|
||||
assert results == []
|
||||
Reference in New Issue
Block a user