buy.py (5.0 KB)


  1 #!/usr/bin/env python3
  2 """
  3 Search for tracks and open purchase links.
  4 
  5 Usage:
  6 	./buy.py <search query>
  7 	./buy.py beatles yesterday
  8 	./buy.py "daft punk revolution 909"
  9 """
 10 
 11 import sys
 12 import os
 13 
 14 # Hop into the shared venv (managed by scan.py) before doing anything else,
 15 # so scan.rescan() has mutagen available when we call it after a purchase.
 16 sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
 17 import scan
 18 scan.bootstrap(__file__)
 19 
 20 import urllib.parse
 21 import urllib.request
 22 import json
 23 import platform
 24 import webbrowser
 25 import shutil
 26 import time
 27 
 28 
 29 ITUNES_MUSIC_DIR = os.path.expanduser(
 30 	"~/Music/Music/Media.localized/Music"
 31 )
 32 MIX_DIR = os.path.join(os.path.dirname(os.path.abspath(__file__)), "mix")
 33 POLL_INTERVAL = 1  # seconds between directory scans
 34 
 35 
 36 def search_itunes(query):
 37 	"""Search iTunes for a track and return results."""
 38 	encoded_query = urllib.parse.quote(query)
 39 	url = f"https://itunes.apple.com/search?term={encoded_query}&media=music&entity=song&limit=10"
 40 
 41 	try:
 42 		with urllib.request.urlopen(url) as response:
 43 			data = json.loads(response.read().decode())
 44 			return data.get('results', [])
 45 	except Exception as e:
 46 		print(f"Error searching iTunes: {e}")
 47 		return []
 48 
 49 
 50 def open_itunes_link(itunes_url, track_id):
 51 	"""Open the iTunes purchase link directly in iTunes app on macOS, or song.link on other platforms."""
 52 	import subprocess
 53 	
 54 	# Convert to geo-aware link and add app=itunes parameter to force iTunes Store
 55 	# This prevents opening in Apple Music (streaming) instead of iTunes Store (purchase)
 56 	
 57 	# Replace music.apple.com with geo.itunes.apple.com for better compatibility
 58 	if 'music.apple.com' in itunes_url:
 59 		itunes_url = itunes_url.replace('music.apple.com', 'geo.itunes.apple.com')
 60 	elif 'itunes.apple.com' in itunes_url and 'geo.' not in itunes_url:
 61 		itunes_url = itunes_url.replace('itunes.apple.com', 'geo.itunes.apple.com')
 62 	
 63 	# Add app=itunes parameter to force iTunes Store (not Apple Music)
 64 	if '?' in itunes_url:
 65 		# URL already has parameters, append with &
 66 		if 'app=' not in itunes_url:
 67 			itunes_url += '&app=itunes'
 68 	else:
 69 		# No parameters yet, add with ?
 70 		itunes_url += '?app=itunes'
 71 	
 72 	# Convert https:// to itms:// to open directly in iTunes app
 73 	itunes_url = itunes_url.replace('https://', 'itms://')
 74 	
 75 	# Use macOS 'open' command to bypass browser entirely
 76 	if platform.system() == 'Darwin':  # macOS
 77 		subprocess.run(['open', itunes_url])
 78 		return True  # opened iTunes Store
 79 	else:
 80 		# For non-macOS systems, open song.link page with all platform options
 81 		songlink_url = f"https://song.link/i/{track_id}"
 82 		webbrowser.open(songlink_url)
 83 		return False  # not iTunes Store
 84 
 85 
 86 def snapshot_audio_files(root):
 87 	"""Return a set of all audio file paths found recursively under root."""
 88 	audio_exts = {'.mp3', '.m4a', '.ogg', '.flac', '.wav', '.aiff', '.aif'}
 89 	found = set()
 90 	for dirpath, _dirnames, filenames in os.walk(root):
 91 		for name in filenames:
 92 			if os.path.splitext(name)[1].lower() in audio_exts:
 93 				found.add(os.path.join(dirpath, name))
 94 	return found
 95 
 96 
 97 def wait_for_new_file(watch_dir, before):
 98 	"""
 99 	Poll watch_dir recursively until an audio file appears that wasn't in the
100 	before snapshot. Returns the full path of the new file.
101 	"""
102 	print(f"Watching for new file...")
103 	print("(complete your purchase in iTunes or press Ctrl+C to cancel)")
104 
105 	while True:
106 		after = snapshot_audio_files(watch_dir)
107 		new_files = after - before
108 		if new_files:
109 			return next(iter(new_files))
110 		time.sleep(POLL_INTERVAL)
111 
112 def main():
113 	"""Main function to run the music search CLI."""
114 	# Get search query from command line args or prompt user
115 	if len(sys.argv) > 1:
116 		query = ' '.join(sys.argv[1:])
117 	else:
118 		query = input("Enter track name, artist, or both: ").strip()
119 	
120 	if not query:
121 		print("No search query provided.")
122 		return
123 	
124 	print(f"Searching for: {query}")
125 	
126 	# Search iTunes
127 	results = search_itunes(query)
128 	
129 	if not results:
130 		print("No results found. Try a different search term.")
131 		return
132 	
133 	# Get the best (first) result
134 	best_track = results[0]
135 	
136 	# Extract track info
137 	artist = best_track.get('artistName', 'Unknown')
138 	track = best_track.get('trackName', 'Unknown')
139 	itunes_url = best_track.get('trackViewUrl')
140 	track_id = best_track.get('trackId')
141 	
142 	if not itunes_url or not track_id:
143 		print("Error: Could not find iTunes URL")
144 		return
145 	
146 	# Open iTunes purchase page (macOS) or song.link (other platforms)
147 	print(f"Opening: {artist} - {track}")
148 	if platform.system() == 'Darwin':
149 		print("Opening iTunes Store directly...")
150 	else:
151 		print("Opening song.link with all platform options...")
152 	
153 	opened_itunes = open_itunes_link(itunes_url, track_id)
154 
155 	# Only watch for the downloaded file when we opened the iTunes Store
156 	if not opened_itunes:
157 		return
158 	
159 	before = snapshot_audio_files(ITUNES_MUSIC_DIR)
160 	new_file = wait_for_new_file(ITUNES_MUSIC_DIR, before)
161 
162 	filename = os.path.basename(new_file)
163 	dest = os.path.join(MIX_DIR, filename)
164 	shutil.copy2(new_file, dest)
165 	scan.rescan(silent=True)
166 	print(f"Added to /mix: {filename}")
167 
168 
169 if __name__ == "__main__":
170 	try:
171 		main()
172 	except KeyboardInterrupt:
173 		print("\n\nExiting...")
174 		sys.exit(0)