Spaces:
Sleeping
Sleeping
File size: 4,580 Bytes
551658a | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 | from fastapi import HTTPException, status
from sqlalchemy.orm import Session
from models.product import Product
from models.user import User
from utils.r2_uploader import get_r2_uploader, R2Config, get_cached_config, set_cached_config
from models.company_r2_config import CompanyR2Config
from typing import Optional
from dotenv import load_dotenv
from pathlib import Path
import os
# 加载.env文件
BASE_DIR = Path(__file__).parent.parent
load_dotenv(BASE_DIR / ".env")
def get_r2_config_from_env() -> Optional[R2Config]:
"""从.env文件获取R2配置"""
r2_account_id = os.getenv('R2_ACCOUNT_ID', '')
r2_access_key_id = os.getenv('R2_ACCESS_KEY_ID', '')
r2_secret_access_key = os.getenv('R2_SECRET_ACCESS_KEY', '')
r2_bucket_name = os.getenv('R2_BUCKET_NAME', 'yomaton')
r2_public_url = os.getenv('R2_PUBLIC_URL', '')
r2_enabled = os.getenv('R2_ENABLED', 'true').lower() == 'true'
if r2_enabled and r2_account_id and r2_access_key_id and r2_secret_access_key:
return R2Config({
'r2_account_id': r2_account_id,
'r2_access_key_id': r2_access_key_id,
'r2_secret_access_key': r2_secret_access_key,
'r2_bucket_name': r2_bucket_name,
'r2_public_url': r2_public_url,
'r2_enabled': 1 if r2_enabled else 0
})
return None
def validate_product_exists(
product_number: str,
current_user: User,
db: Session
) -> Product:
"""
验证产品是否存在并返回产品对象
Args:
product_number: 产品货号
current_user: 当前用户
db: 数据库会话
Returns:
产品对象
Raises:
HTTPException: 当产品不存在时抛出404错误
"""
from routers.auth import is_admin_user
query = db.query(Product).filter(Product.product_number == product_number)
if not is_admin_user(current_user):
query = query.filter(Product.company_code == current_user.company_code)
product = query.first()
if product is None:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="Product not found"
)
return product
def get_company_r2_config(company_code: str, db: Session) -> Optional[R2Config]:
"""
获取公司的 R2 配置
优先从缓存获取，没有则从数据库获取，最后从.env获取并缓存
Args:
company_code: 公司代码
db: 数据库会话
Returns:
R2配置对象，如果不存在则返回None
"""
cached_config = get_cached_config(company_code)
if cached_config:
return cached_config
# 先从当前公司的数据库配置获取
db_config = db.query(CompanyR2Config).filter(
CompanyR2Config.company_code == company_code
).first()
# 如果没有，尝试从默认公司(0000)获取
if not db_config:
db_config = db.query(CompanyR2Config).filter(
CompanyR2Config.company_code == "0000"
).first()
if db_config and db_config.r2_enabled:
r2_config = R2Config({
'r2_account_id': db_config.r2_account_id,
'r2_access_key_id': db_config.r2_access_key_id,
'r2_secret_access_key': db_config.r2_secret_access_key,
'r2_bucket_name': db_config.r2_bucket_name,
'r2_public_url': db_config.r2_public_url,
'r2_enabled': db_config.r2_enabled
})
set_cached_config(company_code, r2_config)
return r2_config
# 最后从.env获取
env_config = get_r2_config_from_env()
if env_config:
set_cached_config(company_code, env_config)
return env_config
return None
def delete_r2_and_local_files(
r2_key: Optional[str],
thumbnail_r2_key: Optional[str],
file_path: Optional[str],
company_code: str,
db: Session
) -> None:
"""
删除R2文件（不再删除本地文件）
Args:
r2_key: R2文件键
thumbnail_r2_key: R2缩略图文件键
file_path: 本地文件路径（已不再使用）
company_code: 公司代码
db: 数据库会话
"""
r2_config = get_company_r2_config(company_code, db)
if r2_config:
r2_uploader = get_r2_uploader()
if r2_uploader.is_available(r2_config):
if r2_key:
r2_uploader.delete_file(r2_key, r2_config)
if thumbnail_r2_key:
r2_uploader.delete_file(thumbnail_r2_key, r2_config)
|