diff --git a/.gitignore b/.gitignore index 83972fa..741d87a 100644 --- a/.gitignore +++ b/.gitignore @@ -1,6 +1,61 @@ +# Created by https://www.toptal.com/developers/gitignore/api/python,emacs +# Edit at https://www.toptal.com/developers/gitignore?templates=python,emacs + +### Emacs ### +# -*- mode: gitignore; -*- +*~ +\#*\# +/.emacs.desktop +/.emacs.desktop.lock +*.elc +auto-save-list +tramp +.\#* + +# Org-mode +.org-id-locations +*_archive + +# flymake-mode +*_flymake.* + +# eshell files +/eshell/history +/eshell/lastdir + +# elpa packages +/elpa/ + +# reftex files +*.rel + +# AUCTeX auto folder +/auto/ + +# cask packages +.cask/ +dist/ + +# Flycheck +flycheck_*.el + +# server auth directory +/server/ + +# projectiles files +.projectile + +# directory configuration +.dir-locals.el + +# network security +/network-security.data + + +### Python ### # Byte-compiled / optimized / DLL files __pycache__/ -*.py[codz] +*.py[cod] *$py.class # C extensions @@ -10,7 +65,6 @@ __pycache__/ .Python build/ develop-eggs/ -dist/ downloads/ eggs/ .eggs/ @@ -27,8 +81,8 @@ share/python-wheels/ MANIFEST # PyInstaller -# Usually these files are written by a python script from a template -# before PyInstaller builds the exe, so as to inject date/other infos into it. +# Usually these files are written by a python script from a template +# before PyInstaller builds the exe, so as to inject date/other infos into it. *.manifest *.spec @@ -46,7 +100,7 @@ htmlcov/ nosetests.xml coverage.xml *.cover -*.py.cover +*.py,cover .hypothesis/ .pytest_cache/ cover/ @@ -92,37 +146,22 @@ ipython_config.py # However, in case of collaboration, if having platform-specific dependencies or dependencies # having no cross-platform support, pipenv may install dependencies that don't work, or not # install all needed dependencies. -# Pipfile.lock - -# UV -# Similar to Pipfile.lock, it is generally recommended to include uv.lock in version control. -# This is especially recommended for binary packages to ensure reproducibility, and is more -# commonly ignored for libraries. -# uv.lock +#Pipfile.lock # poetry # Similar to Pipfile.lock, it is generally recommended to include poetry.lock in version control. # This is especially recommended for binary packages to ensure reproducibility, and is more # commonly ignored for libraries. # https://python-poetry.org/docs/basic-usage/#commit-your-poetrylock-file-to-version-control -# poetry.lock -# poetry.toml +#poetry.lock # pdm # Similar to Pipfile.lock, it is generally recommended to include pdm.lock in version control. -# pdm recommends including project-wide configuration in pdm.toml, but excluding .pdm-python. -# https://pdm-project.org/en/latest/usage/project/#working-with-version-control -# pdm.lock -# pdm.toml -.pdm-python -.pdm-build/ - -# pixi -# Similar to Pipfile.lock, it is generally recommended to include pixi.lock in version control. -# pixi.lock -# Pixi creates a virtual environment in the .pixi directory, just like venv module creates one -# in the .venv directory. It is recommended not to include this directory in version control. -.pixi +#pdm.lock +# pdm stores project-wide configurations in .pdm.toml, but it is recommended to not include it +# in version control. +# https://pdm.fming.dev/#use-with-ide +.pdm.toml # PEP 582; used by e.g. github.com/David-OConnor/pyflow and github.com/pdm-project/pdm __pypackages__/ @@ -131,25 +170,11 @@ __pypackages__/ celerybeat-schedule celerybeat.pid -# Redis -*.rdb -*.aof -*.pid - -# RabbitMQ -mnesia/ -rabbitmq/ -rabbitmq-data/ - -# ActiveMQ -activemq-data/ - # SageMath parsed files *.sage.py # Environments .env -.envrc .venv env/ venv/ @@ -182,37 +207,22 @@ dmypy.json cython_debug/ # PyCharm -# JetBrains specific template is maintained in a separate JetBrains.gitignore that can -# be found at https://github.com/github/gitignore/blob/main/Global/JetBrains.gitignore -# and can be added to the global gitignore or merged into this file. For a more nuclear -# option (not recommended) you can uncomment the following to ignore the entire idea folder. -# .idea/ - -# Abstra -# Abstra is an AI-powered process automation framework. -# Ignore directories containing user credentials, local state, and settings. -# Learn more at https://abstra.io/docs -.abstra/ - -# Visual Studio Code -# Visual Studio Code specific template is maintained in a separate VisualStudioCode.gitignore -# that can be found at https://github.com/github/gitignore/blob/main/Global/VisualStudioCode.gitignore -# and can be added to the global gitignore or merged into this file. However, if you prefer, -# you could uncomment the following to ignore the entire vscode folder -# .vscode/ -# Temporary file for partial code execution -tempCodeRunnerFile.py - -# Ruff stuff: +# JetBrains specific template is maintained in a separate JetBrains.gitignore that can +# be found at https://github.com/github/gitignore/blob/main/Global/JetBrains.gitignore +# and can be added to the global gitignore or merged into this file. For a more nuclear +# option (not recommended) you can uncomment the following to ignore the entire idea folder. +#.idea/ + +### Python Patch ### +# Poetry local configuration file - https://python-poetry.org/docs/configuration/#local-configuration +poetry.toml + +# ruff .ruff_cache/ -# PyPI configuration file -.pypirc +# LSP config files +pyrightconfig.json -# Marimo -marimo/_static/ -marimo/_lsp/ -__marimo__/ +# End of https://www.toptal.com/developers/gitignore/api/python,emacs -# Streamlit -.streamlit/secrets.toml +config.yaml \ No newline at end of file diff --git a/Dockerfile b/Dockerfile new file mode 100644 index 0000000..09dfce4 --- /dev/null +++ b/Dockerfile @@ -0,0 +1,27 @@ +FROM nvidia/cuda:12.8.2-runtime-ubuntu24.04 + +ENV PYTHONDONTWRITEBYTECODE=1 \ + PYTHONUNBUFFERED=1 \ + PIP_NO_CACHE_DIR=1 \ + DEBIAN_FRONTEND=noninteractive + +WORKDIR /app + +RUN apt-get update \ + && apt-get install -y --no-install-recommends ffmpeg \ + python3 \ + python3-pip \ + && rm -rf /var/lib/apt/lists/* + +COPY requirements.txt ./ +RUN pip install --no-cache-dir --break-system-packages -r requirements.txt + +COPY call_archive ./call_archive +COPY pyproject.toml . +COPY transcribe.py . + +RUN pip install --no-cache-dir --break-system-packages -e . + +ENV PYTHONPATH=/app +ENTRYPOINT ["call-archive"] +CMD ["--help"] diff --git a/README.md b/README.md index a97d9bf..a473177 100644 --- a/README.md +++ b/README.md @@ -1 +1,134 @@ -# call-archive \ No newline at end of file +# call-archive + +Local, privacy-first pipeline for processing phone call recordings: +- transcription (faster-whisper), +- semantic summarization (local LLM), +- categorization, +- retention decision support. + +No data leaves your machine. + +## Context + +This project is designed to work with **BCR (Basic Call Recorder)**. + +Recordings are produced on the phone and synchronized to the computer (e.g. via Syncthing, rsync, etc.). +That synchronization step is **out of scope** for this project. + +The only assumption: +- new recordings (audio + JSON metadata) appear in configured `ingest_dir`. + +## Goals + +* Reduce storage usage of call recordings over time +* Keep **structured knowledge** (transcripts + notes) +* Retain only recordings that have **real value** (legal, business, commitments, disputes, etc.) +* Maintain full **local control and privacy** + +## Directory layout + +```text +/data/calls/ + new/ # incoming recordings (from phone sync) + 2026/ # organized by year (after scan) + rm/ # "deleted" recordings (manual cleanup) + transcripts/ + notes/ + calls.sqlite +``` + +### Lifecycle + +1. Files appear in `new/` +2. `scan` moves them to `{year}/` +3. `process`: + + * transcribes audio + * analyzes content using local LLM + * assigns category + retention proposal +4. `review`: + + * you confirm or override decision +5. `delete-approved`: + + * audio is moved to `rm/` (never hard-deleted) + +## Features + +* Directory-based ingestion model (`ingest_dir/ → storage_dir/{year}/ → storage_dir/rm/`) +* Full local processing (no cloud dependency) +* Structured outputs (JSON + SQLite) +* Extensible **categories** (configured in YAML) +* Per-number retention heuristics +* Safe deletion (only after manual review) + +## Categories + +Categories are defined in `config.yaml`: + +```yaml +categories: + - work + - family + - spam + - friends + - unknown +``` + +The local LLM: + +* always receives the current list +* must choose exactly one category + +You can freely extend this list. + +## Requirements + +* Python 3.11+ +* faster-whisper +* Local LLM (recommended: Ollama) + +## Installation + +```bash +python3 -m venv .venv +. .venv/bin/activate +pip install -e . +``` + +## Usage + +```bash +call-archive init --config config.yaml +call-archive scan --config config.yaml +call-archive process --config config.yaml +call-archive list --config config.yaml +call-archive show --config config.yaml 1 +call-archive review --config config.yaml 1 --decision delete_audio_keep_transcript +call-archive delete-approved --config config.yaml +``` + +## Safety model + +* No automatic deletion +* All decisions require explicit review +* Audio is moved to `rm/`, not deleted +* Full audit trail in SQLite + +## Intended use + +This is useful if: + +* you record calls for **memory / accountability** +* you want **automatic summarization** +* you want to **control storage growth** +* you need to keep everything **local** + +## Non-goals + +* Call recording itself (handled by BCR) +* File synchronization (external tools) +* Cloud integrations + +## License +GPL-3.0 license diff --git a/call_archive/__init__.py b/call_archive/__init__.py new file mode 100644 index 0000000..a05eb9a --- /dev/null +++ b/call_archive/__init__.py @@ -0,0 +1,3 @@ +__all__ = ["__version__"] + +__version__ = "0.1.0" diff --git a/call_archive/cli.py b/call_archive/cli.py new file mode 100644 index 0000000..c87cc90 --- /dev/null +++ b/call_archive/cli.py @@ -0,0 +1,426 @@ +from __future__ import annotations + +import argparse +import json +import sqlite3 +import time +from datetime import datetime, timezone +from pathlib import Path +from typing import Any + +from .config import AppConfig, load_config, normalize_phone_number +from .db import connect, init_db, row_to_dict +from .files import ( + AUDIO_SUFFIXES, + ensure_layout, + first_call, + load_metadata, + metadata_phone_number, + metadata_year, + move_audio_to_rm, + move_pair_to_year, + sha256_file, + sibling_metadata_path, +) +from .llm import analyze_with_ollama, build_prompt +from .models import ProcessingStatus, RetentionDecision, ReviewStatus +from .transcriber import run_transcription + + +def utc_now() -> str: + return datetime.now(timezone.utc).isoformat() + + +def build_parser() -> argparse.ArgumentParser: + parser = argparse.ArgumentParser(prog="call-archive") + parser.add_argument("--config", type=Path, default=Path("config.yaml")) + + subparsers = parser.add_subparsers(dest="command", required=True) + + subparsers.add_parser("init") + subparsers.add_parser("scan") + + process_parser = subparsers.add_parser("process") + process_parser.add_argument("--limit", type=int, default=0) + + list_parser = subparsers.add_parser("list") + list_parser.add_argument("--status", choices=["pending", "reviewed"], default=None) + list_parser.add_argument("--category", default=None) + + show_parser = subparsers.add_parser("show") + show_parser.add_argument("call_id", type=int) + + review_parser = subparsers.add_parser("review") + review_parser.add_argument("call_id", type=int) + review_parser.add_argument( + "--decision", + choices=[decision.value for decision in RetentionDecision], + required=True, + ) + + subparsers.add_parser("delete-approved") + + return parser + + +def command_init(config: AppConfig) -> int: + ensure_layout(config.ingest_dir, config.storage_dir, config.transcripts_dir, config.notes_dir) + init_db(config.database_path) + return 0 + + +def command_scan(config: AppConfig) -> int: + ensure_layout(config.ingest_dir, config.storage_dir, config.transcripts_dir, config.notes_dir) + init_db(config.database_path) + + audio_files = sorted(path for path in config.ingest_dir.iterdir() if path.suffix.lower() in AUDIO_SUFFIXES) + + with connect(config.database_path) as connection: + for audio_path in audio_files: + metadata_path = sibling_metadata_path(audio_path) + if not metadata_path.exists(): + print(f"SKIP missing metadata: {audio_path}") + continue + + try: + metadata = load_metadata(metadata_path) + except (json.JSONDecodeError, OSError, ValueError) as error: + print(f"WARNING invalid metadata JSON: {metadata_path} ({error})") + continue + + target_audio, target_metadata = move_pair_to_year(audio_path, config.storage_dir, metadata) + if not target_audio.exists() or not target_metadata.exists(): + print(f"SKIP incomplete move: {audio_path}") + continue + + try: + metadata = load_metadata(target_metadata) + except (json.JSONDecodeError, OSError, ValueError) as error: + print(f"WARNING invalid metadata JSON after move: {target_metadata} ({error})") + continue + + call = first_call(metadata) + output = metadata.get("output", {}) + recording = output.get("recording", {}) if isinstance(output, dict) else {} + duration = recording.get("duration_secs_total") if isinstance(recording, dict) else None + phone_number = metadata_phone_number(metadata) + phone_number_formatted = call.get("phone_number_formatted") + contact_name = call.get("contact_name") or call.get("caller_name") + timestamp = str(metadata.get("timestamp", "")) + year = metadata_year(metadata) + now = utc_now() + + connection.execute( + """ + INSERT OR IGNORE INTO calls ( + audio_path, metadata_path, storage_stem, + phone_number, phone_number_formatted, contact_name, + direction, timestamp, year, duration_secs, + audio_sha256, metadata_sha256, + transcription_status, analysis_status, + review_status, category, + created_at, updated_at + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + """, + ( + str(target_audio), + str(target_metadata), + target_audio.stem, + phone_number, + str(phone_number_formatted) if phone_number_formatted is not None else None, + str(contact_name) if contact_name is not None else None, + str(metadata.get("direction", "")), + timestamp, + year, + float(duration) if isinstance(duration, int | float) else None, + sha256_file(target_audio), + sha256_file(target_metadata), + ProcessingStatus.NEW.value, + ProcessingStatus.NEW.value, + ReviewStatus.PENDING.value, + "unknown", + now, + now, + ), + ) + + connection.execute( + """ + UPDATE calls + SET audio_path = ?, metadata_path = ?, updated_at = ? + WHERE audio_path IN (?, ?) OR metadata_path IN (?, ?) + """, + ( + str(target_audio), + str(target_metadata), + now, + str(audio_path), + str(target_audio), + str(sibling_metadata_path(audio_path)), + str(target_metadata), + ), + ) + print(f"SCANNED {target_audio}") + + return 0 + + +def command_process(config: AppConfig, limit: int) -> int: + ensure_layout(config.ingest_dir, config.storage_dir, config.transcripts_dir, config.notes_dir) + init_db(config.database_path) + + with connect(config.database_path) as connection: + query = """ + SELECT * FROM calls + WHERE transcription_status != ? OR analysis_status != ? + ORDER BY timestamp ASC + """ + parameters: list[Any] = [ProcessingStatus.DONE.value, ProcessingStatus.DONE.value] + if limit > 0: + query += " LIMIT ?" + parameters.append(limit) + + rows = connection.execute(query, parameters).fetchall() + + for row in rows: + call = row_to_dict(row) + call_id = int(call["id"]) + audio_path = Path(str(call["audio_path"])) + metadata_path = Path(str(call["metadata_path"])) + now = utc_now() + + try: + started_at = time.perf_counter() + storage_stem = str(call["storage_stem"] or audio_path.stem) + transcript_path = config.transcripts_dir / f"{storage_stem}.txt" + if call["transcription_status"] != ProcessingStatus.DONE.value: + generated_transcript_path = run_transcription(audio_path, config.transcription) + transcript_path.write_text( + generated_transcript_path.read_text(encoding="utf-8"), + encoding="utf-8", + ) + connection.execute( + """ + UPDATE calls + SET storage_stem = ?, transcription_status = ?, updated_at = ?, error = NULL + WHERE id = ? + """, + (storage_stem, ProcessingStatus.DONE.value, now, call_id), + ) + + if transcript_path is None: + raise ValueError("Missing transcript path.") + + metadata = load_metadata(metadata_path) + transcript = transcript_path.read_text(encoding="utf-8") + prompt = build_prompt( + metadata=metadata, + transcript=transcript, + categories=config.categories, + ) + note = analyze_with_ollama( + config=config.llm, + prompt=prompt, + categories=config.categories, + ) + + note_path = config.notes_dir / f"{storage_stem}.txt" + note_path.write_text( + note.model_dump_json(indent=2), + encoding="utf-8", + ) + + connection.execute( + """ + UPDATE calls + SET storage_stem = ?, + analysis_status = ?, + proposed_retention = ?, + category = ?, + reason = ?, + updated_at = ?, + error = NULL + WHERE id = ? + """, + ( + storage_stem, + ProcessingStatus.DONE.value, + note.recommended_retention.value, + note.category, + note.reason, + now, + call_id, + ), + ) + elapsed_secs = time.perf_counter() - started_at + print(f"PROCESSED #{call_id} {audio_path.name} in {elapsed_secs:.2f}s") + + except Exception as exception: + connection.execute( + """ + UPDATE calls + SET error = ?, updated_at = ?, + transcription_status = CASE + WHEN transcription_status = ? THEN transcription_status + ELSE ? + END, + analysis_status = ? + WHERE id = ? + """, + ( + repr(exception), + now, + ProcessingStatus.DONE.value, + ProcessingStatus.FAILED.value, + ProcessingStatus.FAILED.value, + call_id, + ), + ) + print(f"FAILED #{call_id}: {exception}") + + return 0 + + +def command_list(config: AppConfig, status: str | None, category: str | None) -> int: + init_db(config.database_path) + clauses: list[str] = [] + parameters: list[Any] = [] + + if status is not None: + clauses.append("review_status = ?") + parameters.append(status) + if category is not None: + clauses.append("category = ?") + parameters.append(category) + + query = "SELECT * FROM calls" + if clauses: + query += " WHERE " + " AND ".join(clauses) + query += " ORDER BY timestamp DESC" + + with connect(config.database_path) as connection: + rows = connection.execute(query, parameters).fetchall() + for row in rows: + call = row_to_dict(row) + print( + f"#{call['id']} {call['timestamp']} " + f"{call['phone_number_formatted'] or call['phone_number']} " + f"cat={call['category']} proposed={call['proposed_retention']} " + f"review={call['review_status']} audio={call['audio_path']}" + ) + return 0 + + +def command_show(config: AppConfig, call_id: int) -> int: + init_db(config.database_path) + with connect(config.database_path) as connection: + row = connection.execute("SELECT * FROM calls WHERE id = ?", (call_id,)).fetchone() + if row is None: + raise ValueError(f"Call not found: {call_id}") + + call = row_to_dict(row) + print(json.dumps(call, ensure_ascii=False, indent=2)) + + note_path_raw = call.get("note_path") + storage_stem = call.get("storage_stem") + if isinstance(storage_stem, str) and storage_stem: + note_path = config.notes_dir / f"{storage_stem}.txt" + if note_path.exists(): + print("\nNOTE:") + print(note_path.read_text(encoding="utf-8")) + + return 0 + + +def command_review(config: AppConfig, call_id: int, decision: str) -> int: + init_db(config.database_path) + with connect(config.database_path) as connection: + row = connection.execute("SELECT id FROM calls WHERE id = ?", (call_id,)).fetchone() + if row is None: + raise ValueError(f"Call not found: {call_id}") + + connection.execute( + """ + UPDATE calls + SET reviewed_retention = ?, review_status = ?, updated_at = ? + WHERE id = ? + """, + (decision, ReviewStatus.REVIEWED.value, utc_now(), call_id), + ) + print(f"REVIEWED #{call_id}: {decision}") + return 0 + + +def command_delete_approved(config: AppConfig) -> int: + init_db(config.database_path) + with connect(config.database_path) as connection: + rows = connection.execute( + """ + SELECT * FROM calls + WHERE review_status = ? + ORDER BY timestamp ASC + """, + (ReviewStatus.REVIEWED.value,), + ).fetchall() + + for row in rows: + call = row_to_dict(row) + reviewed_retention = str(call.get("reviewed_retention") or "") + phone_number = normalize_phone_number(str(call.get("phone_number") or "")) + is_normally_deleted = phone_number in config.normalized_delete_numbers() + + should_delete = ( + reviewed_retention == RetentionDecision.DELETE.value + or ( + reviewed_retention == RetentionDecision.DEFAULT.value + and is_normally_deleted + ) + ) + if not should_delete: + continue + + audio_path = Path(str(call["audio_path"])) + if not audio_path.exists(): + print(f"SKIP missing audio #{call['id']}: {audio_path}") + continue + + target = move_audio_to_rm(audio_path, config.storage_dir) + connection.execute( + """ + UPDATE calls + SET audio_path = ?, updated_at = ? + WHERE id = ? + """, + (str(target), utc_now(), int(call["id"])), + ) + print(f"MOVED_TO_RM #{call['id']}: {target}") + + return 0 + + +def main() -> int: + parser = build_parser() + args = parser.parse_args() + config = load_config(args.config) + + if args.command == "init": + return command_init(config) + if args.command == "scan": + return command_scan(config) + if args.command == "process": + return command_process(config, int(args.limit)) + if args.command == "list": + return command_list(config, args.status, args.category) + if args.command == "show": + return command_show(config, int(args.call_id)) + if args.command == "review": + return command_review(config, int(args.call_id), str(args.decision)) + if args.command == "delete-approved": + return command_delete_approved(config) + + parser.error(f"Unknown command: {args.command}") + return 2 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/call_archive/config.py b/call_archive/config.py new file mode 100644 index 0000000..da94c0d --- /dev/null +++ b/call_archive/config.py @@ -0,0 +1,62 @@ +from __future__ import annotations + +from pathlib import Path +from typing import Any + +import yaml +from pydantic import BaseModel, Field + + +class TranscriptionConfig(BaseModel): + module: str = "transcribe" + model: str = "medium" + device: str = "cpu" + compute_type: str = "int8" + beam_size: int = 5 + vad_filter: bool = True + timestamps: bool = False + + +class LlmConfig(BaseModel): + provider: str = "ollama" + base_url: str = "http://127.0.0.1:11434" + model: str + timeout_secs: int = 180 + + +class RetentionConfig(BaseModel): + normally_delete_numbers: list[str] = Field(default_factory=list) + + +class AppConfig(BaseModel): + ingest_dir: Path + storage_dir: Path + database_path: Path + transcripts_dir: Path + notes_dir: Path + transcription: TranscriptionConfig + llm: LlmConfig + categories: list[str] + retention: RetentionConfig + + def normalized_delete_numbers(self) -> set[str]: + return {normalize_phone_number(number) for number in self.retention.normally_delete_numbers} + + +def normalize_phone_number(value: str | None) -> str: + if value is None: + return "" + return "".join(character for character in value if character.isdigit()) + + +def load_config(path: Path) -> AppConfig: + data: Any + with path.open("rt", encoding="utf-8") as handle: + data = yaml.safe_load(handle) + + config = AppConfig.model_validate(data) + if not config.categories: + raise ValueError("Config must define at least one category.") + if "unknown" not in config.categories: + raise ValueError("Config categories must include 'unknown'.") + return config diff --git a/call_archive/db.py b/call_archive/db.py new file mode 100644 index 0000000..6064f0e --- /dev/null +++ b/call_archive/db.py @@ -0,0 +1,70 @@ +from __future__ import annotations + +import sqlite3 +from collections.abc import Iterator +from contextlib import contextmanager +from pathlib import Path +from typing import Any + + +SCHEMA: str = """ +CREATE TABLE IF NOT EXISTS calls ( + id INTEGER PRIMARY KEY, + audio_path TEXT UNIQUE NOT NULL, + metadata_path TEXT NOT NULL, + storage_stem TEXT, + + phone_number TEXT, + phone_number_formatted TEXT, + contact_name TEXT, + direction TEXT, + timestamp TEXT, + year TEXT, + duration_secs REAL, + + audio_sha256 TEXT NOT NULL, + metadata_sha256 TEXT NOT NULL, + + transcription_status TEXT NOT NULL, + analysis_status TEXT NOT NULL, + proposed_retention TEXT, + reviewed_retention TEXT, + review_status TEXT NOT NULL, + + category TEXT NOT NULL DEFAULT 'unknown', + reason TEXT, + error TEXT, + + created_at TEXT NOT NULL, + updated_at TEXT NOT NULL +); + +CREATE INDEX IF NOT EXISTS idx_calls_review_status ON calls(review_status); +CREATE INDEX IF NOT EXISTS idx_calls_phone_number ON calls(phone_number); +CREATE INDEX IF NOT EXISTS idx_calls_category ON calls(category); +CREATE INDEX IF NOT EXISTS idx_calls_timestamp ON calls(timestamp); +""" + + +@contextmanager +def connect(database_path: Path) -> Iterator[sqlite3.Connection]: + database_path.parent.mkdir(parents=True, exist_ok=True) + connection = sqlite3.connect(database_path) + connection.row_factory = sqlite3.Row + try: + yield connection + connection.commit() + finally: + connection.close() + + +def init_db(database_path: Path) -> None: + with connect(database_path) as connection: + connection.executescript(SCHEMA) + columns = {row[1] for row in connection.execute("PRAGMA table_info(calls)").fetchall()} + if "storage_stem" not in columns: + connection.execute("ALTER TABLE calls ADD COLUMN storage_stem TEXT") + + +def row_to_dict(row: sqlite3.Row) -> dict[str, Any]: + return {key: row[key] for key in row.keys()} diff --git a/call_archive/files.py b/call_archive/files.py new file mode 100644 index 0000000..be21a00 --- /dev/null +++ b/call_archive/files.py @@ -0,0 +1,115 @@ +from __future__ import annotations + +import hashlib +import json +import shutil +from datetime import datetime +from pathlib import Path +from typing import Any + +from .config import normalize_phone_number + + +AUDIO_SUFFIXES: set[str] = {".oga", ".ogg", ".opus", ".m4a", ".mp3", ".wav", ".flac"} + + +def ensure_layout(ingest_dir: Path, storage_dir: Path, transcripts_dir: Path, notes_dir: Path) -> None: + for directory in [ + ingest_dir, + storage_dir, + storage_dir / "rm", + transcripts_dir, + notes_dir, + ]: + directory.mkdir(parents=True, exist_ok=True) + + +def sha256_file(path: Path) -> str: + digest = hashlib.sha256() + with path.open("rb") as handle: + while True: + chunk = handle.read(1024 * 1024) + if not chunk: + break + digest.update(chunk) + return digest.hexdigest() + + +def load_metadata(path: Path) -> dict[str, Any]: + with path.open("rt", encoding="utf-8") as handle: + data: Any = json.load(handle) + if not isinstance(data, dict): + raise ValueError(f"Metadata is not an object: {path}") + return data + + +def metadata_year(metadata: dict[str, Any]) -> str: + timestamp = str(metadata.get("timestamp", "")) + if len(timestamp) >= 4 and timestamp[:4].isdigit(): + return timestamp[:4] + unix_ms = metadata.get("timestamp_unix_ms") + if isinstance(unix_ms, int): + return datetime.fromtimestamp(unix_ms / 1000).strftime("%Y") + return datetime.now().strftime("%Y") + + +def first_call(metadata: dict[str, Any]) -> dict[str, Any]: + calls = metadata.get("calls") + if isinstance(calls, list) and calls and isinstance(calls[0], dict): + return calls[0] + return {} + + +def metadata_phone_number(metadata: dict[str, Any]) -> str: + call = first_call(metadata) + raw = call.get("phone_number") + return normalize_phone_number(str(raw)) if raw is not None else "" + + +def sibling_metadata_path(audio_path: Path) -> Path: + return audio_path.with_suffix(".json") + + +def transcript_path_for_audio(audio_path: Path) -> Path: + return audio_path.with_suffix(".txt") + + +def move_pair_to_year(audio_path: Path, storage_dir: Path, metadata: dict[str, Any]) -> tuple[Path, Path]: + year = metadata_year(metadata) + target_dir = storage_dir / year + target_dir.mkdir(parents=True, exist_ok=True) + + metadata_path = sibling_metadata_path(audio_path) + target_audio = unique_target(target_dir / audio_path.name) + target_metadata = target_audio.with_suffix(".json") + + if audio_path.exists(): + shutil.move(str(audio_path), str(target_audio)) + if metadata_path.exists(): + shutil.move(str(metadata_path), str(target_metadata)) + + return target_audio, target_metadata + + +def unique_target(path: Path) -> Path: + if not path.exists(): + return path + + stem = path.stem + suffix = path.suffix + parent = path.parent + + counter = 1 + while True: + candidate = parent / f"{stem}.{counter}{suffix}" + if not candidate.exists(): + return candidate + counter += 1 + + +def move_audio_to_rm(audio_path: Path, storage_dir: Path) -> Path: + rm_dir = storage_dir / "rm" + rm_dir.mkdir(parents=True, exist_ok=True) + target = unique_target(rm_dir / audio_path.name) + shutil.move(str(audio_path), str(target)) + return target diff --git a/call_archive/llm.py b/call_archive/llm.py new file mode 100644 index 0000000..249f61e --- /dev/null +++ b/call_archive/llm.py @@ -0,0 +1,144 @@ +from __future__ import annotations + +import json +from typing import Any + +import requests + +from .config import LlmConfig +from .models import CallNote + + +def build_prompt( + *, + metadata: dict[str, Any], + transcript: str, + categories: list[str], +) -> str: + return f""" +Jesteś lokalnym asystentem do analizy prywatnych transkrypcji rozmów telefonicznych. +Nie wolno Ci zakładać informacji, których nie ma w metadanych lub transkrypcji. +Masz zwrócić wyłącznie poprawny JSON zgodny ze schematem. + +Dostępne kategorie: +{json.dumps(categories, ensure_ascii=False)} + +Wybierz dokładnie jedną kategorię z powyższej listy. + +Zasady retencji: +- Jeżeli rozmowa zawiera istotne ustalenia, spór, zobowiązania, kwestie prawne, finansowe, techniczne, reklamacyjne albo dowodowe i ma być na pewno zachowana niezależnie od innych reguł, rekomenduj keep. +- Jeżeli rozmowa jest nieistotna, spamowa, pomyłkowa, pusta albo czysto organizacyjna bez wartości dowodowej i ma być usunięta niezależnie od innych reguł, rekomenduj delete. +- Jeżeli decyzja ma zostać podjęta wg domyślnej polityki systemu, rekomenduj default. +- Jeżeli transkrypcja jest zbyt słaba albo nie da się ocenić treści, rekomenduj review. + +Dozwolone wartości recommended_retention: +- default +- keep +- delete +- review + +Dozwolone wartości importance: +- low +- medium +- high +- unknown + +Wymagany JSON: +{{ + "participants": ["..."], + "caller_or_contact": "...", + "category": "...", + "topic": "...", + "summary": "...", + "agreements": ["..."], + "action_items": ["..."], + "importance": "low|medium|high|unknown", + "contains_sensitive_or_legal_content": false, + "recommended_retention": "default|keep|delete|review", + "reason": "..." +}} + +Metadane: +{json.dumps(metadata, ensure_ascii=False, indent=2)} + +Transkrypcja: +{transcript} +""".strip() + + +def analyze_with_ollama( + *, + config: LlmConfig, + prompt: str, + categories: list[str], +) -> CallNote: + if config.provider != "ollama": + raise ValueError(f"Unsupported LLM provider: {config.provider}") + + schema: dict[str, Any] = { + "type": "object", + "properties": { + "participants": {"type": "array", "items": {"type": "string"}}, + "caller_or_contact": {"type": "string"}, + "category": {"type": "string", "enum": categories}, + "topic": {"type": "string"}, + "summary": {"type": "string"}, + "agreements": {"type": "array", "items": {"type": "string"}}, + "action_items": {"type": "array", "items": {"type": "string"}}, + "importance": {"type": "string", "enum": ["low", "medium", "high", "unknown"]}, + "contains_sensitive_or_legal_content": {"type": "boolean"}, + "recommended_retention": { + "type": "string", + "enum": [ + "default", + "keep", + "delete", + "review", + ], + }, + "reason": {"type": "string"}, + }, + "required": [ + "participants", + "caller_or_contact", + "category", + "topic", + "summary", + "agreements", + "action_items", + "importance", + "contains_sensitive_or_legal_content", + "recommended_retention", + "reason", + ], + "additionalProperties": False, + } + + response = requests.post( + f"{config.base_url.rstrip('/')}/api/generate", + json={ + "model": config.model, + "prompt": prompt, + "stream": False, + "format": schema, + "think": False, + "keep_alive": "10m", + }, + timeout=config.timeout_secs, + ) + response.raise_for_status() + + response_data: Any = response.json() + raw = response_data.get("response") + if not isinstance(raw, str): + raise ValueError("Ollama response does not contain string field 'response'.") + + try: + parsed: Any = json.loads(raw) + except json.decoder.JSONDecodeError: + print("Call JSON:", repr(response_data)) + raise + note = CallNote.model_validate(parsed) + if note.category not in categories: + note.category = "unknown" + return note diff --git a/call_archive/models.py b/call_archive/models.py new file mode 100644 index 0000000..7e4bccd --- /dev/null +++ b/call_archive/models.py @@ -0,0 +1,45 @@ +from __future__ import annotations + +from enum import StrEnum + +from pydantic import BaseModel, Field, field_validator + + +class RetentionDecision(StrEnum): + DEFAULT = "default" + KEEP = "keep" + DELETE = "delete" + REVIEW = "review" + + +class ReviewStatus(StrEnum): + PENDING = "pending" + REVIEWED = "reviewed" + + +class ProcessingStatus(StrEnum): + NEW = "new" + DONE = "done" + FAILED = "failed" + + +class CallNote(BaseModel): + participants: list[str] = Field(default_factory=list) + caller_or_contact: str = "unknown" + category: str + topic: str + summary: str + agreements: list[str] = Field(default_factory=list) + action_items: list[str] = Field(default_factory=list) + importance: str = "unknown" + contains_sensitive_or_legal_content: bool = False + recommended_retention: RetentionDecision + reason: str + + @field_validator("importance") + @classmethod + def validate_importance(cls, value: str) -> str: + allowed: set[str] = {"low", "medium", "high", "unknown"} + if value not in allowed: + return "unknown" + return value diff --git a/call_archive/transcriber.py b/call_archive/transcriber.py new file mode 100644 index 0000000..ec6e6c1 --- /dev/null +++ b/call_archive/transcriber.py @@ -0,0 +1,45 @@ +from __future__ import annotations + +import importlib +from collections.abc import Callable +from pathlib import Path +from typing import Any, Protocol + +from .config import TranscriptionConfig + + +class TranscribeFile(Protocol): + def __call__( + self, + input_file: Path, + *, + model_name: str, + device: str, + compute_type: str, + beam_size: int, + vad_filter: bool, + timestamps: bool, + ) -> Any: ... + + +def run_transcription(input_file: Path, config: TranscriptionConfig) -> Path: + module = importlib.import_module(config.module) + transcribe_file_any = getattr(module, "transcribe_file") + transcribe_file: TranscribeFile = transcribe_file_any + + transcribe_file( + input_file, + model_name=config.model, + device=config.device, + compute_type=config.compute_type, + beam_size=config.beam_size, + vad_filter=config.vad_filter, + timestamps=config.timestamps, + ) + + transcript_path = input_file.with_suffix(".txt") + if not transcript_path.exists(): + raise FileNotFoundError( + f"Transcription finished, but expected transcript was not created: {transcript_path}" + ) + return transcript_path diff --git a/config.example.yaml b/config.example.yaml new file mode 100644 index 0000000..a5d31af --- /dev/null +++ b/config.example.yaml @@ -0,0 +1,32 @@ +ingest_dir: /data/calls/new +storage_dir: /data/calls +database_path: /data/calls/calls.sqlite +transcripts_dir: /data/calls/transcripts +notes_dir: /data/calls/notes + +transcription: + module: transcribe + model: medium + device: cpu + compute_type: int8 + beam_size: 5 + vad_filter: true + timestamps: false + +llm: + provider: ollama + base_url: http://127.0.0.1:11434 + model: qwen2.5:14b-instruct + timeout_secs: 180 + +categories: + - work + - family + - spam + - friends + - unknown + +retention: + normally_delete_numbers: + - "123456789" + - "+48123456789" diff --git a/docker-compose.yml b/docker-compose.yml new file mode 100644 index 0000000..5c6e5ea --- /dev/null +++ b/docker-compose.yml @@ -0,0 +1,9 @@ +services: + call-archive: + image: call-archive + working_dir: /app + volumes: + - ./config.yaml:/app/config.yaml:ro + - /mnt/Mondoshawan/mcanswer/Tachikoma/recordedCalls:/data/calls/new + - /mnt/Mondoshawan/mcanswer/RecordedCalls:/data/calls + command: ["list", "--config", "/app/config.yaml"] diff --git a/pyproject.toml b/pyproject.toml new file mode 100644 index 0000000..31b7a12 --- /dev/null +++ b/pyproject.toml @@ -0,0 +1,30 @@ +[project] +name = "call-archive" +version = "0.1.0" +description = "Local call recording transcription and retention workflow" +requires-python = ">=3.11" +dependencies = [ + "pydantic>=2.7", + "pyyaml>=6.0", + "requests>=2.31", +] + +[project.scripts] +call-archive = "call_archive.cli:main" + +[tool.mypy] +python_version = "3.12" +strict = true +warn_unused_configs = true +warn_return_any = true +disallow_any_generics = true +disallow_subclassing_any = true +disallow_untyped_defs = true +disallow_incomplete_defs = true +check_untyped_defs = true +disallow_untyped_decorators = true +no_implicit_optional = true +warn_redundant_casts = true +warn_unused_ignores = true +warn_no_return = true +show_error_codes = true diff --git a/requirements.txt b/requirements.txt new file mode 100644 index 0000000..03a9701 --- /dev/null +++ b/requirements.txt @@ -0,0 +1,4 @@ +pydantic>=2.7 +pyyaml>=6.0 +requests>=2.31 +faster-whisper diff --git a/transcribe.py b/transcribe.py new file mode 100644 index 0000000..84ae831 --- /dev/null +++ b/transcribe.py @@ -0,0 +1,130 @@ +from __future__ import annotations + +import argparse +from pathlib import Path +from typing import Iterable + +from faster_whisper import WhisperModel +from faster_whisper.transcribe import Segment + + +def build_parser() -> argparse.ArgumentParser: + parser = argparse.ArgumentParser(description="Transcribe audio/video with faster-whisper.") + parser.add_argument( + "input_file", + type=Path, + help="Path to the input audio or video file.", + ) + parser.add_argument( + "--model", + default="small", + help="Whisper model name, e.g. tiny, base, small, medium, large-v3. Default: small.", + ) + parser.add_argument( + "--device", + default="cuda", + choices=["cpu", "cuda", "auto"], + help="Execution device. Default: cuda.", + ) + parser.add_argument( + "--compute-type", + default="float32", + help="CTranslate2 compute type. On GTX 1080 Ti, float32 is the safe default.", + ) + parser.add_argument( + "--beam-size", + type=int, + default=5, + help="Beam size for decoding. Default: 5.", + ) + parser.add_argument( + "--timestamps", + action="store_true", + help="Print segment start/end timestamps before text.", + ) + parser.add_argument( + "--vad-filter", + action="store_true", + help="Enable voice activity detection filter.", + ) + return parser + + +def format_segment(segment: Segment, include_timestamps: bool) -> str: + if include_timestamps: + return f"[{segment.start:.2f} -> {segment.end:.2f}] {segment.text.strip()}" + return segment.text.strip() + + +def iter_output_lines(segments: Iterable[Segment], include_timestamps: bool) -> Iterable[str]: + for segment in segments: + text: str = format_segment(segment, include_timestamps) + if text: + yield text + + +def transcribe_file( + input_file: Path, + *, + model_name: str, + device: str, + compute_type: str, + beam_size: int, + vad_filter: bool, + timestamps: bool, +) -> Path: + output_file: Path = input_file.with_suffix(".txt") + + model = WhisperModel( + model_name, + device=device, + compute_type=compute_type, + download_root="/data/models/faster-whisper", + ) + + transcribe_kwargs: dict[str, object] = { + "beam_size": beam_size, + "condition_on_previous_text": False, + } + + if vad_filter: + transcribe_kwargs["vad_filter"] = True + + segments, info = model.transcribe(str(input_file), **transcribe_kwargs) + + with output_file.open("wt", encoding="utf-8") as f: + f.write( + f"# language={info.language} " + f"probability={info.language_probability:.3f}\n" + ) + + for line in iter_output_lines(segments, timestamps): + f.write(f"{line}\n") + + return output_file + + +def main() -> int: + parser = build_parser() + args = parser.parse_args() + + input_file: Path = args.input_file + + if not input_file.exists(): + parser.error(f"Input file does not exist: {input_file}") + + transcribe_file( + input_file, + model_name=args.model, + device=args.device, + compute_type=args.compute_type, + beam_size=args.beam_size, + vad_filter=args.vad_filter, + timestamps=args.timestamps, + ) + + return 0 + + +if __name__ == "__main__": + raise SystemExit(main())