- async def _fetch(self, url: str, out: str) -> None:
- out = os.path.join(self.config.dlcache, out)
- res = cast(Any, urlopen(url))
- with open(out, "wb") as f:
- msg: HTTPMessage = res.info()
- file_size = int(str(msg.get("Content-length")))
- print(f"downloading {file_size} bytes...")
- progress = 0
- block = 16 * 1024
- while True:
- buf: bytes = res.read(block)
- if not buf:
- break
- progress += len(buf)
- print(f"downloaded {progress}/{file_size} bytes")
- f.write(buf)
- await asyncio.sleep(0)
+ async def ls(self) -> List[str]:
+ class Parser(HTMLParser):
+ __slots__ = "paks"
+
+ def __init__(self) -> None:
+ super().__init__()
+ self.paks: List[str] = []
+
+ def handle_starttag(self, tag: str, attrs: List[Tuple[str, str]]) -> None:
+ if tag == "a":
+ file: Optional[str] = next((unquote(v) for (k, v) in attrs if k == "href"), None)
+ if file and file.endswith(".pk3"):
+ self.paks.append(file)
+
+ parser = Parser()
+ arr = bytearray()
+ async for buf in fetch(self.config.upstream):
+ arr.extend(buf)
+ buf = arr
+ parser.feed(buf.decode(UTF_8))
+ return parser.paks