发布 1.0 版本

README.md

@@ -6,39 +6,48 @@
 <img alt="GitHub Repo stars" src="https://img.shields.io/github/stars/JoeanAmier/KS-Downloader?style=for-the-badge&color=ff4d4f">
 <img alt="GitHub code size in bytes" src="https://img.shields.io/github/languages/code-size/JoeanAmier/KS-Downloader?style=for-the-badge&color=13c2c2">
 <br>
-
-[//]: # (<img alt="GitHub release &#40;with filter&#41;" src="https://img.shields.io/github/v/release/JoeanAmier/KS-Downloader?style=for-the-badge&color=f759ab">)
+<img alt="GitHub release (with filter)" src="https://img.shields.io/github/v/release/JoeanAmier/KS-Downloader?style=for-the-badge&color=f759ab">
 <img src="https://img.shields.io/badge/Sourcery-enabled-884898?style=for-the-badge&color=1890ff" alt="">
-
-[//]: # (<img alt="GitHub all releases" src="https://img.shields.io/github/downloads/JoeanAmier/KS-Downloader/total?style=for-the-badge&color=52c41a">)
+<img alt="GitHub all releases" src="https://img.shields.io/github/downloads/JoeanAmier/KS-Downloader/total?style=for-the-badge&color=52c41a">
 <br>
 <p>🔥 <b>快手作品下载工具：</b>完全开源，基于 AIOHTTP 模块实现，下载快手无水印视频、图片文件！</p>
 </div>
 <hr>
 <h1>📑 项目功能</h1>
 <ul>
-<li>✅ 采集快手作品数据</li>
 <li>✅ 下载快手无水印作品文件</li>
 <li>✅ 下载快手作品封面图片</li>
+<li>✅ 下载快手作品音乐文件</li>
 <li>✅ 自动跳过已下载的作品文件</li>
 <li>✅ 作品文件完整性处理机制</li>
 <li>✅ 持久化储存作品信息至文件</li>
 <li>✅ 记录已下载作品 ID</li>
 <li>✅ 作品文件储存至单独文件夹</li>
-<li>☑️ 下载快手作品音乐文件</li>
+<li>☑️ 自定义作品文件名称格式</li>
 <li>☑️ 后台监听剪贴板下载作品</li>
 <li>☑️ 支持命令行下载作品文件</li>
 <li>☑️ 支持 API 调用功能</li>
 </ul>
 <h1>📸 程序截图</h1>
 <img src="docs/项目运行截图.png" alt="">
+<h1>🥣 使用方法</h1>
+<h2>🖱 程序运行</h2>
+<p>Windows 10 及以上用户可前往 <a href="https://github.com/JoeanAmier/KS-Downloader/releases/latest">Releases</a> 下载程序压缩包，解压后打开程序文件夹，双击运行 <code>main.exe</code> 即可使用。</p>
+<p>若通过此方式使用程序，文件默认下载路径为：<code>.\_internal\Download</code>；配置文件路径为：<code>.\_internal\config.yaml</code></p>
+<h2>⌨️ 源码运行</h2>
+<ol>
+<li>安装版本号不低于 <code>3.12</code> 的 Python 解释器</li>
+<li>运行 <code>pip install -i https://pypi.tuna.tsinghua.edu.cn/simple -r requirements.txt</code> 命令安装程序所需模块</li>
+<li>下载本项目最新的源码或 <a href="https://github.com/JoeanAmier/KS-Downloader/releases/latest">Releases</a> 发布的源码至本地</li>
+<li>运行 <code>main.py</code> 即可使用</li>
+</ol>
 <h1>🔗 支持链接</h1>
 <ul>
 <li><code>https://www.kuaishou.com/f/分享码</code></li>
 <li><code>https://v.kuaishou.com/分享码</code></li>
 <li><code>https://www.kuaishou.com/short-video/作品ID</code></li>
 <br/>
-<p><b>支持单次输入多个作品链接，链接之间使用空格分隔。</b></p>
+<p><b>推荐使用分享链接；支持单次输入多个作品链接，链接之间使用空格分隔。</b></p>
 </ul>
 <h1>🪟 关于终端</h1>
 <p>⭐ 推荐使用 <a href="https://learn.microsoft.com/zh-cn/windows/terminal/install">Windows 终端</a> （Windows 11 默认终端）运行程序以便获得最佳显示效果！</p>
@@ -97,6 +106,12 @@
 <td align="center">空字符串</td>
 </tr>
 <tr>
+<td align="center">music</td>
+<td align="center">bool</td>
+<td align="center">是否下载作品音乐</td>
+<td align="center">false</td>
+</tr>
+<tr>
 <td align="center">max_retry</td>
 <td align="center">int</td>
 <td align="center">请求数据失败时，重试的最大次数，单位：秒</td>

source/app/app.py

@@ -7,6 +7,7 @@
     VERSION_MAJOR,
     VERSION_MINOR,
     VERSION_BETA,
+    DISCLAIMER_TEXT,
 )
 from source.downloader import Downloader
 from source.extract import APIExtractor
@@ -80,7 +81,8 @@ async def run(self):
         self.__welcome()
         await self.__update_version()
         # await self.async_init()
-        await self.__main_menu()
+        if await self.disclaimer():
+            await self.__main_menu()
 
     async def __detail_enquire(self):
         while self.running:
@@ -121,7 +123,7 @@ async def __update_version(self):
                 f"{VERSION_MAJOR}.{VERSION_MINOR}", target, VERSION_BETA)
             self.console.print(
                 self.version.STATUS_CODE[state],
-                INFO if state == 1 else WARNING)
+                style=INFO if state == 1 else WARNING)
         else:
             self.console.print("检测新版本失败", style=ERROR)
         self.console.print()
@@ -156,12 +158,14 @@ async def detail(self, detail: str):
             case True:
                 items = [await self.detail_request.run(i) for i in urls]
                 data = self.api_extractor.run([i for i in items if i])
+                data = self.__check_extract_data(data)
                 await self.__save_data(data, "Download")
             case False:
                 items = await self.detail_page.run(urls)
                 data = [
                     self.detail_extractor.run(
                         h, i, ) for i, h in items if h]
+                data = self.__check_extract_data(data)
             case _:
                 return
         await self.__download_file(data, app=app, )
@@ -178,9 +182,28 @@ async def __download_file(self, data: list[dict], type_="detail", app=True, ):
         if data:
             await self.download.run(data, type_, app, )
 
+    def __check_extract_data(self, data: list) -> list:
+        if data := [i for i in data if i]:
+            return data
+        self.console.error("获取数据失败")
+        return []
+
     async def user(self):
         pass
 
+    async def disclaimer(self):
+        if self.config["Disclaimer"]:
+            return True
+        self.console.print(
+            "\n".join(DISCLAIMER_TEXT),
+            style=MASTER)
+        if self.console.input(
+                "是否已仔细阅读上述免责声明(YES/NO): ").upper() != "YES":
+            return False
+        await self.database.update_config_data("Disclaimer", 1)
+        self.console.print()
+        return True
+
     async def close(self):
         await self.manager.close()
 

source/config/config.py

@@ -25,7 +25,6 @@ class Config:
         "timeout": TIMEOUT,
         "chunk": 1024 * 1024,
         "folder_mode": False,
-        "server": False,
         # "cookie": "",
     }
 

source/config/parameter.py

@@ -29,7 +29,6 @@ def __init__(self,
                  data_record: bool = False,
                  chunk=1024 * 1024,
                  folder_mode: bool = False,
-                 server=False,
                  max_workers=4,
                  ):
         self.root = PROJECT_ROOT
@@ -42,6 +41,7 @@ def __init__(self,
         self.work_path = self.__check_work_path(work_path)
         # self.cookie = self.__check_cookie(cookie)
         self.cover = self.__check_cover(cover)
+        self.music = self.check_bool(music, False)
         self.download_record = self.check_bool(download_record, True)
         self.data_record = self.check_bool(data_record, False)
         self.chunk = self.__check_chunk(chunk)
@@ -60,6 +60,7 @@ def run(self) -> dict:
             "folder_name": self.folder_name,
             # "cookie": self.cookie,
             "cover": self.cover,
+            "music": self.music,
             "download_record": self.download_record,
             "data_record": self.data_record,
             "max_workers": self.max_workers,
@@ -80,8 +81,7 @@ def __check_max_retry(self, max_retry: int) -> int:
         return max_retry
 
     async def check_proxy(self) -> None:
-        if self.proxy:
-            self.proxy = await self.__check_proxy(self.proxy)
+        self.proxy = await self.__check_proxy(self.proxy) if self.proxy else None
 
     def __check_max_workers(self, max_workers: int) -> int:
         if isinstance(max_workers, int) and max_workers > 0:

source/custom/internal.py

@@ -1,9 +1,9 @@
 from pathlib import Path
 
 PROJECT_ROOT = Path(__file__).resolve().parent.parent.parent
-VERSION_MAJOR = 0
-VERSION_MINOR = 5
-VERSION_BETA = True
+VERSION_MAJOR = 1
+VERSION_MINOR = 0
+VERSION_BETA = False
 PROJECT_NAME = f"KS-Downloader V{VERSION_MAJOR}.{
 VERSION_MINOR}{" Beta" if VERSION_BETA else ""}"
 

source/downloader/downloader.py

@@ -18,6 +18,7 @@
 )
 from source.tools import capture_error_request
 from source.tools import retry_request
+from source.tools import truncation
 
 if TYPE_CHECKING:
     from source.manager import Manager
@@ -31,6 +32,7 @@ class Downloader:
         "image/webp": "webp",
         "video/mp4": "mp4",
         "video/quicktime": "mov",
+        "audio/mp4": "m4a"
     }
 
     def __init__(self, manager: "Manager", database: "Database"):
@@ -40,6 +42,7 @@ def __init__(self, manager: "Manager", database: "Database"):
         self.headers = manager.app_download_headers
         self.cleaner = manager.cleaner
         self.cover = manager.cover
+        self.music = manager.music
         self.console = manager.console
         self.proxy = manager.proxy
         self.retry = manager.max_retry
@@ -92,8 +95,22 @@ async def __handle_detail(self, data: list[dict], app: bool, ):
                         await self.__handle_atlas(tasks, name, item, progress, )
                     case _:
                         self.console.error("未知的作品类型")
+                await self.__handle_music(tasks, name, item, progress, )
             await gather(*tasks)
 
+    async def __handle_music(self, tasks: list, name: str, data: dict, progress: Progress, ):
+        if not (m := data.get("audioUrls")):
+            return
+        file = self.__generate_path(name)
+        if not self.__file_exists(file, "m4a"):
+            tasks.append(self.__download_file(
+                m.split()[0],
+                file,
+                progress,
+                data["detailID"],
+                "音乐",
+            ))
+
     async def __handle_video(self, tasks: list, name: str, data: dict, progress: Progress, ):
         file = self.__generate_path(name)
         if not self.__file_exists(file, "mp4"):
@@ -139,32 +156,33 @@ async def __handle_cover(self, tasks: list, path: "Path", data: dict, progress:
     async def __download_file(self, url: str, path: "Path", progress: Progress, id_: str, tip: str = ""):
         async with self.semaphore:
             if not url:
-                self.console.warning(f"{path.name} {tip}下载链接为空")
+                self.console.warning(f"【{tip}】{truncation(path.name)} 下载链接为空")
                 return True
             try:
                 async with self.session.get(url, headers=self.headers, proxy=self.proxy, ) as response:
                     if response.status != 200:
-                        self.console.error(f"{path.name} 响应码异常：{
+                        self.console.error(f"【{tip}】{truncation(path.name)} 响应码异常：{
                         response.status}")
                         return False
-                    temp = self.temp.joinpath(path.name)
                     suffix = self.__extract_type(
                         response.headers.get("Content-Type"))
+                    temp = self.temp.joinpath(f"{path.name}.{suffix}")
                     path = path.with_name(f"{path.name}.{suffix}")
                     task_id = progress.add_task(
-                        path.name, total=int(
+                        f"【{tip}】{truncation(path.name)}", total=int(
                             response.headers.get(
                                 "Content-Length", "0")) or None)
                     with temp.open("wb") as f:
                         async for chunk in response.content.iter_chunked(self.chunk):
                             f.write(chunk)
                             progress.update(task_id, advance=len(chunk))
             except StopAsyncIteration:
-                self.console.error(f"网络异常，{path.name} 下载中断！")
+                self.console.error(
+                    f"【{tip}】{truncation(path.name)} 网络异常，下载中断！")
                 await self.database.delete_download_data(id_)
                 return False
             self.move(temp, path)
-            self.console.info(f"{path.name} 下载完成")
+            self.console.info(f"【{tip}】{truncation(path.name)} 下载完成")
             await self.database.write_download_data(id_)
             return True
 

source/extract/extractor.py

@@ -153,8 +153,10 @@ def __extract_items(self, container: list, data: SimpleNamespace) -> None:
         self.__extract_photo(item, data)
         match item["photoType"]:
             case "视频":
+                self.__extract_music(item, data, True)
                 self.__extract_mp4(item, data)
             case "图片":
+                self.__extract_music(item, data, False)
                 self.__extract_atlas(item, data)
             case _:
                 item["download"] = ""
@@ -192,6 +194,17 @@ def __extract_photo(self, item: dict, data: SimpleNamespace) -> None:
         item["detailID"] = self.__extract_id(
             self.safe_extract(photo, "share_info"))
 
+    def __extract_music(self, item: dict, data: SimpleNamespace, video=True, ) -> None:
+        if video:
+            music = self.safe_extract(data, "photo.soundTrack")
+        else:
+            music = self.safe_extract(data, "photo.music")
+        item["music_name"] = self.safe_extract(music, "name")
+        item["audioUrls"] = []
+        for i in self.safe_extract(music, "audioUrls", []):
+            item["audioUrls"].append(i.url)
+        item["audioUrls"] = " ".join(i for i in item["audioUrls"] if i)
+
     @staticmethod
     def __extract_id(share: str):
         parsed = parse_qs(share)

source/manager/manager.py

@@ -28,6 +28,7 @@ def __init__(self,
                  folder_name: str,
                  # cookie: str,
                  cover: str,
+                 music: bool,
                  download_record: bool,
                  data_record: bool,
                  chunk: int,
@@ -52,6 +53,7 @@ def __init__(self,
         self.max_retry = max_retry
         self.proxy = proxy
         self.cover = cover
+        self.music = music
         self.download_record = download_record
         self.data_record = data_record
         self.folder_mode = folder_mode

source/record/manager.py

@@ -35,6 +35,8 @@ class RecordManager:
         ("download", "下载链接", "TEXT"),
         ("coverUrls", "封面链接", "TEXT"),
         ("webpCoverUrls", "WEBP封面链接", "TEXT"),
+        ("music_name", "音乐标题", "TEXT"),
+        ("audioUrls", "音乐链接", "TEXT"),
         ("collectionCount", "收藏数量", "INTEGER"),
         ("commentCount", "评论数量", "INTEGER"),
         ("viewCount", "浏览数量", "INTEGER"),

source/tools/__init__.py

@@ -13,4 +13,6 @@
 from .namespace import Namespace
 from .retry import retry_request
 from .session import base_session
+from .string import is_chinese_char
+from .string import truncation
 from .version import Version

source/tools/string.py

@@ -0,0 +1,18 @@
+def is_chinese_char(char):
+    """判断一个字符是否是中文字符"""
+    if "\u4e00" <= char <= "\u9fff":
+        return True
+    return False
+
+
+def truncation(string: str, length=64) -> str:
+    result = ""
+    for s in string:
+        result += s
+        if is_chinese_char(s):
+            length -= 2
+        else:
+            length -= 1
+        if length < 1:
+            break
+    return result