| 1 | from __future__ import annotations |
| 2 | |
| 3 | import json |
| 4 | import posixpath |
| 5 | from datetime import datetime |
| 6 | from typing import Any, Dict, Optional |
| 7 | from urllib.parse import urlparse |
| 8 | |
| 9 | from core.downloader_base import BaseDownloader, DownloadResult |
| 10 | from core.metadata import extract_author_sec_uid |
| 11 | from utils.logger import setup_logger |
| 12 | from utils.naming import ( |
| 13 | DEFAULT_FILE_TEMPLATE, |
| 14 | DEFAULT_FOLDER_TEMPLATE, |
| 15 | build_music_context, |
| 16 | render_template, |
| 17 | ) |
| 18 | |
| 19 | logger = setup_logger("MusicDownloader") |
| 20 | |
| 21 | |
| 22 | class MusicDownloader(BaseDownloader): |
| 23 | async def download(self, parsed_url: Dict[str, Any]) -> DownloadResult: |
| 24 | result = DownloadResult() |
| 25 | |
| 26 | music_id = parsed_url.get("music_id") |
| 27 | if not music_id: |
| 28 | logger.error("No music_id found in parsed URL") |
| 29 | return result |
| 30 | |
| 31 | result.total = 1 |
| 32 | self._progress_set_item_total(1, "单音乐下载") |
| 33 | self._progress_update_step("下载音乐", f"music_id={music_id}") |
| 34 | |
| 35 | detail = await self._get_music_detail(str(music_id)) |
| 36 | music_url = self._extract_music_url(detail) |
| 37 | if music_url: |
| 38 | success = await self._download_music_asset(str(music_id), detail, music_url) |
| 39 | if success: |
| 40 | result.success += 1 |
| 41 | self._progress_advance_item("success", str(music_id)) |
| 42 | else: |
| 43 | result.failed += 1 |
| 44 | self._progress_advance_item("failed", str(music_id)) |
| 45 | return result |
| 46 | |
| 47 | # 回退：音乐详情无法直接拿到音频链接时，尝试下载该音乐下的首条作品 |
| 48 | aweme = await self._get_first_music_aweme(str(music_id)) |
| 49 | if aweme and aweme.get("aweme_id"): |
| 50 | if not await self._should_download(str(aweme.get("aweme_id"))): |
| 51 | result.skipped += 1 |
| 52 | self._progress_advance_item("skipped", str(aweme.get("aweme_id"))) |
| 53 | return result |
| 54 | |
| 55 | aweme_author = (aweme.get("author") or {}).get("nickname", "music") |
| 56 | success = await self._download_aweme_assets(aweme, aweme_author, mode="music") |
| 57 | if success: |
| 58 | result.success += 1 |
| 59 | self._progress_advance_item("success", str(aweme.get("aweme_id"))) |
| 60 | else: |
| 61 | result.failed += 1 |
| 62 | self._progress_advance_item("failed", str(aweme.get("aweme_id"))) |
| 63 | return result |
| 64 | |
| 65 | logger.error("No playable music source found for music_id=%s", music_id) |
| 66 | result.failed += 1 |
| 67 | self._progress_advance_item("failed", str(music_id)) |
| 68 | return result |
| 69 | |
| 70 | async def _download_music_asset( |
| 71 | self, music_id: str, detail: Optional[Dict[str, Any]], music_url: str |
| 72 | ) -> bool: |
| 73 | session = await self.api_client.get_session() |
| 74 | detail = detail or {} |
| 75 | |
| 76 | title = ( |
| 77 | detail.get("title") |
| 78 | or detail.get("music_name") |
| 79 | or (detail.get("music") or {}).get("title") |
| 80 | or f"music_{music_id}" |
| 81 | ) |
| 82 | author_name = ( |
| 83 | detail.get("author_name") or (detail.get("owner") or {}).get("nickname") or "music" |
| 84 | ) |
| 85 | publish_date = datetime.now().strftime("%Y-%m-%d") |
| 86 | record_id = f"music_{music_id}" |
| 87 | template_context = build_music_context( |
| 88 | music_id=str(music_id), |
| 89 | title=title, |
| 90 | author_name=author_name, |
| 91 | publish_date=publish_date, |
| 92 | ) |
| 93 | filename_template = self.config.get("filename_template") or DEFAULT_FILE_TEMPLATE |
| 94 | folder_template = self.config.get("folder_template") or DEFAULT_FOLDER_TEMPLATE |
| 95 | file_stem = render_template( |
| 96 | filename_template, |
| 97 | template_context, |
| 98 | fallback=f"{publish_date}_{record_id}", |
| 99 | ) |
| 100 | folder_name = render_template( |
| 101 | folder_template, |
| 102 | template_context, |
| 103 | fallback=f"{publish_date}_{record_id}", |
| 104 | ) |
| 105 | |
| 106 | save_dir = self.file_manager.get_save_path( |
| 107 | author_name=author_name, |
| 108 | mode="music", |
| 109 | aweme_title=title, |
| 110 | aweme_id=record_id, |
| 111 | folderstyle=self.config.get("folderstyle", True), |
| 112 | download_date=publish_date, |
| 113 | folder_name=folder_name, |
| 114 | author_sec_uid=None, |
| 115 | author_dir_style=self.config.get("author_dir") or "nickname", |
| 116 | ) |
| 117 | |
| 118 | music_ext = self._infer_audio_extension(music_url) |
| 119 | music_path = save_dir / f"{file_stem}{music_ext}" |
| 120 | if self.file_manager.file_exists(music_path): |
| 121 | logger.info("Music already exists locally: %s", music_path.name) |
| 122 | return True |
| 123 | |
| 124 | success = await self._download_with_retry( |
| 125 | music_url, |
| 126 | music_path, |
| 127 | session, |
| 128 | headers=self._download_headers(), |
| 129 | ) |
| 130 | if not success: |
| 131 | return False |
| 132 | |
| 133 | cover_url = self._extract_first_url( |
| 134 | detail.get("cover_large") |
| 135 | or detail.get("cover_thumb") |
| 136 | or (detail.get("music") or {}).get("cover_large") |
| 137 | ) |
| 138 | if cover_url and self.config.get("cover"): |
| 139 | cover_path = save_dir / f"{file_stem}_cover.jpg" |
| 140 | await self._download_with_retry( |
| 141 | cover_url, |
| 142 | cover_path, |
| 143 | session, |
| 144 | headers=self._download_headers(), |
| 145 | optional=True, |
| 146 | ) |
| 147 | |
| 148 | if self.config.get("json"): |
| 149 | await self.metadata_handler.save_metadata( |
| 150 | detail or {"music_id": music_id}, save_dir / f"{file_stem}_data.json" |
| 151 | ) |
| 152 | |
| 153 | if self.database: |
| 154 | await self.database.add_aweme( |
| 155 | { |
| 156 | "aweme_id": record_id, |
| 157 | "aweme_type": "music", |
| 158 | "title": title, |
| 159 | "author_id": None, |
| 160 | "author_name": author_name, |
| 161 | "create_time": None, |
| 162 | "file_path": str(save_dir), |
| 163 | "metadata": json.dumps(detail or {}, ensure_ascii=False), |
| 164 | }, |
| 165 | author_sec_uid=extract_author_sec_uid(detail), |
| 166 | ) |
| 167 | |
| 168 | await self.metadata_handler.append_download_manifest( |
| 169 | self.file_manager.base_path, |
| 170 | { |
| 171 | "date": publish_date, |
| 172 | "aweme_id": record_id, |
| 173 | "author_name": author_name, |
| 174 | "desc": title, |
| 175 | "media_type": "music", |
| 176 | "file_names": [music_path.name], |
| 177 | "file_paths": [self._to_manifest_path(music_path)], |
| 178 | }, |
| 179 | ) |
| 180 | return True |
| 181 | |
| 182 | async def _get_music_detail(self, music_id: str) -> Optional[Dict[str, Any]]: |
| 183 | getter = getattr(self.api_client, "get_music_detail", None) |
| 184 | if not callable(getter): |
| 185 | return None |
| 186 | try: |
| 187 | return await getter(music_id) |
| 188 | except Exception as exc: |
| 189 | logger.warning("Get music detail failed: %s", exc) |
| 190 | return None |
| 191 | |
| 192 | async def _get_first_music_aweme(self, music_id: str) -> Optional[Dict[str, Any]]: |
| 193 | getter = getattr(self.api_client, "get_music_aweme", None) |
| 194 | if not callable(getter): |
| 195 | return None |
| 196 | try: |
| 197 | data = await getter(music_id, cursor=0, count=1) |
| 198 | except Exception as exc: |
| 199 | logger.warning("Get music aweme failed: %s", exc) |
| 200 | return None |
| 201 | |
| 202 | if not isinstance(data, dict): |
| 203 | return None |
| 204 | items = data.get("items") |
| 205 | if not isinstance(items, list): |
| 206 | items = data.get("aweme_list") |
| 207 | if not isinstance(items, list) or not items: |
| 208 | return None |
| 209 | first_item = items[0] |
| 210 | if isinstance(first_item, dict) and first_item.get("aweme_id"): |
| 211 | return first_item |
| 212 | nested_aweme = first_item.get("aweme") if isinstance(first_item, dict) else None |
| 213 | if isinstance(nested_aweme, dict) and nested_aweme.get("aweme_id"): |
| 214 | return nested_aweme |
| 215 | return None |
| 216 | |
| 217 | def _extract_music_url(self, detail: Optional[Dict[str, Any]]) -> Optional[str]: |
| 218 | if not isinstance(detail, dict): |
| 219 | return None |
| 220 | |
| 221 | candidates = ( |
| 222 | detail.get("play_url"), |
| 223 | detail.get("play_url_lowbr"), |
| 224 | detail.get("audio_url"), |
| 225 | (detail.get("music") or {}).get("play_url"), |
| 226 | (detail.get("music") or {}).get("play_url_lowbr"), |
| 227 | (detail.get("music_info") or {}).get("play_url"), |
| 228 | ) |
| 229 | |
| 230 | for candidate in candidates: |
| 231 | url = self._extract_first_url(candidate) |
| 232 | if url: |
| 233 | return url |
| 234 | return None |
| 235 | |
| 236 | @staticmethod |
| 237 | def _infer_audio_extension(music_url: str) -> str: |
| 238 | if not music_url: |
| 239 | return ".mp3" |
| 240 | |
| 241 | raw_path = urlparse(music_url).path or "" |
| 242 | ext = posixpath.splitext(raw_path)[1].lower() |
| 243 | allowed_exts = {".mp3", ".m4a", ".aac", ".wav", ".flac", ".ogg", ".opus"} |
| 244 | if ext in allowed_exts: |
| 245 | return ext |
| 246 | return ".mp3" |
| 247 |