2f26f2d836
- onboard-client.py: client replica scaffolding CLI - gws/: Google Workspace sync (Gmail, Calendar, Drive) - lifestream/: life event stream collector - muscriptor-mcp/: audio → MIDI MCP server - music-mcp/: music library MCP server - data_sharing/: consent-gated data sharing (Python + TS) - sync-mirrors.py: GitHub → Forgejo mirror engine - brain-to-gbrain.py, vault-sync.py, test-all.sh - shared/: TS data-sharing library + index - dirac: provider registry update - .gitignore: exclude Rust build artifacts Co-authored-by: Álvaro de Campos <campos@portugalfuturista.org>
92 lines
3.3 KiB
Python
92 lines
3.3 KiB
Python
#!/usr/bin/env python3
|
|
"""Standalone runner for MuScriptor weekly MIDI conversion.
|
|
|
|
Runs outside the MCP server context — called by Hermes cron job.
|
|
Fetches top tracks from Maloja, downloads from Navidrome, transcribes to MIDI.
|
|
|
|
Usage:
|
|
python3 run_weekly_midi.py [--count 3] [--period week] [--model medium]
|
|
"""
|
|
import argparse
|
|
import asyncio
|
|
import os
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
# Add the package to path
|
|
sys.path.insert(0, str(Path(__file__).parent / "src"))
|
|
|
|
# Load .env if present
|
|
env_file = Path(__file__).parent / ".env"
|
|
if env_file.exists():
|
|
for line in env_file.read_text().splitlines():
|
|
line = line.strip()
|
|
if line and not line.startswith("#") and "=" in line:
|
|
key, _, value = line.partition("=")
|
|
os.environ.setdefault(key.strip(), value.strip())
|
|
|
|
from muscriptor_mcp.maloja_client import get_top_tracks
|
|
from muscriptor_mcp.navidrome_client import search_track, download_track
|
|
from muscriptor_mcp.transcriber import transcribe
|
|
|
|
|
|
async def run(count: int = 3, period: str = "week", model_size: str = "medium") -> str:
|
|
"""Run the weekly MIDI conversion pipeline. Returns a formatted report."""
|
|
# Step 1: Get top tracks from Maloja
|
|
tracks = await get_top_tracks(count=count, period=period)
|
|
if not tracks:
|
|
return f"No scrobble data found for period '{period}'."
|
|
|
|
results = []
|
|
success = 0
|
|
dl_dir = Path(os.environ.get("MUSSCRIPTOR_DOWNLOAD_DIR", "/tmp/muscriptor")) / f"weekly-{period}"
|
|
|
|
for t in tracks:
|
|
label = f"{t.artist} — {t.title}"
|
|
try:
|
|
# Step 2: Search Navidrome
|
|
query = f"{t.title} {t.artist}"
|
|
search_results = await search_track(query)
|
|
if not search_results:
|
|
results.append(f"⏭️ {label} — not found in library")
|
|
continue
|
|
|
|
# Step 3: Download audio
|
|
audio_path = await download_track(search_results[0]["id"], dl_dir)
|
|
|
|
# Step 4: Transcribe to MIDI
|
|
midi_path = await transcribe(audio_path=audio_path, model_size=model_size)
|
|
results.append(f"✅ {label} → {midi_path}")
|
|
success += 1
|
|
except Exception as e:
|
|
results.append(f"❌ {label} — {e}")
|
|
|
|
report = f"🎵 Weekly MIDI Report ({period})\n"
|
|
report += f"Converted {success}/{len(tracks)} tracks\n\n"
|
|
report += "\n".join(results)
|
|
|
|
# Also list the MIDI output directory
|
|
midi_dir = dl_dir
|
|
midi_files = list(midi_dir.glob("*.mid")) if midi_dir.exists() else []
|
|
if midi_files:
|
|
report += f"\n\n📁 MIDI files saved in: {midi_dir}"
|
|
for mf in sorted(midi_files):
|
|
report += f"\n • {mf.name} ({mf.stat().st_size // 1024} KB)"
|
|
|
|
return report
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser(description="MuScriptor weekly MIDI pipeline")
|
|
parser.add_argument("--count", type=int, default=3, help="Number of top tracks")
|
|
parser.add_argument("--period", default="week", help="Time period (week/month/year/all)")
|
|
parser.add_argument("--model", default="medium", choices=["small", "medium", "large"],
|
|
help="MuScriptor model size")
|
|
args = parser.parse_args()
|
|
|
|
report = asyncio.run(run(count=args.count, period=args.period, model_size=args.model))
|
|
print(report)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|