""" YouTube viewer — local HTTP server. Static files: /viewer.html, /control.html, etc. (served from src/ directory). API: GET /api/state — current state JSON POST /api/state — update state (companion -> server) GET /api/search?q=... — YouTube Data API v3 proxy (server-side key) OPTIONS /api/* — CORS preflight State persisted in state.json (alongside server.py). API key — в config.py (gitignored). Stdlib only — no pip install needed. Run: python server.py (default port 8080; override with PORT env var) """ import json import os import threading import time import urllib.error import urllib.parse import urllib.request from http.server import HTTPServer, SimpleHTTPRequestHandler try: from config import YT_API_KEY except ImportError: YT_API_KEY = None STATE_FILE = 'state.json' DEFAULT_STATE = { 'video_id': None, 'title': '', 'ts': 0, 'audio_only': False, 'playback_rate': 1.0, 'start_seconds': 0, 'watched': {}, } state_lock = threading.Lock() def load_state(): if os.path.exists(STATE_FILE): try: with open(STATE_FILE, 'r', encoding='utf-8') as f: loaded = json.load(f) merged = dict(DEFAULT_STATE) merged.update(loaded) return merged except (json.JSONDecodeError, OSError): pass return dict(DEFAULT_STATE) def save_state(state): tmp = STATE_FILE + '.tmp' with open(tmp, 'w', encoding='utf-8') as f: json.dump(state, f, ensure_ascii=False) os.replace(tmp, STATE_FILE) state = load_state() class Handler(SimpleHTTPRequestHandler): extensions_map = dict(SimpleHTTPRequestHandler.extensions_map, **{ '.webmanifest': 'application/manifest+json', '.svg': 'image/svg+xml', }) def _send_json(self, status, payload): body = json.dumps(payload, ensure_ascii=False).encode('utf-8') self.send_response(status) self.send_header('Content-Type', 'application/json; charset=utf-8') self.send_header('Cache-Control', 'no-store') self.send_header('Access-Control-Allow-Origin', '*') self.send_header('Content-Length', str(len(body))) self.end_headers() self.wfile.write(body) def do_GET(self): if self.path.startswith('/api/state'): with state_lock: self._send_json(200, state) return if self.path.startswith('/api/search'): self._handle_search() return if self.path.startswith('/api/progress'): with state_lock: self._send_json(200, {'watched': state.get('watched', {})}) return return super().do_GET() def _handle_search(self): if not YT_API_KEY: self._send_json(500, {'error': 'config_missing', 'detail': 'YT_API_KEY not configured'}) return parsed = urllib.parse.urlparse(self.path) qs = urllib.parse.parse_qs(parsed.query) query = (qs.get('q') or [''])[0].strip() if not query: self._send_json(400, {'error': 'missing_query'}) return max_results = (qs.get('max') or ['12'])[0] try: max_results = max(1, min(25, int(max_results))) except ValueError: max_results = 12 order = (qs.get('order') or ['date'])[0] if order not in ('date', 'relevance', 'rating', 'title', 'viewCount'): order = 'date' params = urllib.parse.urlencode({ 'part': 'snippet', 'type': 'video', 'maxResults': max_results, 'q': query, 'order': order, 'safeSearch': 'none', 'key': YT_API_KEY, }) url = 'https://www.googleapis.com/youtube/v3/search?' + params req = urllib.request.Request(url, headers={'Accept': 'application/json'}) try: with urllib.request.urlopen(req, timeout=10) as resp: raw = resp.read().decode('utf-8') data = json.loads(raw) except urllib.error.HTTPError as e: body = e.read().decode('utf-8', errors='replace') try: err = json.loads(body) except json.JSONDecodeError: err = {'raw': body} self._send_json(e.code, {'error': 'youtube_api_error', 'status': e.code, 'detail': err}) return except Exception as e: self._send_json(502, {'error': 'fetch_failed', 'detail': str(e)}) return items = [] for it in data.get('items', []): vid = (it.get('id') or {}).get('videoId') sn = it.get('snippet') or {} if not vid: continue thumbs = sn.get('thumbnails') or {} thumb = (thumbs.get('medium') or thumbs.get('default') or thumbs.get('high') or {}).get('url', '') items.append({ 'video_id': vid, 'title': sn.get('title', ''), 'channel': sn.get('channelTitle', ''), 'thumbnail': thumb, 'published_at': sn.get('publishedAt', ''), }) self._send_json(200, {'query': query, 'items': items}) def do_POST(self): if self.path == '/api/state': data = self._read_json_body() if data is None: return with state_lock: if 'video_id' in data and data['video_id']: state['video_id'] = data['video_id'] state['title'] = data.get('title', '') state['ts'] = int(time.time() * 1000) try: rate = float(data.get('playback_rate', 1.0)) except (TypeError, ValueError): rate = 1.0 if rate not in (1.0, 1.5, 2.0): rate = 1.0 state['playback_rate'] = rate try: state['start_seconds'] = max(0, int(data.get('start_seconds', 0))) except (TypeError, ValueError): state['start_seconds'] = 0 if 'audio_only' in data: state['audio_only'] = bool(data['audio_only']) save_state(state) self._send_json(200, state) return if self.path == '/api/progress': data = self._read_json_body() if data is None: return vid = data.get('video_id') try: position = max(0, int(data.get('position', 0))) except (TypeError, ValueError): position = 0 if not vid: self._send_json(400, {'error': 'missing_video_id'}) return with state_lock: watched = state.setdefault('watched', {}) watched[vid] = { 'position': position, 'updated_at': int(time.time() * 1000), 'title': data.get('title') or (watched.get(vid, {}) or {}).get('title', ''), } # LRU eviction — keep last 200 entries if len(watched) > 200: items = sorted(watched.items(), key=lambda kv: kv[1].get('updated_at', 0), reverse=True) state['watched'] = dict(items[:200]) save_state(state) self._send_json(200, {'ok': True}) return self._send_json(404, {'error': 'not_found'}) def _read_json_body(self): length = int(self.headers.get('Content-Length', '0')) try: raw = self.rfile.read(length).decode('utf-8') if length else '{}' return json.loads(raw) if raw else {} except (json.JSONDecodeError, UnicodeDecodeError) as e: self._send_json(400, {'error': 'invalid_json', 'detail': str(e)}) return None def do_OPTIONS(self): if self.path.startswith('/api/'): self.send_response(204) self.send_header('Access-Control-Allow-Origin', '*') self.send_header('Access-Control-Allow-Methods', 'GET, POST, OPTIONS') self.send_header('Access-Control-Allow-Headers', 'Content-Type') self.end_headers() return self._send_json(405, {'error': 'method_not_allowed'}) def log_message(self, fmt, *args): print('[{}] {}'.format(self.log_date_time_string(), fmt % args), flush=True) def main(): port = int(os.environ.get('PORT', 8080)) print('YouTube viewer server') print('Serving directory: {}'.format(os.getcwd())) print('Endpoints:') print(' GET /viewer.html — viewer (для очков через Meta AI app)') print(' GET /control.html — companion (для Safari iPhone)') print(' GET /api/state — current state JSON') print(' POST /api/state — update state (companion -> server)') print('Listening on http://0.0.0.0:{}'.format(port)) HTTPServer(('0.0.0.0', port), Handler).serve_forever() if __name__ == '__main__': main()