-
-
Save 5E7EN/612d25f06909ff41bd131eb597ef2372 to your computer and use it in GitHub Desktop.
Download all the videos between two events from Ring
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| import json | |
| import getpass | |
| import time | |
| import requests | |
| import logging | |
| from pathlib import Path | |
| from pprint import pprint | |
| # https://github.com/tchellomello/python-ring-doorbell | |
| from ring_doorbell import Ring, Auth, RingStickUpCam | |
| from oauthlib.oauth2 import MissingTokenError | |
| # Constants | |
| OLDER_THAN = 7267899867159244279 | |
| OLDEST_ONE = 7267176238184312311 | |
| CHUNK_SIZE = 500 | |
| MAX_RETRIES = 3 | |
| CACHE_FILE_PATH = Path("token.cache") | |
| USER_AGENT = "VideoDownloader/1.0" | |
| RETRY_SLEEP_TIME = 5 | |
| logging.getLogger('ring_doorbell').setLevel(logging.CRITICAL) | |
| def token_updated(token): | |
| """Callback function to update token cache.""" | |
| with CACHE_FILE_PATH.open('w') as cache_file: | |
| json.dump(token, cache_file) | |
| def otp_callback(): | |
| """Handle 2-factor authentication.""" | |
| return input("2FA code (check your email/phone/authenticator): ") | |
| def initialize_auth(): | |
| """Initialize authentication.""" | |
| if CACHE_FILE_PATH.is_file(): | |
| with CACHE_FILE_PATH.open('r') as cache_file: | |
| auth_token = json.load(cache_file) | |
| return Auth(USER_AGENT, auth_token, token_updated) | |
| username = input("Enter your Ring email address: ") | |
| password = getpass.getpass("Password: ") | |
| auth = Auth(USER_AGENT, None, token_updated) | |
| try: | |
| auth.fetch_token(username, password) | |
| except MissingTokenError: | |
| auth.fetch_token(username, password, otp_callback()) | |
| return auth | |
| def download(cam: RingStickUpCam): | |
| """Download videos from a camera.""" | |
| count = 0 | |
| eid = OLDER_THAN | |
| while True: | |
| # Fetch the camera event history | |
| events = cam.history(older_than=eid, limit=CHUNK_SIZE) | |
| # Loop through the events and download each video (newest -> oldest) | |
| for event in events: | |
| eid = event['id'] | |
| if eid < OLDEST_ONE: | |
| return | |
| success = download_event(cam, eid) | |
| if success: | |
| count += 1 | |
| print(f'Downloaded #{count} -> {eid}') | |
| def download_event(cam: RingStickUpCam, eid): | |
| """Attempt to download a specific event. Returns True if successful.""" | |
| retries = 0 | |
| video_path = Path(f'videos/{cam.name}') | |
| video_path.mkdir(parents=True, exist_ok=True) | |
| while retries < MAX_RETRIES: | |
| try: | |
| cam.recording_download(eid, filename=video_path/f'{eid}.mp4') | |
| return True | |
| except requests.exceptions.RequestException as e: | |
| if '404' in str(e): | |
| retries += 1 | |
| print(f"Download failed for {eid}. Retrying {retries}/{MAX_RETRIES}...") | |
| time.sleep(RETRY_SLEEP_TIME) | |
| else: | |
| raise | |
| print(f"- Failed to download {eid} after {MAX_RETRIES} attempts!") | |
| return False | |
| def main(): | |
| """Main function.""" | |
| auth = initialize_auth() | |
| ring = Ring(auth) | |
| ring.update_data() | |
| devices = ring.devices() | |
| pprint(devices) | |
| cameras = devices['stickup_cams'] | |
| # Determine the index of the camera to download from the list above and edit this (0 = first camera, 1 = second camera, etc.) | |
| targetCam = cameras[0] | |
| print(f'Downloading videos from {targetCam.name}...') | |
| download(targetCam) | |
| print('\nDONE.') | |
| if __name__ == "__main__": | |
| main() |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment