60 lines
1.8 KiB
Python
60 lines
1.8 KiB
Python
import asyncio
|
|
import os
|
|
from datetime import datetime
|
|
from pathlib import Path
|
|
from aiohttp import ClientSession
|
|
from blinkpy.blinkpy import Blink
|
|
from blinkpy.auth import Auth
|
|
|
|
class BlinkRecorder:
|
|
def __init__(self, credentials_file: str, save_path: str):
|
|
self.credentials_file = credentials_file
|
|
self.save_path = Path(save_path)
|
|
self.blink: Blink | None = None
|
|
|
|
async def setup(self) -> None:
|
|
self.blink = Blink(session=ClientSession())
|
|
auth = Auth({"username": os.getenv("BLINK_USERNAME"),
|
|
"password": os.getenv("BLINK_PASSWORD")})
|
|
self.blink.auth = auth
|
|
await self.blink.start()
|
|
|
|
async def download_new_videos(self) -> None:
|
|
if not self.blink:
|
|
return
|
|
|
|
# Create dated folder
|
|
date_folder = self.save_path / datetime.now().strftime("%Y-%m-%d")
|
|
date_folder.mkdir(parents=True, exist_ok=True)
|
|
|
|
await self.blink.download_videos(
|
|
str(date_folder),
|
|
since=datetime.now().strftime("%Y/%m/%d 00:00"),
|
|
delay=2
|
|
)
|
|
|
|
async def main():
|
|
# Setup paths
|
|
base_path = Path(__file__).parent
|
|
recordings_path = base_path / "recordings"
|
|
credentials_path = base_path / "credentials.json"
|
|
|
|
recorder = BlinkRecorder(
|
|
credentials_file=str(credentials_path),
|
|
save_path=str(recordings_path)
|
|
)
|
|
|
|
await recorder.setup()
|
|
|
|
while True:
|
|
try:
|
|
await recorder.download_new_videos()
|
|
# Wait 5 minutes before checking for new videos
|
|
await asyncio.sleep(300)
|
|
except Exception as err:
|
|
print(f"Error occurred: {err}")
|
|
await asyncio.sleep(60)
|
|
|
|
if __name__ == "__main__":
|
|
asyncio.run(main())
|