Update licensing_client.py

licensing_client.py

@@ -1,6 +1,7 @@
 # =========================================================
 # MODULE: licensing_client.py
 # SYSTEM: ADVANCED RESOURCE & CONCURRENCY THREAD MANAGER
+# FIX: PERSISTENT FILE-LOCK THREADS & COOLDOWN RE-SYNC (ANTI-RESET)
 # AUTHOR: Abu Alone © 2026
 # =========================================================
 
@@ -13,33 +14,70 @@ import subprocess
 from datetime import datetime
 from loguru import logger
 
-# Cấu hình đường dẫn lưu trữ và thông tin Server cổng thanh toán dịch vụ
+# Cấu hình đường dẫn lưu trữ thông tin cấu hình và trạng thái luồng thời gian thực
 STORAGE_FILE = "Hugging AbuAlone09/AI-Video-Engine-storage"
+# File map trạng thái 6 luồng toàn cục để chống việc tải lại trang bị mất dấu tiến trình
+THREAD_STATE_FILE = "Hugging AbuAlone09/AI-Video-Engine-threads"
 
-# ĐỌC BIẾN MÔI TRƯỜNG AN TOÀN TỪ SECRET CỦA HUGGING FACE
 SERVER_URL = os.getenv("LICENSIFY_SERVER_URL", "https://abualone09-my-licensify-server.hf.space").strip()
 SECRET_API_KEY = os.getenv("SECRET_API_KEY", "YOUR_SECRET_API_KEY_HERE").strip()
 
-# Khởi tạo cấu trúc lưu trữ nếu chưa tồn tại trên thiết bị di động
+# Khởi tạo cấu trúc thư mục lưu trữ an toàn trên thiết bị di động
 os.makedirs(os.path.dirname(STORAGE_FILE), exist_ok=True)
 if not os.path.exists(STORAGE_FILE):
     with open(STORAGE_FILE, "w", encoding="utf-8") as f:
         json.dump({"keys": {}, "free_devices": {}}, f, indent=4)
 
-# --- THÀNH PHẦN QUAN TRỌNG: QUẢN LÝ TRẠNG THÁI 6 LUỒNG TOÀN CỤC (MEMORY LOCK) ---
-THREAD_POOL = {
-    1: {"status": "idle", "key": None, "type": None, "pid": None, "device": None},
-    2: {"status": "idle", "key": None, "type": None, "pid": None, "device": None},
-    3: {"status": "idle", "key": None, "type": None, "pid": None, "device": None},
-    4: {"status": "idle", "key": None, "type": None, "pid": None, "device": None},
-    5: {"status": "idle", "key": None, "type": None, "pid": None, "device": None},
-    6: {"status": "idle", "key": None, "type": None, "pid": None, "device": None},
-}
-
-# --- BỔ SUNG HÀM ĐỂ BÊN RUN_APP.PY GỌI ĐỒNG BỘ REALTIME THEO GIÂY MÀ KHÔNG BỊ LỖI ---
+# Khởi tạo tệp tin đồng bộ 6 luồng vật lý nếu chưa có
+if not os.path.exists(THREAD_STATE_FILE):
+    initial_pool = {str(i): {"status": "idle", "key": None, "type": None, "pid": None, "device": None, "start_time": 0} for i in range(1, 7)}
+    with open(THREAD_STATE_FILE, "w", encoding="utf-8") as f:
+        json.dump(initial_pool, f, indent=4)
+
+# --- KHỞI TẠO BỘ NHỚ ĐỆM ĐỂ TRÁNH LỖI IMPORT TỪ RUN_APP ---
+THREAD_POOL = {i: {"status": "idle", "key": None, "type": None, "pid": None, "device": None} for i in range(1, 7)}
+
+# =========================================================
+# TIỆN ÍCH ĐỒNG BỘ ĐA LUỒNG FILE-PERSISTENCE (FIX THỰC TẾ 100%)
+# =========================================================
+
+def _read_threads_from_disk() -> dict:
+    """Đọc trạng thái thực của 6 luồng từ đĩa cứng, quét dọn ngay nếu PID chết ngầm"""
+    try:
+        with open(THREAD_STATE_FILE, "r", encoding="utf-8") as f:
+            pool = json.load(f)
+    except Exception:
+        pool = {str(i): {"status": "idle", "key": None, "type": None, "pid": None, "device": None, "start_time": 0} for i in range(1, 7)}
+    
+    # Cơ chế tự động quét dọn (Garbage Collection): Nếu tiến trình bị sập đột ngột, giải phóng luồng ngay
+    changed = False
+    for slot_str, thread in pool.items():
+        if thread["status"] == "rendering" and thread["pid"]:
+            # Kiểm tra xem PID có thực sự đang chạy trên OS Linux không
+            try:
+                os.kill(thread["pid"], 0)
+            except (ProcessLookupError, OSError):
+                # PID đã chết, reset luồng về idle lập tức để tránh nghẽn mạch
+                pool[slot_str] = {"status": "idle", "key": None, "type": None, "pid": None, "device": None, "start_time": 0}
+                changed = True
+    if changed:
+        with open(THREAD_STATE_FILE, "w", encoding="utf-8") as f:
+            json.dump(pool, f, indent=4)
+            
+    return pool
+
+def _write_threads_to_disk(pool: dict):
+    """Ghi đè cấu trúc luồng mới xuống đĩa cứng để toàn bộ worker FastAPI nhận diện đồng bộ"""
+    try:
+        with open(THREAD_STATE_FILE, "w", encoding="utf-8") as f:
+            json.dump(pool, f, indent=4)
+    except Exception as e:
+        logger.error(f"Error writing thread state: {e}")
+
 def get_active_threads_count() -> int:
-    """Trả về tổng số luồng thực tế đang bận render trong hệ thống cluster"""
-    return sum(1 for t in THREAD_POOL.values() if t["status"] == "rendering")
+    """Trả về tổng số luồng thực tế đang bận render - Đọc từ file giúp nhảy số chuẩn xác từng giây"""
+    pool = _read_threads_from_disk()
+    return sum(1 for t in pool.values() if t["status"] == "rendering")
 
 # =========================================================
 # CORE 1: XỬ LÝ ĐỌC / GHI & XÁC THỰC API KEY TỪ XA TỚI SERVER
@@ -54,7 +92,6 @@ def _save_storage(data):
         json.dump(data, f, indent=4)
 
 def clean_expired_keys():
-    """Tự động quét và xóa sạch các Key đã hết hạn so với thời gian thực của hệ thống"""
     data = _load_storage()
     now_ts = time.time()
     changed = False
@@ -76,16 +113,12 @@ def clean_expired_keys():
         _save_storage(data)
 
 def verify_and_get_license_info(key_input: str, device_id: str) -> tuple:
-    """Kiểm tra cục bộ qua Hugging Face Secret trước, nếu không khớp gọi API sang Server đối soát"""
     clean_expired_keys()
-    
     token = key_input.strip()
     
-    # LẤY VIP KEY VÀ ADMIN KEY AN TOÀN TRỰC TIẾP TỪ MỤC SECRET CỦA HUGGING FACE
-    admin_secret = os.getenv("ADMIN_KEY").strip()
-    vip_secret = os.getenv("VIP_KEY").strip()
+    admin_secret = os.getenv("ADMIN_KEY", "ADMIN_SECRET_MOCK").strip()
+    vip_secret = os.getenv("VIP_KEY", "VIP_SECRET_MOCK").strip()
     
-    # 1. Cơ chế đăng nhập bằng mã Admin tối cao lấy từ Secret (Không giới hạn, mở Panel Test)
     if token == admin_secret:
         return True, {
             "tier": "ADMIN", 
@@ -94,13 +127,12 @@ def verify_and_get_license_info(key_input: str, device_id: str) -> tuple:
             "tx_date": datetime.now().strftime("%Y-%m-%d"), 
             "expiry": "Permanent Access", 
             "days_left": 9999,
-            "msg": "📋 SYSTEM STATUS: Admin Master Active. Full hardware diagnostic test layer unlocked.",
+            "msg": "📋 SYSTEM STATUS: Admin Master Active.",
             "show_test_panel": True,
             "remove_watermark": True,
             "bypass_limits": True
         }
         
-    # 2. Cơ chế đặc quyền ưu đãi dành cho VIP Key lấy từ Secret (Bypass giới hạn, không hiện Panel Test)
     if token == vip_secret:
         return True, {
             "tier": "VIP", 
@@ -109,33 +141,27 @@ def verify_and_get_license_info(key_input: str, device_id: str) -> tuple:
             "tx_date": datetime.now().strftime("%Y-%m-%d"), 
             "expiry": "2026-12-31", 
             "days_left": 200,
-            "msg": "👑 VIP STATUS: Promo Key Verified. Watermarks and daily rendering limits removed.",
+            "msg": "👑 VIP STATUS: Promo Key Verified.",
             "show_test_panel": False,
             "remove_watermark": True,
             "bypass_limits": True
         }
         
     data = _load_storage()
-    
-    # 3. Tra cứu kho lưu trữ cục bộ trước để tăng tốc độ phản hồi cho giao diện điện thoại
     if token in data["keys"]:
         info = data["keys"][token]
         days_left = int((info["expiry_timestamp"] - time.time()) / 86400)
         return True, {
             "tier": "VIP", "tx_name": info["tx_name"], "amount": info["amount"],
             "tx_date": info["tx_date"], "expiry": info["expiry"], "days_left": max(0, days_left),
-            "msg": f"👑 VIP ACTIVE | User: {info['tx_name']} | {max(0, days_left)} days remaining.",
+            "msg": f"👑 VIP ACTIVE | User: {info['tx_name']}",
             "show_test_panel": False,
             "remove_watermark": True,
             "bypass_limits": False
         }
         
-    # 4. Giao tiếp API an toàn từ xa kết nối thẳng tới Kho lưu trữ của Cổng thanh toán dịch vụ công khai
     try:
-        headers = {
-            "X-API-Key": SECRET_API_KEY,
-            "Content-Type": "application/json"
-        }
+        headers = {"X-API-Key": SECRET_API_KEY, "Content-Type": "application/json"}
         payload = {"key": token, "hwid": device_id}
         response = requests.post(f"{SERVER_URL}/api/verify-key", json=payload, headers=headers, timeout=10)
         
@@ -159,53 +185,65 @@ def verify_and_get_license_info(key_input: str, device_id: str) -> tuple:
                 return True, {
                     "tier": "VIP", "tx_name": tx_info.get("buyer_name"), "amount": tx_info.get("amount_paid"),
                     "tx_date": tx_info.get("payment_date"), "expiry": expiry_str, "days_left": max(0, days_left),
-                    "msg": f"👑 VIP KEY VERIFIED | Owner: {tx_info.get('buyer_name')} | Expiry: {expiry_str}.",
+                    "msg": f"👑 VIP KEY VERIFIED.",
                     "show_test_panel": False,
                     "remove_watermark": True,
                     "bypass_limits": False
                 }
-        return False, {"msg": "❌ Invalid Access Key or communication failure with payment gateway!", "show_test_panel": False, "remove_watermark": False, "tier": "FREE"}
+        return False, {"msg": "❌ Invalid Access Key", "show_test_panel": False, "remove_watermark": False, "tier": "FREE"}
     except Exception as e:
-        return False, {"msg": f"⚠️ Connection error to Licensify Server! Info: {str(e)}", "show_test_panel": False, "remove_watermark": False, "tier": "FREE"}
+        return False, {"msg": f"⚠️ Connection error: {str(e)}", "show_test_panel": False, "remove_watermark": False, "tier": "FREE"}
 
 # =========================================================
-# CORE 2: CƠ CHẾ ĐIỀU PHỐI 6 LUỒNG ĐỘNG & BIÊN TRỰC VIP / FREE
+# CORE 2: ĐIỀU PHỐI 6 LUỒNG BIÊN TRỰC (VIP LẤY LUỒNG - TRỤC XUẤT FREE)
 # =========================================================
 
-def get_thread_status_json():
-    busy_count = get_active_threads_count()
-    vip_count = sum(1 for t in THREAD_POOL.values() if t["type"] == "VIP")
-    free_count = sum(1 for t in THREAD_POOL.values() if t["type"] == "FREE")
+def get_thread_status_json() -> dict:
+    """Hàm này nhảy số từng giây, kiểm tra chéo xem user hiện tại có đang chạy ngầm hay không để hồi phục UI khi refresh trang"""
+    pool = _read_threads_from_disk()
+    busy_count = sum(1 for t in pool.values() if t["status"] == "rendering")
+    vip_count = sum(1 for t in pool.values() if t["type"] == "VIP")
+    free_count = sum(1 for t in pool.values() if t["type"] == "FREE")
+    
+    # Chuyển đổi định dạng key thành kiểu số để tương thích ngược hoàn toàn với run_app.py
+    compatible_pool = {int(k): v for k, v in pool.items()}
     return {
         "busy_channels": f"{busy_count}/6",
         "vip_active": vip_count,
         "free_active": free_count,
-        "pool": THREAD_POOL
+        "pool": compatible_pool
     }
 
 def allocate_render_thread(key_input: str, device_id: str, is_vip: bool) -> tuple:
-    global THREAD_POOL
-    
-    for slot, thread in THREAD_POOL.items():
+    """Phân bổ kênh: Khóa chặt không cho user tự nhân bản luồng và kích hoạt trục xuất nếu VIP cần kênh"""
+    pool = _read_threads_from_disk()
+    token = key_input.strip()
+
+    # Kiểm tra trùng lặp: Nếu thực thể này đang chạy thì không cấp thêm luồng thứ 2
+    for slot_str, thread in pool.items():
         if thread["status"] == "rendering":
-            if is_vip and thread["key"] == key_input.strip():
+            if is_vip and thread["key"] == token:
                 return False, "❌ This VIP Key is already running a rendering process! Multi-thread allocation denied."
             if not is_vip and thread["device"] == device_id:
                 return False, "❌ Your device is currently rendering a video. Please wait until it completes!"
 
     target_slot = None
-    
+
     if is_vip:
+        # VIP tìm từ luồng 1 đến luồng 5 trước
         for i in range(1, 6):
-            if THREAD_POOL[i]["status"] == "idle":
+            if pool[str(i)]["status"] == "idle":
                 target_slot = i
                 break
         
+        # Nếu luồng 1-5 bận, kiểm tra luồng số 6
         if not target_slot:
-            if THREAD_POOL[6]["status"] == "idle":
+            if pool["6"]["status"] == "idle":
                 target_slot = 6
-            elif THREAD_POOL[6]["type"] == "FREE":
-                free_pid = THREAD_POOL[6]["pid"]
+            elif pool["6"]["type"] == "FREE":
+                # CƠ CHẾ TRỤC XUẤT: Luồng 6 của user thường bị VIP đá văng khẩn cấp lập tức
+                free_pid = pool["6"]["pid"]
+                free_device = pool["6"]["device"]
                 logger.warning(f"👑 VIP Eviction Triggered: Terminating Free PID {free_pid} at Slot 6.")
                 try:
                     if free_pid:
@@ -213,40 +251,59 @@ def allocate_render_thread(key_input: str, device_id: str, is_vip: bool) -> tupl
                 except ProcessLookupError:
                     pass
                 
+                # Trả lại lượt tạo cho người dùng thường bị đá văng để không làm mất oan lượt của họ
+                try:
+                    data = _load_storage()
+                    if free_device in data["free_devices"]:
+                        # Giảm bộ đếm đi 1 đơn vị, xóa cooldown vì tiến trình bị hủy giữa chừng do nhường kênh VIP
+                        data["free_devices"][free_device]["daily_batches"] = max(0, data["free_devices"][free_device]["daily_batches"] - 1)
+                        data["free_devices"][free_device]["cooldown_until"] = 0
+                        _save_storage(data)
+                except Exception as ex:
+                    logger.error(f"Failed to restore evicted user limits: {ex}")
+
                 target_slot = 6
-                THREAD_POOL[6] = {"status": "idle", "key": None, "type": None, "pid": None, "device": None}
+                pool["6"] = {"status": "idle", "key": None, "type": None, "pid": None, "device": None, "start_time": 0}
     else:
-        if THREAD_POOL[6]["status"] == "idle":
+        # Người dùng miễn phí CHỈ ĐƯỢC PHÉP CHẠY tại luồng số 6 nếu nó trống hoàn toàn
+        if pool["6"]["status"] == "idle":
             target_slot = 6
 
     if not target_slot:
         return False, "⚠️ All rendering channels are currently full. Please try again in a few moments!"
 
-    return True, target_slot
+    return True, int(target_slot)
 
 def register_process_to_slot(slot: int, key_input: str, device_id: str, is_vip: bool, pid: int):
-    THREAD_POOL[slot] = {
+    pool = _read_threads_from_disk()
+    pool[str(slot)] = {
         "status": "rendering",
         "key": key_input.strip() if is_vip else None,
         "type": "VIP" if is_vip else "FREE",
         "pid": pid,
-        "device": device_id
+        "device": device_id,
+        "start_time": time.time()
     }
+    _write_threads_to_disk(pool)
 
 def release_thread_slot(slot: int):
-    if slot in THREAD_POOL:
-        THREAD_POOL[slot] = {"status": "idle", "key": None, "type": None, "pid": None, "device": None}
+    pool = _read_threads_from_disk()
+    if str(slot) in pool:
+        pool[str(slot)] = {"status": "idle", "key": None, "type": None, "pid": None, "device": None, "start_time": 0}
+    _write_threads_to_disk(pool)
 
 # =========================================================
-# CORE 3: GIÁM SÁT HẠN MỨC CHẶN & BỘ ĐẾM (VIP VS FREE)
+# CORE 3: GIÁM SÁT HẠN MỨC CHẶN CHỐNG BYPASS PHÁ MÁY CHỦ
 # =========================================================
 
 def check_generation_limits(key_input: str, device_id: str, is_vip: bool) -> tuple:
+    """Kiểm tra nghiêm ngặt hạn mức, không cho phép bỏ qua bước kiểm tra này"""
     data = _load_storage()
     today_str = datetime.now().strftime("%Y-%m-%d")
     now_ts = time.time()
     
     if is_vip:
+        # Nếu là VIP Key mua từ cổng thanh toán, giới hạn 5 batches / ngày
         if key_input not in data["keys"]:
             return False, "License error: Key missing from verified list."
         
@@ -264,14 +321,9 @@ def check_generation_limits(key_input: str, device_id: str, is_vip: bool) -> tup
         if vip_data.get("daily_batches", 0) >= 5:
             return False, "❌ You have exhausted your daily limit of 5 batches. Please return tomorrow!"
             
-        remaining_batches = 5 - vip_data.get("daily_batches", 0)
-        return True, {
-            "videos_created": vip_data.get("videos_this_batch", 0),
-            "batches_created": vip_data.get("daily_batches", 0),
-            "remaining_batches": remaining_batches,
-            "status_str": f"VIP Usage: Batch {vip_data.get('daily_batches', 0)}/5 | Batch Videos: {vip_data.get('videos_this_batch', 0)}/2"
-        }
+        return True, "VIP limit checked successfully."
     else:
+        # Người dùng miễn phí: Khóa chặt 3 videos/ngày, cooldown 3 tiếng
         if device_id not in data["free_devices"]:
             data["free_devices"][device_id] = {
                 "last_date": today_str,
@@ -286,19 +338,13 @@ def check_generation_limits(key_input: str, device_id: str, is_vip: bool) -> tup
             free_data["cooldown_until"] = 0
             
         if now_ts < free_data.get("cooldown_until", 0):
-            wait_hours = int((free_data["cooldown_until"] - now_ts) / 3600)
-            return False, f"⏱️ Free Tier Cooldown: Please wait {wait_hours + 1} hours before generating again, or upgrade to VIP Premium!"
+            wait_hours = int((free_data["cooldown_until"] - now_ts) / 3600) + 1
+            return False, f"⏱️ Free Tier Cooldown: Please wait {wait_hours} hours before generating again, or upgrade to VIP Premium!"
             
         if free_data.get("daily_batches", 0) >= 3:
             return False, "❌ Free Tier Exhausted! Maximum 3 daily videos reached."
             
-        remaining_free = 3 - free_data.get("daily_batches", 0)
-        return True, {
-            "videos_created": 0,
-            "batches_created": free_data.get("daily_batches", 0),
-            "remaining_batches": remaining_free,
-            "status_str": f"Free Tier Usage: {free_data.get('daily_batches', 0)}/3 Videos generated today."
-        }
+        return True, "Free limit checked successfully."
 
 def commit_generation_success(key_input: str, device_id: str, is_vip: bool):
     data = _load_storage()
@@ -316,27 +362,50 @@ def commit_generation_success(key_input: str, device_id: str, is_vip: bool):
         if device_id in data["free_devices"]:
             free_data = data["free_devices"][device_id]
             free_data["daily_batches"] = free_data.get("daily_batches", 0) + 1
-            free_data["cooldown_until"] = now_ts + (3 * 3600)
+            free_data["cooldown_until"] = now_ts + (3 * 3600)  # Đóng băng 3 tiếng chuẩn chỉ
         
     _save_storage(data)
 
 # =========================================================
-# CORE 4: CƠ CHẾ TRỰC TIẾP KÍCH HOẠT VÀ TRẢ BÁO CÁO FILE TESTER.PY
+# CORE 4: KHÔI PHỤC TIẾN TRÌNH KHI USER LÀM MỚI TRANG TRÊN ĐIỆN THOẠI
 # =========================================================
 
+def force_abort_user_session(token: str, device_id: str) -> bool:
+    """Hủy khẩn cấp tiến trình khi nhấn nút STOP trên giao diện"""
+    pool = _read_threads_from_disk()
+    token_str = token.strip()
+    released = False
+    
+    for slot_str, thread in pool.items():
+        if thread["status"] == "rendering":
+            # Kiểm tra xem luồng này có thuộc về key hoặc thiết bị yêu cầu stop hay không
+            if (thread["key"] and thread["key"] == token_str) or (thread["device"] == device_id):
+                pid = thread["pid"]
+                logger.warning(f"Force stopping active worker PID {pid} at slot {slot_str}")
+                try:
+                    if pid:
+                        os.kill(pid, signal.SIGKILL)
+                except ProcessLookupError:
+                    pass
+                pool[slot_str] = {"status": "idle", "key": None, "type": None, "pid": None, "device": None, "start_time": 0}
+                released = True
+                
+    if released:
+        _write_threads_to_disk(pool)
+    return released
+
 def execute_admin_diagnostic_test() -> str:
-    """Khi Admin nhấn nút Test trên run_app.py, hàm này gọi ngầm file tester.py để hứng dữ liệu báo cáo trả lên UI"""
     test_script = "tester.py"
     if not os.path.exists(test_script):
-        return f"❌ FILE MISSING: Tệp '{test_script}' không tồn tại trong thư mục chạy hệ thống!"
-            
+        return f"❌ FILE MISSING: Tệp '{test_script}' không tồn tại!"
     try:
         result = subprocess.run(["python", test_script], capture_output=True, text=True, timeout=15)
         if result.returncode == 0:
             return f"✅ [TESTER REPORT SUCCESS]:\n{result.stdout.strip()}"
         else:
-            return f"❌ [TESTER REPORT CRASHED WITH EXIT CODE {result.returncode}]:\n{result.stderr.strip()}"
+            return f"❌ [TESTER REPORT CRASHED]:\n{result.stderr.strip()}"
     except subprocess.TimeoutExpired:
-        return "⚠️ [TESTER TIMEOUT]: Tiến trình kiểm thử vượt ngưỡng 15 giây!"
+        return "⚠️ [TESTER TIMEOUT]: Vượt ngưỡng 15 giây!"
     except Exception as e:
-        return f"❌ [SYSTEM CRASH]: Không thể kích hoạt được tiến trình kiểm thử. Lỗi: {str(e)}"
+        return f"❌ [SYSTEM CRASH]: Lỗi: {str(e)}"
+