transport.py

1"""Fixed-origin Lichess transport. Bounded asynchronous control requests."""
2import json,queue,re,threading,time
3from pathlib import Path
4import requests
5
6class Events:
7    """Game updates cannot be starved by a burst of public challenges."""
8    def __init__(self):self.game=queue.Queue(128);self.account=queue.Queue(32)
9    def put_nowait(self,item):
10        (self.account if item[0]=='account' else self.game).put_nowait(item)
11    def get_nowait(self):
12        try:return self.game.get_nowait()
13        except queue.Empty:return self.account.get_nowait()
14
15class Cooldown:
16    def __init__(self):self.until=0.;self.lock=threading.Lock()
17    def defer(self,seconds=60):
18        with self.lock:self.until=max(self.until,time.monotonic()+max(60,seconds))
19    def remaining(self):return max(0.,self.until-time.monotonic())
20
21class Transport:
22    def __init__(self,token_file,request=requests.request):
23        token=Path(token_file).read_text(encoding='utf-8-sig').strip()
24        if not re.fullmatch(r'[A-Za-z0-9_]+',token):raise ValueError('invalid token format')
25        self.headers={'Authorization':'Bearer '+token,'User-Agent':'LittlePly/1.0','Accept':'application/json'}
26        self.request=request;self.cooldown=Cooldown();self.stop=threading.Event()
27        self.jobs=queue.PriorityQueue(32);self.results=queue.Queue(64);self.counter=0;self.lock=threading.Lock()
28        self.thread=threading.Thread(target=self._run,daemon=True);self.thread.start()
29    def submit(self,key,method,path,data=None,priority=1,guard=None):
30        if not re.fullmatch(r'/api/[A-Za-z0-9_/?=&.-]+',path):raise ValueError('invalid API path')
31        with self.lock:self.counter+=1;n=self.counter
32        self.jobs.put_nowait((priority,n,(key,method,path,data,guard)))
33    def _run(self):
34        while not self.stop.is_set():
35            if self.stop.wait(min(.2,self.cooldown.remaining())):return
36            if self.cooldown.remaining():continue
37            try:priority,n,job=self.jobs.get(timeout=.2)
38            except queue.Empty:continue
39            key,method,path,data,guard=job
40            if guard and not guard():self.results.put((key,'cancelled',None));continue
41            try:
42                with self.request(method,'https://lichess.org'+path,headers=self.headers,data=data,
43                                  timeout=(3,5),allow_redirects=False) as r:
44                    code=r.status_code
45                    if code==429:
46                        try:retry=float(r.headers.get('Retry-After',60))
47                        except (TypeError,ValueError):retry=60
48                        self.cooldown.defer(retry)
49                        # Known rejected: return for state-aware retry after shared cooldown.
50                        result=('rate_limited',{'retry_after':max(60,retry)})
51                    elif code in (401,403):result=('auth_error',code)
52                    elif 200<=code<300:
53                        try:body=r.json()
54                        except ValueError:body={}
55                        result=('ok',body)
56                    elif 400<=code<500:result=('rejected',code)
57                    else:result=('uncertain' if method=='POST' else 'retryable',code)
58            except requests.RequestException as exc:
59                result=('uncertain' if method=='POST' else 'retryable',type(exc).__name__)
60            except Exception as exc:result=('retryable',type(exc).__name__)
61            self.results.put((key,*result))
62    def close(self):self.stop.set();self.thread.join(timeout=9)
63
64class Stream:
65    """One stream thread, monotonically increasing epochs, bounded queue."""
66    def __init__(self,transport,path,tag,events,epochs):
67        self.transport=transport;self.path=path;self.tag=tag;self.events=events;self.epochs=epochs
68        self.stop=threading.Event();self.resync=threading.Event();self.last_event=0.;self.response=None
69        self.thread=threading.Thread(target=self.run,daemon=True);self.thread.start()
70    def emit(self,kind,data,epoch):
71        try:self.events.put_nowait((self.tag,kind,data,epoch,time.monotonic()))
72        except queue.Full:self.resync.set()
73    def run(self):
74        delay=2
75        while not self.stop.is_set():
76            if self.stop.wait(max(0,self.transport.cooldown.remaining())):return
77            epoch=next(self.epochs);self.resync.clear()
78            try:
79                with requests.get('https://lichess.org'+self.path,headers=self.transport.headers,
80                                  stream=True,timeout=(5,15),allow_redirects=False) as r:
81                    self.response=r
82                    if r.status_code==429:self.transport.cooldown.defer();raise RuntimeError('rate_limited')
83                    if r.status_code in (401,403):self.emit('auth_error',r.status_code,epoch);return
84                    if r.status_code!=200:raise RuntimeError('stream_http')
85                    self.last_event=time.monotonic();self.emit('connected',{},epoch);delay=2
86                    for line in r.iter_lines(chunk_size=1):
87                        self.last_event=time.monotonic()
88                        if self.stop.is_set() or self.resync.is_set():break
89                        if len(line)>262144:raise ValueError('oversized stream record')
90                        if line:
91                            data=json.loads(line)
92                            if data.get('type') in ('gameFull','gameState','gameStart','gameFinish','challenge','challengeCanceled','challengeDeclined'):
93                                self.emit('data',data,epoch)
94            except Exception as exc:self.emit('disconnected',type(exc).__name__,epoch)
95            finally:self.response=None
96            if self.stop.wait(delay):return
97            delay=min(60,delay*2)
98    def close(self):
99        self.stop.set()
100        if self.response:self.response.close()
101        self.thread.join(timeout=16)
102