feat(tools): 애니메이션 자동화 파이프라인 구현
- tools/anissia_client.py: Anissia API 클라이언트 (편성표/자막) - tools/nyaa_client.py: Nyaa.si RSS 토렌트 검색 - tools/qbit_client.py: qBittorrent Web API 클라이언트 - tools/subtitle_downloader.py: Google Drive/Tistory/Naver 자막 파서 - tools/title_matcher.py: 제목 매칭 + NAS 폴더명 생성 - tools/anime_pipeline.py: 전체 파이프라인 오케스트레이터 - tools/nas_scanner.py: NAS 폴더/파일 스캔 - prompts/unified.md: anime 모드 추가 (AI 평문 의도 분류) - api/discord_bot.py: AI 평문 anime 핸들러 + /anime 슬래시 커맨드 - config.py: qBittorrent/NAS 설정 추가 - .agents/: agent_guide 워크플로우 통합 - docs/devlog: 세션 기록
This commit is contained in:
152
tools/nas_scanner.py
Normal file
152
tools/nas_scanner.py
Normal file
@@ -0,0 +1,152 @@
|
||||
"""NAS 폴더 스캐너 — 다운로드된 애니 목록 + 파일 정보 조회.
|
||||
|
||||
NAS Animation 폴더 구조:
|
||||
\\192.168.10.10\NasData\Video\Animation\
|
||||
[26_1분기]장송의프리렌2기\
|
||||
[ASW] Sousou no Frieren S2 - 07.mkv
|
||||
subtitles\...
|
||||
[25_4분기]그노시아\
|
||||
...
|
||||
"""
|
||||
|
||||
import logging
|
||||
import os
|
||||
import re
|
||||
from dataclasses import dataclass, field
|
||||
from pathlib import Path
|
||||
from typing import Optional
|
||||
|
||||
import config
|
||||
|
||||
logger = logging.getLogger("variet.tools.nas")
|
||||
|
||||
|
||||
@dataclass
|
||||
class AnimeFolder:
|
||||
"""NAS에 있는 애니 폴더 정보."""
|
||||
folder_name: str # [26_1분기]장송의프리렌2기
|
||||
full_path: str
|
||||
title: str # 장송의프리렌2기
|
||||
year: int # 26
|
||||
quarter: int # 1
|
||||
video_count: int = 0
|
||||
subtitle_count: int = 0
|
||||
total_size_gb: float = 0.0
|
||||
video_files: list[str] = field(default_factory=list)
|
||||
subtitle_files: list[str] = field(default_factory=list)
|
||||
|
||||
|
||||
VIDEO_EXTS = {".mkv", ".mp4", ".avi", ".webm", ".m4v"}
|
||||
SUB_EXTS = {".ass", ".srt", ".ssa", ".sub", ".smi"}
|
||||
|
||||
|
||||
def _parse_folder_name(name: str) -> tuple[int, int, str]:
|
||||
"""폴더명에서 연도, 분기, 제목 추출.
|
||||
|
||||
[26_1분기]장송의프리렌2기 → (26, 1, '장송의프리렌2기')
|
||||
"""
|
||||
m = re.match(r'\[(\d{2})_(\d)분기\](.+)', name)
|
||||
if m:
|
||||
return int(m.group(1)), int(m.group(2)), m.group(3)
|
||||
return 0, 0, name
|
||||
|
||||
|
||||
class NasScanner:
|
||||
"""NAS Animation 폴더 스캐너."""
|
||||
|
||||
def __init__(self, base_path: str = ""):
|
||||
self.base_path = Path(
|
||||
base_path or getattr(config, "NAS_ANIME_PATH",
|
||||
r"\\192.168.10.10\NasData\Video\Animation")
|
||||
)
|
||||
|
||||
def is_accessible(self) -> bool:
|
||||
"""NAS 접근 가능 여부."""
|
||||
try:
|
||||
return self.base_path.exists() and self.base_path.is_dir()
|
||||
except (OSError, PermissionError):
|
||||
return False
|
||||
|
||||
def list_anime_folders(
|
||||
self,
|
||||
year: Optional[int] = None,
|
||||
quarter: Optional[int] = None,
|
||||
) -> list[AnimeFolder]:
|
||||
"""애니 폴더 목록 조회 (분기별 필터 가능)."""
|
||||
if not self.is_accessible():
|
||||
logger.error(f"NAS 경로 접근 불가: {self.base_path}")
|
||||
return []
|
||||
|
||||
results = []
|
||||
try:
|
||||
for entry in sorted(self.base_path.iterdir()):
|
||||
if not entry.is_dir():
|
||||
continue
|
||||
|
||||
y, q, title = _parse_folder_name(entry.name)
|
||||
|
||||
# 필터링
|
||||
if year is not None and y != year:
|
||||
continue
|
||||
if quarter is not None and q != quarter:
|
||||
continue
|
||||
|
||||
folder = AnimeFolder(
|
||||
folder_name=entry.name,
|
||||
full_path=str(entry),
|
||||
title=title,
|
||||
year=y,
|
||||
quarter=q,
|
||||
)
|
||||
|
||||
# 파일 스캔
|
||||
self._scan_folder(entry, folder)
|
||||
results.append(folder)
|
||||
except (OSError, PermissionError) as e:
|
||||
logger.error(f"NAS 스캔 오류: {e}")
|
||||
|
||||
return results
|
||||
|
||||
def _scan_folder(self, path: Path, folder: AnimeFolder):
|
||||
"""폴더 내 영상/자막 파일 집계."""
|
||||
try:
|
||||
for item in path.rglob("*"):
|
||||
if not item.is_file():
|
||||
continue
|
||||
ext = item.suffix.lower()
|
||||
size = item.stat().st_size
|
||||
|
||||
if ext in VIDEO_EXTS:
|
||||
folder.video_count += 1
|
||||
folder.video_files.append(item.name)
|
||||
folder.total_size_gb += size / (1024 ** 3)
|
||||
elif ext in SUB_EXTS:
|
||||
folder.subtitle_count += 1
|
||||
folder.subtitle_files.append(item.name)
|
||||
except (OSError, PermissionError) as e:
|
||||
logger.warning(f"파일 스캔 오류 ({path}): {e}")
|
||||
|
||||
def get_current_quarter_anime(self) -> list[AnimeFolder]:
|
||||
"""이번 분기 다운로드된 애니 목록."""
|
||||
from datetime import date
|
||||
today = date.today()
|
||||
year = today.year % 100
|
||||
quarter = (today.month - 1) // 3 + 1
|
||||
return self.list_anime_folders(year=year, quarter=quarter)
|
||||
|
||||
def search(self, keyword: str) -> list[AnimeFolder]:
|
||||
"""키워드로 NAS 폴더 검색."""
|
||||
all_folders = self.list_anime_folders()
|
||||
kw = keyword.lower()
|
||||
return [f for f in all_folders if kw in f.title.lower() or kw in f.folder_name.lower()]
|
||||
|
||||
def get_summary(self, year: Optional[int] = None, quarter: Optional[int] = None) -> dict:
|
||||
"""요약 통계."""
|
||||
folders = self.list_anime_folders(year=year, quarter=quarter)
|
||||
return {
|
||||
"total_anime": len(folders),
|
||||
"total_videos": sum(f.video_count for f in folders),
|
||||
"total_subtitles": sum(f.subtitle_count for f in folders),
|
||||
"total_size_gb": round(sum(f.total_size_gb for f in folders), 2),
|
||||
"folders": folders,
|
||||
}
|
||||
Reference in New Issue
Block a user