#!/usr/bin/env python3 import os import sys import time import shlex import subprocess import tempfile from pathlib import Path from typing import List, Tuple, Optional # --- Configuration --- HOME = Path.home() SOURCES = [ HOME / "Music", HOME / "Downloads", ] DEST_ON_PHONE = "/sdcard/Music" # Model tag as seen in `adb devices -l` (e.g., ... model:AGM_M7 ...) ADB_MODEL_MATCH = ["AGM_M7", "AGM", "M7"] POLL_INTERVAL_SEC = 60 SIZE_LIMIT_BYTES = 7 * 1024 ** 3 # 7 GiB MUSIC_EXTS = {".mp3", ".m4a", ".ogg", ".aac", ".opus", ".alac", ".flac"} # If you have a static Wi-Fi ADB endpoint, set it here (e.g., "192.168.1.50:5555"). ADB_IP_PORT: Optional[str] = None # --- Helpers --- def run(cmd: List[str], check: bool = False, text: bool = True, timeout: Optional[int] = None) -> subprocess.CompletedProcess: try: return subprocess.run(cmd, check=check, text=text, capture_output=True, timeout=timeout) except Exception as e: return subprocess.CompletedProcess(cmd, returncode=1, stdout="", stderr=str(e)) def ensure_adb_server() -> None: run(["adb", "start-server"]) # best-effort def list_adb_devices() -> List[str]: proc = run(["adb", "devices", "-l"]) # do not check, we parse regardless if proc.returncode != 0: return [] lines = [ln.strip() for ln in proc.stdout.splitlines()] # skip header line: "List of devices attached" if lines and lines[0].lower().startswith("list of devices"): lines = lines[1:] return [ln for ln in lines if ln] def adb_connect_if_needed(): if not ADB_IP_PORT: return devices = list_adb_devices() if any(ADB_IP_PORT in ln for ln in devices): return # already connected run(["adb", "connect", ADB_IP_PORT]) def is_agm_connected() -> bool: devices = list_adb_devices() for ln in devices: if "device" in ln and not ln.endswith("offline"): if any(tag in ln for tag in ADB_MODEL_MATCH): return True return False def gather_music_files() -> List[Tuple[Path, Path, float, int]]: # Returns list of tuples: (root_dir, file_path, mtime, size) entries: List[Tuple[Path, Path, float, int]] = [] for root in SOURCES: try: if not root.exists(): continue for dirpath, _, filenames in os.walk(root): dpath = Path(dirpath) for fn in filenames: p = dpath / fn if p.suffix.lower() not in MUSIC_EXTS: continue try: st = p.stat() entries.append((root, p, st.st_mtime, st.st_size)) except Exception: continue except Exception: continue # newest first by mtime entries.sort(key=lambda t: t[2], reverse=True) return entries def select_up_to_size(entries: List[Tuple[Path, Path, float, int]], limit_bytes: int) -> List[Tuple[Path, Path, int]]: selected: List[Tuple[Path, Path, int]] = [] total = 0 for root, p, _mt, sz in entries: if total >= limit_bytes: break selected.append((root, p, sz)) total += sz return selected def adb_shell(cmd: str) -> subprocess.CompletedProcess: # Single string executed via adb shell return run(["adb", "shell", cmd]) def remote_size(adb_path: str) -> Optional[int]: # Uses stat if available; returns None if missing # Escape path for shell qpath = shlex.quote(adb_path) proc = adb_shell(f"stat -c %s {qpath}") if proc.returncode != 0 or not proc.stdout.strip(): return None out = proc.stdout.strip() if "No such file" in out or "not found" in out: return None try: return int(out.splitlines()[-1].strip()) except Exception: return None def ensure_remote_dir(adb_path: str) -> None: # mkdir -p dirname dirname = os.path.dirname(adb_path) if not dirname: return adb_shell(f"mkdir -p {shlex.quote(dirname)}") def have_ffmpeg() -> bool: proc = run(["ffmpeg", "-version"]) return proc.returncode == 0 def transcode_flac_to_mp3(src: Path) -> Optional[Path]: try: tmp = tempfile.NamedTemporaryFile(suffix=".mp3", delete=False) tmp_path = Path(tmp.name) tmp.close() except Exception: return None cmd = [ "ffmpeg", "-y", "-i", str(src), "-vn", "-c:a", "libmp3lame", "-q:a", "2", str(tmp_path), ] proc = run(cmd) if proc.returncode != 0: try: tmp_path.unlink(missing_ok=True) # type: ignore[arg-type] except Exception: pass return None return tmp_path def push_file(local_path: Path, remote_path: str, size: int) -> bool: # Skip if same size exists rsz = remote_size(remote_path) if rsz is not None and rsz == size: print(f"[=] Skip (same size): {local_path} -> {remote_path}") return True ensure_remote_dir(remote_path) proc = run(["adb", "push", "-p", str(local_path), remote_path]) if proc.returncode == 0: print(f"[+] Pushed: {local_path} -> {remote_path}") return True else: print(f"[!] Push failed: {local_path}\n{proc.stderr}") return False def sync_latest_batch(): entries = gather_music_files() if not entries: print("[i] No music files found.") return batch = select_up_to_size(entries, SIZE_LIMIT_BYTES) print(f"[i] Selected {len(batch)} files up to {SIZE_LIMIT_BYTES/(1024**3):.2f} GiB") for root, p, sz in batch: try: rel = p.relative_to(root) except Exception: rel = p.name # fallback if p.suffix.lower() == ".flac": if not have_ffmpeg(): print(f"[!] ffmpeg not available, skipping: {p}") continue mp3_tmp = transcode_flac_to_mp3(p) if not mp3_tmp: print(f"[!] Transcode failed, skipping: {p}") continue try: if isinstance(rel, Path): rel_mp3 = rel.with_suffix(".mp3") remote = f"{DEST_ON_PHONE.rstrip('/')}/{rel_mp3.as_posix()}" else: remote = f"{DEST_ON_PHONE.rstrip('/')}/{Path(rel).with_suffix('.mp3').as_posix()}" msz = mp3_tmp.stat().st_size push_file(mp3_tmp, remote, msz) finally: try: mp3_tmp.unlink() except Exception: pass else: remote = f"{DEST_ON_PHONE.rstrip('/')}/{rel.as_posix()}" push_file(p, remote, sz) def main(): print("[i] adb-music-sync daemon starting...") ensure_adb_server() connected_prev = False while True: try: ensure_adb_server() adb_connect_if_needed() connected = is_agm_connected() if connected and not connected_prev: print("[+] AGM M7 connected. Starting sync of latest ~7 GiB...") sync_latest_batch() elif not connected and connected_prev: print("[-] AGM M7 disconnected.") connected_prev = connected time.sleep(POLL_INTERVAL_SEC) except KeyboardInterrupt: print("[i] Exiting on user interrupt.") sys.exit(0) except Exception as e: print(f"[!] Error in main loop: {e}") time.sleep(POLL_INTERVAL_SEC) if __name__ == "__main__": main()