made shit faster

app.py

@@ -1,144 +1,130 @@
 from fastapi import FastAPI
 from fastapi.middleware.cors import CORSMiddleware
-import requests
+import httpx
 import json
 import logging
 
 app = FastAPI()
-
-# Set up logging for debugging
 logging.basicConfig(level=logging.INFO)
 
 # Allow all CORS so browser apps can call this API
-app.add_middleware(
-    CORSMiddleware,
-    allow_origins=["*"],
-    allow_methods=["*"],
-    allow_headers=["*"],
-)
-
-# Base62 conversion helper
-BASE_62_CHAR_SET = "0123456789ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz"
-def base62_to_int(e):
-    t = 0
-    for char in e:
-        t = t * 62 + BASE_62_CHAR_SET.index(char)
-    return t
-
-# Construct the base URL for the shared album from the token
-def get_base_url(token: str) -> str:
-    first_char = token[0]
-    n = base62_to_int(token[1]) if first_char == "A" else base62_to_int(token[1:3])
-    base_url = f"https://p{n:02d}-sharedstreams.icloud.com/{token}/sharedstreams/"
-    return base_url
+def configure_cors(app: FastAPI):
+    app.add_middleware(
+        CORSMiddleware,
+        allow_origins=["*"],
+        allow_methods=["*"],
+        allow_headers=["*"],
+    )
+configure_cors(app)
+
+# Precompute Base62 index mapping for O(1) lookup\BASE_62_MAP = {c: i for i, c in enumerate("0123456789ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz")}
+
+async def get_client() -> httpx.AsyncClient:
+    if not hasattr(app.state, 'client'):
+        app.state.client = httpx.AsyncClient(timeout=10.0)
+    return app.state.client
+
+# Convert base62 to int using dict lookup
+def base62_to_int(token: str) -> int:
+    result = 0
+    for ch in token:
+        result = result * 62 + BASE_62_MAP[ch]
+    return result
 
-# Handle 330 redirect from iCloud
-def get_redirected_base_url(base_url: str, token: str) -> str:
-    url = base_url + "webstream"
-    headers = {
-        'Origin': 'https://www.icloud.com',
-        'Content-Type': 'text/plain',
-    }
-    data = '{"streamCtag":null}'
-    response = requests.post(url, headers=headers, data=data, allow_redirects=False)
-    if response.status_code == 330:
-        body = response.json()
-        new_base_url = f"https://{body['X-Apple-MMe-Host']}/{token}/sharedstreams/"
-        return new_base_url
+# Get base URL from token
+async def get_base_url(token: str) -> str:
+    first = token[0]
+    n = base62_to_int(token[1]) if first == 'A' else base62_to_int(token[1:3])
+    return f"https://p{n:02d}-sharedstreams.icloud.com/{token}/sharedstreams/"
+
+# Static request details
+ICLOUD_HEADERS = {'Origin': 'https://www.icloud.com', 'Content-Type': 'text/plain'}
+ICLOUD_PAYLOAD = '{"streamCtag":null}'
+
+# Handle possible 330 redirect
+async def get_redirected_base_url(base_url: str, token: str) -> str:
+    client = await get_client()
+    resp = await client.post(base_url + 'webstream', headers=ICLOUD_HEADERS, data=ICLOUD_PAYLOAD, follow_redirects=False)
+    if resp.status_code == 330:
+        body = resp.json()
+        host = body.get('X-Apple-MMe-Host')
+        return f"https://{host}/{token}/sharedstreams/"
     return base_url
 
-# Fetch metadata (the "webstream" data)
-def get_metadata(base_url: str):
-    url = base_url + "webstream"
-    headers = {
-        'Origin': 'https://www.icloud.com',
-        'Content-Type': 'text/plain',
-    }
-    data = '{"streamCtag":null}'
-    response = requests.post(url, headers=headers, data=data)
-    return response.json().get("photos", [])
-
-# Fetch asset URLs mapping for a list of photoGuids
-def get_asset_urls(base_url: str, guids: list):
-    url = base_url + "webasseturls"
-    headers = {
-        'Origin': 'https://www.icloud.com',
-        'Content-Type': 'text/plain',
-    }
+# Generic function to POST and parse JSON
+async def post_json(path: str, base_url: str, payload: str) -> dict:
+    client = await get_client()
+    resp = await client.post(f"{base_url}{path}", headers=ICLOUD_HEADERS, data=payload)
+    resp.raise_for_status()
+    return resp.json()
+
+# Fetch metadata
+async def get_metadata(base_url: str) -> list:
+    return (await post_json('webstream', base_url, ICLOUD_PAYLOAD)).get('photos', [])
+
+# Fetch asset URL mappings
+async def get_asset_urls(base_url: str, guids: list) -> dict:
     data = json.dumps({"photoGuids": guids})
-    response = requests.post(url, headers=headers, data=data)
-    return response.json().get("items", {})
+    return (await post_json('webasseturls', base_url, data)).get('items', {})
 
-# Main endpoint to extract video URLs and include the poster (first frame)
-@app.get("/album/{token}")
-def get_album(token: str):
+# Main endpoint\@app.get('/album/{token}')
+async def get_album(token: str):
     try:
-        base_url = get_base_url(token)
-        base_url = get_redirected_base_url(base_url, token)
-        metadata = get_metadata(base_url)
-        guids = [photo["photoGuid"] for photo in metadata]
-        asset_urls = get_asset_urls(base_url, guids)
-
-        video_list = []
-        # Process each photo in the album
-        for photo in metadata:
-            # Check if the overall photo is a video
-            if photo.get("mediaAssetType", "").lower() == "video":
-                best_video = None
-                best_size = 0
-                # Look at each derivative except the poster frame
-                derivatives = photo.get("derivatives", {})
-                for key, derivative in derivatives.items():
-                    if key.lower() == "posterframe":
-                        continue
-                    try:
-                        file_size = int(derivative.get("fileSize", 0))
-                    except (ValueError, TypeError):
-                        file_size = 0
-                    if file_size > best_size:
-                        best_size = file_size
-                        best_video = derivative
-
-                # Also get the poster frame derivative if available
-                poster_url = None
-                poster_derivative = photo.get("derivatives", {}).get("PosterFrame")
-                if poster_derivative:
-                    poster_checksum = poster_derivative.get("checksum")
-                    if poster_checksum in asset_urls:
-                        poster_info = asset_urls[poster_checksum]
-                        poster_url = f"https://{poster_info['url_location']}{poster_info['url_path']}"
-
-                # Build the video URL if found
-                if best_video:
-                    checksum = best_video.get("checksum")
-                    if checksum in asset_urls:
-                        url_info = asset_urls[checksum]
-                        video_url = f"https://{url_info['url_location']}{url_info['url_path']}"
-                        video_list.append({
-                            "caption": photo.get("caption", ""),
-                            "url": video_url,
-                            "poster": poster_url
-                        })
-                    else:
-                        logging.info(f"Checksum {checksum} not found in asset_urls for photo {photo.get('photoGuid')}")
-                else:
-                    logging.info(f"No video derivative found for photo {photo.get('photoGuid')}")
-            else:
-                logging.info(f"Photo {photo.get('photoGuid')} is not a video. mediaAssetType: {photo.get('mediaAssetType')}")
-        
-        return {"videos": video_list}
+        base_url = await get_base_url(token)
+        base_url = await get_redirected_base_url(base_url, token)
+
+        metadata = await get_metadata(base_url)
+        guids = [p['photoGuid'] for p in metadata]
+        asset_map = await get_asset_urls(base_url, guids)
+
+        videos = []
+        for p in metadata:
+            if p.get('mediaAssetType', '').lower() != 'video':
+                continue
+
+            # select the largest derivative (exclude poster)
+            derivatives = p.get('derivatives', {})
+            best = max((d for k, d in derivatives.items() if k.lower() != 'posterframe'),
+                       key=lambda d: int(d.get('fileSize') or 0),
+                       default=None)
+            if not best:
+                continue
+
+            checksum = best.get('checksum')
+            if checksum not in asset_map:
+                logging.info(f"Missing asset for checksum {checksum}")
+                continue
+
+            info = asset_map[checksum]
+            video_url = f"https://{info['url_location']}{info['url_path']}"
+
+            poster = None
+            pf = derivatives.get('PosterFrame')
+            if pf and pf.get('checksum') in asset_map:
+                pi = asset_map[pf['checksum']]
+                poster = f"https://{pi['url_location']}{pi['url_path']}"
+
+            videos.append({
+                'caption': p.get('caption', ''),
+                'url': video_url,
+                'poster': poster
+            })
+
+        return {'videos': videos}
+
     except Exception as e:
-        return {"error": str(e)}
+        logging.error(f"Error fetching album {token}: {e}")
+        return {'error': str(e)}
 
-# Debug endpoint to return raw metadata for further inspection
-@app.get("/album/{token}/raw")
-def get_album_raw(token: str):
+# Debug endpoint
+@app.get('/album/{token}/raw')
+async def get_album_raw(token: str):
     try:
-        base_url = get_base_url(token)
-        base_url = get_redirected_base_url(base_url, token)
-        metadata = get_metadata(base_url)
-        guids = [photo["photoGuid"] for photo in metadata]
-        asset_urls = get_asset_urls(base_url, guids)
-        return {"metadata": metadata, "asset_urls": asset_urls}
+        base_url = await get_base_url(token)
+        base_url = await get_redirected_base_url(base_url, token)
+        metadata = await get_metadata(base_url)
+        guids = [p['photoGuid'] for p in metadata]
+        asset_map = await get_asset_urls(base_url, guids)
+        return {'metadata': metadata, 'asset_urls': asset_map}
     except Exception as e:
-        return {"error": str(e)}
+        return {'error': str(e)}