+1 -1

apps/import/routes.py

reviewed

··· 12 12 13 13 from convey import emit, state 14 14 from think.detect_created import detect_created 15 15 - from think.importer_utils import ( 15 15 + from think.importers.utils import ( 16 16 build_import_info, 17 17 get_import_details, 18 18 list_import_timestamps,

+1 -1

docs/CALLOSUM.md

reviewed

··· 97 97 **Event Log:** Observe, dream, and activity tract events with `day` + `segment` are logged to `<day>/<segment>/events.jsonl` by supervisor 98 98 99 99 ### `importer` - Media import processing 100 100 - **Source:** `think/importer.py` 100 100 + **Source:** `think/importers/cli.py` 101 101 **Events:** `started`, `status`, `completed`, `error` 102 102 **Key fields:** `import_id` (correlates all events), `stage`, `segments` (created segment keys), `stream` (stream name, e.g., `"import.apple"`) 103 103 **Stages:** `initialization`, `segmenting`, `transcribing`, `summarizing`

-1

observe/plaud.py think/importers/plaud.py

reviewed

··· 1 1 - #!/usr/bin/env python3 2 1 # SPDX-License-Identifier: AGPL-3.0-only 3 2 # Copyright (c) 2026 sol pbc 4 3

+4 -4

sol.py

reviewed

··· 6 6 Usage: 7 7 sol Show status and available commands 8 8 sol <command> [args] Run a subcommand 9 9 - sol <module> [args] Run by module path (e.g., sol think.importer) 9 9 + sol <module> [args] Run by module path (e.g., sol think.importers.cli) 10 10 11 11 Examples: 12 12 sol import data.json Import data into journal ··· 38 38 39 39 COMMANDS: dict[str, str] = { 40 40 # think package - daily processing and analysis 41 41 - "import": "think.importer", 41 41 + "import": "think.importers.cli", 42 42 "dream": "think.dream", 43 43 "planner": "think.planner", 44 44 "indexer": "think.indexer", ··· 194 194 print() 195 195 196 196 print("Direct module syntax: sol <module.path> [args]") 197 197 - print("Example: sol think.importer --help") 197 197 + print("Example: sol think.importers.cli --help") 198 198 199 199 200 200 def resolve_command(name: str) -> tuple[str, list[str]]: ··· 234 234 """Import and run a module's main() function. 235 235 236 236 Args: 237 237 - module_path: Dotted module path (e.g., "think.importer") 237 237 + module_path: Dotted module path (e.g., "think.importers.cli") 238 238 239 239 Returns: 240 240 Exit code (0 for success)

+18 -10

tests/test_activities.py

reviewed

··· 141 141 142 142 # Check browsing (predefined with instructions override) 143 143 browsing = next(a for a in loaded if a["id"] == "browsing") 144 144 - assert browsing["instructions"] == "Custom browsing instructions for this facet" 144 144 + assert ( 145 145 + browsing["instructions"] 146 146 + == "Custom browsing instructions for this facet" 147 147 + ) 145 148 146 149 # Check custom activity with instructions 147 150 custom = next(a for a in loaded if a["id"] == "custom_activity") ··· 207 210 for a in get_facet_activities("test_facet") 208 211 if a["id"] == "3d_modeling" 209 212 ) 210 210 - assert modeling["instructions"] == "Detect via: Blender, FreeCAD, OpenSCAD windows" 213 213 + assert ( 214 214 + modeling["instructions"] 215 215 + == "Detect via: Blender, FreeCAD, OpenSCAD windows" 216 216 + ) 211 217 212 218 # Remove it 213 219 removed = remove_activity_from_facet("test_facet", "meeting") ··· 256 262 instructions="Only detect scheduled meetings, not ad-hoc calls", 257 263 ) 258 264 assert updated is not None 259 259 - assert updated["instructions"] == "Only detect scheduled meetings, not ad-hoc calls" 265 265 + assert ( 266 266 + updated["instructions"] 267 267 + == "Only detect scheduled meetings, not ad-hoc calls" 268 268 + ) 260 269 # Other fields should be preserved 261 270 assert updated["priority"] == "low" 262 271 263 272 # Verify via lookup 264 273 activity = get_activity_by_id("test_facet", "meeting") 265 274 assert activity["priority"] == "low" 266 266 - assert activity["instructions"] == "Only detect scheduled meetings, not ad-hoc calls" 275 275 + assert ( 276 276 + activity["instructions"] 277 277 + == "Only detect scheduled meetings, not ad-hoc calls" 278 278 + ) 267 279 268 280 # Reset instructions to default via empty string 269 281 from think.activities import DEFAULT_ACTIVITIES ··· 271 283 default_instructions = next( 272 284 a["instructions"] for a in DEFAULT_ACTIVITIES if a["id"] == "meeting" 273 285 ) 274 274 - updated = update_activity_in_facet( 275 275 - "test_facet", "meeting", instructions="" 276 276 - ) 286 286 + updated = update_activity_in_facet("test_facet", "meeting", instructions="") 277 287 assert updated is not None 278 288 assert updated["instructions"] == default_instructions 279 289 ··· 281 291 default_desc = next( 282 292 a["description"] for a in DEFAULT_ACTIVITIES if a["id"] == "meeting" 283 293 ) 284 284 - updated = update_activity_in_facet( 285 285 - "test_facet", "meeting", description="" 286 286 - ) 294 294 + updated = update_activity_in_facet("test_facet", "meeting", description="") 287 295 assert updated is not None 288 296 assert updated["description"] == default_desc 289 297

+12 -11

tests/test_importer.py

reviewed

··· 13 13 14 14 def test_slice_audio_segment(tmp_path): 15 15 """Test slice_audio_segment extracts audio with stream copy.""" 16 16 - mod = importlib.import_module("think.importer") 16 16 + mod = importlib.import_module("think.importers.audio") 17 17 18 18 source = tmp_path / "source.mp3" 19 19 source.write_bytes(b"fake audio") ··· 34 34 35 35 def test_slice_audio_segment_fallback(tmp_path): 36 36 """Test slice_audio_segment falls back to re-encode on copy failure.""" 37 37 - mod = importlib.import_module("think.importer") 37 37 + mod = importlib.import_module("think.importers.audio") 38 38 39 39 source = tmp_path / "source.mp3" 40 40 source.write_bytes(b"fake audio") ··· 60 60 61 61 def test_importer_text(tmp_path, monkeypatch): 62 62 """Test importing a text transcript file.""" 63 63 - mod = importlib.import_module("think.importer") 63 63 + mod = importlib.import_module("think.importers.cli") 64 64 + text_mod = importlib.import_module("think.importers.text") 64 65 65 66 transcript = "hello\nworld" 66 67 txt = tmp_path / "sample.txt" ··· 75 76 def mock_detect_segment(text, start_time): 76 77 return [("12:00:00", "seg1"), ("12:05:00", "seg2")] 77 78 78 78 - monkeypatch.setattr(mod, "detect_transcript_segment", mock_detect_segment) 79 79 + monkeypatch.setattr(text_mod, "detect_transcript_segment", mock_detect_segment) 79 80 80 81 # Mock JSON conversion: returns entries with absolute timestamps 81 82 def mock_detect_json(text, segment_start): 82 83 return [{"start": segment_start, "speaker": "Unknown", "text": text}] 83 84 84 84 - monkeypatch.setattr(mod, "detect_transcript_json", mock_detect_json) 85 85 + monkeypatch.setattr(text_mod, "detect_transcript_json", mock_detect_json) 85 86 86 87 # Mock CallosumConnection and status emitter to avoid real sockets/threads 87 88 monkeypatch.setattr(mod, "CallosumConnection", lambda **kwargs: MagicMock()) ··· 125 126 126 127 def test_get_audio_duration(tmp_path): 127 128 """Test _get_audio_duration calls ffprobe correctly.""" 128 128 - mod = importlib.import_module("think.importer") 129 129 + mod = importlib.import_module("think.importers.audio") 129 130 130 131 audio_file = tmp_path / "test.mp3" 131 132 audio_file.write_bytes(b"fake audio") ··· 146 147 147 148 def test_get_audio_duration_failure(tmp_path): 148 149 """Test _get_audio_duration returns None on error.""" 149 149 - mod = importlib.import_module("think.importer") 150 150 + mod = importlib.import_module("think.importers.audio") 150 151 151 152 audio_file = tmp_path / "test.mp3" 152 153 audio_file.write_bytes(b"fake audio") ··· 160 161 161 162 def test_prepare_audio_segments(tmp_path, monkeypatch): 162 163 """Test prepare_audio_segments creates segment directories with audio slices.""" 163 163 - mod = importlib.import_module("think.importer") 164 164 + mod = importlib.import_module("think.importers.audio") 164 165 165 166 monkeypatch.setenv("JOURNAL_PATH", str(tmp_path)) 166 167 ··· 213 214 214 215 def test_prepare_audio_segments_with_collision(tmp_path, monkeypatch): 215 216 """Test prepare_audio_segments handles segment key collisions.""" 216 216 - mod = importlib.import_module("think.importer") 217 217 + mod = importlib.import_module("think.importers.audio") 217 218 218 219 monkeypatch.setenv("JOURNAL_PATH", str(tmp_path)) 219 220 ··· 258 259 259 260 def test_run_import_summary(tmp_path, monkeypatch): 260 261 """Test _run_import_summary calls cortex_request correctly.""" 261 261 - mod = importlib.import_module("think.importer") 262 262 + mod = importlib.import_module("think.importers.shared") 262 263 263 264 import_dir = tmp_path / "imports" / "20240101_120000" 264 265 import_dir.mkdir(parents=True) ··· 300 301 301 302 def test_run_import_summary_no_segments(tmp_path): 302 303 """Test _run_import_summary returns False with no segments.""" 303 303 - mod = importlib.import_module("think.importer") 304 304 + mod = importlib.import_module("think.importers.shared") 304 305 305 306 import_dir = tmp_path / "imports" / "20240101_120000" 306 307 import_dir.mkdir(parents=True)

+2 -2

tests/test_importer_jsonl.py

reviewed

··· 1 1 # SPDX-License-Identifier: AGPL-3.0-only 2 2 # Copyright (c) 2026 sol pbc 3 3 4 4 - """Tests for think.importer JSONL format writing.""" 4 4 + """Tests for think.importers.shared JSONL format writing.""" 5 5 6 6 import json 7 7 import tempfile 8 8 from pathlib import Path 9 9 10 10 - from think.importer import _write_import_jsonl 10 10 + from think.importers.shared import _write_import_jsonl 11 11 12 12 13 13 def test_write_import_jsonl_with_entries():

+2 -2

tests/test_importer_utils.py

reviewed

··· 1 1 # SPDX-License-Identifier: AGPL-3.0-only 2 2 # Copyright (c) 2026 sol pbc 3 3 4 4 - """Tests for think.importer_utils module.""" 4 4 + """Tests for think.importers.utils module.""" 5 5 6 6 import json 7 7 import tempfile ··· 9 9 10 10 import pytest 11 11 12 12 - from think.importer_utils import ( 12 12 + from think.importers.utils import ( 13 13 build_import_info, 14 14 calculate_duration_from_files, 15 15 get_import_details,

+3 -3

tests/test_sol.py

reviewed

··· 17 17 def test_resolve_known_command(self): 18 18 """Test resolving a known command from registry.""" 19 19 module_path, preset_args = sol.resolve_command("import") 20 20 - assert module_path == "think.importer" 20 20 + assert module_path == "think.importers.cli" 21 21 assert preset_args == [] 22 22 23 23 def test_resolve_direct_module_path(self): 24 24 """Test resolving a direct module path with dot.""" 25 25 - module_path, preset_args = sol.resolve_command("think.importer") 26 26 - assert module_path == "think.importer" 25 25 + module_path, preset_args = sol.resolve_command("think.importers.cli") 26 26 + assert module_path == "think.importers.cli" 27 27 assert preset_args == [] 28 28 29 29 def test_resolve_nested_module_path(self):

+2 -2

think/activities.py

reviewed

··· 335 335 entry["description"] = activity["description"] 336 336 337 337 # Store instructions only if different from default 338 338 - if activity.get("instructions") and activity[ 338 338 + if activity.get("instructions") and activity["instructions"] != default.get( 339 339 "instructions" 340 340 - ] != default.get("instructions"): 340 340 + ): 341 341 entry["instructions"] = activity["instructions"] 342 342 343 343 # Store priority if set

-1063

think/importer.py

reviewed

··· 1 1 - # SPDX-License-Identifier: AGPL-3.0-only 2 2 - # Copyright (c) 2026 sol pbc 3 3 - 4 4 - import argparse 5 5 - import datetime as dt 6 6 - import json 7 7 - import logging 8 8 - import os 9 9 - import queue 10 10 - import re 11 11 - import shutil 12 12 - import subprocess 13 13 - import threading 14 14 - import time 15 15 - from datetime import timedelta 16 16 - from pathlib import Path 17 17 - 18 18 - from observe.utils import find_available_segment 19 19 - from think.callosum import CallosumConnection 20 20 - from think.detect_created import detect_created 21 21 - from think.detect_transcript import detect_transcript_json, detect_transcript_segment 22 22 - from think.importer_utils import ( 23 23 - save_import_file, 24 24 - save_import_segments, 25 25 - write_import_metadata, 26 26 - ) 27 27 - from think.streams import stream_name, update_stream, write_segment_stream 28 28 - from think.utils import ( 29 29 - day_path, 30 30 - get_journal, 31 31 - get_rev, 32 32 - now_ms, 33 33 - segment_key, 34 34 - setup_cli, 35 35 - ) 36 36 - 37 37 - try: 38 38 - from pypdf import PdfReader 39 39 - except Exception: # pragma: no cover - optional dependency 40 40 - PdfReader = None 41 41 - 42 42 - logger = logging.getLogger(__name__) 43 43 - 44 44 - TIME_RE = re.compile(r"\d{8}_\d{6}") 45 45 - 46 46 - # Importer tract state 47 47 - _callosum: CallosumConnection | None = None 48 48 - _message_queue: queue.Queue | None = None 49 49 - _import_id: str | None = None 50 50 - _current_stage: str = "initialization" 51 51 - _start_time: float = 0.0 52 52 - _stage_start_time: float = 0.0 53 53 - _stages_run: list[str] = [] 54 54 - _status_thread: threading.Thread | None = None 55 55 - _status_running: bool = False 56 56 - 57 57 - 58 58 - def _get_relative_path(path: str) -> str: 59 59 - """Get path relative to journal, or return as-is if not under journal.""" 60 60 - journal_path = get_journal() 61 61 - try: 62 62 - return os.path.relpath(path, journal_path) 63 63 - except ValueError: 64 64 - return path 65 65 - 66 66 - 67 67 - def _set_stage(stage: str) -> None: 68 68 - """Update current stage and track timing.""" 69 69 - global _current_stage, _stage_start_time 70 70 - _current_stage = stage 71 71 - _stage_start_time = time.monotonic() 72 72 - if stage not in _stages_run: 73 73 - _stages_run.append(stage) 74 74 - logger.debug(f"Stage changed to: {stage}") 75 75 - 76 76 - 77 77 - def _status_emitter() -> None: 78 78 - """Background thread that emits status events every 5 seconds.""" 79 79 - while _status_running: 80 80 - if _callosum and _import_id: 81 81 - elapsed_ms = int((time.monotonic() - _start_time) * 1000) 82 82 - stage_elapsed_ms = int((time.monotonic() - _stage_start_time) * 1000) 83 83 - _callosum.emit( 84 84 - "importer", 85 85 - "status", 86 86 - import_id=_import_id, 87 87 - stage=_current_stage, 88 88 - elapsed_ms=elapsed_ms, 89 89 - stage_elapsed_ms=stage_elapsed_ms, 90 90 - ) 91 91 - time.sleep(5) 92 92 - 93 93 - 94 94 - def _write_import_jsonl( 95 95 - file_path: str, 96 96 - entries: list[dict], 97 97 - *, 98 98 - import_id: str, 99 99 - raw_filename: str | None = None, 100 100 - facet: str | None = None, 101 101 - setting: str | None = None, 102 102 - ) -> None: 103 103 - """Write imported transcript entries in JSONL format. 104 104 - 105 105 - First line contains imported metadata, subsequent lines contain entries. 106 106 - Each entry gets source="import" added to match the imported_audio.jsonl convention. 107 107 - 108 108 - Args: 109 109 - file_path: Path to write JSONL file 110 110 - entries: List of transcript entries 111 111 - import_id: Import identifier 112 112 - raw_filename: Source file name (relative path from segment to imports/) 113 113 - facet: Optional facet name 114 114 - setting: Optional setting description 115 115 - """ 116 116 - imported_meta: dict[str, str] = {"id": import_id} 117 117 - if facet: 118 118 - imported_meta["facet"] = facet 119 119 - if setting: 120 120 - imported_meta["setting"] = setting 121 121 - 122 122 - # Build top-level metadata with imported info 123 123 - metadata: dict[str, object] = {"imported": imported_meta} 124 124 - 125 125 - # Add raw file reference (path relative from segment to imports directory) 126 126 - if raw_filename: 127 127 - metadata["raw"] = f"../../imports/{import_id}/{raw_filename}" 128 128 - 129 129 - # Write JSONL: metadata first, then entries with source field 130 130 - jsonl_lines = [json.dumps(metadata)] 131 131 - for entry in entries: 132 132 - # Add source field if not already present (skip metadata entries like topics/setting) 133 133 - if "text" in entry and "source" not in entry: 134 134 - entry = {**entry, "source": "import"} 135 135 - jsonl_lines.append(json.dumps(entry)) 136 136 - 137 137 - with open(file_path, "w", encoding="utf-8") as f: 138 138 - f.write("\n".join(jsonl_lines) + "\n") 139 139 - 140 140 - 141 141 - def str2bool(value: str) -> bool: 142 142 - if isinstance(value, bool): 143 143 - return value 144 144 - val = value.lower() 145 145 - if val in {"y", "yes", "true", "t", "1"}: 146 146 - return True 147 147 - if val in {"n", "no", "false", "f", "0"}: 148 148 - return False 149 149 - raise argparse.ArgumentTypeError("boolean value expected") 150 150 - 151 151 - 152 152 - def slice_audio_segment( 153 153 - source_path: str, 154 154 - output_path: str, 155 155 - start_seconds: float, 156 156 - duration_seconds: float, 157 157 - ) -> str: 158 158 - """Extract an audio segment from source file, preserving original format. 159 159 - 160 160 - Uses stream copy for lossless extraction when possible. 161 161 - 162 162 - Args: 163 163 - source_path: Path to source audio file 164 164 - output_path: Path for output segment file 165 165 - start_seconds: Start offset in seconds 166 166 - duration_seconds: Duration to extract in seconds 167 167 - 168 168 - Returns: 169 169 - Output path on success 170 170 - 171 171 - Raises: 172 172 - subprocess.CalledProcessError: If ffmpeg fails 173 173 - """ 174 174 - cmd = [ 175 175 - "ffmpeg", 176 176 - "-ss", 177 177 - str(start_seconds), 178 178 - "-i", 179 179 - source_path, 180 180 - "-t", 181 181 - str(duration_seconds), 182 182 - "-vn", # No video 183 183 - "-c:a", 184 184 - "copy", # Stream copy for lossless extraction 185 185 - "-y", # Overwrite output 186 186 - output_path, 187 187 - ] 188 188 - try: 189 189 - subprocess.run(cmd, check=True, capture_output=True, text=True) 190 190 - except subprocess.CalledProcessError: 191 191 - # Fallback: re-encode if stream copy fails (some formats don't support it) 192 192 - logger.debug(f"Stream copy failed, re-encoding: {output_path}") 193 193 - cmd_reencode = [ 194 194 - "ffmpeg", 195 195 - "-ss", 196 196 - str(start_seconds), 197 197 - "-i", 198 198 - source_path, 199 199 - "-t", 200 200 - str(duration_seconds), 201 201 - "-vn", 202 202 - "-y", 203 203 - output_path, 204 204 - ] 205 205 - subprocess.run(cmd_reencode, check=True, capture_output=True, text=True) 206 206 - 207 207 - logger.info(f"Created audio segment: {output_path}") 208 208 - return output_path 209 209 - 210 210 - 211 211 - def _get_audio_duration(audio_path: str) -> float | None: 212 212 - """Get audio duration in seconds using ffprobe. 213 213 - 214 214 - Args: 215 215 - audio_path: Path to audio file 216 216 - 217 217 - Returns: 218 218 - Duration in seconds, or None if unable to determine 219 219 - """ 220 220 - try: 221 221 - cmd = [ 222 222 - "ffprobe", 223 223 - "-v", 224 224 - "error", 225 225 - "-show_entries", 226 226 - "format=duration", 227 227 - "-of", 228 228 - "default=noprint_wrappers=1:nokey=1", 229 229 - audio_path, 230 230 - ] 231 231 - result = subprocess.run(cmd, capture_output=True, text=True, check=True) 232 232 - return float(result.stdout.strip()) 233 233 - except (subprocess.CalledProcessError, ValueError) as e: 234 234 - logger.warning(f"Could not determine audio duration: {e}") 235 235 - return None 236 236 - 237 237 - 238 238 - def prepare_audio_segments( 239 239 - media_path: str, 240 240 - day_dir: str, 241 241 - base_dt: dt.datetime, 242 242 - import_id: str, 243 243 - stream: str, 244 244 - ) -> list[tuple[str, Path, list[str]]]: 245 245 - """Slice audio into 5-minute segments for observe pipeline. 246 246 - 247 247 - Creates segment directories with audio slices, ready for transcription 248 248 - via observe.observing events. 249 249 - 250 250 - Args: 251 251 - media_path: Path to source audio file 252 252 - day_dir: Day directory path (YYYYMMDD) 253 253 - base_dt: Base datetime for timestamp calculation 254 254 - import_id: Import identifier 255 255 - stream: Stream name for directory layout (day/stream/segment/) 256 256 - 257 257 - Returns: 258 258 - List of (segment_key, segment_dir, files_list) tuples 259 259 - where files_list contains the audio filename(s) created 260 260 - """ 261 261 - media = Path(media_path) 262 262 - source_ext = media.suffix.lower() 263 263 - stream_dir = Path(day_dir) / stream 264 264 - 265 265 - # Get audio duration to calculate number of segments 266 266 - duration = _get_audio_duration(media_path) 267 267 - if duration is None: 268 268 - raise RuntimeError(f"Could not determine duration of {media_path}") 269 269 - 270 270 - # Calculate number of 5-minute segments (ceiling division) 271 271 - segment_duration = 300 # 5 minutes 272 272 - num_segments = int((duration + segment_duration - 1) // segment_duration) 273 273 - if num_segments == 0: 274 274 - num_segments = 1 # At least one segment for very short audio 275 275 - 276 276 - segments: list[tuple[str, Path, list[str]]] = [] 277 277 - 278 278 - for chunk_index in range(num_segments): 279 279 - # Calculate timestamp for this segment 280 280 - ts = base_dt + timedelta(minutes=chunk_index * 5) 281 281 - time_part = ts.strftime("%H%M%S") 282 282 - 283 283 - # Create segment key with 5-minute duration 284 284 - segment_key_candidate = f"{time_part}_{segment_duration}" 285 285 - 286 286 - # Check for collision and deconflict if needed 287 287 - available_key = find_available_segment(stream_dir, segment_key_candidate) 288 288 - if available_key is None: 289 289 - logger.warning( 290 290 - f"Could not find available segment key near {segment_key_candidate}" 291 291 - ) 292 292 - continue 293 293 - 294 294 - if available_key != segment_key_candidate: 295 295 - logger.info( 296 296 - f"Segment collision: {segment_key_candidate} -> {available_key}" 297 297 - ) 298 298 - 299 299 - # Create segment directory under stream 300 300 - segment_dir = stream_dir / available_key 301 301 - segment_dir.mkdir(parents=True, exist_ok=True) 302 302 - 303 303 - # Slice audio for this segment 304 304 - audio_filename = f"imported_audio{source_ext}" 305 305 - audio_path = segment_dir / audio_filename 306 306 - start_seconds = chunk_index * segment_duration 307 307 - 308 308 - # For the last segment, use remaining duration 309 309 - if chunk_index == num_segments - 1: 310 310 - chunk_duration = duration - start_seconds 311 311 - else: 312 312 - chunk_duration = segment_duration 313 313 - 314 314 - try: 315 315 - slice_audio_segment( 316 316 - media_path, 317 317 - str(audio_path), 318 318 - start_seconds, 319 319 - chunk_duration, 320 320 - ) 321 321 - segments.append((available_key, segment_dir, [audio_filename])) 322 322 - logger.info(f"Created segment: {available_key} with {audio_filename}") 323 323 - except subprocess.CalledProcessError as e: 324 324 - logger.warning(f"Failed to slice segment {available_key}: {e}") 325 325 - # Clean up empty directory 326 326 - if segment_dir.exists() and not any(segment_dir.iterdir()): 327 327 - segment_dir.rmdir() 328 328 - 329 329 - return segments 330 330 - 331 331 - 332 332 - def _read_transcript(path: str) -> str: 333 333 - """Return transcript text from a .txt/.md/.pdf file.""" 334 334 - ext = os.path.splitext(path)[1].lower() 335 335 - if ext in {".txt", ".md"}: 336 336 - with open(path, "r", encoding="utf-8") as f: 337 337 - return f.read() 338 338 - if ext == ".pdf": 339 339 - if PdfReader is None: 340 340 - raise RuntimeError("pypdf required for PDF support") 341 341 - reader = PdfReader(path) 342 342 - parts = [] 343 343 - for page in reader.pages: 344 344 - text = page.extract_text() or "" 345 345 - parts.append(text) 346 346 - return "\n".join(parts) 347 347 - raise ValueError("unsupported transcript format") 348 348 - 349 349 - 350 350 - def _time_to_seconds(time_str: str) -> int: 351 351 - """Convert HH:MM:SS time string to seconds from midnight.""" 352 352 - h, m, s = map(int, time_str.split(":")) 353 353 - return h * 3600 + m * 60 + s 354 354 - 355 355 - 356 356 - def process_transcript( 357 357 - path: str, 358 358 - day_dir: str, 359 359 - base_dt: dt.datetime, 360 360 - *, 361 361 - import_id: str, 362 362 - stream: str, 363 363 - facet: str | None = None, 364 364 - setting: str | None = None, 365 365 - audio_duration: int | None = None, 366 366 - ) -> list[str]: 367 367 - """Process a transcript file and write imported JSONL segments. 368 368 - 369 369 - Args: 370 370 - path: Path to transcript file 371 371 - day_dir: Journal day directory 372 372 - base_dt: Base datetime for the import 373 373 - import_id: Import identifier 374 374 - stream: Stream name for directory layout (day/stream/segment/) 375 375 - facet: Optional facet name 376 376 - setting: Optional setting description 377 377 - audio_duration: Optional total audio duration in seconds (for last segment) 378 378 - 379 379 - Returns: 380 380 - List of created file paths. 381 381 - """ 382 382 - created_files = [] 383 383 - text = _read_transcript(path) 384 384 - stream_dir = os.path.join(day_dir, stream) 385 385 - 386 386 - # Get start time from base_dt for segmentation 387 387 - start_time = base_dt.strftime("%H:%M:%S") 388 388 - 389 389 - # Get segments with their absolute start times 390 390 - segments = detect_transcript_segment(text, start_time) 391 391 - 392 392 - for idx, (start_at, seg_text) in enumerate(segments): 393 393 - # Convert segment text to structured JSON with absolute timestamps 394 394 - json_data = detect_transcript_json(seg_text, start_at) 395 395 - if not json_data: 396 396 - continue 397 397 - 398 398 - # Parse absolute time for segment directory name 399 399 - time_part = start_at.replace(":", "") # "12:05:30" -> "120530" 400 400 - 401 401 - # Compute segment duration from absolute times 402 402 - start_seconds = _time_to_seconds(start_at) 403 403 - if idx + 1 < len(segments): 404 404 - next_start_at, _ = segments[idx + 1] 405 405 - next_seconds = _time_to_seconds(next_start_at) 406 406 - duration = next_seconds - start_seconds 407 407 - else: 408 408 - # Last segment: use remaining audio duration or default +5s 409 409 - if audio_duration: 410 410 - # audio_duration is total length, start_seconds is time-of-day 411 411 - # Need to calculate offset from recording start 412 412 - recording_start_seconds = _time_to_seconds(start_time) 413 413 - segment_offset = start_seconds - recording_start_seconds 414 414 - duration = audio_duration - segment_offset 415 415 - else: 416 416 - duration = 5 417 417 - 418 418 - # Negative duration indicates corrupted/invalid timestamp data 419 419 - if duration < 0: 420 420 - raise ValueError( 421 421 - f"Invalid segment duration: {duration}s for segment at {time_part}. " 422 422 - "Timestamps may be out of order or audio_duration is incorrect." 423 423 - ) 424 424 - 425 425 - # Ensure minimum duration of 1 second 426 426 - duration = max(1, duration) 427 427 - 428 428 - segment_name = f"{time_part}_{duration}" 429 429 - ts_dir = os.path.join(stream_dir, segment_name) 430 430 - os.makedirs(ts_dir, exist_ok=True) 431 431 - json_path = os.path.join(ts_dir, "imported_audio.jsonl") 432 432 - 433 433 - _write_import_jsonl( 434 434 - json_path, 435 435 - json_data, 436 436 - import_id=import_id, 437 437 - raw_filename=os.path.basename(path), 438 438 - facet=facet, 439 439 - setting=setting, 440 440 - ) 441 441 - logger.info(f"Added transcript segment to journal: {json_path}") 442 442 - created_files.append(json_path) 443 443 - 444 444 - return created_files 445 445 - 446 446 - 447 447 - def _run_import_summary( 448 448 - import_dir: Path, 449 449 - day: str, 450 450 - segments: list[str], 451 451 - ) -> bool: 452 452 - """Create a summary for imported segments using cortex generator. 453 453 - 454 454 - Args: 455 455 - import_dir: Directory where the summary will be saved 456 456 - day: Day string (YYYYMMDD format) 457 457 - segments: List of segment keys to summarize 458 458 - 459 459 - Returns: 460 460 - True if summary was created successfully, False otherwise 461 461 - """ 462 462 - from think.cortex_client import cortex_request, wait_for_agents 463 463 - 464 464 - if not segments: 465 465 - logger.info("No segments to summarize") 466 466 - return False 467 467 - 468 468 - summary_path = import_dir / "summary.md" 469 469 - 470 470 - try: 471 471 - logger.info(f"Creating summary for {len(segments)} segments via cortex") 472 472 - 473 473 - # Spawn generator via cortex 474 474 - agent_id = cortex_request( 475 475 - prompt="", # Generators don't use prompt 476 476 - name="importer", 477 477 - config={ 478 478 - "day": day, 479 479 - "span": segments, 480 480 - "output": "md", 481 481 - "output_path": str(summary_path), 482 482 - }, 483 483 - ) 484 484 - 485 485 - if agent_id is None: 486 486 - logger.error("Failed to send cortex request for import summary") 487 487 - return False 488 488 - 489 489 - # Wait for completion 490 490 - completed, timed_out = wait_for_agents([agent_id], timeout=300) 491 491 - 492 492 - if timed_out: 493 493 - logger.error(f"Import summary timed out (ID: {agent_id})") 494 494 - return False 495 495 - 496 496 - if completed: 497 497 - end_state = completed.get(agent_id, "unknown") 498 498 - if end_state == "finish" and summary_path.exists(): 499 499 - logger.info(f"Created import summary: {summary_path}") 500 500 - return True 501 501 - else: 502 502 - logger.warning( 503 503 - f"Generator ended with state {end_state}, " 504 504 - f"summary exists: {summary_path.exists()}" 505 505 - ) 506 506 - return False 507 507 - 508 508 - logger.error("Generator did not complete") 509 509 - return False 510 510 - 511 511 - except Exception as e: 512 512 - logger.error(f"Failed to create summary: {e}") 513 513 - return False 514 514 - 515 515 - 516 516 - # MIME type mapping for import metadata 517 517 - _MIME_TYPES = { 518 518 - ".m4a": "audio/mp4", 519 519 - ".mp3": "audio/mpeg", 520 520 - ".wav": "audio/wav", 521 521 - ".flac": "audio/flac", 522 522 - ".ogg": "audio/ogg", 523 523 - ".mp4": "video/mp4", 524 524 - ".webm": "video/webm", 525 525 - ".mov": "video/quicktime", 526 526 - ".txt": "text/plain", 527 527 - ".md": "text/markdown", 528 528 - ".pdf": "application/pdf", 529 529 - } 530 530 - 531 531 - 532 532 - def _is_in_imports(media_path: str) -> bool: 533 533 - """Check if file path is already under journal/imports/.""" 534 534 - imports_dir = os.path.join(get_journal(), "imports") 535 535 - abs_media = os.path.abspath(media_path) 536 536 - abs_imports = os.path.abspath(imports_dir) 537 537 - return abs_media.startswith(abs_imports + os.sep) 538 538 - 539 539 - 540 540 - def _setup_import( 541 541 - media_path: str, 542 542 - timestamp: str, 543 543 - facet: str | None, 544 544 - setting: str | None, 545 545 - detection_result: dict | None, 546 546 - force: bool = False, 547 547 - ) -> str: 548 548 - """Copy file to imports/ and write metadata. Returns new file path.""" 549 549 - journal_root = Path(get_journal()) 550 550 - import_dir = journal_root / "imports" / timestamp 551 551 - 552 552 - # Check for conflict 553 553 - if import_dir.exists(): 554 554 - if force: 555 555 - logger.info(f"Removing existing import directory: {import_dir}") 556 556 - shutil.rmtree(import_dir) 557 557 - else: 558 558 - raise SystemExit( 559 559 - f"Error: Import already exists for timestamp {timestamp}\n" 560 560 - f"To re-import, use --force to delete existing data and start over" 561 561 - ) 562 562 - 563 563 - # Copy file to imports/ 564 564 - filename = os.path.basename(media_path) 565 565 - new_path = save_import_file( 566 566 - journal_root=journal_root, 567 567 - timestamp=timestamp, 568 568 - source_path=Path(media_path), 569 569 - filename=filename, 570 570 - ) 571 571 - 572 572 - # Build metadata matching app structure 573 573 - upload_ts = now_ms() 574 574 - ext = os.path.splitext(filename)[1].lower() 575 575 - metadata = { 576 576 - "original_filename": filename, 577 577 - "upload_timestamp": upload_ts, 578 578 - "upload_datetime": dt.datetime.fromtimestamp(upload_ts / 1000).isoformat(), 579 579 - "detection_result": detection_result, 580 580 - "detected_timestamp": timestamp, 581 581 - "user_timestamp": timestamp, 582 582 - "file_size": new_path.stat().st_size if new_path.exists() else 0, 583 583 - "mime_type": _MIME_TYPES.get(ext, "application/octet-stream"), 584 584 - "facet": facet, 585 585 - "setting": setting, 586 586 - "file_path": str(new_path), 587 587 - } 588 588 - 589 589 - write_import_metadata( 590 590 - journal_root=journal_root, 591 591 - timestamp=timestamp, 592 592 - metadata=metadata, 593 593 - ) 594 594 - 595 595 - logger.info(f"Copied to journal: {new_path}") 596 596 - return str(new_path) 597 597 - 598 598 - 599 599 - def _format_timestamp_display(timestamp: str) -> str: 600 600 - """Format timestamp for human-readable display.""" 601 601 - try: 602 602 - dt_obj = dt.datetime.strptime(timestamp, "%Y%m%d_%H%M%S") 603 603 - return dt_obj.strftime("%a %b %d %Y, %-I:%M %p") 604 604 - except ValueError: 605 605 - return timestamp 606 606 - 607 607 - 608 608 - def main() -> None: 609 609 - global _callosum, _message_queue, _import_id, _current_stage, _start_time 610 610 - global _stage_start_time, _stages_run, _status_thread, _status_running 611 611 - 612 612 - parser = argparse.ArgumentParser(description="Chunk a media file into the journal") 613 613 - parser.add_argument("media", help="Path to video or audio file") 614 614 - parser.add_argument( 615 615 - "--timestamp", help="Timestamp YYYYMMDD_HHMMSS for journal entry" 616 616 - ) 617 617 - parser.add_argument( 618 618 - "--summarize", 619 619 - type=str2bool, 620 620 - default=True, 621 621 - help="Create summary.md after transcription completes", 622 622 - ) 623 623 - parser.add_argument( 624 624 - "--facet", 625 625 - type=str, 626 626 - default=None, 627 627 - help="Facet name for this import", 628 628 - ) 629 629 - parser.add_argument( 630 630 - "--setting", 631 631 - type=str, 632 632 - default=None, 633 633 - help="Contextual setting description to store with import metadata", 634 634 - ) 635 635 - parser.add_argument( 636 636 - "--source", 637 637 - type=str, 638 638 - default=None, 639 639 - help="Import source type (apple, plaud, audio, text). Auto-detected if omitted.", 640 640 - ) 641 641 - parser.add_argument( 642 642 - "--skip-summary", 643 643 - action="store_true", 644 644 - help="Skip waiting for transcription and summary generation", 645 645 - ) 646 646 - parser.add_argument( 647 647 - "--force", 648 648 - action="store_true", 649 649 - help="Force re-import by deleting existing import directory", 650 650 - ) 651 651 - parser.add_argument( 652 652 - "--auto", 653 653 - action="store_true", 654 654 - help="Auto-accept detected timestamp and proceed with import", 655 655 - ) 656 656 - args, extra = setup_cli(parser, parse_known=True) 657 657 - if extra and not args.timestamp: 658 658 - args.timestamp = extra[0] 659 659 - 660 660 - # Track detection result for metadata 661 661 - detection_result = None 662 662 - 663 663 - # If no timestamp provided, detect it 664 664 - if not args.timestamp: 665 665 - # Pass the original filename for better detection 666 666 - detection_result = detect_created( 667 667 - args.media, original_filename=os.path.basename(args.media) 668 668 - ) 669 669 - if ( 670 670 - detection_result 671 671 - and detection_result.get("day") 672 672 - and detection_result.get("time") 673 673 - ): 674 674 - detected_timestamp = f"{detection_result['day']}_{detection_result['time']}" 675 675 - display = _format_timestamp_display(detected_timestamp) 676 676 - if args.auto: 677 677 - print( 678 678 - f"Detected timestamp: {detected_timestamp} ({display}) — auto-importing" 679 679 - ) 680 680 - args.timestamp = detected_timestamp 681 681 - else: 682 682 - print(f"Detected timestamp: {detected_timestamp} ({display})") 683 683 - print("\nRun:") 684 684 - print(f" sol import {args.media} --timestamp {detected_timestamp}") 685 685 - return 686 686 - else: 687 687 - raise SystemExit( 688 688 - "Could not detect timestamp. Please provide --timestamp YYYYMMDD_HHMMSS" 689 689 - ) 690 690 - 691 691 - if not TIME_RE.fullmatch(args.timestamp): 692 692 - raise SystemExit("timestamp must be in YYYYMMDD_HHMMSS format") 693 693 - 694 694 - # Check if file needs setup (not already in imports/) 695 695 - needs_setup = not _is_in_imports(args.media) 696 696 - 697 697 - # Copy to imports/ if file is not already there 698 698 - if needs_setup: 699 699 - args.media = _setup_import( 700 700 - args.media, 701 701 - args.timestamp, 702 702 - args.facet, 703 703 - args.setting, 704 704 - detection_result, 705 705 - force=args.force, 706 706 - ) 707 707 - print("Starting import...") 708 708 - 709 709 - base_dt = dt.datetime.strptime(args.timestamp, "%Y%m%d_%H%M%S") 710 710 - day = base_dt.strftime("%Y%m%d") 711 711 - logger.info(f"Using provided timestamp: {args.timestamp}") 712 712 - day_dir = str(day_path(day)) 713 713 - 714 714 - # Derive stream identity for this import 715 715 - if args.source: 716 716 - import_source = args.source 717 717 - else: 718 718 - # Auto-detect from file extension 719 719 - _ext = os.path.splitext(args.media)[1].lower() 720 720 - if _ext == ".m4a": 721 721 - import_source = "apple" 722 722 - elif _ext in {".txt", ".md", ".pdf"}: 723 723 - import_source = "text" 724 724 - else: 725 725 - import_source = "audio" 726 726 - stream = stream_name(import_source=import_source) 727 727 - 728 728 - # Initialize importer tract state 729 729 - _import_id = args.timestamp 730 730 - _start_time = time.monotonic() 731 731 - _stage_start_time = _start_time 732 732 - _current_stage = "initialization" 733 733 - _stages_run = ["initialization"] 734 734 - 735 735 - # Start Callosum connection with message queue for receiving events 736 736 - _message_queue = queue.Queue() 737 737 - _callosum = CallosumConnection(defaults={"rev": get_rev()}) 738 738 - _callosum.start(callback=lambda msg: _message_queue.put(msg)) 739 739 - 740 740 - # Start status emitter thread 741 741 - _status_running = True 742 742 - _status_thread = threading.Thread(target=_status_emitter, daemon=True) 743 743 - _status_thread.start() 744 744 - 745 745 - # Emit started event 746 746 - ext = os.path.splitext(args.media)[1].lower() 747 747 - _callosum.emit( 748 748 - "importer", 749 749 - "started", 750 750 - import_id=_import_id, 751 751 - input_file=os.path.basename(args.media), 752 752 - file_type=ext.lstrip("."), 753 753 - day=day, 754 754 - facet=args.facet, 755 755 - setting=args.setting, 756 756 - options={ 757 757 - "summarize": args.summarize, 758 758 - "skip_summary": args.skip_summary, 759 759 - }, 760 760 - stage=_current_stage, 761 761 - stream=stream, 762 762 - ) 763 763 - 764 764 - # Track all created files and processing metadata 765 765 - all_created_files: list[str] = [] 766 766 - created_segments: list[str] = [] 767 767 - journal_root = Path(get_journal()) 768 768 - processing_results = { 769 769 - "processed_timestamp": args.timestamp, 770 770 - "target_day": base_dt.strftime("%Y%m%d"), 771 771 - "target_day_path": day_dir, 772 772 - "input_file": args.media, 773 773 - "processing_started": dt.datetime.now().isoformat(), 774 774 - "facet": args.facet, 775 775 - "setting": args.setting, 776 776 - "outputs": [], 777 777 - } 778 778 - 779 779 - # Get parent directory for saving metadata 780 780 - media_path = Path(args.media) 781 781 - import_dir = media_path.parent 782 782 - failed_segments: list[str] = [] 783 783 - 784 784 - try: 785 785 - if ext in {".txt", ".md", ".pdf"}: 786 786 - # Text transcript processing — no observe pipeline 787 787 - _set_stage("segmenting") 788 788 - 789 789 - created_files = process_transcript( 790 790 - args.media, 791 791 - day_dir, 792 792 - base_dt, 793 793 - import_id=args.timestamp, 794 794 - stream=stream, 795 795 - facet=args.facet, 796 796 - setting=args.setting, 797 797 - ) 798 798 - all_created_files.extend(created_files) 799 799 - processing_results["outputs"].append( 800 800 - { 801 801 - "type": "transcript", 802 802 - "format": "imported_audio.jsonl", 803 803 - "description": "Transcript segments", 804 804 - "files": created_files, 805 805 - "count": len(created_files), 806 806 - } 807 807 - ) 808 808 - 809 809 - # Extract segment keys for text imports 810 810 - for file_path in created_files: 811 811 - seg = segment_key(file_path) 812 812 - if seg and seg not in created_segments: 813 813 - created_segments.append(seg) 814 814 - 815 815 - # Write stream markers for text import segments 816 816 - for seg in created_segments: 817 817 - try: 818 818 - seg_dir = day_path(day) / stream / seg 819 819 - result = update_stream(stream, day, seg, type="import", host=None) 820 820 - write_segment_stream( 821 821 - seg_dir, 822 822 - stream, 823 823 - result["prev_day"], 824 824 - result["prev_segment"], 825 825 - result["seq"], 826 826 - ) 827 827 - except Exception as e: 828 828 - logger.warning(f"Failed to write stream identity: {e}") 829 829 - 830 830 - # Emit observe.observed for text imports (already processed) 831 831 - for seg in created_segments: 832 832 - _callosum.emit( 833 833 - "observe", "observed", segment=seg, day=day, stream=stream 834 834 - ) 835 835 - logger.info(f"Emitted observe.observed for segment: {day}/{seg}") 836 836 - 837 837 - else: 838 838 - # Audio processing via observe pipeline 839 839 - _set_stage("segmenting") 840 840 - 841 841 - # Prepare audio segments (slice into 5-minute chunks) 842 842 - segments = prepare_audio_segments( 843 843 - args.media, 844 844 - day_dir, 845 845 - base_dt, 846 846 - args.timestamp, 847 847 - stream, 848 848 - ) 849 849 - 850 850 - if not segments: 851 851 - raise RuntimeError("No segments created from audio file") 852 852 - 853 853 - # Track created files and segment keys, write stream markers 854 854 - for seg_key, seg_dir, files in segments: 855 855 - created_segments.append(seg_key) 856 856 - for f in files: 857 857 - all_created_files.append(str(seg_dir / f)) 858 858 - try: 859 859 - result = update_stream( 860 860 - stream, day, seg_key, type="import", host=None 861 861 - ) 862 862 - write_segment_stream( 863 863 - seg_dir, 864 864 - stream, 865 865 - result["prev_day"], 866 866 - result["prev_segment"], 867 867 - result["seq"], 868 868 - ) 869 869 - except Exception as e: 870 870 - logger.warning(f"Failed to write stream identity: {e}") 871 871 - 872 872 - # Save segment list for tracking 873 873 - save_import_segments(journal_root, args.timestamp, created_segments, day) 874 874 - 875 875 - processing_results["outputs"].append( 876 876 - { 877 877 - "type": "audio_segments", 878 878 - "description": "Audio segments queued for transcription", 879 879 - "segments": created_segments, 880 880 - "count": len(created_segments), 881 881 - } 882 882 - ) 883 883 - 884 884 - # Build meta dict for observe.observing events 885 885 - meta: dict[str, str] = {"import_id": args.timestamp, "stream": stream} 886 886 - if args.facet: 887 887 - meta["facet"] = args.facet 888 888 - if args.setting: 889 889 - meta["setting"] = args.setting 890 890 - 891 891 - # Emit observe.observing per segment to trigger sense.py transcription 892 892 - for seg_key, seg_dir, files in segments: 893 893 - _callosum.emit( 894 894 - "observe", 895 895 - "observing", 896 896 - segment=seg_key, 897 897 - day=day, 898 898 - files=files, 899 899 - meta=meta, 900 900 - stream=stream, 901 901 - ) 902 902 - logger.info(f"Emitted observe.observing for segment: {day}/{seg_key}") 903 903 - 904 904 - # Wait for transcription to complete (unless --no-wait) 905 905 - if not args.skip_summary: 906 906 - _set_stage("transcribing") 907 907 - pending = set(created_segments) 908 908 - segment_timeout = 600 # 10 minutes since last progress 909 909 - last_progress = time.monotonic() 910 910 - 911 911 - logger.info(f"Waiting for {len(pending)} segments to complete") 912 912 - 913 913 - while pending: 914 914 - # Check for timeout since last progress 915 915 - if time.monotonic() - last_progress > segment_timeout: 916 916 - timed_out = sorted(pending) 917 917 - logger.error(f"Timed out waiting for segments: {timed_out}") 918 918 - failed_segments.extend(timed_out) 919 919 - break 920 920 - 921 921 - # Poll for observe.observed events from message queue 922 922 - try: 923 923 - msg = _message_queue.get(timeout=5.0) 924 924 - except queue.Empty: 925 925 - continue 926 926 - 927 927 - tract = msg.get("tract") 928 928 - event = msg.get("event") 929 929 - seg = msg.get("segment") 930 930 - 931 931 - if tract == "observe" and event == "observed" and seg in pending: 932 932 - pending.discard(seg) 933 933 - last_progress = time.monotonic() 934 934 - if msg.get("error"): 935 935 - errors = msg.get("errors", []) 936 936 - logger.warning( 937 937 - f"Segment {seg} failed: {errors} " 938 938 - f"({len(pending)} remaining)" 939 939 - ) 940 940 - failed_segments.append(seg) 941 941 - else: 942 942 - logger.info( 943 943 - f"Segment {seg} transcribed " 944 944 - f"({len(pending)} remaining)" 945 945 - ) 946 946 - 947 947 - if failed_segments: 948 948 - logger.warning( 949 949 - f"{len(failed_segments)} of {len(created_segments)} " 950 950 - f"segments failed: {failed_segments}" 951 951 - ) 952 952 - else: 953 953 - logger.info("All segments transcribed successfully") 954 954 - 955 955 - # Complete processing metadata 956 956 - processing_results["processing_completed"] = dt.datetime.now().isoformat() 957 957 - processing_results["total_files_created"] = len(all_created_files) 958 958 - processing_results["all_created_files"] = all_created_files 959 959 - processing_results["segments"] = created_segments 960 960 - if failed_segments: 961 961 - processing_results["failed_segments"] = failed_segments 962 962 - 963 963 - # Write imported.json with all processing metadata 964 964 - imported_path = import_dir / "imported.json" 965 965 - try: 966 966 - with open(imported_path, "w", encoding="utf-8") as f: 967 967 - json.dump(processing_results, f, indent=2) 968 968 - logger.info(f"Saved import processing metadata: {imported_path}") 969 969 - except Exception as e: 970 970 - logger.warning(f"Failed to save imported.json: {e}") 971 971 - 972 972 - # Update import.json with processing summary if it exists 973 973 - import_metadata_path = import_dir / "import.json" 974 974 - if import_metadata_path.exists(): 975 975 - try: 976 976 - with open(import_metadata_path, "r", encoding="utf-8") as f: 977 977 - import_meta = json.load(f) 978 978 - import_meta["processing_completed"] = processing_results[ 979 979 - "processing_completed" 980 980 - ] 981 981 - import_meta["total_files_created"] = processing_results[ 982 982 - "total_files_created" 983 983 - ] 984 984 - import_meta["imported_json_path"] = str(imported_path) 985 985 - import_meta["segments"] = created_segments 986 986 - with open(import_metadata_path, "w", encoding="utf-8") as f: 987 987 - json.dump(import_meta, f, indent=2) 988 988 - logger.info(f"Updated import metadata: {import_metadata_path}") 989 989 - except Exception as e: 990 990 - logger.warning(f"Failed to update import metadata: {e}") 991 991 - 992 992 - # Create summary if requested and we have segments 993 993 - if args.summarize and created_segments and not args.skip_summary: 994 994 - _set_stage("summarizing") 995 995 - _run_import_summary(import_dir, day, created_segments) 996 996 - 997 997 - # Emit completed event 998 998 - duration_ms = int((time.monotonic() - _start_time) * 1000) 999 999 - output_files_relative = [_get_relative_path(f) for f in all_created_files] 1000 1000 - metadata_file_relative = _get_relative_path(str(imported_path)) 1001 1001 - 1002 1002 - _callosum.emit( 1003 1003 - "importer", 1004 1004 - "completed", 1005 1005 - import_id=_import_id, 1006 1006 - stage=_current_stage, 1007 1007 - duration_ms=duration_ms, 1008 1008 - total_files_created=len(all_created_files), 1009 1009 - output_files=output_files_relative, 1010 1010 - metadata_file=metadata_file_relative, 1011 1011 - stages_run=_stages_run, 1012 1012 - segments=created_segments, 1013 1013 - stream=stream, 1014 1014 - ) 1015 1015 - 1016 1016 - except Exception as e: 1017 1017 - # Write error state to imported.json for persistent failure tracking 1018 1018 - duration_ms = int((time.monotonic() - _start_time) * 1000) 1019 1019 - partial_outputs = [_get_relative_path(f) for f in all_created_files] 1020 1020 - imported_path = import_dir / "imported.json" 1021 1021 - 1022 1022 - error_results = { 1023 1023 - **processing_results, # Include all the metadata we have 1024 1024 - "processing_failed": dt.datetime.now().isoformat(), 1025 1025 - "error": str(e), 1026 1026 - "error_stage": _current_stage, 1027 1027 - "duration_ms": duration_ms, 1028 1028 - "total_files_created": len(all_created_files), 1029 1029 - "all_created_files": all_created_files, 1030 1030 - "stages_run": _stages_run, 1031 1031 - } 1032 1032 - 1033 1033 - try: 1034 1034 - with open(imported_path, "w", encoding="utf-8") as f: 1035 1035 - json.dump(error_results, f, indent=2) 1036 1036 - logger.info(f"Saved error state: {imported_path}") 1037 1037 - except Exception as write_err: 1038 1038 - logger.warning(f"Failed to write error state: {write_err}") 1039 1039 - 1040 1040 - # Emit error event 1041 1041 - if _callosum: 1042 1042 - _callosum.emit( 1043 1043 - "importer", 1044 1044 - "error", 1045 1045 - import_id=_import_id, 1046 1046 - stage=_current_stage, 1047 1047 - error=str(e), 1048 1048 - duration_ms=duration_ms, 1049 1049 - partial_outputs=partial_outputs, 1050 1050 - ) 1051 1051 - 1052 1052 - logger.error(f"Import failed: {e}") 1053 1053 - raise 1054 1054 - 1055 1055 - finally: 1056 1056 - # Stop status thread 1057 1057 - _status_running = False 1058 1058 - if _status_thread: 1059 1059 - _status_thread.join(timeout=6) 1060 1060 - 1061 1061 - 1062 1062 - if __name__ == "__main__": 1063 1063 - main()

think/importer_utils.py think/importers/utils.py

reviewed

+2

think/importers/__init__.py

reviewed

··· 1 1 + # SPDX-License-Identifier: AGPL-3.0-only 2 2 + # Copyright (c) 2026 sol pbc

+192

think/importers/audio.py

reviewed

··· 1 1 + # SPDX-License-Identifier: AGPL-3.0-only 2 2 + # Copyright (c) 2026 sol pbc 3 3 + 4 4 + import datetime as dt 5 5 + import logging 6 6 + import subprocess 7 7 + from datetime import timedelta 8 8 + from pathlib import Path 9 9 + 10 10 + from observe.utils import find_available_segment 11 11 + 12 12 + logger = logging.getLogger(__name__) 13 13 + 14 14 + 15 15 + def slice_audio_segment( 16 16 + source_path: str, 17 17 + output_path: str, 18 18 + start_seconds: float, 19 19 + duration_seconds: float, 20 20 + ) -> str: 21 21 + """Extract an audio segment from source file, preserving original format. 22 22 + 23 23 + Uses stream copy for lossless extraction when possible. 24 24 + 25 25 + Args: 26 26 + source_path: Path to source audio file 27 27 + output_path: Path for output segment file 28 28 + start_seconds: Start offset in seconds 29 29 + duration_seconds: Duration to extract in seconds 30 30 + 31 31 + Returns: 32 32 + Output path on success 33 33 + 34 34 + Raises: 35 35 + subprocess.CalledProcessError: If ffmpeg fails 36 36 + """ 37 37 + cmd = [ 38 38 + "ffmpeg", 39 39 + "-ss", 40 40 + str(start_seconds), 41 41 + "-i", 42 42 + source_path, 43 43 + "-t", 44 44 + str(duration_seconds), 45 45 + "-vn", # No video 46 46 + "-c:a", 47 47 + "copy", # Stream copy for lossless extraction 48 48 + "-y", # Overwrite output 49 49 + output_path, 50 50 + ] 51 51 + try: 52 52 + subprocess.run(cmd, check=True, capture_output=True, text=True) 53 53 + except subprocess.CalledProcessError: 54 54 + # Fallback: re-encode if stream copy fails (some formats don't support it) 55 55 + logger.debug(f"Stream copy failed, re-encoding: {output_path}") 56 56 + cmd_reencode = [ 57 57 + "ffmpeg", 58 58 + "-ss", 59 59 + str(start_seconds), 60 60 + "-i", 61 61 + source_path, 62 62 + "-t", 63 63 + str(duration_seconds), 64 64 + "-vn", 65 65 + "-y", 66 66 + output_path, 67 67 + ] 68 68 + subprocess.run(cmd_reencode, check=True, capture_output=True, text=True) 69 69 + 70 70 + logger.info(f"Created audio segment: {output_path}") 71 71 + return output_path 72 72 + 73 73 + 74 74 + def _get_audio_duration(audio_path: str) -> float | None: 75 75 + """Get audio duration in seconds using ffprobe. 76 76 + 77 77 + Args: 78 78 + audio_path: Path to audio file 79 79 + 80 80 + Returns: 81 81 + Duration in seconds, or None if unable to determine 82 82 + """ 83 83 + try: 84 84 + cmd = [ 85 85 + "ffprobe", 86 86 + "-v", 87 87 + "error", 88 88 + "-show_entries", 89 89 + "format=duration", 90 90 + "-of", 91 91 + "default=noprint_wrappers=1:nokey=1", 92 92 + audio_path, 93 93 + ] 94 94 + result = subprocess.run(cmd, capture_output=True, text=True, check=True) 95 95 + return float(result.stdout.strip()) 96 96 + except (subprocess.CalledProcessError, ValueError) as e: 97 97 + logger.warning(f"Could not determine audio duration: {e}") 98 98 + return None 99 99 + 100 100 + 101 101 + def prepare_audio_segments( 102 102 + media_path: str, 103 103 + day_dir: str, 104 104 + base_dt: dt.datetime, 105 105 + import_id: str, 106 106 + stream: str, 107 107 + ) -> list[tuple[str, Path, list[str]]]: 108 108 + """Slice audio into 5-minute segments for observe pipeline. 109 109 + 110 110 + Creates segment directories with audio slices, ready for transcription 111 111 + via observe.observing events. 112 112 + 113 113 + Args: 114 114 + media_path: Path to source audio file 115 115 + day_dir: Day directory path (YYYYMMDD) 116 116 + base_dt: Base datetime for timestamp calculation 117 117 + import_id: Import identifier 118 118 + stream: Stream name for directory layout (day/stream/segment/) 119 119 + 120 120 + Returns: 121 121 + List of (segment_key, segment_dir, files_list) tuples 122 122 + where files_list contains the audio filename(s) created 123 123 + """ 124 124 + media = Path(media_path) 125 125 + source_ext = media.suffix.lower() 126 126 + stream_dir = Path(day_dir) / stream 127 127 + 128 128 + # Get audio duration to calculate number of segments 129 129 + duration = _get_audio_duration(media_path) 130 130 + if duration is None: 131 131 + raise RuntimeError(f"Could not determine duration of {media_path}") 132 132 + 133 133 + # Calculate number of 5-minute segments (ceiling division) 134 134 + segment_duration = 300 # 5 minutes 135 135 + num_segments = int((duration + segment_duration - 1) // segment_duration) 136 136 + if num_segments == 0: 137 137 + num_segments = 1 # At least one segment for very short audio 138 138 + 139 139 + segments: list[tuple[str, Path, list[str]]] = [] 140 140 + 141 141 + for chunk_index in range(num_segments): 142 142 + # Calculate timestamp for this segment 143 143 + ts = base_dt + timedelta(minutes=chunk_index * 5) 144 144 + time_part = ts.strftime("%H%M%S") 145 145 + 146 146 + # Create segment key with 5-minute duration 147 147 + segment_key_candidate = f"{time_part}_{segment_duration}" 148 148 + 149 149 + # Check for collision and deconflict if needed 150 150 + available_key = find_available_segment(stream_dir, segment_key_candidate) 151 151 + if available_key is None: 152 152 + logger.warning( 153 153 + f"Could not find available segment key near {segment_key_candidate}" 154 154 + ) 155 155 + continue 156 156 + 157 157 + if available_key != segment_key_candidate: 158 158 + logger.info( 159 159 + f"Segment collision: {segment_key_candidate} -> {available_key}" 160 160 + ) 161 161 + 162 162 + # Create segment directory under stream 163 163 + segment_dir = stream_dir / available_key 164 164 + segment_dir.mkdir(parents=True, exist_ok=True) 165 165 + 166 166 + # Slice audio for this segment 167 167 + audio_filename = f"imported_audio{source_ext}" 168 168 + audio_path = segment_dir / audio_filename 169 169 + start_seconds = chunk_index * segment_duration 170 170 + 171 171 + # For the last segment, use remaining duration 172 172 + if chunk_index == num_segments - 1: 173 173 + chunk_duration = duration - start_seconds 174 174 + else: 175 175 + chunk_duration = segment_duration 176 176 + 177 177 + try: 178 178 + slice_audio_segment( 179 179 + media_path, 180 180 + str(audio_path), 181 181 + start_seconds, 182 182 + chunk_duration, 183 183 + ) 184 184 + segments.append((available_key, segment_dir, [audio_filename])) 185 185 + logger.info(f"Created segment: {available_key} with {audio_filename}") 186 186 + except subprocess.CalledProcessError as e: 187 187 + logger.warning(f"Failed to slice segment {available_key}: {e}") 188 188 + # Clean up empty directory 189 189 + if segment_dir.exists() and not any(segment_dir.iterdir()): 190 190 + segment_dir.rmdir() 191 191 + 192 192 + return segments

+547

think/importers/cli.py

reviewed

··· 1 1 + # SPDX-License-Identifier: AGPL-3.0-only 2 2 + # Copyright (c) 2026 sol pbc 3 3 + 4 4 + import argparse 5 5 + import datetime as dt 6 6 + import json 7 7 + import logging 8 8 + import os 9 9 + import queue 10 10 + import re 11 11 + import threading 12 12 + import time 13 13 + from pathlib import Path 14 14 + 15 15 + from think.callosum import CallosumConnection 16 16 + from think.detect_created import detect_created 17 17 + from think.importers.audio import prepare_audio_segments 18 18 + from think.importers.shared import ( 19 19 + _get_relative_path, 20 20 + _is_in_imports, 21 21 + _run_import_summary, 22 22 + _setup_import, 23 23 + ) 24 24 + from think.importers.text import process_transcript 25 25 + from think.importers.utils import save_import_segments 26 26 + from think.streams import stream_name, update_stream, write_segment_stream 27 27 + from think.utils import day_path, get_journal, get_rev, segment_key, setup_cli 28 28 + 29 29 + logger = logging.getLogger(__name__) 30 30 + 31 31 + TIME_RE = re.compile(r"\d{8}_\d{6}") 32 32 + 33 33 + # Importer tract state 34 34 + _callosum: CallosumConnection | None = None 35 35 + _message_queue: queue.Queue | None = None 36 36 + _import_id: str | None = None 37 37 + _current_stage: str = "initialization" 38 38 + _start_time: float = 0.0 39 39 + _stage_start_time: float = 0.0 40 40 + _stages_run: list[str] = [] 41 41 + _status_thread: threading.Thread | None = None 42 42 + _status_running: bool = False 43 43 + 44 44 + 45 45 + def _set_stage(stage: str) -> None: 46 46 + """Update current stage and track timing.""" 47 47 + global _current_stage, _stage_start_time 48 48 + _current_stage = stage 49 49 + _stage_start_time = time.monotonic() 50 50 + if stage not in _stages_run: 51 51 + _stages_run.append(stage) 52 52 + logger.debug(f"Stage changed to: {stage}") 53 53 + 54 54 + 55 55 + def _status_emitter() -> None: 56 56 + """Background thread that emits status events every 5 seconds.""" 57 57 + while _status_running: 58 58 + if _callosum and _import_id: 59 59 + elapsed_ms = int((time.monotonic() - _start_time) * 1000) 60 60 + stage_elapsed_ms = int((time.monotonic() - _stage_start_time) * 1000) 61 61 + _callosum.emit( 62 62 + "importer", 63 63 + "status", 64 64 + import_id=_import_id, 65 65 + stage=_current_stage, 66 66 + elapsed_ms=elapsed_ms, 67 67 + stage_elapsed_ms=stage_elapsed_ms, 68 68 + ) 69 69 + time.sleep(5) 70 70 + 71 71 + 72 72 + def str2bool(value: str) -> bool: 73 73 + if isinstance(value, bool): 74 74 + return value 75 75 + val = value.lower() 76 76 + if val in {"y", "yes", "true", "t", "1"}: 77 77 + return True 78 78 + if val in {"n", "no", "false", "f", "0"}: 79 79 + return False 80 80 + raise argparse.ArgumentTypeError("boolean value expected") 81 81 + 82 82 + 83 83 + def _format_timestamp_display(timestamp: str) -> str: 84 84 + """Format timestamp for human-readable display.""" 85 85 + try: 86 86 + dt_obj = dt.datetime.strptime(timestamp, "%Y%m%d_%H%M%S") 87 87 + return dt_obj.strftime("%a %b %d %Y, %-I:%M %p") 88 88 + except ValueError: 89 89 + return timestamp 90 90 + 91 91 + 92 92 + def main() -> None: 93 93 + global _callosum, _message_queue, _import_id, _current_stage, _start_time 94 94 + global _stage_start_time, _stages_run, _status_thread, _status_running 95 95 + 96 96 + parser = argparse.ArgumentParser(description="Chunk a media file into the journal") 97 97 + parser.add_argument("media", help="Path to video or audio file") 98 98 + parser.add_argument( 99 99 + "--timestamp", help="Timestamp YYYYMMDD_HHMMSS for journal entry" 100 100 + ) 101 101 + parser.add_argument( 102 102 + "--summarize", 103 103 + type=str2bool, 104 104 + default=True, 105 105 + help="Create summary.md after transcription completes", 106 106 + ) 107 107 + parser.add_argument( 108 108 + "--facet", 109 109 + type=str, 110 110 + default=None, 111 111 + help="Facet name for this import", 112 112 + ) 113 113 + parser.add_argument( 114 114 + "--setting", 115 115 + type=str, 116 116 + default=None, 117 117 + help="Contextual setting description to store with import metadata", 118 118 + ) 119 119 + parser.add_argument( 120 120 + "--source", 121 121 + type=str, 122 122 + default=None, 123 123 + help="Import source type (apple, plaud, audio, text). Auto-detected if omitted.", 124 124 + ) 125 125 + parser.add_argument( 126 126 + "--skip-summary", 127 127 + action="store_true", 128 128 + help="Skip waiting for transcription and summary generation", 129 129 + ) 130 130 + parser.add_argument( 131 131 + "--force", 132 132 + action="store_true", 133 133 + help="Force re-import by deleting existing import directory", 134 134 + ) 135 135 + parser.add_argument( 136 136 + "--auto", 137 137 + action="store_true", 138 138 + help="Auto-accept detected timestamp and proceed with import", 139 139 + ) 140 140 + args, extra = setup_cli(parser, parse_known=True) 141 141 + if extra and not args.timestamp: 142 142 + args.timestamp = extra[0] 143 143 + 144 144 + # Track detection result for metadata 145 145 + detection_result = None 146 146 + 147 147 + # If no timestamp provided, detect it 148 148 + if not args.timestamp: 149 149 + # Pass the original filename for better detection 150 150 + detection_result = detect_created( 151 151 + args.media, original_filename=os.path.basename(args.media) 152 152 + ) 153 153 + if ( 154 154 + detection_result 155 155 + and detection_result.get("day") 156 156 + and detection_result.get("time") 157 157 + ): 158 158 + detected_timestamp = f"{detection_result['day']}_{detection_result['time']}" 159 159 + display = _format_timestamp_display(detected_timestamp) 160 160 + if args.auto: 161 161 + print( 162 162 + f"Detected timestamp: {detected_timestamp} ({display}) — auto-importing" 163 163 + ) 164 164 + args.timestamp = detected_timestamp 165 165 + else: 166 166 + print(f"Detected timestamp: {detected_timestamp} ({display})") 167 167 + print("\nRun:") 168 168 + print(f" sol import {args.media} --timestamp {detected_timestamp}") 169 169 + return 170 170 + else: 171 171 + raise SystemExit( 172 172 + "Could not detect timestamp. Please provide --timestamp YYYYMMDD_HHMMSS" 173 173 + ) 174 174 + 175 175 + if not TIME_RE.fullmatch(args.timestamp): 176 176 + raise SystemExit("timestamp must be in YYYYMMDD_HHMMSS format") 177 177 + 178 178 + # Check if file needs setup (not already in imports/) 179 179 + needs_setup = not _is_in_imports(args.media) 180 180 + 181 181 + # Copy to imports/ if file is not already there 182 182 + if needs_setup: 183 183 + args.media = _setup_import( 184 184 + args.media, 185 185 + args.timestamp, 186 186 + args.facet, 187 187 + args.setting, 188 188 + detection_result, 189 189 + force=args.force, 190 190 + ) 191 191 + print("Starting import...") 192 192 + 193 193 + base_dt = dt.datetime.strptime(args.timestamp, "%Y%m%d_%H%M%S") 194 194 + day = base_dt.strftime("%Y%m%d") 195 195 + logger.info(f"Using provided timestamp: {args.timestamp}") 196 196 + day_dir = str(day_path(day)) 197 197 + 198 198 + # Derive stream identity for this import 199 199 + if args.source: 200 200 + import_source = args.source 201 201 + else: 202 202 + # Auto-detect from file extension 203 203 + _ext = os.path.splitext(args.media)[1].lower() 204 204 + if _ext == ".m4a": 205 205 + import_source = "apple" 206 206 + elif _ext in {".txt", ".md", ".pdf"}: 207 207 + import_source = "text" 208 208 + else: 209 209 + import_source = "audio" 210 210 + stream = stream_name(import_source=import_source) 211 211 + 212 212 + # Initialize importer tract state 213 213 + _import_id = args.timestamp 214 214 + _start_time = time.monotonic() 215 215 + _stage_start_time = _start_time 216 216 + _current_stage = "initialization" 217 217 + _stages_run = ["initialization"] 218 218 + 219 219 + # Start Callosum connection with message queue for receiving events 220 220 + _message_queue = queue.Queue() 221 221 + _callosum = CallosumConnection(defaults={"rev": get_rev()}) 222 222 + _callosum.start(callback=lambda msg: _message_queue.put(msg)) 223 223 + 224 224 + # Start status emitter thread 225 225 + _status_running = True 226 226 + _status_thread = threading.Thread(target=_status_emitter, daemon=True) 227 227 + _status_thread.start() 228 228 + 229 229 + # Emit started event 230 230 + ext = os.path.splitext(args.media)[1].lower() 231 231 + _callosum.emit( 232 232 + "importer", 233 233 + "started", 234 234 + import_id=_import_id, 235 235 + input_file=os.path.basename(args.media), 236 236 + file_type=ext.lstrip("."), 237 237 + day=day, 238 238 + facet=args.facet, 239 239 + setting=args.setting, 240 240 + options={ 241 241 + "summarize": args.summarize, 242 242 + "skip_summary": args.skip_summary, 243 243 + }, 244 244 + stage=_current_stage, 245 245 + stream=stream, 246 246 + ) 247 247 + 248 248 + # Track all created files and processing metadata 249 249 + all_created_files: list[str] = [] 250 250 + created_segments: list[str] = [] 251 251 + journal_root = Path(get_journal()) 252 252 + processing_results = { 253 253 + "processed_timestamp": args.timestamp, 254 254 + "target_day": base_dt.strftime("%Y%m%d"), 255 255 + "target_day_path": day_dir, 256 256 + "input_file": args.media, 257 257 + "processing_started": dt.datetime.now().isoformat(), 258 258 + "facet": args.facet, 259 259 + "setting": args.setting, 260 260 + "outputs": [], 261 261 + } 262 262 + 263 263 + # Get parent directory for saving metadata 264 264 + media_path = Path(args.media) 265 265 + import_dir = media_path.parent 266 266 + failed_segments: list[str] = [] 267 267 + 268 268 + try: 269 269 + if ext in {".txt", ".md", ".pdf"}: 270 270 + # Text transcript processing — no observe pipeline 271 271 + _set_stage("segmenting") 272 272 + 273 273 + created_files = process_transcript( 274 274 + args.media, 275 275 + day_dir, 276 276 + base_dt, 277 277 + import_id=args.timestamp, 278 278 + stream=stream, 279 279 + facet=args.facet, 280 280 + setting=args.setting, 281 281 + ) 282 282 + all_created_files.extend(created_files) 283 283 + processing_results["outputs"].append( 284 284 + { 285 285 + "type": "transcript", 286 286 + "format": "imported_audio.jsonl", 287 287 + "description": "Transcript segments", 288 288 + "files": created_files, 289 289 + "count": len(created_files), 290 290 + } 291 291 + ) 292 292 + 293 293 + # Extract segment keys for text imports 294 294 + for file_path in created_files: 295 295 + seg = segment_key(file_path) 296 296 + if seg and seg not in created_segments: 297 297 + created_segments.append(seg) 298 298 + 299 299 + # Write stream markers for text import segments 300 300 + for seg in created_segments: 301 301 + try: 302 302 + seg_dir = day_path(day) / stream / seg 303 303 + result = update_stream(stream, day, seg, type="import", host=None) 304 304 + write_segment_stream( 305 305 + seg_dir, 306 306 + stream, 307 307 + result["prev_day"], 308 308 + result["prev_segment"], 309 309 + result["seq"], 310 310 + ) 311 311 + except Exception as e: 312 312 + logger.warning(f"Failed to write stream identity: {e}") 313 313 + 314 314 + # Emit observe.observed for text imports (already processed) 315 315 + for seg in created_segments: 316 316 + _callosum.emit( 317 317 + "observe", "observed", segment=seg, day=day, stream=stream 318 318 + ) 319 319 + logger.info(f"Emitted observe.observed for segment: {day}/{seg}") 320 320 + 321 321 + else: 322 322 + # Audio processing via observe pipeline 323 323 + _set_stage("segmenting") 324 324 + 325 325 + # Prepare audio segments (slice into 5-minute chunks) 326 326 + segments = prepare_audio_segments( 327 327 + args.media, 328 328 + day_dir, 329 329 + base_dt, 330 330 + args.timestamp, 331 331 + stream, 332 332 + ) 333 333 + 334 334 + if not segments: 335 335 + raise RuntimeError("No segments created from audio file") 336 336 + 337 337 + # Track created files and segment keys, write stream markers 338 338 + for seg_key, seg_dir, files in segments: 339 339 + created_segments.append(seg_key) 340 340 + for f in files: 341 341 + all_created_files.append(str(seg_dir / f)) 342 342 + try: 343 343 + result = update_stream( 344 344 + stream, day, seg_key, type="import", host=None 345 345 + ) 346 346 + write_segment_stream( 347 347 + seg_dir, 348 348 + stream, 349 349 + result["prev_day"], 350 350 + result["prev_segment"], 351 351 + result["seq"], 352 352 + ) 353 353 + except Exception as e: 354 354 + logger.warning(f"Failed to write stream identity: {e}") 355 355 + 356 356 + # Save segment list for tracking 357 357 + save_import_segments(journal_root, args.timestamp, created_segments, day) 358 358 + 359 359 + processing_results["outputs"].append( 360 360 + { 361 361 + "type": "audio_segments", 362 362 + "description": "Audio segments queued for transcription", 363 363 + "segments": created_segments, 364 364 + "count": len(created_segments), 365 365 + } 366 366 + ) 367 367 + 368 368 + # Build meta dict for observe.observing events 369 369 + meta: dict[str, str] = {"import_id": args.timestamp, "stream": stream} 370 370 + if args.facet: 371 371 + meta["facet"] = args.facet 372 372 + if args.setting: 373 373 + meta["setting"] = args.setting 374 374 + 375 375 + # Emit observe.observing per segment to trigger sense.py transcription 376 376 + for seg_key, seg_dir, files in segments: 377 377 + _callosum.emit( 378 378 + "observe", 379 379 + "observing", 380 380 + segment=seg_key, 381 381 + day=day, 382 382 + files=files, 383 383 + meta=meta, 384 384 + stream=stream, 385 385 + ) 386 386 + logger.info(f"Emitted observe.observing for segment: {day}/{seg_key}") 387 387 + 388 388 + # Wait for transcription to complete (unless --no-wait) 389 389 + if not args.skip_summary: 390 390 + _set_stage("transcribing") 391 391 + pending = set(created_segments) 392 392 + segment_timeout = 600 # 10 minutes since last progress 393 393 + last_progress = time.monotonic() 394 394 + 395 395 + logger.info(f"Waiting for {len(pending)} segments to complete") 396 396 + 397 397 + while pending: 398 398 + # Check for timeout since last progress 399 399 + if time.monotonic() - last_progress > segment_timeout: 400 400 + timed_out = sorted(pending) 401 401 + logger.error(f"Timed out waiting for segments: {timed_out}") 402 402 + failed_segments.extend(timed_out) 403 403 + break 404 404 + 405 405 + # Poll for observe.observed events from message queue 406 406 + try: 407 407 + msg = _message_queue.get(timeout=5.0) 408 408 + except queue.Empty: 409 409 + continue 410 410 + 411 411 + tract = msg.get("tract") 412 412 + event = msg.get("event") 413 413 + seg = msg.get("segment") 414 414 + 415 415 + if tract == "observe" and event == "observed" and seg in pending: 416 416 + pending.discard(seg) 417 417 + last_progress = time.monotonic() 418 418 + if msg.get("error"): 419 419 + errors = msg.get("errors", []) 420 420 + logger.warning( 421 421 + f"Segment {seg} failed: {errors} " 422 422 + f"({len(pending)} remaining)" 423 423 + ) 424 424 + failed_segments.append(seg) 425 425 + else: 426 426 + logger.info( 427 427 + f"Segment {seg} transcribed " 428 428 + f"({len(pending)} remaining)" 429 429 + ) 430 430 + 431 431 + if failed_segments: 432 432 + logger.warning( 433 433 + f"{len(failed_segments)} of {len(created_segments)} " 434 434 + f"segments failed: {failed_segments}" 435 435 + ) 436 436 + else: 437 437 + logger.info("All segments transcribed successfully") 438 438 + 439 439 + # Complete processing metadata 440 440 + processing_results["processing_completed"] = dt.datetime.now().isoformat() 441 441 + processing_results["total_files_created"] = len(all_created_files) 442 442 + processing_results["all_created_files"] = all_created_files 443 443 + processing_results["segments"] = created_segments 444 444 + if failed_segments: 445 445 + processing_results["failed_segments"] = failed_segments 446 446 + 447 447 + # Write imported.json with all processing metadata 448 448 + imported_path = import_dir / "imported.json" 449 449 + try: 450 450 + with open(imported_path, "w", encoding="utf-8") as f: 451 451 + json.dump(processing_results, f, indent=2) 452 452 + logger.info(f"Saved import processing metadata: {imported_path}") 453 453 + except Exception as e: 454 454 + logger.warning(f"Failed to save imported.json: {e}") 455 455 + 456 456 + # Update import.json with processing summary if it exists 457 457 + import_metadata_path = import_dir / "import.json" 458 458 + if import_metadata_path.exists(): 459 459 + try: 460 460 + with open(import_metadata_path, "r", encoding="utf-8") as f: 461 461 + import_meta = json.load(f) 462 462 + import_meta["processing_completed"] = processing_results[ 463 463 + "processing_completed" 464 464 + ] 465 465 + import_meta["total_files_created"] = processing_results[ 466 466 + "total_files_created" 467 467 + ] 468 468 + import_meta["imported_json_path"] = str(imported_path) 469 469 + import_meta["segments"] = created_segments 470 470 + with open(import_metadata_path, "w", encoding="utf-8") as f: 471 471 + json.dump(import_meta, f, indent=2) 472 472 + logger.info(f"Updated import metadata: {import_metadata_path}") 473 473 + except Exception as e: 474 474 + logger.warning(f"Failed to update import metadata: {e}") 475 475 + 476 476 + # Create summary if requested and we have segments 477 477 + if args.summarize and created_segments and not args.skip_summary: 478 478 + _set_stage("summarizing") 479 479 + _run_import_summary(import_dir, day, created_segments) 480 480 + 481 481 + # Emit completed event 482 482 + duration_ms = int((time.monotonic() - _start_time) * 1000) 483 483 + output_files_relative = [_get_relative_path(f) for f in all_created_files] 484 484 + metadata_file_relative = _get_relative_path(str(imported_path)) 485 485 + 486 486 + _callosum.emit( 487 487 + "importer", 488 488 + "completed", 489 489 + import_id=_import_id, 490 490 + stage=_current_stage, 491 491 + duration_ms=duration_ms, 492 492 + total_files_created=len(all_created_files), 493 493 + output_files=output_files_relative, 494 494 + metadata_file=metadata_file_relative, 495 495 + stages_run=_stages_run, 496 496 + segments=created_segments, 497 497 + stream=stream, 498 498 + ) 499 499 + 500 500 + except Exception as e: 501 501 + # Write error state to imported.json for persistent failure tracking 502 502 + duration_ms = int((time.monotonic() - _start_time) * 1000) 503 503 + partial_outputs = [_get_relative_path(f) for f in all_created_files] 504 504 + imported_path = import_dir / "imported.json" 505 505 + 506 506 + error_results = { 507 507 + **processing_results, # Include all the metadata we have 508 508 + "processing_failed": dt.datetime.now().isoformat(), 509 509 + "error": str(e), 510 510 + "error_stage": _current_stage, 511 511 + "duration_ms": duration_ms, 512 512 + "total_files_created": len(all_created_files), 513 513 + "all_created_files": all_created_files, 514 514 + "stages_run": _stages_run, 515 515 + } 516 516 + 517 517 + try: 518 518 + with open(imported_path, "w", encoding="utf-8") as f: 519 519 + json.dump(error_results, f, indent=2) 520 520 + logger.info(f"Saved error state: {imported_path}") 521 521 + except Exception as write_err: 522 522 + logger.warning(f"Failed to write error state: {write_err}") 523 523 + 524 524 + # Emit error event 525 525 + if _callosum: 526 526 + _callosum.emit( 527 527 + "importer", 528 528 + "error", 529 529 + import_id=_import_id, 530 530 + stage=_current_stage, 531 531 + error=str(e), 532 532 + duration_ms=duration_ms, 533 533 + partial_outputs=partial_outputs, 534 534 + ) 535 535 + 536 536 + logger.error(f"Import failed: {e}") 537 537 + raise 538 538 + 539 539 + finally: 540 540 + # Stop status thread 541 541 + _status_running = False 542 542 + if _status_thread: 543 543 + _status_thread.join(timeout=6) 544 544 + 545 545 + 546 546 + if __name__ == "__main__": 547 547 + main()

+222

think/importers/shared.py

reviewed

··· 1 1 + # SPDX-License-Identifier: AGPL-3.0-only 2 2 + # Copyright (c) 2026 sol pbc 3 3 + 4 4 + import datetime as dt 5 5 + import json 6 6 + import logging 7 7 + import os 8 8 + import shutil 9 9 + from pathlib import Path 10 10 + 11 11 + from think.importers.utils import save_import_file, write_import_metadata 12 12 + from think.utils import get_journal, now_ms 13 13 + 14 14 + logger = logging.getLogger(__name__) 15 15 + 16 16 + 17 17 + def _get_relative_path(path: str) -> str: 18 18 + """Get path relative to journal, or return as-is if not under journal.""" 19 19 + journal_path = get_journal() 20 20 + try: 21 21 + return os.path.relpath(path, journal_path) 22 22 + except ValueError: 23 23 + return path 24 24 + 25 25 + 26 26 + def _write_import_jsonl( 27 27 + file_path: str, 28 28 + entries: list[dict], 29 29 + *, 30 30 + import_id: str, 31 31 + raw_filename: str | None = None, 32 32 + facet: str | None = None, 33 33 + setting: str | None = None, 34 34 + ) -> None: 35 35 + """Write imported transcript entries in JSONL format. 36 36 + 37 37 + First line contains imported metadata, subsequent lines contain entries. 38 38 + Each entry gets source="import" added to match the imported_audio.jsonl convention. 39 39 + 40 40 + Args: 41 41 + file_path: Path to write JSONL file 42 42 + entries: List of transcript entries 43 43 + import_id: Import identifier 44 44 + raw_filename: Source file name (relative path from segment to imports/) 45 45 + facet: Optional facet name 46 46 + setting: Optional setting description 47 47 + """ 48 48 + imported_meta: dict[str, str] = {"id": import_id} 49 49 + if facet: 50 50 + imported_meta["facet"] = facet 51 51 + if setting: 52 52 + imported_meta["setting"] = setting 53 53 + 54 54 + # Build top-level metadata with imported info 55 55 + metadata: dict[str, object] = {"imported": imported_meta} 56 56 + 57 57 + # Add raw file reference (path relative from segment to imports directory) 58 58 + if raw_filename: 59 59 + metadata["raw"] = f"../../imports/{import_id}/{raw_filename}" 60 60 + 61 61 + # Write JSONL: metadata first, then entries with source field 62 62 + jsonl_lines = [json.dumps(metadata)] 63 63 + for entry in entries: 64 64 + # Add source field if not already present (skip metadata entries like topics/setting) 65 65 + if "text" in entry and "source" not in entry: 66 66 + entry = {**entry, "source": "import"} 67 67 + jsonl_lines.append(json.dumps(entry)) 68 68 + 69 69 + with open(file_path, "w", encoding="utf-8") as f: 70 70 + f.write("\n".join(jsonl_lines) + "\n") 71 71 + 72 72 + 73 73 + def _run_import_summary( 74 74 + import_dir: Path, 75 75 + day: str, 76 76 + segments: list[str], 77 77 + ) -> bool: 78 78 + """Create a summary for imported segments using cortex generator. 79 79 + 80 80 + Args: 81 81 + import_dir: Directory where the summary will be saved 82 82 + day: Day string (YYYYMMDD format) 83 83 + segments: List of segment keys to summarize 84 84 + 85 85 + Returns: 86 86 + True if summary was created successfully, False otherwise 87 87 + """ 88 88 + from think.cortex_client import cortex_request, wait_for_agents 89 89 + 90 90 + if not segments: 91 91 + logger.info("No segments to summarize") 92 92 + return False 93 93 + 94 94 + summary_path = import_dir / "summary.md" 95 95 + 96 96 + try: 97 97 + logger.info(f"Creating summary for {len(segments)} segments via cortex") 98 98 + 99 99 + # Spawn generator via cortex 100 100 + agent_id = cortex_request( 101 101 + prompt="", # Generators don't use prompt 102 102 + name="importer", 103 103 + config={ 104 104 + "day": day, 105 105 + "span": segments, 106 106 + "output": "md", 107 107 + "output_path": str(summary_path), 108 108 + }, 109 109 + ) 110 110 + 111 111 + if agent_id is None: 112 112 + logger.error("Failed to send cortex request for import summary") 113 113 + return False 114 114 + 115 115 + # Wait for completion 116 116 + completed, timed_out = wait_for_agents([agent_id], timeout=300) 117 117 + 118 118 + if timed_out: 119 119 + logger.error(f"Import summary timed out (ID: {agent_id})") 120 120 + return False 121 121 + 122 122 + if completed: 123 123 + end_state = completed.get(agent_id, "unknown") 124 124 + if end_state == "finish" and summary_path.exists(): 125 125 + logger.info(f"Created import summary: {summary_path}") 126 126 + return True 127 127 + else: 128 128 + logger.warning( 129 129 + f"Generator ended with state {end_state}, " 130 130 + f"summary exists: {summary_path.exists()}" 131 131 + ) 132 132 + return False 133 133 + 134 134 + logger.error("Generator did not complete") 135 135 + return False 136 136 + 137 137 + except Exception as e: 138 138 + logger.error(f"Failed to create summary: {e}") 139 139 + return False 140 140 + 141 141 + 142 142 + # MIME type mapping for import metadata 143 143 + _MIME_TYPES = { 144 144 + ".m4a": "audio/mp4", 145 145 + ".mp3": "audio/mpeg", 146 146 + ".wav": "audio/wav", 147 147 + ".flac": "audio/flac", 148 148 + ".ogg": "audio/ogg", 149 149 + ".mp4": "video/mp4", 150 150 + ".webm": "video/webm", 151 151 + ".mov": "video/quicktime", 152 152 + ".txt": "text/plain", 153 153 + ".md": "text/markdown", 154 154 + ".pdf": "application/pdf", 155 155 + } 156 156 + 157 157 + 158 158 + def _is_in_imports(media_path: str) -> bool: 159 159 + """Check if file path is already under journal/imports/.""" 160 160 + imports_dir = os.path.join(get_journal(), "imports") 161 161 + abs_media = os.path.abspath(media_path) 162 162 + abs_imports = os.path.abspath(imports_dir) 163 163 + return abs_media.startswith(abs_imports + os.sep) 164 164 + 165 165 + 166 166 + def _setup_import( 167 167 + media_path: str, 168 168 + timestamp: str, 169 169 + facet: str | None, 170 170 + setting: str | None, 171 171 + detection_result: dict | None, 172 172 + force: bool = False, 173 173 + ) -> str: 174 174 + """Copy file to imports/ and write metadata. Returns new file path.""" 175 175 + journal_root = Path(get_journal()) 176 176 + import_dir = journal_root / "imports" / timestamp 177 177 + 178 178 + # Check for conflict 179 179 + if import_dir.exists(): 180 180 + if force: 181 181 + logger.info(f"Removing existing import directory: {import_dir}") 182 182 + shutil.rmtree(import_dir) 183 183 + else: 184 184 + raise SystemExit( 185 185 + f"Error: Import already exists for timestamp {timestamp}\n" 186 186 + f"To re-import, use --force to delete existing data and start over" 187 187 + ) 188 188 + 189 189 + # Copy file to imports/ 190 190 + filename = os.path.basename(media_path) 191 191 + new_path = save_import_file( 192 192 + journal_root=journal_root, 193 193 + timestamp=timestamp, 194 194 + source_path=Path(media_path), 195 195 + filename=filename, 196 196 + ) 197 197 + 198 198 + # Build metadata matching app structure 199 199 + upload_ts = now_ms() 200 200 + ext = os.path.splitext(filename)[1].lower() 201 201 + metadata = { 202 202 + "original_filename": filename, 203 203 + "upload_timestamp": upload_ts, 204 204 + "upload_datetime": dt.datetime.fromtimestamp(upload_ts / 1000).isoformat(), 205 205 + "detection_result": detection_result, 206 206 + "detected_timestamp": timestamp, 207 207 + "user_timestamp": timestamp, 208 208 + "file_size": new_path.stat().st_size if new_path.exists() else 0, 209 209 + "mime_type": _MIME_TYPES.get(ext, "application/octet-stream"), 210 210 + "facet": facet, 211 211 + "setting": setting, 212 212 + "file_path": str(new_path), 213 213 + } 214 214 + 215 215 + write_import_metadata( 216 216 + journal_root=journal_root, 217 217 + timestamp=timestamp, 218 218 + metadata=metadata, 219 219 + ) 220 220 + 221 221 + logger.info(f"Copied to journal: {new_path}") 222 222 + return str(new_path)

+131

think/importers/text.py

reviewed

··· 1 1 + # SPDX-License-Identifier: AGPL-3.0-only 2 2 + # Copyright (c) 2026 sol pbc 3 3 + 4 4 + import datetime as dt 5 5 + import logging 6 6 + import os 7 7 + 8 8 + from think.detect_transcript import detect_transcript_json, detect_transcript_segment 9 9 + from think.importers.shared import _write_import_jsonl 10 10 + 11 11 + try: 12 12 + from pypdf import PdfReader 13 13 + except ImportError: # pragma: no cover - optional dependency 14 14 + PdfReader = None 15 15 + 16 16 + logger = logging.getLogger(__name__) 17 17 + 18 18 + 19 19 + def _read_transcript(path: str) -> str: 20 20 + """Return transcript text from a .txt/.md/.pdf file.""" 21 21 + ext = os.path.splitext(path)[1].lower() 22 22 + if ext in {".txt", ".md"}: 23 23 + with open(path, "r", encoding="utf-8") as f: 24 24 + return f.read() 25 25 + if ext == ".pdf": 26 26 + if PdfReader is None: 27 27 + raise RuntimeError("pypdf required for PDF support") 28 28 + reader = PdfReader(path) 29 29 + parts = [] 30 30 + for page in reader.pages: 31 31 + text = page.extract_text() or "" 32 32 + parts.append(text) 33 33 + return "\n".join(parts) 34 34 + raise ValueError("unsupported transcript format") 35 35 + 36 36 + 37 37 + def _time_to_seconds(time_str: str) -> int: 38 38 + """Convert HH:MM:SS time string to seconds from midnight.""" 39 39 + h, m, s = map(int, time_str.split(":")) 40 40 + return h * 3600 + m * 60 + s 41 41 + 42 42 + 43 43 + def process_transcript( 44 44 + path: str, 45 45 + day_dir: str, 46 46 + base_dt: dt.datetime, 47 47 + *, 48 48 + import_id: str, 49 49 + stream: str, 50 50 + facet: str | None = None, 51 51 + setting: str | None = None, 52 52 + audio_duration: int | None = None, 53 53 + ) -> list[str]: 54 54 + """Process a transcript file and write imported JSONL segments. 55 55 + 56 56 + Args: 57 57 + path: Path to transcript file 58 58 + day_dir: Journal day directory 59 59 + base_dt: Base datetime for the import 60 60 + import_id: Import identifier 61 61 + stream: Stream name for directory layout (day/stream/segment/) 62 62 + facet: Optional facet name 63 63 + setting: Optional setting description 64 64 + audio_duration: Optional total audio duration in seconds (for last segment) 65 65 + 66 66 + Returns: 67 67 + List of created file paths. 68 68 + """ 69 69 + created_files = [] 70 70 + text = _read_transcript(path) 71 71 + stream_dir = os.path.join(day_dir, stream) 72 72 + 73 73 + # Get start time from base_dt for segmentation 74 74 + start_time = base_dt.strftime("%H:%M:%S") 75 75 + 76 76 + # Get segments with their absolute start times 77 77 + segments = detect_transcript_segment(text, start_time) 78 78 + 79 79 + for idx, (start_at, seg_text) in enumerate(segments): 80 80 + # Convert segment text to structured JSON with absolute timestamps 81 81 + json_data = detect_transcript_json(seg_text, start_at) 82 82 + if not json_data: 83 83 + continue 84 84 + 85 85 + # Parse absolute time for segment directory name 86 86 + time_part = start_at.replace(":", "") # "12:05:30" -> "120530" 87 87 + 88 88 + # Compute segment duration from absolute times 89 89 + start_seconds = _time_to_seconds(start_at) 90 90 + if idx + 1 < len(segments): 91 91 + next_start_at, _ = segments[idx + 1] 92 92 + next_seconds = _time_to_seconds(next_start_at) 93 93 + duration = next_seconds - start_seconds 94 94 + else: 95 95 + # Last segment: use remaining audio duration or default +5s 96 96 + if audio_duration: 97 97 + # audio_duration is total length, start_seconds is time-of-day 98 98 + # Need to calculate offset from recording start 99 99 + recording_start_seconds = _time_to_seconds(start_time) 100 100 + segment_offset = start_seconds - recording_start_seconds 101 101 + duration = audio_duration - segment_offset 102 102 + else: 103 103 + duration = 5 104 104 + 105 105 + # Negative duration indicates corrupted/invalid timestamp data 106 106 + if duration < 0: 107 107 + raise ValueError( 108 108 + f"Invalid segment duration: {duration}s for segment at {time_part}. " 109 109 + "Timestamps may be out of order or audio_duration is incorrect." 110 110 + ) 111 111 + 112 112 + # Ensure minimum duration of 1 second 113 113 + duration = max(1, duration) 114 114 + 115 115 + segment_name = f"{time_part}_{duration}" 116 116 + ts_dir = os.path.join(stream_dir, segment_name) 117 117 + os.makedirs(ts_dir, exist_ok=True) 118 118 + json_path = os.path.join(ts_dir, "imported_audio.jsonl") 119 119 + 120 120 + _write_import_jsonl( 121 121 + json_path, 122 122 + json_data, 123 123 + import_id=import_id, 124 124 + raw_filename=os.path.basename(path), 125 125 + facet=facet, 126 126 + setting=setting, 127 127 + ) 128 128 + logger.info(f"Added transcript segment to journal: {json_path}") 129 129 + created_files.append(json_path) 130 130 + 131 131 + return created_files