a2d46a7c6e
Ruff autofix removed pre-existing F401 unused imports across adapters, context, compiler, extractor, scheduler, web, and main. Also reordered models.py so its SQLAlchemy imports sit at the top of the file (E402 was triggered by the inline `utcnow` helper definition). Added a [dependency-groups] dev block (pytest, pytest-cov, ruff) so the zero-check test gauntlet — which runs `pytest --cov --cov-report=term-missing` — can resolve its plugins without a manual `uv pip install pytest-cov`.
308 lines
10 KiB
Python
308 lines
10 KiB
Python
"""
|
|
second-brain CLI entry point.
|
|
|
|
Commands:
|
|
add — queue a source URL (video or article)
|
|
process — run the full pipeline on pending/pulled/transcribed sources
|
|
serve — launch the FastAPI web UI
|
|
compile — run the wiki compiler on accepted sources
|
|
schedule — start the overnight scheduler
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from typing import Optional
|
|
from urllib.parse import urlparse
|
|
|
|
import click
|
|
|
|
from second_brain.config import DOMAINS, load_config
|
|
from second_brain.database import get_database
|
|
from second_brain.models import Source, SourceStatus, SourceType, utcnow
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# CLI group
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@click.group()
|
|
def cli() -> None:
|
|
"""second-brain — personal knowledge extraction pipeline."""
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# add
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@cli.command()
|
|
@click.argument("url")
|
|
@click.option(
|
|
"--domain",
|
|
"-d",
|
|
type=click.Choice(DOMAINS),
|
|
default="development",
|
|
show_default=True,
|
|
help="Knowledge domain for extraction.",
|
|
)
|
|
@click.option("--focus", "-f", default=None, help="Optional focus directive.")
|
|
@click.option("--title", "-t", default=None, help="Override title.")
|
|
def add(url: str, domain: str, focus: Optional[str], title: Optional[str]) -> None:
|
|
"""Queue a source URL for processing."""
|
|
config = load_config()
|
|
config.ensure_dirs()
|
|
db = get_database()
|
|
|
|
source_type = SourceType.ARTICLE if _is_article(url) else SourceType.VIDEO
|
|
|
|
with db.session() as sess:
|
|
existing = sess.query(Source).filter_by(url=url).first()
|
|
if existing:
|
|
click.echo(
|
|
f"[add] Already queued (id={existing.id}, status={existing.status.value})"
|
|
)
|
|
return
|
|
|
|
source = Source(
|
|
url=url,
|
|
title=title,
|
|
domain=domain,
|
|
focus=focus,
|
|
source_type=source_type,
|
|
status=SourceStatus.PENDING,
|
|
ingested_at=utcnow(),
|
|
)
|
|
sess.add(source)
|
|
|
|
click.echo(f"[add] Queued {source_type.value}: {url}")
|
|
click.echo(f" domain={domain}" + (f" focus={focus}" if focus else ""))
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# process
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@cli.command()
|
|
@click.option("--limit", "-n", default=None, type=int, help="Max sources to process.")
|
|
@click.option(
|
|
"--skip-pull", is_flag=True, default=False, help="Skip the pull/download step."
|
|
)
|
|
@click.option(
|
|
"--skip-transcribe",
|
|
is_flag=True,
|
|
default=False,
|
|
help="Skip the transcription step.",
|
|
)
|
|
@click.option(
|
|
"--skip-extract", is_flag=True, default=False, help="Skip the LLM extraction step."
|
|
)
|
|
def process(
|
|
limit: Optional[int],
|
|
skip_pull: bool,
|
|
skip_transcribe: bool,
|
|
skip_extract: bool,
|
|
) -> None:
|
|
"""Run the pipeline on pending sources."""
|
|
from second_brain.adapters.youtube import YouTubeAdapter
|
|
from second_brain.adapters.article import ArticleAdapter
|
|
from second_brain.context.assembler import ContextAssembler
|
|
from second_brain.extractor.engine import ExtractionEngine
|
|
from second_brain.models import Extraction
|
|
|
|
config = load_config()
|
|
config.ensure_dirs()
|
|
db = get_database()
|
|
|
|
yt_adapter = YouTubeAdapter(config)
|
|
art_adapter = ArticleAdapter(config)
|
|
assembler = ContextAssembler(config)
|
|
|
|
engine: Optional[ExtractionEngine] = None
|
|
if not skip_extract:
|
|
ex_cfg = config.extractor
|
|
try:
|
|
engine = ExtractionEngine(
|
|
backend=ex_cfg["backend"],
|
|
model=ex_cfg["model"],
|
|
api_key=config.anthropic_api_key,
|
|
cli_binary=ex_cfg["cli_binary"],
|
|
timeout_sec=ex_cfg["timeout_sec"],
|
|
)
|
|
except ValueError as exc:
|
|
click.echo(f"[process] Warning: {exc}")
|
|
click.echo("[process] Extraction step will be skipped.")
|
|
skip_extract = True
|
|
|
|
with db.session() as sess:
|
|
query = sess.query(Source.id).filter(
|
|
Source.status.in_([
|
|
SourceStatus.PENDING,
|
|
SourceStatus.PULLED,
|
|
SourceStatus.TRANSCRIBED,
|
|
])
|
|
)
|
|
if limit:
|
|
query = query.limit(limit)
|
|
source_ids = [row[0] for row in query.all()]
|
|
|
|
if not source_ids:
|
|
click.echo("[process] No sources to process.")
|
|
return
|
|
|
|
click.echo(f"[process] Processing {len(source_ids)} source(s)…")
|
|
|
|
for source_id in source_ids:
|
|
with db.session() as sess:
|
|
source = sess.get(Source, source_id)
|
|
click.echo(f"\n [{source.id}] {source.url[:80]}")
|
|
|
|
# Pull step
|
|
if not skip_pull and source.status == SourceStatus.PENDING:
|
|
click.echo(" → pull")
|
|
if source.source_type == SourceType.VIDEO:
|
|
ok = yt_adapter.pull(source)
|
|
else:
|
|
ok = art_adapter.pull(source)
|
|
if not ok:
|
|
click.echo(f" ✗ pull failed: {source.error_message}")
|
|
continue
|
|
|
|
# Transcribe step (videos only — articles come out transcribed)
|
|
if (
|
|
not skip_transcribe
|
|
and source.status == SourceStatus.PULLED
|
|
and source.source_type == SourceType.VIDEO
|
|
):
|
|
click.echo(" → transcribe")
|
|
ok = yt_adapter.transcribe(source)
|
|
if not ok:
|
|
click.echo(f" ✗ transcribe failed: {source.error_message}")
|
|
continue
|
|
|
|
# Extract step
|
|
if not skip_extract and source.status == SourceStatus.TRANSCRIBED and engine:
|
|
click.echo(" → extract")
|
|
try:
|
|
prompt = assembler.assemble(source)
|
|
result = engine.extract(prompt, source)
|
|
|
|
existing_ext = sess.query(Extraction).filter_by(
|
|
source_id=source.id
|
|
).first()
|
|
if existing_ext:
|
|
for k, v in result.to_db_dict().items():
|
|
setattr(existing_ext, k, v)
|
|
existing_ext.raw_response = getattr(result, "_raw_response", None)
|
|
existing_ext.input_tokens = getattr(result, "_input_tokens", None)
|
|
existing_ext.output_tokens = getattr(result, "_output_tokens", None)
|
|
ext_row = existing_ext
|
|
else:
|
|
ext_row = Extraction(
|
|
source_id=source.id,
|
|
raw_response=getattr(result, "_raw_response", None),
|
|
input_tokens=getattr(result, "_input_tokens", None),
|
|
output_tokens=getattr(result, "_output_tokens", None),
|
|
**result.to_db_dict(),
|
|
)
|
|
sess.add(ext_row)
|
|
|
|
source.status = SourceStatus.ANALYZED
|
|
source.updated_at = utcnow()
|
|
# Flush so the new extraction gets its PK before we kick
|
|
# off embedding (we key public.embeddings rows by it).
|
|
sess.flush()
|
|
extraction_id = ext_row.id
|
|
extraction_summary = ext_row.summary
|
|
click.echo(" ✓ analyzed")
|
|
|
|
if config.embeddings.get("enabled", True):
|
|
from second_brain.embeddings import embed_extraction
|
|
|
|
n = embed_extraction(extraction_id, extraction_summary)
|
|
if n is not None:
|
|
click.echo(f" ✓ embedded ({n} chunk(s))")
|
|
except Exception as exc:
|
|
source.error_message = str(exc)
|
|
source.status = SourceStatus.FAILED
|
|
click.echo(f" ✗ extraction failed: {exc}")
|
|
|
|
click.echo("\n[process] Done.")
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# serve
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@cli.command()
|
|
@click.option("--host", default="127.0.0.1", show_default=True)
|
|
@click.option("--port", default=8000, show_default=True, type=int)
|
|
@click.option("--reload", is_flag=True, default=False)
|
|
def serve(host: str, port: int, reload: bool) -> None:
|
|
"""Launch the web UI."""
|
|
import uvicorn
|
|
|
|
click.echo(f"[serve] Starting on http://{host}:{port}")
|
|
uvicorn.run(
|
|
"second_brain.web.app:app",
|
|
host=host,
|
|
port=port,
|
|
reload=reload,
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# compile
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@cli.command("compile")
|
|
def compile_wiki() -> None:
|
|
"""Run the wiki compiler on accepted sources."""
|
|
from second_brain.compiler.wiki import WikiCompiler
|
|
|
|
config = load_config()
|
|
compiler = WikiCompiler(config)
|
|
stats = compiler.run()
|
|
click.echo(
|
|
f"[compile] processed={stats['processed']} "
|
|
f"created={stats['created']} updated={stats['updated']} "
|
|
f"errors={stats['errors']}"
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# schedule
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@cli.command()
|
|
@click.option("--dry-run", is_flag=True, default=False)
|
|
def schedule(dry_run: bool) -> None:
|
|
"""Start the overnight scheduler."""
|
|
from second_brain.scheduler.runner import Scheduler
|
|
|
|
config = load_config()
|
|
scheduler = Scheduler(config)
|
|
scheduler.run(dry_run=dry_run)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Helpers
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def _is_article(url: str) -> bool:
|
|
parsed = urlparse(url)
|
|
video_hosts = {
|
|
"youtube.com", "www.youtube.com", "youtu.be",
|
|
"vimeo.com", "www.vimeo.com",
|
|
}
|
|
return parsed.netloc not in video_hosts
|
|
|
|
|
|
if __name__ == "__main__":
|
|
cli()
|