138 lines
4.4 KiB
Python
138 lines
4.4 KiB
Python
import asyncio
|
|
import re
|
|
from datetime import datetime
|
|
from pathlib import Path
|
|
|
|
from sqlalchemy import select
|
|
|
|
from ..core.config import BASE_DIR, THUMBS, VIDEO_EXTENSIONS, YT_DIR
|
|
from ..core.db import Base, SessionLocal, engine
|
|
from ..models.video_model import Channel, Playlist, Video
|
|
|
|
|
|
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 = BASE_DIR / YT_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_channels = {ch.name:ch for ch in (await session.execute(select(Channel))).scalars()}
|
|
cached_playlists = {(pl.channel_id, pl.name): pl for pl in (await session.execute(select(Playlist))).scalars()}
|
|
|
|
existing_videos: set[tuple[int, str]] = set(
|
|
(await session.execute(select(Video.channel_id, Video.filename))).all()
|
|
)
|
|
|
|
videos = []
|
|
|
|
for file_path in DIRECTORY.rglob("*"):
|
|
if file_path.is_file():
|
|
relative_path = str(file_path.relative_to(DIRECTORY))
|
|
|
|
data = relative_path.split('/')
|
|
|
|
length = len(data)
|
|
|
|
if file_path.suffix.lower() not in VIDEO_EXTENSIONS:
|
|
continue
|
|
|
|
if length < 2:
|
|
print ("error invalid channel")
|
|
continue
|
|
|
|
elif length == 2:
|
|
if data[0] == "_thumbs":
|
|
continue
|
|
|
|
channel_name = data[0]
|
|
filename = data[1]
|
|
name = Path(filename).stem
|
|
playlist_name = None
|
|
preview_path = f"{THUMBS}/{channel_name}/{name}.jpg"
|
|
path = str(relative_path)
|
|
title, date = parse_preview_path(preview_path)
|
|
|
|
elif length == 3:
|
|
if data[0] == "_thumbs":
|
|
continue
|
|
|
|
channel_name = data[0]
|
|
filename = data[2]
|
|
name = Path(filename).stem
|
|
playlist_name = data[1]
|
|
preview_path = f"{THUMBS}/{channel_name}/{playlist_name}/{name}.jpg"
|
|
path = str(relative_path)
|
|
title, date = parse_preview_path(preview_path)
|
|
|
|
else:
|
|
print(f"skip (too deep): {relative_path}")
|
|
|
|
channel = cached_channels.get(channel_name)
|
|
if channel is None:
|
|
channel = Channel(name=channel_name)
|
|
session.add(channel)
|
|
await session.flush()
|
|
cached_channels[channel_name] = channel
|
|
|
|
if not playlist_name is None:
|
|
playlist = cached_playlists.get((channel.id, playlist_name))
|
|
if playlist is None:
|
|
playlist = Playlist(name=playlist_name, channel=channel)
|
|
session.add(playlist)
|
|
await session.flush()
|
|
cached_playlists[(channel.id, playlist_name)] = playlist
|
|
|
|
vkey = (channel.id, filename)
|
|
if vkey in existing_videos:
|
|
continue
|
|
|
|
video = Video(filename=filename, path=path, preview_path=preview_path,
|
|
playlist=playlist, channel=channel, title=title, date=date)
|
|
videos.append(video)
|
|
existing_videos.add(vkey)
|
|
|
|
else:
|
|
|
|
vkey = (channel.id, filename)
|
|
if vkey in existing_videos:
|
|
continue
|
|
|
|
video = Video(filename=filename, path=path, preview_path=preview_path,
|
|
playlist=None, channel=channel, title=title, date=date)
|
|
videos.append(video)
|
|
|
|
if videos:
|
|
session.add_all(videos)
|
|
|
|
await session.commit()
|
|
|
|
|
|
async def main():
|
|
await init_db()
|
|
await update_yt()
|
|
|
|
if __name__ == "__main__":
|
|
asyncio.run(main()) |