Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/fsspec/utils.py: 15%

1from __future__ import annotations 

2 

3import contextlib 

4import ctypes 

5import logging 

6import math 

7import os 

8import re 

9import sys 

10import tempfile 

11from collections.abc import Callable, Iterable, Iterator, Mapping, Sequence 

12from functools import partial 

13from hashlib import md5 

14from importlib.metadata import version 

15from typing import IO, TYPE_CHECKING, Any, TypeVar 

16from urllib.parse import unquote, urlsplit 

17 

18if TYPE_CHECKING: 

19 import pathlib 

20 from typing import TypeGuard 

21 

22 from fsspec.spec import AbstractFileSystem 

23 

24 

25DEFAULT_BLOCK_SIZE = 5 * 2**20 

26 

27T = TypeVar("T") 

28 

29 

30def _unquote_userinfo(value: str) -> str: 

31 # Percent-decode userinfo; keep the raw value on any decode error (all-or-nothing). 

32 try: 

33 return unquote(value, errors="strict") 

34 except UnicodeDecodeError: 

35 return value 

36 

37 

38def infer_storage_options( 

39 urlpath: str, inherit_storage_options: dict[str, Any] | None = None 

40) -> dict[str, Any]: 

41 """Infer storage options from URL path and merge it with existing storage 

42 options. 

43 

44 Parameters 

45 ---------- 

46 urlpath: str or unicode 

47 Either local absolute file path or URL (hdfs://namenode:8020/file.csv) 

48 inherit_storage_options: dict (optional) 

49 Its contents will get merged with the inferred information from the 

50 given path 

51 

52 Returns 

53 ------- 

54 Storage options dict. 

55 

56 Examples 

57 -------- 

58 >>> infer_storage_options('/mnt/datasets/test.csv') # doctest: +SKIP 

59 {"protocol": "file", "path", "/mnt/datasets/test.csv"} 

60 >>> infer_storage_options( 

61 ... 'hdfs://username:pwd@node:123/mnt/datasets/test.csv?q=1', 

62 ... inherit_storage_options={'extra': 'value'}, 

63 ... ) # doctest: +SKIP 

64 {"protocol": "hdfs", "username": "username", "password": "pwd", 

65 "host": "node", "port": 123, "path": "/mnt/datasets/test.csv", 

66 "url_query": "q=1", "extra": "value"} 

67 """ 

68 

69 # Discover Windows paths including disk name in this special case. 

70 is_filesystem = re.match(r"^[a-zA-Z]:[\\/]", urlpath) 

71 

72 # Discover URI according to RFC 3986: Scheme names consist of a 

73 # sequence of characters beginning with a letter and followed by 

74 # any combination of letters, digits, plus ("+"), period ("."), 

75 # or hyphen ("-"). 

76 # https://datatracker.ietf.org/doc/html/rfc3986#section-3.1 

77 is_uri = re.match(r"^[a-zA-Z0-9+.-]+://", urlpath) 

78 

79 if is_filesystem or is_uri is None: 

80 return {"protocol": "file", "path": urlpath} 

81 

82 parsed_path = urlsplit(urlpath) 

83 protocol = parsed_path.scheme or "file" 

84 if parsed_path.fragment: 

85 path = "#".join([parsed_path.path, parsed_path.fragment]) 

86 else: 

87 path = parsed_path.path 

88 if protocol == "file": 

89 # Special case parsing file protocol URL on Windows according to: 

90 # https://msdn.microsoft.com/en-us/library/jj710207.aspx 

91 windows_path = re.match(r"^/([a-zA-Z])[:|]([\\/].*)$", path) 

92 if windows_path: 

93 drive, path = windows_path.groups() 

94 path = f"{drive}:{path}" 

95 

96 if protocol in ["http", "https"]: 

97 # for HTTP, we don't want to parse, as requests will anyway 

98 return {"protocol": protocol, "path": urlpath} 

99 

100 options: dict[str, Any] = {"protocol": protocol, "path": path} 

101 

102 if parsed_path.netloc: 

103 # Parse `hostname` from netloc manually because `parsed_path.hostname` 

104 # lowercases the hostname which is not always desirable (e.g. in S3): 

105 # https://github.com/dask/dask/issues/1417 

106 host = parsed_path.netloc.rsplit("@", 1)[-1] 

107 if host.startswith("[") and "]" in host: 

108 # An IPv6 literal carries colons of its own, so only a colon after 

109 # the closing bracket separates the port. 

110 options["host"] = host[: host.index("]") + 1] 

111 else: 

112 options["host"] = host.rsplit(":", 1)[0] 

113 

114 if protocol in ("s3", "s3a", "gcs", "gs"): 

115 options["path"] = options["host"] + options["path"] 

116 else: 

117 options["host"] = options["host"] 

118 if parsed_path.port: 

119 options["port"] = parsed_path.port 

120 if parsed_path.username: 

121 options["username"] = _unquote_userinfo(parsed_path.username) 

122 if parsed_path.password: 

123 options["password"] = _unquote_userinfo(parsed_path.password) 

124 

125 if parsed_path.query: 

126 options["url_query"] = parsed_path.query 

127 if parsed_path.fragment: 

128 options["url_fragment"] = parsed_path.fragment 

129 

130 if inherit_storage_options: 

131 update_storage_options(options, inherit_storage_options) 

132 

133 return options 

134 

135 

136def update_storage_options( 

137 options: dict[str, Any], inherited: dict[str, Any] | None = None 

138) -> None: 

139 if not inherited: 

140 inherited = {} 

141 collisions = set(options) & set(inherited) 

142 if collisions: 

143 for collision in collisions: 

144 if options.get(collision) != inherited.get(collision): 

145 raise KeyError( 

146 f"Collision between inferred and specified storage " 

147 f"option:\n{collision}" 

148 ) 

149 options.update(inherited) 

150 

151 

152# Compression extensions registered via fsspec.compression.register_compression 

153compressions: dict[str, str] = {} 

154 

155 

156def infer_compression(filename: str) -> str | None: 

157 """Infer compression, if available, from filename. 

158 

159 Infer a named compression type, if registered and available, from filename 

160 extension. This includes builtin (gz, bz2, zip) compressions, as well as 

161 optional compressions. See fsspec.compression.register_compression. 

162 """ 

163 extension = os.path.splitext(filename)[-1].strip(".").lower() 

164 if extension in compressions: 

165 return compressions[extension] 

166 return None 

167 

168 

169def build_name_function(max_int: float) -> Callable[[int], str]: 

170 """Returns a function that receives a single integer 

171 and returns it as a string padded by enough zero characters 

172 to align with maximum possible integer 

173 

174 >>> name_f = build_name_function(57) 

175 

176 >>> name_f(7) 

177 '07' 

178 >>> name_f(31) 

179 '31' 

180 >>> build_name_function(1000)(42) 

181 '0042' 

182 >>> build_name_function(999)(42) 

183 '042' 

184 >>> build_name_function(0)(0) 

185 '0' 

186 """ 

187 if max_int < 0: 

188 raise ValueError("max_int must be non-negative") 

189 pad_length = len(str(int(max_int))) 

190 

191 def name_function(i: int) -> str: 

192 return str(i).zfill(pad_length) 

193 

194 return name_function 

195 

196 

197def seek_delimiter(file: IO[bytes], delimiter: bytes, blocksize: int) -> bool: 

198 r"""Seek current file to file start, file end, or byte after delimiter seq. 

199 

200 Seeks file to next chunk delimiter, where chunks are defined on file start, 

201 a delimiting sequence, and file end. Use file.tell() to see location afterwards. 

202 Note that file start is a valid split, so must be at offset > 0 to seek for 

203 delimiter. 

204 

205 Parameters 

206 ---------- 

207 file: a file 

208 delimiter: bytes 

209 a delimiter like ``b'\n'`` or message sentinel, matching file .read() type 

210 blocksize: int 

211 Number of bytes to read from the file at once. 

212 

213 

214 Returns 

215 ------- 

216 Returns True if a delimiter was found, False if at file start or end. 

217 

218 """ 

219 

220 if file.tell() == 0: 

221 # beginning-of-file, return without seek 

222 return False 

223 

224 # Interface is for binary IO, with delimiter as bytes, but initialize last 

225 # with result of file.read to preserve compatibility with text IO. 

226 last: bytes | None = None 

227 while True: 

228 current = file.read(blocksize) 

229 if not current: 

230 # end-of-file without delimiter 

231 return False 

232 full = last + current if last else current 

233 try: 

234 if delimiter in full: 

235 i = full.index(delimiter) 

236 file.seek(file.tell() - (len(full) - i) + len(delimiter)) 

237 return True 

238 elif len(current) < blocksize: 

239 # end-of-file without delimiter 

240 return False 

241 except (OSError, ValueError): 

242 pass 

243 last = full[-len(delimiter) :] 

244 

245 

246def read_block( 

247 f: IO[bytes], 

248 offset: int, 

249 length: int | None, 

250 delimiter: bytes | None = None, 

251 split_before: bool = False, 

252) -> bytes: 

253 """Read a block of bytes from a file 

254 

255 Parameters 

256 ---------- 

257 f: File 

258 Open file 

259 offset: int 

260 Byte offset to start read 

261 length: int or None 

262 Number of bytes to read, read through end of file if None 

263 delimiter: bytes (optional) 

264 Ensure reading starts and stops at delimiter bytestring 

265 split_before: bool (optional) 

266 Start/stop read *before* delimiter bytestring. 

267 

268 

269 If using the ``delimiter=`` keyword argument we ensure that the read 

270 starts and stops at delimiter boundaries that follow the locations 

271 ``offset`` and ``offset + length``. If ``offset`` is zero then we 

272 start at zero, regardless of delimiter. The bytestring returned WILL 

273 include the terminating delimiter string. 

274 

275 Examples 

276 -------- 

277 

278 >>> from io import BytesIO # doctest: +SKIP 

279 >>> f = BytesIO(b'Alice, 100\\nBob, 200\\nCharlie, 300') # doctest: +SKIP 

280 >>> read_block(f, 0, 13) # doctest: +SKIP 

281 b'Alice, 100\\nBo' 

282 

283 >>> read_block(f, 0, 13, delimiter=b'\\n') # doctest: +SKIP 

284 b'Alice, 100\\nBob, 200\\n' 

285 

286 >>> read_block(f, 10, 10, delimiter=b'\\n') # doctest: +SKIP 

287 b'Bob, 200\\nCharlie, 300' 

288 """ 

289 if delimiter: 

290 f.seek(offset) 

291 found_start_delim = seek_delimiter(f, delimiter, 2**16) 

292 if length is None: 

293 return f.read() 

294 start = f.tell() 

295 length -= start - offset 

296 

297 f.seek(start + length) 

298 found_end_delim = seek_delimiter(f, delimiter, 2**16) 

299 end = f.tell() 

300 

301 # Adjust split location to before delimiter if seek found the 

302 # delimiter sequence, not start or end of file. 

303 if found_start_delim and split_before: 

304 start -= len(delimiter) 

305 

306 if found_end_delim and split_before: 

307 end -= len(delimiter) 

308 

309 offset = start 

310 length = end - start 

311 

312 f.seek(offset) 

313 

314 return f.read(length) 

315 

316 

317def tokenize(*args: Any, **kwargs: Any) -> str: 

318 """Deterministic token 

319 

320 (modified from dask.base) 

321 

322 >>> tokenize([1, 2, '3']) 

323 '9d71491b50023b06fc76928e6eddb952' 

324 

325 >>> tokenize('Hello') == tokenize('Hello') 

326 True 

327 """ 

328 if kwargs: 

329 args += (kwargs,) 

330 try: 

331 h = md5(str(args).encode()) 

332 except ValueError: 

333 # FIPS systems: https://github.com/fsspec/filesystem_spec/issues/380 

334 h = md5(str(args).encode(), usedforsecurity=False) 

335 return h.hexdigest() 

336 

337 

338def stringify_path(filepath: str | os.PathLike[str] | pathlib.Path) -> str: 

339 """Attempt to convert a path-like object to a string. 

340 

341 Parameters 

342 ---------- 

343 filepath: object to be converted 

344 

345 Returns 

346 ------- 

347 filepath_str: maybe a string version of the object 

348 

349 Notes 

350 ----- 

351 Objects supporting the fspath protocol are coerced according to its 

352 __fspath__ method. 

353 

354 For backwards compatibility with older Python version, pathlib.Path 

355 objects are specially coerced. 

356 

357 Any other object is passed through unchanged, which includes bytes, 

358 strings, buffers, or anything else that's not even path-like. 

359 """ 

360 if isinstance(filepath, str): 

361 return filepath 

362 elif hasattr(filepath, "__fspath__"): 

363 return filepath.__fspath__() 

364 elif hasattr(filepath, "path"): 

365 return filepath.path 

366 else: 

367 return filepath # type: ignore[return-value] 

368 

369 

370def make_instance( 

371 cls: Callable[..., T], args: Sequence[Any], kwargs: dict[str, Any] 

372) -> T: 

373 inst = cls(*args, **kwargs) 

374 inst._determine_worker() # type: ignore[attr-defined] 

375 return inst 

376 

377 

378def common_prefix(paths: Iterable[str]) -> str: 

379 """For a list of paths, find the shortest prefix common to all""" 

380 parts = [p.split("/") for p in paths] 

381 lmax = min(len(p) for p in parts) 

382 end = 0 

383 for i in range(lmax): 

384 end = all(p[i] == parts[0][i] for p in parts) 

385 if not end: 

386 break 

387 i += end 

388 return "/".join(parts[0][:i]) 

389 

390 

391def other_paths( 

392 paths: list[str], 

393 path2: str | list[str], 

394 exists: bool = False, 

395 flatten: bool = False, 

396) -> list[str]: 

397 """In bulk file operations, construct a new file tree from a list of files 

398 

399 Parameters 

400 ---------- 

401 paths: list of str 

402 The input file tree 

403 path2: str or list of str 

404 Root to construct the new list in. If this is already a list of str, we just 

405 assert it has the right number of elements. 

406 exists: bool (optional) 

407 For a str destination, it is already exists (and is a dir), files should 

408 end up inside. 

409 flatten: bool (optional) 

410 Whether to flatten the input directory tree structure so that the output files 

411 are in the same directory. 

412 

413 Returns 

414 ------- 

415 list of str 

416 """ 

417 

418 if isinstance(path2, str): 

419 path2 = path2.rstrip("/") 

420 

421 if flatten: 

422 path2 = ["/".join((path2, p.split("/")[-1])) for p in paths] 

423 else: 

424 cp = common_prefix(paths) 

425 if exists: 

426 cp = cp.rpartition("/")[0] 

427 if not cp and all(not s.startswith("/") for s in paths): 

428 path2 = ["/".join([path2, p]) for p in paths] 

429 else: 

430 path2 = [p.replace(cp, path2, 1) for p in paths] 

431 else: 

432 assert len(paths) == len(path2) 

433 return path2 

434 

435 

436def check_contained(root: str, paths: list[str]) -> None: 

437 """Raise if any of ``paths`` lies outside the destination ``root``. 

438 

439 Bulk copies build their destination names by joining source names onto a 

440 destination root. Those names come from the source listing, so a name 

441 holding ".." segments resolves above the root and writes outside the 

442 destination the caller asked for. 

443 

444 Parameters 

445 ---------- 

446 root: str 

447 The destination the caller passed. 

448 paths: list of str 

449 The destination names built for that root. 

450 """ 

451 root_abs = os.path.abspath(root) 

452 # normcase so that a case-insensitive platform does not report a false 

453 # escape, while the message keeps the paths as the caller would see them. 

454 root_key = os.path.normcase(root_abs) 

455 prefix = root_key.rstrip(os.sep) + os.sep 

456 for path in paths: 

457 path_abs = os.path.abspath(path) 

458 path_key = os.path.normcase(path_abs) 

459 if path_key != root_key and not path_key.startswith(prefix): 

460 raise ValueError( 

461 f"path {path!r} would be copied to {path_abs!r}, which is " 

462 f"outside the destination {root!r}" 

463 ) 

464 

465 

466def is_exception(obj: Any) -> bool: 

467 return isinstance(obj, BaseException) 

468 

469 

470def isfilelike(f: Any) -> TypeGuard[IO[bytes]]: 

471 return all(hasattr(f, attr) for attr in ["read", "close", "tell"]) 

472 

473 

474def get_protocol(url: str) -> str: 

475 url = stringify_path(url) 

476 parts = re.split(r"(\:\:|\://)", url, maxsplit=1) 

477 if len(parts) > 1: 

478 return parts[0] 

479 return "file" 

480 

481 

482def get_file_extension(url: str) -> str: 

483 url = stringify_path(url) 

484 # Only consider the final path component: a "." in a parent directory name 

485 # (e.g. "/path/to.dir/file") is not the file's extension. 

486 name = url.rsplit("/", 1)[-1] 

487 # A leading dot marks a hidden file rather than an extension, so ".bashrc" 

488 # has none, while ".hidden.txt" still has "txt". 

489 stem, dot, extension = name.lstrip(".").rpartition(".") 

490 if stem and dot: 

491 return extension 

492 return "" 

493 

494 

495def can_be_local(path: str) -> bool: 

496 """Can the given URL be used with open_local?""" 

497 from fsspec import get_filesystem_class 

498 

499 try: 

500 return getattr(get_filesystem_class(get_protocol(path)), "local_file", False) 

501 except (ValueError, ImportError): 

502 # not in registry or import failed 

503 return False 

504 

505 

506def get_package_version_without_import(name: str) -> str | None: 

507 """For given package name, try to find the version without importing it 

508 

509 Import and package.__version__ is still the backup here, so an import 

510 *might* happen. 

511 

512 Returns either the version string, or None if the package 

513 or the version was not readily found. 

514 """ 

515 if name in sys.modules: 

516 mod = sys.modules[name] 

517 if hasattr(mod, "__version__"): 

518 return mod.__version__ 

519 try: 

520 return version(name) 

521 except: # noqa: E722 

522 pass 

523 try: 

524 import importlib 

525 

526 mod = importlib.import_module(name) 

527 return mod.__version__ 

528 except (ImportError, AttributeError): 

529 return None 

530 

531 

532def setup_logging( 

533 logger: logging.Logger | None = None, 

534 logger_name: str | None = None, 

535 level: str = "DEBUG", 

536 clear: bool = True, 

537) -> logging.Logger: 

538 if logger is None and logger_name is None: 

539 raise ValueError("Provide either logger object or logger name") 

540 logger = logger or logging.getLogger(logger_name) 

541 handle = logging.StreamHandler() 

542 formatter = logging.Formatter( 

543 "%(asctime)s - %(name)s - %(levelname)s - %(funcName)s -- %(message)s" 

544 ) 

545 handle.setFormatter(formatter) 

546 if clear: 

547 logger.handlers.clear() 

548 logger.addHandler(handle) 

549 logger.setLevel(level) 

550 return logger 

551 

552 

553def _unstrip_protocol(name: str, fs: AbstractFileSystem) -> str: 

554 return fs.unstrip_protocol(name) 

555 

556 

557def mirror_from( 

558 origin_name: str, methods: Iterable[str] 

559) -> Callable[[type[T]], type[T]]: 

560 """Mirror attributes and methods from the given 

561 origin_name attribute of the instance to the 

562 decorated class""" 

563 

564 def origin_getter(method: str, self: Any) -> Any: 

565 origin = getattr(self, origin_name) 

566 return getattr(origin, method) 

567 

568 def wrapper(cls: type[T]) -> type[T]: 

569 for method in methods: 

570 wrapped_method = partial(origin_getter, method) 

571 setattr(cls, method, property(wrapped_method)) 

572 return cls 

573 

574 return wrapper 

575 

576 

577@contextlib.contextmanager 

578def nullcontext(obj: T) -> Iterator[T]: 

579 yield obj 

580 

581 

582def parse_content_range_header( 

583 headers: Mapping[str, str], 

584) -> tuple[int | None, int | None, int | None]: 

585 """Return ``(start, end, total)`` from a ``Content-Range`` header.""" 

586 m = re.match(r"bytes (\d+-\d+|\*)/(\d+|\*)", headers.get("Content-Range", "")) 

587 if not m: 

588 return None, None, None 

589 if m[1] == "*": 

590 start = end = None 

591 else: 

592 start, end = [int(x) for x in m[1].split("-")] 

593 return start, end, None if m[2] == "*" else int(m[2]) 

594 

595 

596def response_is_requested_range( 

597 status: int, headers: Mapping[str, str], start: int | None, end: int | None 

598) -> bool: 

599 """Whether the response is the requested range, not the whole file.""" 

600 if status == 206: 

601 return True 

602 if start is not None and parse_content_range_header(headers)[0] == start: 

603 return True 

604 if start is None or start < 0 or end is None: 

605 return False 

606 return int(headers.get("Content-Length", end + 1)) <= end - start 

607 

608 

609def range_response_window( 

610 out: bytes, 

611 status: int, 

612 headers: Mapping[str, str], 

613 start: int | None, 

614 end: int | None, 

615) -> bytes: 

616 """The requested window of a response body that may be the whole file.""" 

617 lo = 0 if start is None else start 

618 if lo < 0 and (end is None or end >= len(out)): 

619 # a suffix range asks for the tail of either response 

620 return out[lo:] 

621 if status != 206: 

622 window = end - lo if end is not None and (lo >= 0) == (end >= 0) else None 

623 content_range_start = parse_content_range_header(headers)[0] 

624 if lo >= 0 and content_range_start is not None and content_range_start != lo: 

625 window = 0 

626 if window is not None and len(out) > window: 

627 if lo > 0: 

628 raise ValueError( 

629 "The HTTP server doesn't appear to support range requests. " 

630 "Only reading this file from the beginning is supported. " 

631 "Open with block_size=0 for a streaming file interface." 

632 ) 

633 return out[:end] if lo == 0 else out[len(out) + lo : end] 

634 return out 

635 

636 

637def merge_offset_ranges( 

638 paths: list[str], 

639 starts: list[int | None] | int | None, 

640 ends: list[int | None] | int | None, 

641 max_gap: int = 0, 

642 max_block: int | None = None, 

643 sort: bool = True, 

644) -> tuple[list[str], list[int], list[int | None]]: 

645 """Merge adjacent byte-offset ranges when the inter-range 

646 gap is <= `max_gap`, and when the merged byte range does not 

647 exceed `max_block` (if specified). Every input range is covered by 

648 at least one returned range. Overlapping input ranges are merged 

649 where `max_block` allows it, so returned ranges may overlap once a 

650 chain of overlapping inputs reaches `max_block`; a single input 

651 range larger than `max_block` is still returned whole. 

652 

653 An `end` of `None` means to the end of the file. By default, this 

654 function will re-order the input paths and byte ranges to ensure 

655 sorted order. 

656 

657 Passing `sort=False` skips the re-ordering, which is only worthwhile 

658 when the inputs are already grouped by path and ascending by start 

659 within each path. Ranges that break that order still appear in the 

660 output, as their own range rather than merged, so coverage holds for 

661 any input order. 

662 """ 

663 # Check input 

664 if not isinstance(paths, list): 

665 raise TypeError 

666 if not isinstance(starts, list): 

667 starts = [starts] * len(paths) 

668 if not isinstance(ends, list): 

669 ends = [ends] * len(paths) 

670 if len(starts) != len(paths) or len(ends) != len(paths): 

671 raise ValueError 

672 

673 starts_i: list[int] = [s or 0 for s in starts] 

674 ends_i: list[int | None] = ends 

675 

676 # Early Return 

677 if len(starts_i) <= 1: 

678 return paths, starts_i, ends_i 

679 

680 # Sort by paths and then ranges if `sort=True` 

681 if sort: 

682 ranges = sorted( 

683 zip(paths, starts_i, ends_i), 

684 # None end sorts last (covers furthest into the file) 

685 key=lambda pse: (pse[0], pse[1], math.inf if pse[2] is None else pse[2]), 

686 ) 

687 paths = [r[0] for r in ranges] 

688 starts_i = [r[1] for r in ranges] 

689 ends_i = [r[2] for r in ranges] 

690 

691 # Loop through the coupled `paths`, `starts`, and 

692 # `ends`, and merge adjacent blocks when appropriate 

693 new_paths = paths[:1] 

694 new_starts = starts_i[:1] 

695 new_ends = ends_i[:1] 

696 for path, start, end in zip(paths[1:], starts_i[1:], ends_i[1:]): 

697 prev_end = new_ends[-1] 

698 if path != new_paths[-1]: 

699 # Cannot merge with previous block 

700 new_paths.append(path) 

701 new_starts.append(start) 

702 new_ends.append(end) 

703 elif start < new_starts[-1]: 

704 # Out of order (only possible when `sort=False`). Walking the 

705 # current block start backwards would uncover bytes already 

706 # attributed to it, so give this range its own block 

707 new_paths.append(path) 

708 new_starts.append(start) 

709 new_ends.append(end) 

710 elif prev_end is None: 

711 # Previous block already covers the rest of the file 

712 continue 

713 elif start < prev_end: 

714 # Overlap / nested 

715 if end is not None and end <= prev_end: 

716 # Already covered by the current block 

717 continue 

718 elif ( 

719 end is not None 

720 and max_block is not None 

721 and (end - new_starts[-1]) > max_block 

722 ): 

723 # Extending would exceed `max_block`. Start a new block, 

724 # which overlaps the previous one, rather than letting a 

725 # chain of overlaps grow the block without bound 

726 new_paths.append(path) 

727 new_starts.append(start) 

728 new_ends.append(end) 

729 else: 

730 # An `end` of None extends the block to EOF; a separate 

731 # block would subsume the current one anyway 

732 new_ends[-1] = end 

733 elif (start - prev_end) > max_gap or ( 

734 max_block is not None 

735 and (end is None or (end - new_starts[-1]) > max_block) 

736 ): 

737 # Gap too large, or merging would exceed `max_block` 

738 new_paths.append(path) 

739 new_starts.append(start) 

740 new_ends.append(end) 

741 else: 

742 # Merge with the previous block 

743 new_ends[-1] = end 

744 

745 return new_paths, new_starts, new_ends 

746 

747 

748def file_size(filelike: IO[bytes]) -> int: 

749 """Find length of any open read-mode file-like""" 

750 pos = filelike.tell() 

751 try: 

752 return filelike.seek(0, 2) 

753 finally: 

754 filelike.seek(pos) 

755 

756 

757@contextlib.contextmanager 

758def atomic_write(path: str, mode: str = "wb"): 

759 """ 

760 A context manager that opens a temporary file next to `path` and, on exit, 

761 replaces `path` with the temporary file, thereby updating `path` 

762 atomically. 

763 """ 

764 fd, fn = tempfile.mkstemp( 

765 dir=os.path.dirname(path), prefix=os.path.basename(path) + "-" 

766 ) 

767 try: 

768 with open(fd, mode) as fp: 

769 yield fp 

770 except BaseException: 

771 with contextlib.suppress(FileNotFoundError): 

772 os.unlink(fn) 

773 raise 

774 else: 

775 os.replace(fn, path) 

776 

777 

778def _translate(pat, STAR, QUESTION_MARK): 

779 # Copied from: https://github.com/python/cpython/pull/106703. 

780 res: list[str] = [] 

781 add = res.append 

782 i, n = 0, len(pat) 

783 while i < n: 

784 c = pat[i] 

785 i = i + 1 

786 if c == "*": 

787 # compress consecutive `*` into one 

788 if (not res) or res[-1] is not STAR: 

789 add(STAR) 

790 elif c == "?": 

791 add(QUESTION_MARK) 

792 elif c == "[": 

793 j = i 

794 if j < n and pat[j] == "!": 

795 j = j + 1 

796 if j < n and pat[j] == "]": 

797 j = j + 1 

798 while j < n and pat[j] != "]": 

799 j = j + 1 

800 if j >= n: 

801 add("\\[") 

802 else: 

803 stuff = pat[i:j] 

804 if "-" not in stuff: 

805 stuff = stuff.replace("\\", r"\\") 

806 else: 

807 chunks = [] 

808 k = i + 2 if pat[i] == "!" else i + 1 

809 while True: 

810 k = pat.find("-", k, j) 

811 if k < 0: 

812 break 

813 chunks.append(pat[i:k]) 

814 i = k + 1 

815 k = k + 3 

816 chunk = pat[i:j] 

817 if chunk: 

818 chunks.append(chunk) 

819 else: 

820 chunks[-1] += "-" 

821 # Remove empty ranges -- invalid in RE. 

822 for k in range(len(chunks) - 1, 0, -1): 

823 if chunks[k - 1][-1] > chunks[k][0]: 

824 chunks[k - 1] = chunks[k - 1][:-1] + chunks[k][1:] 

825 del chunks[k] 

826 # Escape backslashes and hyphens for set difference (--). 

827 # Hyphens that create ranges shouldn't be escaped. 

828 stuff = "-".join( 

829 s.replace("\\", r"\\").replace("-", r"\-") for s in chunks 

830 ) 

831 # Escape set operations (&&, ~~ and ||). 

832 stuff = re.sub(r"([&~|])", r"\\\1", stuff) 

833 i = j + 1 

834 if not stuff: 

835 # Empty range: never match. 

836 add("(?!)") 

837 elif stuff == "!": 

838 # Negated empty range: match any character. 

839 add(".") 

840 else: 

841 if stuff[0] == "!": 

842 stuff = "^" + stuff[1:] 

843 elif stuff[0] in ("^", "["): 

844 stuff = "\\" + stuff 

845 add(f"[{stuff}]") 

846 else: 

847 add(re.escape(c)) 

848 assert i == n 

849 return res