121 lines
5.2 KiB
Python
121 lines
5.2 KiB
Python
import os
|
||
import datetime
|
||
from pathlib import Path
|
||
|
||
from service.device_service import get_unique_client_ips
|
||
def create_directory_structure():
|
||
"""创建项目所需的目录结构，为所有客户端IP预创建基础目录"""
|
||
try:
|
||
# 1. 创建根目录下的resource文件夹（存在则跳过，不覆盖子内容）
|
||
resource_dir = Path("resource")
|
||
resource_dir.mkdir(exist_ok=True)
|
||
print(f"确保resource目录存在: {resource_dir.absolute()}")
|
||
|
||
# 2. 在resource下创建dect文件夹
|
||
dect_dir = resource_dir / "dect"
|
||
dect_dir.mkdir(exist_ok=True)
|
||
print(f"确保dect目录存在: {dect_dir.absolute()}")
|
||
|
||
# 3. 在dect下创建三个模型文件夹
|
||
model_dirs = ["ocr", "face", "yolo"]
|
||
for model in model_dirs:
|
||
model_dir = dect_dir / model
|
||
model_dir.mkdir(exist_ok=True)
|
||
print(f"确保{model}模型目录存在: {model_dir.absolute()}")
|
||
|
||
# 4. 调用外部方法获取所有客户端IP地址
|
||
try:
|
||
# 调用外部ip_read()方法获取所有客户端IP地址列表
|
||
all_ip_addresses = get_unique_client_ips()
|
||
|
||
# 确保返回的是列表类型
|
||
if not isinstance(all_ip_addresses, list):
|
||
all_ip_addresses = [all_ip_addresses]
|
||
|
||
# 过滤有效IP（去除空字符串和空格）
|
||
valid_ips = [ip.strip() for ip in all_ip_addresses if ip.strip()]
|
||
|
||
if not valid_ips:
|
||
print("警告: 未获取到有效的客户端IP地址")
|
||
return
|
||
|
||
print(f"获取到的所有客户端IP地址: {valid_ips}")
|
||
|
||
# 5. 获取当前日期（年、月）
|
||
now = datetime.datetime.now()
|
||
current_year = str(now.year)
|
||
current_month = str(now.month)
|
||
|
||
# 6. 为每个客户端IP在每个模型文件夹下创建年->月的基础目录结构
|
||
for ip in valid_ips:
|
||
# 处理IP地址中的特殊字符（将.替换为_，避免路径问题）
|
||
safe_ip = ip.replace(".", "_")
|
||
|
||
for model in model_dirs:
|
||
# 构建路径: resource/dect/{model}/{safe_ip}/{year}/{month}
|
||
ip_dir = dect_dir / model / safe_ip
|
||
year_dir = ip_dir / current_year
|
||
month_dir = year_dir / current_month
|
||
|
||
# 递归创建目录（存在则跳过，不覆盖）
|
||
month_dir.mkdir(parents=True, exist_ok=True)
|
||
print(f"为客户端IP {ip} 创建/确保目录存在: {month_dir.absolute()}")
|
||
|
||
except Exception as e:
|
||
print(f"处理客户端IP和日期目录时发生错误: {str(e)}")
|
||
|
||
except Exception as e:
|
||
print(f"创建基础目录结构时发生错误: {str(e)}")
|
||
|
||
|
||
def get_image_save_path(model_type: str, client_ip: str) -> tuple:
|
||
"""
|
||
获取图片保存的「完整路径」和「显示用短路径」
|
||
|
||
参数:
|
||
model_type: 模型类型，应为"ocr"、"face"或"yolo"
|
||
client_ip: 检测到违禁的客户端IP地址（原始格式，如192.168.1.101）
|
||
|
||
返回:
|
||
元组 (完整保存路径, 显示用短路径)；若出错则返回 ("", "")
|
||
"""
|
||
try:
|
||
# 1. 验证客户端IP有效性（检查是否在已知IP列表中）
|
||
all_ip_addresses = get_unique_client_ips()
|
||
if not isinstance(all_ip_addresses, list):
|
||
all_ip_addresses = [all_ip_addresses]
|
||
valid_ips = [ip.strip() for ip in all_ip_addresses if ip.strip()]
|
||
|
||
if client_ip.strip() not in valid_ips:
|
||
raise ValueError(f"客户端IP {client_ip} 不在已知IP列表中，无法保存文件")
|
||
|
||
# 2. 处理IP地址（与目录创建逻辑一致，将.替换为_）
|
||
safe_ip = client_ip.strip().replace(".", "_")
|
||
|
||
# 3. 获取当前日期和毫秒级时间戳（确保文件名唯一）
|
||
now = datetime.datetime.now()
|
||
current_year = str(now.year)
|
||
current_month = str(now.month)
|
||
current_day = str(now.day)
|
||
# 时间戳格式：年月日时分秒毫秒（如20250910143050123）
|
||
timestamp = now.strftime("%Y%m%d%H%M%S%f")[:-3]
|
||
|
||
# 4. 定义基础目录（用于生成相对路径）
|
||
base_dir = Path("resource") / "dect" # 显示路径会去掉这个前缀
|
||
# 构建日级目录（完整路径：resource/dect/{model}/{safe_ip}/{年}/{月}/{日}）
|
||
day_dir = base_dir / model_type / safe_ip / current_year / current_month / current_day
|
||
day_dir.mkdir(parents=True, exist_ok=True) # 确保日目录存在
|
||
|
||
# 5. 构建唯一文件名
|
||
image_filename = f"dect_{model_type}_{safe_ip}_{current_year}{current_month}{current_day}_{timestamp}.jpg"
|
||
|
||
# 6. 生成完整路径（用于实际保存图片）和显示路径（用于打印）
|
||
full_path = day_dir / image_filename # 完整路径：resource/dect/.../xxx.jpg
|
||
display_path = full_path.relative_to(base_dir) # 短路径：{model}/.../xxx.jpg（去掉resource/dect）
|
||
|
||
return str(full_path), str(display_path)
|
||
|
||
except Exception as e:
|
||
print(f"获取图片保存路径时发生错误: {str(e)}")
|
||
return "", ""
|