114 lines
3.0 KiB
Python
114 lines
3.0 KiB
Python
import asyncio
|
|
import re
|
|
from datetime import datetime
|
|
from pathlib import Path
|
|
|
|
from sqlalchemy import select
|
|
|
|
from ..core.config import AUDIO_DIR, AUDIO_EXTENSIONS, COVERS
|
|
from ..core.db import Base, SessionLocal, engine
|
|
from ..models.audio_model import Audio, AudioPlaylist
|
|
|
|
|
|
def parse_preview_path(path: str) -> tuple[str, datetime | None]:
|
|
p = Path(path)
|
|
stem = p.stem
|
|
|
|
match = re.match(r"^(\d{8})\s*-\s*(.+)$", stem)
|
|
if not match:
|
|
return stem, None
|
|
|
|
date_str, title = match.group(1), match.group(2).strip()
|
|
|
|
try:
|
|
date = datetime.strptime(date_str, "%Y%m%d")
|
|
except ValueError:
|
|
return stem, None
|
|
|
|
return title, date
|
|
|
|
|
|
DIRECTORY = AUDIO_DIR
|
|
|
|
|
|
async def init_db():
|
|
async with engine.begin() as conn:
|
|
await conn.run_sync(Base.metadata.create_all)
|
|
|
|
|
|
async def update_yt():
|
|
async with SessionLocal() as session:
|
|
cached_playlists = {
|
|
(pl.title): pl for pl in (await session.execute(select(AudioPlaylist))).scalars()
|
|
}
|
|
|
|
existing_audios: set[str] = set(
|
|
(await session.execute(select(Audio.filename))).scalars()
|
|
)
|
|
|
|
audios = []
|
|
|
|
for file_path in DIRECTORY.rglob("*"):
|
|
if not file_path.is_file():
|
|
continue
|
|
if file_path.suffix.lower() not in AUDIO_EXTENSIONS:
|
|
continue
|
|
|
|
relative_path = str(file_path.relative_to(DIRECTORY))
|
|
data = relative_path.split('/')
|
|
length = len(data)
|
|
|
|
if data[0] == "_covers":
|
|
continue
|
|
|
|
if length < 2:
|
|
print(f"error invalid path: {relative_path}")
|
|
continue
|
|
|
|
filename = data[-1]
|
|
name = Path(filename).stem
|
|
playlist_name = data[0] if length >= 2 else None
|
|
|
|
if playlist_name:
|
|
preview_path = f"{COVERS}/{playlist_name}/{name}.jpg"
|
|
else:
|
|
preview_path = f"{COVERS}/{name}.jpg"
|
|
|
|
path = str(relative_path)
|
|
title, date = parse_preview_path(preview_path)
|
|
|
|
if filename in existing_audios:
|
|
continue
|
|
|
|
playlist = None
|
|
if playlist_name:
|
|
playlist = cached_playlists.get(playlist_name)
|
|
if playlist is None:
|
|
playlist = AudioPlaylist(title=playlist_name)
|
|
session.add(playlist)
|
|
await session.flush()
|
|
cached_playlists[playlist_name] = playlist
|
|
|
|
audio = Audio(
|
|
filename=filename,
|
|
path=path,
|
|
preview_path=preview_path,
|
|
name=name,
|
|
playlist=playlist
|
|
)
|
|
audios.append(audio)
|
|
existing_audios.add(filename)
|
|
|
|
if audios:
|
|
session.add_all(audios)
|
|
|
|
await session.commit()
|
|
|
|
|
|
async def main():
|
|
await init_db()
|
|
await update_yt()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
asyncio.run(main()) |