refactor: remove unused get_models method from LoraRoutes

This commit is contained in:
Will Miao
2025-07-25 18:23:52 +08:00
parent 981e367bf1
commit ea8a64fafc
3 changed files with 16 additions and 47 deletions

View File

@@ -17,15 +17,15 @@ class Config:
def __init__(self):
self.templates_path = os.path.join(os.path.dirname(os.path.dirname(__file__)), 'templates')
self.static_path = os.path.join(os.path.dirname(os.path.dirname(__file__)), 'static')
# 路径映射字典, target to link mapping
# Path mapping dictionary, target to link mapping
self._path_mappings = {}
# 静态路由映射字典, target to route mapping
# Static route mapping dictionary, target to route mapping
self._route_mappings = {}
self.loras_roots = self._init_lora_paths()
self.checkpoints_roots = None
self.unet_roots = None
self.base_models_roots = self._init_checkpoint_paths()
# 在初始化时扫描符号链接
# Scan symbolic links during initialization
self._scan_symbolic_links()
if not standalone_mode:
@@ -83,7 +83,7 @@ class Config:
return False
def _scan_symbolic_links(self):
"""扫描所有 LoRA Checkpoint 根目录中的符号链接"""
"""Scan all symbolic links in LoRA and Checkpoint root directories"""
for root in self.loras_roots:
self._scan_directory_links(root)
@@ -91,7 +91,7 @@ class Config:
self._scan_directory_links(root)
def _scan_directory_links(self, root: str):
"""递归扫描目录中的符号链接"""
"""Recursively scan symbolic links in a directory"""
try:
with os.scandir(root) as it:
for entry in it:
@@ -106,40 +106,40 @@ class Config:
logger.error(f"Error scanning links in {root}: {e}")
def add_path_mapping(self, link_path: str, target_path: str):
"""添加符号链接路径映射
target_path: 实际目标路径
link_path: 符号链接路径
"""Add a symbolic link path mapping
target_path: actual target path
link_path: symbolic link path
"""
normalized_link = os.path.normpath(link_path).replace(os.sep, '/')
normalized_target = os.path.normpath(target_path).replace(os.sep, '/')
# 保持原有的映射关系:目标路径 -> 链接路径
# Keep the original mapping: target path -> link path
self._path_mappings[normalized_target] = normalized_link
logger.info(f"Added path mapping: {normalized_target} -> {normalized_link}")
def add_route_mapping(self, path: str, route: str):
"""添加静态路由映射"""
"""Add a static route mapping"""
normalized_path = os.path.normpath(path).replace(os.sep, '/')
self._route_mappings[normalized_path] = route
# logger.info(f"Added route mapping: {normalized_path} -> {route}")
def map_path_to_link(self, path: str) -> str:
"""将目标路径映射回符号链接路径"""
"""Map a target path back to its symbolic link path"""
normalized_path = os.path.normpath(path).replace(os.sep, '/')
# 检查路径是否包含在任何映射的目标路径中
# Check if the path is contained in any mapped target path
for target_path, link_path in self._path_mappings.items():
if normalized_path.startswith(target_path):
# 如果路径以目标路径开头,则替换为链接路径
# If the path starts with the target path, replace with link path
mapped_path = normalized_path.replace(target_path, link_path, 1)
return mapped_path
return path
def map_link_to_path(self, link_path: str) -> str:
"""将符号链接路径映射回实际路径"""
"""Map a symbolic link path back to the actual path"""
normalized_link = os.path.normpath(link_path).replace(os.sep, '/')
# 检查路径是否包含在任何映射的目标路径中
# Check if the path is contained in any mapped target path
for target_path, link_path in self._path_mappings.items():
if normalized_link.startswith(target_path):
# 如果路径以目标路径开头,则替换为实际路径
# If the path starts with the target path, replace with actual path
mapped_path = normalized_link.replace(target_path, link_path, 1)
return mapped_path
return link_path

View File

@@ -199,35 +199,6 @@ class LoraRoutes(BaseModelRoutes):
'error': str(e)
}, status=500)
# Override get_models to add LoRA-specific response data
async def get_models(self, request: web.Request) -> web.Response:
"""Get paginated LoRA data with LoRA-specific fields"""
try:
# Parse common query parameters
params = self._parse_common_params(request)
# Get data from service
result = await self.service.get_paginated_data(**params)
# Get all available folders from cache for LoRA-specific response
cache = await self.service.scanner.get_cached_data()
# Format response items with LoRA-specific structure
formatted_result = {
'items': [await self.service.format_response(item) for item in result['items']],
'folders': cache.folders, # LoRA-specific: include folders in response
'total': result['total'],
'page': result['page'],
'page_size': result['page_size'],
'total_pages': result['total_pages']
}
return web.json_response(formatted_result)
except Exception as e:
logger.error(f"Error in get_loras: {e}", exc_info=True)
return web.json_response({"error": str(e)}, status=500)
# CivitAI integration methods
async def get_civitai_versions_lora(self, request: web.Request) -> web.Response:
"""Get available versions for a Civitai LoRA model with local availability info"""

View File

@@ -1,13 +1,11 @@
from datetime import datetime
import aiohttp
import os
import json
import logging
import asyncio
from email.parser import Parser
from typing import Optional, Dict, Tuple, List
from urllib.parse import unquote
from ..utils.models import LoraMetadata
logger = logging.getLogger(__name__)