Coverage for adrs/data/utils.py: 36%
22 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-08-29 15:31 +0800
« prev ^ index » next coverage.py v7.15.2, created at 2026-08-29 15:31 +0800
1import re
3from pathlib import Path
4from datetime import datetime, timezone
6from adrs.types import Topic
8DATE_RE = re.compile(r"(\d{4}-\d{2}-\d{2})")
11def make_filename(topic: Topic) -> str:
12 return (
13 (topic.query_params_str() or "data")
14 .replace("-", "_")
15 .replace("|", "_")
16 .replace("?", "_")
17 .replace("/", "_")
18 .replace("=", "_")
19 .replace("&", "_")
20 )
23def make_filepath(topic: Topic, fmt: str, suffix: str | None = None) -> Path:
24 filename = make_filename(topic)
25 if suffix:
26 filename = f"{filename}_{suffix}"
27 path = f"{topic.provider}|{topic.endpoint}".replace("coinglass-v4", "coinglass")
28 transformed = path.replace("/", "_").replace("|", "/")
29 return Path(transformed) / f"{filename}.{fmt}"
32def parse_date_from_filename(filename: str) -> datetime | None:
33 m = DATE_RE.search(filename)
34 if m:
35 try:
36 return datetime.fromisoformat(m.group(1)).replace(tzinfo=timezone.utc)
37 except ValueError:
38 return None
39 return None