GeezerVisonThe World Wide Web, restored to 800 pixels of glory. |
| Home|About|Gaming cabinet|Archives|How it works|Starter sites |
GeezerVison is showing github.com — python-scripts/util.py at master · fireattack/python-scripts Expand file tree/ Copy pathutil.pyMore file actions 1288 lines (1128 loc) · 50.9 KB / Copy pathutil.pyFile metadata and controls1288 lines (1128 loc) · 50.9 KB 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 279 280 281 282 283 284 285 286 287 288 289 290 291 292 293 294 295 296 297 298 299 300 301 302 303 304 305 306 307 308 309 310 311 312 313 314 315 316 317 318 319 320 321 322 323 324 325 326 327 328 329 330 331 332 333 334 335 336 337 338 339 340 341 342 343 344 345 346 347 348 349 350 351 352 353 354 355 356 357 358 359 360 361 362 363 364 365 366 367 368 369 370 371 372 373 374 375 376 377 378 379 380 381 382 383 384 385 386 387 388 389 390 391 392 393 394 395 396 397 398 399 400 401 402 403 404 405 406 407 408 409 410 411 412 413 414 415 416 417 418 419 420 421 422 423 424 425 426 427 428 429 430 431 432 433 434 435 436 437 438 439 440 441 442 443 444 445 446 447 448 449 450 451 452 453 454 455 456 457 458 459 460 461 462 463 464 465 466 467 468 469 470 471 472 473 474 475 476 477 478 479 480 481 482 483 484 485 486 487 488 489 490 491 492 493 494 495 496 497 498 499 500 501 502 503 504 505 506 507 508 509 510 511 512 513 514 515 516 517 518 519 520 521 522 523 524 525 526 527 528 529 530 531 532 533 534 535 536 537 538 539 540 541 542 543 544 545 546 547 548 549 550 551 552 553 554 555 556 557 558 559 560 561 562 563 564 565 566 567 568 569 570 571 572 573 574 575 576 577 578 579 580 581 582 583 584 585 586 587 588 589 590 591 592 593 594 595 596 597 598 599 600 601 602 603 604 605 606 607 608 609 610 611 612 613 614 615 616 617 618 619 620 621 622 623 624 625 626 627 628 629 630 631 632 633 634 635 636 637 638 639 640 641 642 643 644 645 646 647 648 649 650 651 652 653 654 655 656 657 658 659 660 661 662 663 664 665 666 667 668 669 670 671 672 673 674 675 676 677 678 679 680 681 682 683 684 685 686 687 688 689 690 691 692 693 694 695 696 697 698 699 700 701 702 703 704 705 706 707 708 709 710 711 712 713 714 715 716 717 718 719 720 721 722 723 724 725 726 727 728 729 730 731 732 733 734 735 736 737 738 739 740 741 742 743 744 745 746 747 748 749 750 751 752 753 754 755 756 757 758 759 760 761 762 763 764 765 766 767 768 769 770 771 772 773 774 775 776 777 778 779 780 781 782 783 784 785 786 787 788 789 790 791 792 793 794 795 796 797 798 799 800 801 802 803 804 805 806 807 808 809 810 811 812 813 814 815 816 817 818 819 820 821 822 823 824 825 826 827 828 829 830 831 832 833 834 835 836 837 838 839 840 841 842 843 844 845 846 847 848 849 850 851 852 853 854 855 856 857 858 859 860 861 862 863 864 865 866 867 868 869 870 871 872 873 874 875 876 877 878 879 880 881 882 883 884 885 886 887 888 889 890 891 892 893 894 895 896 897 898 899 900 901 902 903 904 905 906 907 908 909 910 911 912 913 914 915 916 917 918 919 920 921 922 923 924 925 926 927 928 929 930 931 932 933 934 935 936 937 938 939 940 941 942 943 944 945 946 947 948 949 950 951 952 953 954 955 956 957 958 959 960 961 962 963 964 965 966 967 968 969 970 971 972 973 974 975 976 977 978 979 980 981 982 983 984 985 986 987 988 989 990 991 992 993 994 995 996 997 998 999 1000 import builtins import json import os import re import sys import time import hashlib import shutil import unicodedata from datetime import datetime from pathlib import Path from urllib.parse import unquote # all the external dependencies are imported inside the functions, # so we can use this file in other projects without installing them. # or to copy and paste the functions to other public projects directly. # to install them all, you can do: # pip install requests lxml beautifulsoup4 python-dateutil pytz pyperclip wcwidth rich browsercookie # ==================== CONSTANTS ==================== '''Twitter media name handle The filename format I use, inherited from good old twMediaDownloader (RIP: here is a mirror: https://github.com/fireattack/twMediaDownloader). The first version must start with screen_name and ended with type[index][ dupe].suffix The second one allows optional arbitrary prefix or suffix. NOTE: to make it simpler, the returned m['extra'] and m['dupe'] will have leading space or hyphen with it. ''' TWITTER_FILENAME_RE = re.compile(r'^(?P<screen_name>\w+)-(?P<id>\d+)-(?P<date>\d{8})_(?P<time>\d{6})-(?P<type>[^-.]+?)(?P<index>\d*)(?P<dupe> *\(\d+\))?(?P<suffix>\.(?:mp4|zip|jpg|png))$') TWITTER_FILENAME_RELEXED_RE = re.compile(r'^(?:(?P<prefix>.+?)(?: +?|[-]??))??(?P<screen_name>\w+)-(?P<id>\d+)-(?P<date>\d{8})_(?P<time>\d{6})-(?P<type>[^-.]+?)(?P<index>\d*)(?P<extra>[ _-].+?)??(?P<dupe> *\(\d+\))?(?P<suffix>\.(?:mp4|zip|jpg|png))$') # 240521_osk_airi_C7OgZYqPbyI.jpg # 241101_o_rikachi_o_DBz6UC3TZCL_5.jpg # 241101_o_rikachi_o_STORY_3491384076140533984.jpg INSTAGRAM_FILENAME_RE = re.compile(r'^(?P<date>\d{6})_(?P<user_id>.+?)_(?:STORY_(?P<story_id>\d+)|(?P<post_id>[\-_A-Za-z0-9]{11})(?:_(?P<index>\d+))?)(?P<suffix>\.[^.]+)$') # ==================== data structure manipulation & misc. ==================== def to_list(a): return a if isinstance(a, list) or a is None else [a] def flatten(x): from collections.abc import Iterable if isinstance(x, Iterable) and not isinstance(x, str): return [a for i in x for a in flatten(i)] else: return [x] def print_cmd(cmd, prefix='', actually_print=True): commands_text_form = [f'"{c}"' if re.search(r'[ ?]', str(c)) else str(c) for c in cmd] output = prefix + ' '.join(commands_text_form) if actually_print: print(output) return output def copy(data): # pip install pyperclip import pyperclip pyperclip.copy(data) def get_clipboard_data(): # pip install pyperclip import pyperclip return pyperclip.paste() def safeify(name, ignore_backslash=False): """ Replaces illegal characters in a given name with safe alternatives. Args: name (str): The name to be made safe. ignore_backslash (bool, optional): Whether to ignore backslashes. Defaults to False. Returns: str: The safe version of the name. Raises: AssertionError: If the name is not a string. """ assert isinstance(name, str), f'Name must be a string, not {type(name)}' template = {'\\': '＼', '/': '／', ':': '：', '*': '＊', '?': '？', '"': '＂', '<': '＜', '>': '＞', '|': '｜','\n':'','\r':'','\t':''} if ignore_backslash: template.pop('\\', None) for illegal in template: name = name.replace(illegal, template[illegal]) return name def format_str(s, width=None, align='left', padding=' '): """ Format a string `s` with a specified width, alignment, and padding. Args: s (str): The string to be formatted. width (int, optional): The desired width of the formatted string. If not provided, the original string will be returned as is. Defaults to None. align (str, optional): The alignment of the formatted string. Possible values are 'left', 'right', and 'center'. Defaults to 'left'. padding (str, optional): The padding character used to fill the remaining space in the formatted string. Defaults to ' '. Returns: str: The formatted string. """ # pip install wcwidth import wcwidth if s is None: s = '' else: s = str(s) if not width: return s output = '' length = 0 for char in s: size = wcwidth.wcswidth(char) if length + size > width: break output += char length += size if align == 'left': return output + padding*(width-length) if align == 'right': return padding*(width-length) + output if align == 'center': left_space = (width-length)//2 right_space = width-length-left_space return padding*left_space + output + padding*right_space class Table(): # pip install wcwidth def __init__(self, rows=None, headers=None, max_width=100) -> None: if rows and not headers: self.headers = rows[0] self.data = rows[1:] elif headers: self.headers = headers if rows: self.data = rows else: self.data = [] else: raise Exception('No header or data given!') self.max_width = max_width def rows(self): for row in self.data: row_dict = {} for i, header in enumerate(self.headers): row_dict[header] = row[i] yield row_dict def search(self, keywords, single=True): matched_rows = [] for row in self.data: matched = True for key, value in keywords.items(): assert key in self.headers col_idx = self.headers.index(key) if not value: # Ignore empty values continue if not ((str(row[col_idx]) == str(value)) or (row[col_idx] == value)): matched = False break if matched: matched_rows.append(row) if single: break if matched_rows: if single and len(matched_rows) == 1: return matched_rows[0] else: return matched_rows else: return [] def sort(self, header_name_or_index, reverse=False): if isinstance(header_name_or_index, int): col_idx = header_name_or_index else: col_idx = self.headers.index(header_name_or_index) self.data.sort(key=lambda x: x[col_idx], reverse=reverse) def append(self, new_data): d = [''] * len(self.headers) for key, value in new_data.items(): if not key in self.headers: # silently ignore keys that are not in headers continue col_idx = self.headers.index(key) d[col_idx] = value self.data.append(d) def print(self, formats=None, custom_print=None): import wcwidth def fmt(value, str_format): try: return str_format.format(value) except Exception: return value def calc_width(col_format, idx): maxw = sum(wcwidth.wcwidth(c) for c in str(self.headers[idx])) for row in self.data: if idx > len(row) - 1: continue w =sum(wcwidth.wcwidth(c) for c in fmt(row[idx], col_format['str_format'])) if w > maxw: maxw = w maxw = min(maxw, col_format['max_width']) col_format['width'] = maxw def print_row(row, header_mode=False, custom_print=custom_print): parts = [] for idx in range(len(row)): col_format = col_formats[idx] # override str format back to nothing for headers str_format = "{}" if header_mode else col_format['str_format'] s = format_str(fmt(row[idx], str_format), width=col_format['width'], align=col_format['align']) parts.append(s) line = ' '.join(parts) if custom_print: custom_print(line) else: print(line) col_formats = dict() for idx in range(len(self.headers)): col_format = { "align": "left", "str_format": "{}", "max_width": self.max_width } if formats: if idx in formats: col_format.update(formats[idx]) elif self.headers[idx] in formats: col_format.update(formats[self.headers[idx]]) if not 'width' in col_format: calc_width(col_format, idx) col_formats[idx] = col_format print_row(self.headers, header_mode=True) for row in self.data: print_row(row) # return some useful info total_width = sum(col_format['width'] for col_format in col_formats.values()) + 2 * (len(col_formats) - 1) return {'total_width': total_width, 'col_formats': col_formats} def save(self, f): s = '' f = Path(f) s += '\t'.join(self.headers) + '\n' for row in self.data: s += '\t'.join([str(cell) for cell in row]) + '\n' f.write_text(s, encoding='utf8') def multi_col_print(data, columns=5): '''Print a list of data in a table with columns''' # pip install rich from rich.console import Console from rich.table import Table table = Table(show_header=False, box=None) for _ in range(columns): table.add_column(style='cyan') # add data to table vertically row_count = len(data) // columns + 1 for i in range(row_count): row = data[i::row_count] table.add_row(*row) console = Console() console.print(table) def rprint(*args, **kwargs): ''' Print function that uses rich library to print (This is different from rich.print() because that does not support additional arguments like `style`.) The console obj is created only once and reused. ''' # pip install rich from rich.console import Console if not hasattr(rprint, '_console'): rprint._console = Console() rprint._console.print(*args, **kwargs) def print2(*args, **kwargs): ''' Custom print function that manages line endings. If a previous print call ended without a newline (e.g., end=''), and the current call implies a newline (e.g., default end or end='\n'), an explicit newline is printed first to terminate the previous line. The state _newlined is stored as an attribute of the print2 function itself. ''' from builtins import print as builtin_print if not hasattr(print2, '_newlined'): print2._newlined = True end = kwargs.pop('end', '\n') if end.endswith('\n') and not print2._newlined: builtin_print() print2._newlined = True if end.endswith('\n') else False builtin_print(*args, **kwargs, end=end) def array_to_range_text(a, sep=', ', dash='-'): ''' Convert an array of integers to a range text (e.g., [1,2,3,5,6,8] -> '1-3, 5-6, 8') ''' s = '' prev_seg = None for idx, seg in enumerate(a): if idx == 0: # first segment s += f'{seg}' range_start = seg else: if seg - prev_seg == 1: if idx == len(a) - 1: # last segment s += f'{dash}{seg}' else: if prev_seg > range_start: s += f'{dash}{prev_seg}' s += f'{sep}{seg}' range_start = seg prev_seg = seg return s def pluralize(word, count, suffix='s', alt=None): ''' Pluralize a word based on the count. If count is 1, return the word as is. If count is not 1, return the word with the suffix added, or the alt word if provided (e.g. for irregular plurals). ''' if count == 1: return word else: if alt: return alt else: return word + suffix def compare_obj(value_old, value, print_prefix='ROOT', mute=False): ''' Compare two objects and print the differences. value_old: the old object value: the new object print_prefix: the prefix to print before the key mute: whether to print the differences or not Returns: bool: whether the two objects are equal or not ''' # pip install rich from rich import print as rprint def print(*args, **kwargs): if not mute: rprint(*args, **kwargs) equal = True type_changed = False if type(value_old) != type(value): type_changed = True equal = False elif isinstance(value, dict): for key, v in value.items(): if key not in value_old: equal = False print(f'{print_prefix}.{key}: [green]Added:[/green]', v) else: v_old = value_old[key] equal &= compare_obj(v_old, v, print_prefix=f'{print_prefix}.{key}', mute=mute) for key, v in value_old.items(): if key not in value: equal = False print(f'{print_prefix}.{key}: [red]Removed:[/red]', v) return equal elif isinstance(value, list): for i in range(min(len(value), len(value_old))): equal &= compare_obj(value_old[i], value[i], print_prefix=f'{print_prefix}[{i}]', mute=mute) if len(value_old) < len(value): equal = False for i in range(len(value_old), len(value)): print(f'{print_prefix}[{i}]: [green]Added:[/green]', value[i]) elif len(value_old) > len(value): equal = False for i in range(len(value), len(value_old)): print(f'{print_prefix}[{i}]: [red]Removed:[/red]',value_old[i]) return equal elif isinstance(value, str): equal = re.sub(r'\s+',' ', value_old) == re.sub(r'\s+',' ', value) else: equal = value_old == value if not equal: print(f'{print_prefix}:', end='') s = str(value_old) + str(value) if len(s) < 60 and not '\n' in s: print(f' {value_old} [yellow]->[/yellow] {value}') # print(f'[red]Old:[/red] {value_old} [yellow]->[/yellow] [green]New:[/green] {value}') else: print() print('[red]Old:', value_old) print('[green]New:', value) return equal def normalize(s): '''Normalize a string by replacing full-width characters with half-width characters but keep the full-width dash (～) as it is.''' return unicodedata.normalize('NFKC', s.replace('～', '$dash$')).replace('$dash$', '～') # ==================== datetime related ==================== def parse_to_shortdate(date_str, fmt=None): # pip install python-dateutil from dateutil import parser current_year = datetime.now().year if fmt is None: fmt = '%y%m%d' if isinstance(date_str, datetime): return date_str.strftime(fmt) date_str = re.sub(r'[\s ]+', ' ', date_str).strip() patterns = [r'(?:(?P<y>\d+)年)? *(?P<m>\d+)月 *(?P<d>\d+)日', r'(?P<y>\d{4})[/\-.](?P<m>\d{1,2})[/\-.](?P<d>\d{1,2})'] for pattern in patterns: # Sometimes the string has extra spaces. But matching with spaces removed is dangerous # since things like `2014/3/7 23:52` will be parsed as `20140372`. But it *should* have been # caught by the first re.search already, so 99% of the cases it should be fine. if m := re.search(pattern, date_str) or re.search(pattern, date_str.replace(' ', '')): # year is assumed to be current year if not provided for kanji dates. # for numbers, it is NOT assumed because it has way too many edge cases. date_str = f"{m['y'] or current_year}-{m['m']}-{m['d']}" break try: return parser.parse(date_str, yearfirst=True).strftime(fmt) except: return "" class MyTime: def __init__(self, t=None): import pytz self.tz = pytz.timezone('Asia/Tokyo') if t is None: t = datetime.now() # Convert now to local timezone and then to JST self.naive_time = t.replace(tzinfo=None) self.local_time = t.astimezone() self.jst_time = self.local_time.astimezone(self.tz) def _format_time(self, dt, format_type): if format_type == 'obj': return dt elif format_type == "short": return dt.strftime('%y%m%d_%H%M%S') elif format_type == "pretty": return dt.strftime('%Y-%m-%d (%a) %H:%M:%S') else: raise ValueError("Invalid format_type. Choose from 'obj', 'short', 'pretty'.") def local(self, format_type="obj"): return self._format_time(self.local_time, format_type) def jst(self, format_type="obj"): return self._format_time(self.jst_time, format_type) def naive(self, format_type="obj"): return self._format_time(self.naive_time, format_type) # deprecated, will be removed in the future def get_current_time(now=None): # pip install pytz import pytz tz = pytz.timezone('Asia/Tokyo') if now is None: now = datetime.now() # convert now to local timezone and then to JST now = now.astimezone() jst_now = now.astimezone(tz) return { 'local': { 'obj': now, 'str_pretty': now.strftime('%Y-%m-%d (%a) %H:%M:%S'), 'str_short': now.strftime('%y%m%d_%H%M%S'), }, 'jst': { 'obj': jst_now, 'str_pretty': jst_now.strftime('%Y-%m-%d (%a) %H:%M:%S'), 'str_short': jst_now.strftime('%y%m%d_%H%M%S'), } } # https://stackoverflow.com/a/13756038/3939155 def td_format(td_object_or_sec, long_form=True): if isinstance(td_object_or_sec, int) or isinstance(td_object_or_sec, float): seconds = int(td_object_or_sec) else: seconds = int(td_object_or_sec.total_seconds()) periods = [ ('year', 60*60*24*365), ('month', 60*60*24*30), ('day', 60*60*24), ('hour', 60*60), ('minute', 60), ('second', 1) ] strings=[] for period_name, period_seconds in periods: if seconds > period_seconds: period_value , seconds = divmod(seconds, period_seconds) if long_form: has_s = 's' if period_value > 1 else '' strings.append(f"{period_value} {period_name}{has_s}") else: strings.append(f"{period_value}{period_name[0:1]}") if long_form: return ", ".join(strings) else: return "".join(strings) # deprecated, just use MyTime(dt).jst() def to_jp_time(dt, input_timezone=None): # pip install python-dateutil pytz from dateutil import parser from pytz import timezone if isinstance(dt, str): dt = parser.parse(dt) if input_timezone: dt = timezone(input_timezone).localize(dt) return dt.astimezone(timezone('asia/tokyo')) # ==================== performance related ==================== def tic(): global _start_time _start_time = time.time() def tac(print=True): global _start_time t = time.time() - _start_time if print: builtins.print(f'Time passed: {t:.2f} s') return t # a decorator to time a function def timeme(func): def wrapper(*args, **kwargs): start_time = time.time() result = func(*args, **kwargs) end_time = time.time() print(f"{func.__name__} executed in {end_time - start_time:.02f} seconds") return result return wrapper # ==================== file related ==================== def dump_json(mydict, filename, **kwargs): filename = Path(filename) if filename.suffix.lower() !='.json': filename = filename.with_suffix('.json') filename.parent.mkdir(parents=True, exist_ok=True) default_kwargs = dict(ensure_ascii=False, indent=2) default_kwargs.update(kwargs) with filename.open('w', encoding='utf-8') as f: json.dump(mydict, f, **default_kwargs) def load_json(filename, encoding='utf-8'): filename = Path(filename) with filename.open('r', encoding=encoding) as f: data = json.load(f) return data def dump_html(soup, filename='temp.html', encoding='utf-8', prettify=True): """ Dump the contents of a BeautifulSoup object to an HTML file. Args: soup (BeautifulSoup): The BeautifulSoup object containing the HTML content. filename (str, optional): The name of the output file. Defaults to 'temp.html'. encoding (str, optional): The encoding to use when writing the file. Defaults to 'utf-8'. prettify (bool, optional): Whether to prettify the HTML content. Defaults to True. """ filename = Path(filename) filename.parent.mkdir(parents=True, exist_ok=True) if prettify: text = soup.prettify() else: text = str(soup) with filename.open('w', encoding=encoding) as f: f.write(text) def dump_tsv(data, filename='temp.tsv', verbose=True, print_header=False): """ Dump data into a TSV (Tab-Separated Values) file. Args: data (list): The data to be dumped. It can be a list of lists or a list of dictionaries. filename (str, optional): The name of the output file. Defaults to 'temp.tsv'. verbose (bool, optional): Whether to print the content of the TSV file. Defaults to True. print_header (bool, optional): Whether to print the header in the TSV file. Defaults to False. """ s = '' if isinstance(data[0], dict): headers = data[0].keys() if print_header: s += '\t'.join(str(h) for h in headers) + '\n' for row in data: s += '\t'.join([str(row.get(h, '')) for h in headers]) + '\n' else: if print_header: print('[W] No header provided. Use default header [0, 1, 2, ...]') s += '\t'.join([str(i) for i in range(len(data[0]))]) + '\n' for row in data: s += '\t'.join([str(i) for i in row]) + '\n' if verbose: print(s) filename = Path(filename) filename.parent.mkdir(parents=True, exist_ok=True) filename.write_text(s, encoding='utf8') def get_files(directory, recursive=False, file_filter=None, path_filter=None): '''filter(s): true means include, false means exclude''' directory = Path(directory) assert(directory.is_dir()) # if there is no filter, use os.scandir generator, since it is so much faster. if file_filter is None and path_filter is None: def quick_scan(directory): for entry in os.scandir(directory): if entry.is_file(): yield entry elif recursive and entry.is_dir(follow_symlinks=False): yield from quick_scan(entry.path) return [Path(f) for f in quick_scan(directory)] # else, use pathlib.iterdir and just dynamically change the list. The speed is basically the same (slow). file_list = [] for x in directory.iterdir(): if x.is_file(): if not file_filter or file_filter(x): file_list.append(x) elif recursive and x.is_dir() and (not path_filter or path_filter(x)): file_list.extend(get_files(x, recursive=recursive, file_filter=file_filter, path_filter=path_filter)) return file_list def remove_empty_folders(directory, remove_root=True): #Including root. directory = Path(directory) try: assert directory.is_dir() for x in directory.iterdir(): if x.is_dir(): remove_empty_folders(x, remove_root=True) if remove_root and not list(directory.iterdir()): directory.rmdir() except PermissionError as e: print('Error:', e) def ensure_nonexist(f): ''' Ensure the file does not exist. If it does, rename it to filename_2, filename_3, etc. ''' i = 2 stem = f.stem if m := re.search(r'^(.+?)_(\d)$', stem): # only do so if for i < 10 and has no padding. # so things like file_01, file_1986 etc. won't be renamed to confusing file_2, file_1987 etc. if int(m[2]) < 10 and len(m[2]) == 1: stem = m[1] i = int(m[2]) + 1 while f.exists(): f = f.with_name(f'{stem}_{i}{f.suffix}') i = i + 1 return f def ensure_path_exists(path): ''' ensure the path exists. ''' path = Path(path) if not path.exists(): raise FileNotFoundError(f'Path {path} does not exist!') return path def quickmd5(f): hasher = hashlib.md5() file_size = f.stat().st_size buffer_size = 1024 * 1024 # 1MB with f.open('rb') as f: if file_size <= buffer_size: # If the file is less than or equal to 1MB, read the entire file data = f.read() hasher.update(data) else: # Read the first 1MB data = f.read(buffer_size) hasher.update(data) # Go to the end of the file to read the last 1MB f.seek(-buffer_size, os.SEEK_END) data = f.read(buffer_size) hasher.update(data) return f"{file_size}_{hasher.hexdigest()}" # this way is more readable than just return the hexdigest. def move_or_delete_duplicate(src, dst, verbose=True, conflict='error', hash_method=quickmd5): """ Move a file, or delete it if the same file already exists at the destination. Args: src (pathlib.Path): The path to the source file. dst (pathlib.Path): The path to the destination file. verbose (bool, optional): Whether to print verbose output. Defaults to True. conflict (str, optional): The conflict resolution strategy. Can be one of 'error', 'skip', or 'rename'. Defaults to 'error'. hash_method (function, optional): The hash method to use for comparing files. Defaults to quickmd5. Raises: FileNotFoundError: If the source file does not exist. ValueError: If the source and destination paths are the same. FileExistsError: If the destination file already exists and the conflict resolution strategy is 'error'. Returns: None """ if not src.exists(): raise FileNotFoundError(f"The source file {src} does not exist.") if src == dst: raise ValueError(f"Source and destination are the same: {src}") if dst.exists(): if hash_method(src) == hash_method(dst): print(f'[W] {src.name} is a duplicate. Remove.') src.unlink() return else: if conflict == 'skip': print(f'[W] Destination file {dst} already exists and hash does not match. Skip.') return elif conflict == 'error': raise FileExistsError(f"Destination file {dst} already exists and hash does not match.") elif conflict == 'rename': dst_ = ensure_nonexist(dst) print(f'[W] Destination file {dst} already exists and hash does not match. Rename to {dst_.name} instead.') dst = dst_ if verbose: if src.parent == dst.parent: print(f"Rename {src.name} to {dst.name}.") elif src.name == dst.name: print(f'Move {src.name} into {dst.parent}') else: print(f'Move {src.name} to {dst}') shutil.move(src, dst) def batch_rename(renamings): """ Batch rename files without conflicts. Args: renamings (list): A list of tuples containing the original file paths and the new names. Returns: None Raises: None This function renames multiple files simultaneously without causing conflicts. It checks for duplicate files in the list, duplicate new filenames, and conflicts with existing files. If any conflicts are detected, the function prints an error message and aborts the renaming process. Example usage: renamings = [(Path('file1.txt'), 'new_file1.txt'), (Path('file2.txt'), 'new_file2.txt')] batch_rename(renamings) """ files = [f for f, _ in renamings] dst_files = [f.with_name(name) for f, name in renamings] if len(set(files)) != len(files): print('[E] There are duplicate files in the list! Please check.') return if len(set(dst_files)) != len(dst_files): print('[E] There are duplicate new filenames in the list! Please check.') return # if new filename is conflicting and not in our current files, abort for new_f in dst_files: if new_f.exists() and new_f not in files: print(f'[E] file {new_f.name} already exists. Please rename it first.') return # rename current file(s) to temp filename to make renaming possible real_renamings = [] for f, name in renamings: if f in dst_files: temp_f = ensure_nonexist(f.with_name(f.stem + '_temp' + f.suffix)) print(f'[I] temporarily rename {f.name} to {temp_f.name}') f = f.rename(temp_f) real_renamings.append((f, name)) for f, name in real_renamings: f.rename(f.with_name(name)) # ==================== network related ==================== def requests_retry_session( retries=5, backoff_factor=0.2, status_forcelist=(502, 503, 504), session=None, ): """ Create a session object with retry functionality for making HTTP requests. Modified from https://www.peterbe.com/plog/best-practice-with-retries-with-requests Args: retries (int): The maximum number of retries for each request. Default is 5. backoff_factor (float): The backoff factor between retries. Default is 0.2. status_forcelist (tuple): A tuple of HTTP status codes that should trigger a retry. Default is (502, 503, 504). session (requests.Session): An existing session object to use. If not provided, a new session will be created. Returns: requests.Session: The session object with retry functionality. """ # pip install requests urllib3 import requests from requests.adapters import HTTPAdapter from urllib3.util.retry import Retry session = session or requests.Session() retry = Retry( total=retries, backoff_factor=backoff_factor, status_forcelist=status_forcelist, ) adapter = HTTPAdapter(max_retries=retry) session.mount('http://', adapter) session.mount('https://', adapter) return session def get(url, headers=None, cookies=None, encoding=None, session=None, parser='lxml', timeout=None): """ Sends a GET request to the specified URL and returns the parsed HTML content. Args: url (str): The URL to send the GET request to. headers (dict, optional): The headers to include in the request. Defaults to None. cookies (dict, optional): The cookies to include in the request. Defaults to None. encoding (str, optional): The encoding to use when parsing the HTML content. Defaults to None. session (requests.Session, optional): The session to use for the request. Defaults to None. parser (str, optional): The parser to use for parsing the HTML content. Defaults to 'lxml'. timeout (float, optional): The maximum number of seconds to wait for the request to complete. Defaults to None. Returns: BeautifulSoup: The parsed HTML content. Raises: Any exceptions raised by the underlying requests library. Dependencies: - requests - lxml - beautifulsoup4 """ from bs4 import BeautifulSoup if not session: session = requests_retry_session() r = session.get(url, cookies=cookies, headers=headers, timeout=timeout) if encoding: return BeautifulSoup(r.content, parser, from_encoding=encoding) else: return BeautifulSoup(r.content, parser) def get_webname(url): return unquote(url.split('?')[0].split('/')[-1]) def load_cookie(s): """ Load cookies from various sources and convert them to a `RequestsCookieJar` object. Args: s (str): The input string, file path containing the cookies, or "{browser_name}/{domain_name}" to load cookies from a browser. Returns: requests.cookies.RequestsCookieJar: The converted `RequestsCookieJar` object. Raises: ValueError: If the input string is invalid. Examples: >>> load_cookie('cookie1=value1; cookie2=value2') <RequestsCookieJar[Cookie(version=0, name='cookie1', value='value1', port=None, port_specified=False, domain='', domain_specified=False, domain_initial_dot=False, path='/', path_specified=False, secure=False, expires=None, discard=True, comment=None, comment_url=None, rest={'HttpOnly': None}, rfc2109=False), Cookie(version=0, name='cookie2', value='value2', port=None, port_specified=False, domain='', domain_specified=False, domain_initial_dot=False, path='/', path_specified=False, secure=False, expires=None, discard=True, comment=None, comment_url=None, rest={'HttpOnly': None}, rfc2109=False)]> >>> load_cookie('/path/to/cookies.txt') <RequestsCookieJar[Cookie(version=0, name='cookie1', value='value1', port=None, port_specified=False, domain='example.com', domain_specified=False, domain_initial_dot=False, path='/', path_specified=False, secure=False, expires=None, discard=True, comment=None, comment_url=None, rest={'HttpOnly': None}, rfc2109=False), Cookie(version=0, name='cookie2', value='value2', port=None, port_specified=False, domain='example.com', domain_specified=False, domain_initial_dot=False, path='/', path_specified=False, secure=False, expires=None, discard=True, comment=None, comment_url=None, rest={'HttpOnly': None}, rfc2109=False)]> """ from http.cookiejar import MozillaCookieJar from requests.cookies import RequestsCookieJar, create_cookie import browsercookie # pip install browsercookie def convert(cj): cookies = RequestsCookieJar() for cookie in cj: requests_cookie = create_cookie( name=cookie.name, value=cookie.value, domain=cookie.domain, path=cookie.path, secure=cookie.secure, rest={'HttpOnly': cookie.get_nonstandard_attr('HttpOnly')}, expires=cookie.expires, ) cookies.set_cookie(requests_cookie) return cookies if m := re.search(r'^(chrome|firefox|edge)(/.+)?', str(s), re.IGNORECASE): domain_name = m[2].lstrip('/') if m[2] else None if m[1].lower() == 'chrome': cj = browsercookie.chrome() elif m[1].lower() == 'firefox': cj = browsercookie.firefox() elif m[1].lower() == 'edge': cj = browsercookie.edge() if domain_name: cj = [cookie for cookie in cj if domain_name in cookie.domain] return convert(cj) if Path(s).exists(): cj = MozillaCookieJar(s) cj.load(ignore_expires=True, ignore_discard=True) for cookie in cj: if cookie.expires == 0: cookie.expires = int(time.time()+ 86400) return convert(cj) if re.search(r'^(.+?):\s*(.+?)', str(s)): cookies = RequestsCookieJar() for k, v in re.findall(r'(.+?):\s*([^;]+?)(?:;|$)', s): cookies.set(k, v) return cookies raise ValueError(f'Invalid cookie string: {s}') def download(url, filename=None, save_path='.', cookies=None, session=None, dry_run=False, dupe='skip_same_size', referer=None, headers=None, placeholder=True, prefix='', get_suffix=True, verbose=2, retry_failed=True): """ Downloads a file from the given URL and saves it to the specified location. Args: url (str): The URL of the file to download. filename (str, optional): The name of the file to save. If not provided, the filename will be extracted from the URL or the response header. Defaults to None. save_path (str, optional): The directory path to save the file. Defaults to '.' (current directory). cookies (dict, optional): A dictionary of cookies to include in the request. Defaults to None. session (requests.Session, optional): A requests Session object to use for the request. Defaults to None. dry_run (bool, optional): If True, only prints the URL and does not perform the actual download. Defaults to False. dupe (str, optional): The method to handle duplicate files. Must be one of 'skip', 'overwrite', 'rename', or 'skip_same_size'. Defaults to 'skip_same_size'. referer (str, optional): The referer header to include in the request. Defaults to None. headers (dict, optional): Additional headers to include in the request. Defaults to None. placeholder (bool, optional): If True, creates a placeholder file with a '.broken' extension if the download fails. Defaults to True. prefix (str, optional): A prefix to add to the filename. Useful when fetching the filename from the response headers. Defaults to ''. get_suffix (bool, optional): If True, attempts to determine the file extension from the response headers. Defaults to True. verbose (int, optional): The verbosity level of the download progress. Must be 0, 1, or 2. Defaults to 2. retry_failed (bool, optional): If True, retries the download if it fails. Defaults to True. Returns: str: The status of the download. Can be 'Dry run', 'Exists', or the HTTP status code. """ from mimetypes import guess_extension # it uses requests_retry_session, so # pip install requests if dupe not in ['skip', 'overwrite', 'rename', 'skip_same_size']: raise ValueError(f'[Error] Invalid dupe method: {dupe} (must be either skip, overwrite, rename or skip_same_size).') def print(s, verbose_level, only=False): if (only and verbose == verbose_level) or (not only and verbose >= verbose_level): builtins.print(s) def has_valid_suffix(f): # common suffixes if f.suffix.lower() in ['.jpg', '.png', '.gif', '.webp', '.jpeg', '.bmp', '.svg', '.ico', '.mp4', '.mkv', '.webm', '.heic', '.pdf']: return True if f.suffix.lower() in ['.php', '']: return False # if the suffix is too long, has a space, etc., we assume it is not a valid suffix if len(f.suffix.lower()) > 5 or ' ' in f.suffix.lower(): return False return True def replace_suffix(f, content_type): |
| You are visitor 01402503 — GeezerVison does not store the pages you look up. Best viewed with Netscape Navigator 4.7 or Internet Explorer 6 at 800×600. No cookies required. |