Release v0.4.2: 统一数据目录与缩略图CLI，仅视频缩略图、清理校验与进度条

Author: ChanMo

pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "TikLocal"
-version = "0.4.1"
+version = "0.4.2"
 description = "A local media server that combines the features of TikTok and Pinterest"
 authors = ["ChanMo <chan.mo@outlook.com>"]
 readme = "README.md"

tiklocal/.DS_Store

@@ -10,9 +10,11 @@
 from urllib.parse import quote, unquote
 from importlib.metadata import version, PackageNotFoundError
 import math
+import hashlib
 
 from pathlib import Path
-from flask import Flask, render_template, send_from_directory, request, session, redirect
+from flask import Flask, render_template, send_from_directory, request, session, redirect, send_file
+from tiklocal.paths import get_thumbnails_dir, get_thumbs_map_path
 
 
 try:
@@ -21,6 +23,7 @@
     app_version = '1.0.0'
 
 FAVORITE_FILENAME = 'favorite.json'
+VIDEO_EXTENSIONS = {'.mp4', '.webm', '.mov', '.mkv', '.avi', '.m4v'}
 
 
 def load_favorites(media_root: Path) -> set[str]:
@@ -107,6 +110,153 @@ def create_app(test_config=None):
     except OSError:
         pass
 
+    # 缩略图配置（统一使用全局数据目录 ~/.tiklocal 或 TIKLOCAL_INSTANCE）
+    THUMB_DIR = get_thumbnails_dir()
+    THUMB_MAP = get_thumbs_map_path()
+
+    PLACEHOLDER_PNG = (
+        b"\x89PNG\r\n\x1a\n\x00\x00\x00\rIHDR\x00\x00\x00\x01\x00\x00\x00\x01\x08\x06\x00\x00\x00\x1f\x15\xc4\x89\x00\x00\x00\x0cIDAT\x08\x99c\xf8\xff\xff?\x00\x05\xfe\x02\xfeA\x93\x8a\x1d\x00\x00\x00\x00IEND\xaeB`\x82"
+    )
+
+    def _thumb_key(rel_path: str) -> str:
+        return hashlib.sha1(rel_path.encode('utf-8', errors='ignore')).hexdigest() + '.jpg'
+
+    def _thumb_path(rel_path: str) -> Path:
+        return THUMB_DIR / _thumb_key(rel_path)
+
+    def _load_thumb_map() -> dict:
+        if THUMB_MAP.exists():
+            try:
+                with THUMB_MAP.open('r', encoding='utf-8') as f:
+                    data = json.load(f)
+                    if isinstance(data, dict):
+                        return data
+            except Exception as exc:
+                print(f"读取缩略图映射失败: {exc}", file=sys.stderr)
+        return {}
+
+    def _save_thumb_map(data: dict) -> None:
+        try:
+            with THUMB_MAP.open('w', encoding='utf-8') as f:
+                json.dump(data, f, ensure_ascii=False)
+        except Exception as exc:
+            print(f"保存缩略图映射失败: {exc}", file=sys.stderr)
+
+    def _ffmpeg_capture(input_path: Path, output_path: Path, ts: float | None) -> bool:
+        """用 ffmpeg 截帧到 output_path，返回是否成功。"""
+        candidates = []
+        if ts is not None and ts >= 0:
+            candidates = [ts]
+        else:
+            # 无时长信息时的保守候选
+            candidates = [5.0, 1.0, 0.1]
+
+        for t in candidates:
+            cmd = [
+                'ffmpeg',
+                '-y',
+                '-ss', str(max(0.0, float(t))),
+                '-i', str(input_path),
+                '-frames:v', '1',
+                '-vf', 'scale=-1:360:force_original_aspect_ratio=decrease',
+                '-q:v', '3',
+                str(output_path)
+            ]
+            try:
+                proc = sp.run(cmd, stdout=sp.DEVNULL, stderr=sp.DEVNULL, timeout=30)
+                if output_path.exists() and output_path.stat().st_size > 0:
+                    return True
+            except FileNotFoundError:
+                # 未安装 ffmpeg
+                return False
+            except sp.TimeoutExpired:
+                continue
+            except Exception as exc:
+                print(f"ffmpeg 截帧错误: {exc}", file=sys.stderr)
+                continue
+        return False
+
+    def _is_video(path: Path) -> bool:
+        try:
+            mime = mimetypes.guess_type(path.name)[0] or ''
+        except Exception:
+            mime = ''
+        if path.suffix.lower() in VIDEO_EXTENSIONS:
+            return True
+        return mime.startswith('video/')
+
+    @app.route('/thumb')
+    def thumb_view():
+        """按需返回或生成缩略图，不污染 MEDIA_ROOT。"""
+        uri = request.args.get('uri')
+        if not uri:
+            return send_file(io.BytesIO(PLACEHOLDER_PNG), mimetype='image/png')
+
+        # 解析并校验路径
+        try:
+            media_root: Path = Path(app.config["MEDIA_ROOT"])
+            target = media_root / Path(unquote(uri))
+            target = target.resolve()
+            # 拒绝越权访问
+            if not str(target).startswith(str(media_root.resolve())):
+                return send_file(io.BytesIO(PLACEHOLDER_PNG), mimetype='image/png')
+        except Exception:
+            return send_file(io.BytesIO(PLACEHOLDER_PNG), mimetype='image/png')
+
+        rel_path = str(Path(unquote(uri)))
+        thumb_file = _thumb_path(rel_path)
+
+        # 仅为视频生成缩略图
+        if not _is_video(target):
+            return send_file(io.BytesIO(PLACEHOLDER_PNG), mimetype='image/png')
+
+        if not thumb_file.exists():
+            # 生成缩略图（懒生成）
+            ok = _ffmpeg_capture(target, thumb_file, None)
+            if not ok:
+                return send_file(io.BytesIO(PLACEHOLDER_PNG), mimetype='image/png')
+
+        return send_from_directory(THUMB_DIR, thumb_file.name)
+
+    @app.route('/api/thumbnail/<path:name>', methods=['POST'])
+    def set_thumbnail(name):
+        """将指定时间点设为缩略图。"""
+        try:
+            media_root: Path = Path(app.config["MEDIA_ROOT"])
+            target = (media_root / name).resolve()
+            if not target.exists() or not str(target).startswith(str(media_root.resolve())):
+                return {'success': False, 'error': '文件不存在或非法路径'}, 400
+
+            payload = request.get_json(silent=True) or {}
+            ts = payload.get('time', None)
+            try:
+                ts_val = None if ts is None else max(0.0, float(ts))
+            except (TypeError, ValueError):
+                ts_val = None
+
+            rel_path = str(Path(name))
+            thumb_file = _thumb_path(rel_path)
+            # 重新生成
+            ok = _ffmpeg_capture(target, thumb_file, ts_val)
+            if not ok:
+                return {'success': False, 'error': '缩略图生成失败（可能未安装 ffmpeg）'}, 500
+
+            # 记录映射
+            mapping = _load_thumb_map()
+            mapping[rel_path] = {
+                'ts': ts_val if ts_val is not None else None,
+                'updated_at': datetime.datetime.now().isoformat(timespec='seconds')
+            }
+            _save_thumb_map(mapping)
+
+            return {
+                'success': True,
+                'url': f"/thumb?uri={quote(rel_path)}&v={int(datetime.datetime.now().timestamp())}"
+            }
+        except Exception as exc:
+            print(f"设置缩略图错误: {exc}", file=sys.stderr)
+            return {'success': False, 'error': '内部错误'}, 500
+
     # 添加自定义过滤器
     @app.template_filter('timestamp_to_date')
     def timestamp_to_date(timestamp):
@@ -211,6 +361,26 @@ def get_files(directory, media_type='video'):
     def browse():
         root = Path(app.config["MEDIA_ROOT"])
         videos = list(root.glob('**/*.mp4')) + list(root.glob('**/*.webm'))
+
+        # 大文件快速筛选：通过 query 参数启用（size=big），默认阈值 50MB，可用 min_mb 调整
+        size_mode = request.args.get('size', 'all')
+        try:
+            min_mb = int(request.args.get('min_mb', 50))
+        except ValueError:
+            min_mb = 50
+        has_min_mb = request.args.get('min_mb') is not None
+
+        if size_mode == 'big':
+            threshold = min_mb * 1024 * 1024
+            filtered = []
+            for v in videos:
+                try:
+                    if v.stat().st_size >= threshold:
+                        filtered.append(v)
+                except (FileNotFoundError, PermissionError):
+                    continue
+            videos = filtered
+
         videos = sorted(videos, key=lambda row:row.stat().st_ctime, reverse=True)
         count = len(videos)
         page = int(request.args.get('page', 1))
@@ -228,6 +398,9 @@ def browse():
             length = length,
             files = res,
             menu = 'browse',
+            size_mode = size_mode,
+            min_mb = min_mb,
+            has_min_mb = has_min_mb,
             has_previous = page > 1,
             has_next = len(videos[offset+length:])>1
         )
@@ -337,6 +510,18 @@ def delete_view(name):
                 file_path = Path(app.config["MEDIA_ROOT"]) / name
                 if file_path.exists():
                     os.unlink(file_path)
+                # 清理缩略图及映射
+                try:
+                    rel_path = str(Path(name))
+                    thumb_file = _thumb_path(rel_path)
+                    if thumb_file.exists():
+                        thumb_file.unlink(missing_ok=True)
+                    mapping = _load_thumb_map()
+                    if rel_path in mapping:
+                        mapping.pop(rel_path, None)
+                        _save_thumb_map(mapping)
+                except Exception as _:
+                    pass
                 return redirect('/browse')
             except Exception as e:
                 print(f"删除文件错误: {e}")

tiklocal/app.py

@@ -0,0 +1,26 @@
+import os
+from pathlib import Path
+
+
+def get_data_dir() -> Path:
+    """Return the global data directory for TikLocal.
+    Default to ~/.tiklocal unless TIKLOCAL_INSTANCE is set.
+    """
+    base = os.environ.get('TIKLOCAL_INSTANCE')
+    if base:
+        p = Path(base).expanduser()
+    else:
+        p = Path.home() / '.tiklocal'
+    p.mkdir(parents=True, exist_ok=True)
+    return p
+
+
+def get_thumbnails_dir() -> Path:
+    d = get_data_dir() / 'thumbnails'
+    d.mkdir(parents=True, exist_ok=True)
+    return d
+
+
+def get_thumbs_map_path() -> Path:
+    return get_data_dir() / 'thumbs.json'
+

tiklocal/paths.py

@@ -4,6 +4,8 @@
 from pathlib import Path
 from waitress import serve
 from tiklocal.app import create_app
+from tiklocal.thumbs import generate_thumbnails
+from tiklocal.paths import get_data_dir
 
 try:
     import yaml
@@ -40,39 +42,70 @@ def main():
     # 读取配置文件
     config = load_config()
 
-    # 解析命令行参数
+    # 预处理 argv，支持以下形式：
+    # 1) tiklocal                      -> serve
+    # 2) tiklocal /path                -> serve /path
+    # 3) tiklocal --port 9000          -> serve --port 9000
+    # 4) tiklocal thumbs /path         -> thumbs /path
+    # 5) tiklocal /path thumbs         -> thumbs /path
+    argv = sys.argv[1:]
+    if '-h' not in argv and '--help' not in argv:
+        if 'thumbs' in argv:
+            idx = argv.index('thumbs')
+            if idx != 0:
+                argv.pop(idx)
+                argv.insert(0, 'thumbs')
+        elif len(argv) == 0 or argv[0] not in ('serve', 'thumbs'):
+            # 默认回退 serve（空参数或第一个不是已知子命令）
+            argv.insert(0, 'serve')
+
+    # 解析命令行参数（支持子命令）
     parser = argparse.ArgumentParser(
         description='TikLocal - 本地媒体服务器',
         formatter_class=argparse.RawDescriptionHelpFormatter,
         epilog='''
 示例:
-  tiklocal                           # 使用配置文件或环境变量
-  tiklocal /path/to/media            # 指定媒体目录
-  tiklocal --port 9000               # 使用指定端口
-  tiklocal /path/to/media --port 9000
+  tiklocal                                 # 启动服务（默认）
+  tiklocal /path/to/media                  # 指定媒体目录
+  tiklocal --port 9000                     # 使用指定端口
+  tiklocal serve /path --port 9000         # 显式使用 serve 子命令
+  tiklocal thumbs /path --overwrite        # 批量生成缩略图
         '''
     )
 
-    parser.add_argument(
-        'media_root',
-        nargs='?',
-        help='媒体文件根目录路径'
-    )
-    parser.add_argument(
-        '--host',
-        default=None,
-        help='服务器监听地址 (默认: 0.0.0.0)'
-    )
-    parser.add_argument(
-        '--port',
-        type=int,
-        default=None,
-        help='服务器端口 (默认: 8000)'
-    )
-
-    args = parser.parse_args()
-
-    # 配置优先级: 命令行参数 > 环境变量 > 配置文件 > 默认值
+    subparsers = parser.add_subparsers(dest='command')
+
+    # serve 子命令
+    serve_parser = subparsers.add_parser('serve', help='启动服务器')
+    serve_parser.add_argument('media_root', nargs='?', help='媒体文件根目录路径')
+    serve_parser.add_argument('--host', default=None, help='服务器监听地址 (默认: 0.0.0.0)')
+    serve_parser.add_argument('--port', type=int, default=None, help='服务器端口 (默认: 8000)')
+
+    # thumbs 子命令
+    thumbs_parser = subparsers.add_parser('thumbs', help='批量生成视频缩略图')
+    thumbs_parser.add_argument('media_root', nargs='?', help='媒体文件根目录路径（可省略以使用环境变量/配置文件）')
+    thumbs_parser.add_argument('--overwrite', action='store_true', help='存在时覆盖重建')
+    thumbs_parser.add_argument('--limit', type=int, default=0, help='最多处理多少个（0 表示全部）')
+
+    args = parser.parse_args(argv)
+
+    # 判断命令类型（无子命令时视为 serve）
+    cmd = args.command or 'serve'
+
+    if cmd == 'thumbs':
+        media_root = args.media_root or os.environ.get('MEDIA_ROOT') or config.get('media_root')
+        if not media_root:
+            parser.error('必须指定媒体目录:\n  - tiklocal thumbs /path/to/media\n  - 或设置环境变量: MEDIA_ROOT=/path/to/media')
+        media_path = Path(media_root)
+        if not media_path.exists() or not media_path.is_dir():
+            print(f"错误: 媒体目录不可用: {media_root}", file=sys.stderr)
+            sys.exit(1)
+        print(f"数据目录: {get_data_dir()}")
+        stats = generate_thumbnails(media_path, overwrite=getattr(args, 'overwrite', False), limit=getattr(args, 'limit', 0), show_progress=True)
+        # 完成后退出
+        return
+
+    # serve 路径
     media_root = args.media_root or os.environ.get('MEDIA_ROOT') or config.get('media_root')
     host = args.host or os.environ.get('TIKLOCAL_HOST') or config.get('host', '0.0.0.0')
     port = args.port or int(os.environ.get('TIKLOCAL_PORT', 0)) or config.get('port', 8000)
@@ -96,6 +129,7 @@ def main():
     # 启动服务器
     print(f"启动 TikLocal 服务器...")
     print(f"媒体目录: {media_path.absolute()}")
+    print(f"数据目录: {get_data_dir()}")
     print(f"访问地址: http://{host}:{port}")
 
     serve(create_app(), host=host, port=port)