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