(wip)update file cache pipeline

app.py

@@ -4,6 +4,8 @@ from apscheduler.schedulers.background import BackgroundScheduler
 from concurrent.futures import ThreadPoolExecutor
 from datetime import datetime
 import threading  # Added for locking
+
+from huggingface_hub.hf_api import RepoFile
 from sqlalchemy import or_  # Added for vote counting query
 import hashlib
 
@@ -42,7 +44,6 @@ from flask import (
     redirect,
     url_for,
     session,
-    abort,
 )
 from flask_login import LoginManager, current_user
 from models import *
@@ -61,8 +62,6 @@ import json
 from datetime import datetime, timedelta
 from flask_migrate import Migrate
 import requests
-import functools
-import time  # Added for potential retries
 
 # Load environment variables
 if not IS_SPACES:
@@ -118,6 +117,7 @@ TTS_CACHE_SIZE = int(os.getenv("TTS_CACHE_SIZE", "10"))
 CACHE_AUDIO_SUBDIR = "cache"
 tts_cache = {}  # sentence -> {model_a, model_b, audio_a, audio_b, created_at}
 tts_cache_lock = threading.Lock()
+preload_cache_lock = threading.Lock()
 SMOOTHING_FACTOR_MODEL_SELECTION = 500  # For weighted random model selection
 # Increased max_workers to 8 for concurrent generation/refill
 cache_executor = ThreadPoolExecutor(max_workers=8, thread_name_prefix='CacheReplacer')
@@ -371,6 +371,7 @@ with open("init_sentences.txt", "r") as f:
     initial_sentences = random.sample(all_harvard_sentences,
                                       min(len(all_harvard_sentences), 500))  # Limit initial pass for template
 
+
 @app.route("/")
 def arena():
     # Pass a subset of sentences for the random button fallback
@@ -616,6 +617,11 @@ def generate_tts():
     if not text or len(text) > 1000:
         return jsonify({"error": "Invalid or too long text"}), 400
 
+    prompt_md5 = ''
+    if reference_audio_path and os.path.exists(reference_audio_path):
+        with open(reference_audio_path, 'rb') as f:
+            prompt_md5 = hashlib.md5(f.read()).hexdigest()
+
     # --- Cache Check ---
     cache_hit = False
     session_data_from_cache = None
@@ -662,7 +668,31 @@ def generate_tts():
         return jsonify({"error": "Not enough TTS models available"}), 500
 
     selected_models = get_weighted_random_models(available_models, 2, ModelType.TTS)
+    # 尝试从持久化缓存中查找两个模型的音频
+    audio_a_path = find_cached_audio(str(selected_models[0].id), text, reference_audio_path)
+    audio_b_path = find_cached_audio(str(selected_models[1].id), text, reference_audio_path)
 
+    if audio_a_path and audio_b_path:
+        app.logger.info(f"Persistent Cache HIT for: '{text[:50]}...'. Using files directly.")
+        session_id = str(uuid.uuid4())
+        app.tts_sessions[session_id] = {
+            "model_a": selected_models[0].id,
+            "model_b": selected_models[1].id,
+            "audio_a": audio_a_path,
+            "audio_b": audio_b_path,
+            "text": text,
+            "created_at": datetime.utcnow(),
+            "expires_at": datetime.utcnow() + timedelta(minutes=30),
+            "voted": False,
+        }
+        return jsonify({
+            "session_id": session_id,
+            "audio_a": f"/api/tts/audio/{session_id}/a",
+            "audio_b": f"/api/tts/audio/{session_id}/b",
+            "expires_in": 1800,
+            "cache_hit": True,  # 可以认为这也是一种缓存命中
+        })
+    # --- 持久化缓存检查结束 ---
     try:
         audio_files = []
         model_ids = []
@@ -716,15 +746,16 @@ def generate_tts():
 
         # Check if text and prompt are in predefined libraries
         if text in predefined_texts and prompt_md5 in predefined_prompts.values():
-            preload_key = get_tts_cache_key(str(model_ids[0]), text, reference_audio_path)
-            preload_path = os.path.join(PRELOAD_CACHE_DIR, f"{preload_key}.wav")
-            shutil.copy(audio_files[0], preload_path)
-            app.logger.info(f"Preloaded cache audio saved: {preload_path}")
+            with preload_cache_lock:
+                preload_key = get_tts_cache_key(str(model_ids[0]), text, reference_audio_path)
+                preload_path = os.path.join(PRELOAD_CACHE_DIR, f"{preload_key}.wav")
+                shutil.copy(audio_files[0], preload_path)
+                app.logger.info(f"Preloaded cache audio saved: {preload_path}")
 
-            preload_key = get_tts_cache_key(str(model_ids[1]), text, reference_audio_path)
-            preload_path = os.path.join(PRELOAD_CACHE_DIR, f"{preload_key}.wav")
-            shutil.copy(audio_files[1], preload_path)
-            app.logger.info(f"Preloaded cache audio saved: {preload_path}")
+                preload_key = get_tts_cache_key(str(model_ids[1]), text, reference_audio_path)
+                preload_path = os.path.join(PRELOAD_CACHE_DIR, f"{preload_key}.wav")
+                shutil.copy(audio_files[1], preload_path)
+                app.logger.info(f"Preloaded cache audio saved: {preload_path}")
 
         # Return audio file paths and session
         return jsonify(
@@ -1120,98 +1151,105 @@ def setup_periodic_tasks():
         同步缓存音频到HF dataset并从HF下载更新的缓存��频。
         """
         os.makedirs(PRELOAD_CACHE_DIR, exist_ok=True)
-        try:
-            api = HfApi(token=os.getenv("HF_TOKEN"))
-
-            # 获取带有 etag 的文件列表
-            files_info = api.list_repo_files(repo_id=REFERENCE_AUDIO_DATASET, repo_type="dataset", expand=True)
-            # 只处理cache_audios/下的wav文件
-            wav_files = [f for f in files_info if
-                         f["rfilename"].startswith(CACHE_AUDIO_PATTERN) and f["rfilename"].endswith(".wav")]
-
-            # 获取本地已有文件名及hash集合
-            local_hashes = {}
-            for root, _, filenames in os.walk(PRELOAD_CACHE_DIR):
-                for fname in filenames:
-                    if fname.endswith(".wav"):
-                        rel_path = os.path.relpath(os.path.join(root, fname), PRELOAD_CACHE_DIR)
-                        remote_path = os.path.join(CACHE_AUDIO_PATTERN, rel_path)
-                        local_file_path = os.path.join(root, fname)
-                        # 计算本地文件md5
-                        try:
-                            with open(local_file_path, 'rb') as f:
-                                md5 = hashlib.md5(f.read()).hexdigest()
-                            local_hashes[remote_path] = md5
-                        except Exception:
-                            continue
-
-            download_count = 0
-            for f in wav_files:
-                remote_path = f["rfilename"]
-                etag = f.get("lfs", {}).get("oid") or f.get("etag")  # 优先lfs oid, 其次etag
-                local_md5 = local_hashes.get(remote_path)
-
-                # 如果远端etag为32位md5且与本地一致，跳过下载
-                if etag and len(etag) == 32 and local_md5 == etag:
-                    continue
-
-                # 下载文件
-                local_path = hf_hub_download(
-                    repo_id=REFERENCE_AUDIO_DATASET,
-                    filename=remote_path,
-                    repo_type="dataset",
-                    local_dir=PRELOAD_CACHE_DIR,
-                    token=os.getenv("HF_TOKEN"),
-                    force_download=True if local_md5 else False
-                )
-                print(f"Downloaded cache audio: {local_path}")
-                download_count += 1
+        with preload_cache_lock:
+            try:
+                api = HfApi(token=os.getenv("HF_TOKEN"))
 
-            print(f"Downloaded {download_count} new/updated cache audios from HF to {PRELOAD_CACHE_DIR}")
+                # 获取带有 etag 的文件列表
+                files_info = [
+                    f
+                    for f in api.list_repo_tree(
+                        repo_id=REFERENCE_AUDIO_DATASET, path_in_repo=CACHE_AUDIO_PATTERN.strip("/"), recursive=True,
+                        repo_type="dataset", expand=True
+                    )
+                    if isinstance(f, RepoFile)
+                ]
+                # 只处理cache_audios/下的wav文件
+                wav_files = [f for f in files_info if
+                            f.path.endswith(".wav")]
+
+                # 获取本地已有文件名及hash集合
+                local_hashes = {}
+                for root, _, filenames in os.walk(PRELOAD_CACHE_DIR):
+                    for fname in filenames:
+                        if fname.endswith(".wav"):
+                            rel_path = os.path.relpath(os.path.join(root, fname), PRELOAD_CACHE_DIR)
+                            remote_path = os.path.join(CACHE_AUDIO_PATTERN, rel_path)
+                            local_file_path = os.path.join(root, fname)
+                            # 计算本地文件md5
+                            try:
+                                with open(local_file_path, 'rb') as f:
+                                    file_hash = hashlib.sha256(f.read()).hexdigest()
+                                local_hashes[remote_path] = file_hash
+                            except Exception:
+                                continue
+
+                download_count = 0
+                for f in wav_files:
+                    remote_path = f.path
+                    etag = f.lfs.sha256 if f.lfs else None
+                    local_hash = local_hashes.get(remote_path)
+
+                    # 如果远端etag为32位md5且与本地一致，跳过下载
+                    if local_hash == etag:
+                        continue
+
+                    # 下载文件
+                    local_path = hf_hub_download(
+                        repo_id=REFERENCE_AUDIO_DATASET,
+                        filename=remote_path,
+                        repo_type="dataset",
+                        local_dir=PRELOAD_CACHE_DIR,
+                        token=os.getenv("HF_TOKEN"),
+                        force_download=True if local_hash else False
+                    )
+                    print(f"Downloaded cache audio: {local_path}")
+                    download_count += 1
 
-            # 上传本地文件到HF dataset
-            for root, _, files in os.walk(PRELOAD_CACHE_DIR):
-                for file in files:
-                    if file.endswith('.wav'):
-                        local_path = os.path.join(root, file)
-                        rel_path = os.path.relpath(local_path, PRELOAD_CACHE_DIR)
-                        remote_path = os.path.join(CACHE_AUDIO_PATTERN, rel_path)
+                print(f"Downloaded {download_count} new/updated cache audios from HF to {PRELOAD_CACHE_DIR}")
 
-                        try:
-                            # 计算本地文件MD5，用于检查是否需要上传
-                            with open(local_path, 'rb') as f:
-                                file_md5 = hashlib.md5(f.read()).hexdigest()
+                # 上传本地文件到HF dataset
+                for root, _, files in os.walk(PRELOAD_CACHE_DIR):
+                    for file in files:
+                        if file.endswith('.wav'):
+                            local_path = os.path.join(root, file)
+                            rel_path = os.path.relpath(local_path, PRELOAD_CACHE_DIR)
+                            remote_path = os.path.join(CACHE_AUDIO_PATTERN, rel_path)
 
-                            # 尝试获取远程文件信息
                             try:
-                                remote_info = api.get_file_info(
+                                # 计算本地文件MD5，用于检查是否需要上传
+                                with open(local_path, 'rb') as f:
+                                    file_hash = hashlib.sha256(f.read()).hexdigest()
+
+                                # 尝试获取远程文件信息
+                                try:
+                                    remote_info = api.get_paths_info(
+                                        repo_id=REFERENCE_AUDIO_DATASET,
+                                        repo_type="dataset",
+                                        path=[remote_path],expand=True)
+                                    remote_etag = remote_info[0].lfs.sha256 if remote_info and remote_info[0].lfs else None
+                                    # 如果远程文件存在且hash相同，则跳过
+                                    if remote_etag and remote_etag == file_hash:
+                                        app.logger.debug(f"Skipping upload for {remote_path}: file unchanged")
+                                        continue
+                                except Exception as e:
+                                    app.logger.warning(f"Could not get remote info for {remote_path}: {str(e)}")
+                                # 上传文件
+                                app.logger.info(f"Uploading preload cache file: {remote_path}")
+                                api.upload_file(
+                                    path_or_fileobj=local_path,
+                                    path_in_repo=remote_path,
                                     repo_id=REFERENCE_AUDIO_DATASET,
                                     repo_type="dataset",
-                                    path=remote_path)
-                                remote_etag = remote_info.etag or remote_info.lfs.get("oid", "")
-                                # 如果远程文件存在且hash相同，则跳过
-                                if remote_etag and remote_etag == file_md5:
-                                    app.logger.debug(f"Skipping upload for {remote_path}: file unchanged")
-                                    continue
-                            except Exception:
-                                pass
+                                    commit_message=f"Upload preload cache file: {os.path.basename(file)}"
+                                )
+                                app.logger.info(f"Successfully uploaded {remote_path}")
+                            except Exception as e:
+                                app.logger.error(f"Error uploading {remote_path}: {str(e)}")
 
-                            # 上传文件
-                            app.logger.info(f"Uploading preload cache file: {remote_path}")
-                            api.upload_file(
-                                path_or_fileobj=local_path,
-                                path_in_repo=remote_path,
-                                repo_id=REFERENCE_AUDIO_DATASET,
-                                repo_type="dataset",
-                                commit_message=f"Upload preload cache file: {os.path.basename(file)}"
-                            )
-                            app.logger.info(f"Successfully uploaded {remote_path}")
-                        except Exception as e:
-                            app.logger.error(f"Error uploading {remote_path}: {str(e)}")
-
-        except Exception as e:
-            print(f"Error syncing cache audios with HF: {e}")
-            app.logger.error(f"Error syncing cache audios with HF: {e}")
+            except Exception as e:
+                print(f"Error syncing cache audios with HF: {e}")
+                app.logger.error(f"Error syncing cache audios with HF: {e}")
 
     # Schedule periodic tasks
     scheduler = BackgroundScheduler()
@@ -1354,9 +1392,6 @@ def get_tts_cache_key(model_name, text, prompt_audio_path):
     return hashlib.md5(key_str.encode('utf-8')).hexdigest()
 
 
-
-
-
 if __name__ == "__main__":
     with app.app_context():
         # Ensure ./instance and ./votes directories exist