99 lines
2.6 KiB
Python
99 lines
2.6 KiB
Python
"""
|
|
路径规范化模块：按用户隔离 Cookie 存储
|
|
"""
|
|
from pathlib import Path
|
|
import re
|
|
from typing import Set
|
|
|
|
# 基础目录
|
|
BASE_DIR = Path(__file__).parent.parent.parent
|
|
USER_DATA_DIR = BASE_DIR / "user_data"
|
|
|
|
# 有效的平台列表
|
|
VALID_PLATFORMS: Set[str] = {"bilibili", "douyin", "xiaohongshu"}
|
|
|
|
# UUID 格式正则
|
|
UUID_PATTERN = re.compile(r'^[a-f0-9]{8}-[a-f0-9]{4}-[a-f0-9]{4}-[a-f0-9]{4}-[a-f0-9]{12}$', re.IGNORECASE)
|
|
|
|
|
|
def validate_user_id(user_id: str) -> bool:
|
|
"""验证 user_id 格式 (防止路径遍历攻击)"""
|
|
return bool(UUID_PATTERN.match(user_id))
|
|
|
|
|
|
def validate_platform(platform: str) -> bool:
|
|
"""验证平台名称"""
|
|
return platform in VALID_PLATFORMS
|
|
|
|
|
|
def get_user_data_dir(user_id: str) -> Path:
|
|
"""
|
|
获取用户数据根目录
|
|
|
|
Args:
|
|
user_id: 用户 UUID
|
|
|
|
Returns:
|
|
用户数据目录路径
|
|
|
|
Raises:
|
|
ValueError: user_id 格式无效
|
|
"""
|
|
if not validate_user_id(user_id):
|
|
raise ValueError(f"Invalid user_id format: {user_id}")
|
|
|
|
user_dir = USER_DATA_DIR / user_id
|
|
user_dir.mkdir(parents=True, exist_ok=True)
|
|
return user_dir
|
|
|
|
|
|
def get_user_cookie_dir(user_id: str) -> Path:
|
|
"""
|
|
获取用户 Cookie 目录
|
|
|
|
Args:
|
|
user_id: 用户 UUID
|
|
|
|
Returns:
|
|
Cookie 目录路径
|
|
"""
|
|
cookie_dir = get_user_data_dir(user_id) / "cookies"
|
|
cookie_dir.mkdir(parents=True, exist_ok=True)
|
|
return cookie_dir
|
|
|
|
|
|
def get_platform_cookie_path(user_id: str, platform: str) -> Path:
|
|
"""
|
|
获取平台 Cookie 文件路径
|
|
|
|
Args:
|
|
user_id: 用户 UUID
|
|
platform: 平台名称 (bilibili/douyin/xiaohongshu)
|
|
|
|
Returns:
|
|
Cookie 文件路径
|
|
|
|
Raises:
|
|
ValueError: 平台名称无效
|
|
"""
|
|
if not validate_platform(platform):
|
|
raise ValueError(f"Invalid platform: {platform}. Valid: {VALID_PLATFORMS}")
|
|
|
|
return get_user_cookie_dir(user_id) / f"{platform}_cookies.json"
|
|
|
|
|
|
# === 兼容旧代码的路径 (无用户隔离) ===
|
|
|
|
def get_legacy_cookie_dir() -> Path:
|
|
"""获取旧版 Cookie 目录 (无用户隔离)"""
|
|
cookie_dir = BASE_DIR / "app" / "cookies"
|
|
cookie_dir.mkdir(parents=True, exist_ok=True)
|
|
return cookie_dir
|
|
|
|
|
|
def get_legacy_cookie_path(platform: str) -> Path:
|
|
"""获取旧版 Cookie 路径 (无用户隔离)"""
|
|
if not validate_platform(platform):
|
|
raise ValueError(f"Invalid platform: {platform}")
|
|
return get_legacy_cookie_dir() / f"{platform}_cookies.json"
|