247 lines
7.5 KiB
Python
247 lines
7.5 KiB
Python
#!/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()
|