util.py

importbuiltins
importjson
importos
importre
importsys
importtime
importhashlib
importshutil
importunicodedata
fromdatetimeimportdatetime
frompathlibimportPath
fromurllib.parseimportunquote

# 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. ====================
defto_list(a):
returnaifisinstance(a, list) oraisNoneelse [a]

defflatten(x):
fromcollections.abcimportIterable
ifisinstance(x, Iterable) andnotisinstance(x, str):
return [aforiinxforainflatten(i)]
else:
return [x]

defprint_cmd(cmd, prefix='', actually_print=True):
commands_text_form= [f'"{c}"'ifre.search(r'[ ?]', str(c)) elsestr(c) forcincmd]
output=prefix+' '.join(commands_text_form)
ifactually_print:
print(output)
returnoutput

defcopy(data):
# pip install pyperclip
importpyperclip
pyperclip.copy(data)

defget_clipboard_data():
# pip install pyperclip
importpyperclip
returnpyperclip.paste()

defsafeify(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.
 """

assertisinstance(name, str), f'Name must be a string, not {type(name)}'

template= {'\\': '＼', '/': '／', ':': '：', '*': '＊', '?': '？', '"': '＂', '<': '＜', '>': '＞', '|': '｜','\n':'','\r':'','\t':''}
ifignore_backslash:
template.pop('\\', None)

forillegalintemplate:
name=name.replace(illegal, template[illegal])
returnname

defformat_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
importwcwidth

ifsisNone:
s=''
else:
s=str(s)
ifnotwidth:
returns
output=''
length=0
forcharins:
size=wcwidth.wcswidth(char)
iflength+size>width:
break
output+=char
length+=size
ifalign=='left':
returnoutput+padding*(width-length)
ifalign=='right':
returnpadding*(width-length) +output
ifalign=='center':
left_space= (width-length)//2
right_space=width-length-left_space
returnpadding*left_space+output+padding*right_space

classTable():
# pip install wcwidth

def__init__(self, rows=None, headers=None, max_width=100) ->None:
ifrowsandnotheaders:
self.headers=rows[0]
self.data=rows[1:]
elifheaders:
self.headers=headers
ifrows:
self.data=rows
else:
self.data= []
else:
raiseException('No header or data given!')
self.max_width=max_width

defrows(self):
forrowinself.data:
row_dict= {}
fori, headerinenumerate(self.headers):
row_dict[header] =row[i]
yieldrow_dict

defsearch(self, keywords, single=True):
matched_rows= []
forrowinself.data:
matched=True
forkey, valueinkeywords.items():
assertkeyinself.headers
col_idx=self.headers.index(key)
ifnotvalue: # Ignore empty values
continue
ifnot ((str(row[col_idx]) ==str(value)) or (row[col_idx] ==value)):
matched=False
break
ifmatched:
matched_rows.append(row)
ifsingle:
break
ifmatched_rows:
ifsingleandlen(matched_rows) ==1:
returnmatched_rows[0]
else:
returnmatched_rows
else:
return []

defsort(self, header_name_or_index, reverse=False):
ifisinstance(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=lambdax: x[col_idx], reverse=reverse)

defappend(self, new_data):
d= [''] *len(self.headers)
forkey, valueinnew_data.items():
ifnotkeyinself.headers: # silently ignore keys that are not in headers
continue
col_idx=self.headers.index(key)
d[col_idx] =value
self.data.append(d)

defprint(self, formats=None, custom_print=None):
importwcwidth

deffmt(value, str_format):
try:
returnstr_format.format(value)
exceptException:
returnvalue

defcalc_width(col_format, idx):
maxw=sum(wcwidth.wcwidth(c) forcinstr(self.headers[idx]))
forrowinself.data:
ifidx>len(row) -1:
continue
w=sum(wcwidth.wcwidth(c) forcinfmt(row[idx], col_format['str_format']))
ifw>maxw:
maxw=w
maxw=min(maxw, col_format['max_width'])
col_format['width'] =maxw

defprint_row(row, header_mode=False, custom_print=custom_print):
parts= []
foridxinrange(len(row)):
col_format=col_formats[idx]
# override str format back to nothing for headers
str_format="{}"ifheader_modeelsecol_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)
ifcustom_print:
custom_print(line)
else:
print(line)

col_formats=dict()
foridxinrange(len(self.headers)):
col_format= {
"align": "left",
"str_format": "{}",
"max_width": self.max_width
 }
ifformats:
ifidxinformats:
col_format.update(formats[idx])
elifself.headers[idx] informats:
col_format.update(formats[self.headers[idx]])
ifnot'width'incol_format:
calc_width(col_format, idx)
col_formats[idx] =col_format

print_row(self.headers, header_mode=True)
forrowinself.data:
print_row(row)

# return some useful info
total_width=sum(col_format['width'] forcol_formatincol_formats.values()) +2* (len(col_formats) -1)
return {'total_width': total_width, 'col_formats': col_formats}

defsave(self, f):
s=''
f=Path(f)
s+='\t'.join(self.headers) +'\n'
forrowinself.data:
s+='\t'.join([str(cell) forcellinrow]) +'\n'
f.write_text(s, encoding='utf8')

defmulti_col_print(data, columns=5):
'''Print a list of data in a table with columns'''
# pip install rich
fromrich.consoleimportConsole
fromrich.tableimportTable

table=Table(show_header=False, box=None)
for_inrange(columns):
table.add_column(style='cyan')
# add data to table vertically
row_count=len(data) //columns+1
foriinrange(row_count):
row=data[i::row_count]
table.add_row(*row)
console=Console()
console.print(table)

defrprint(*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
fromrich.consoleimportConsole
ifnothasattr(rprint, '_console'):
rprint._console=Console()
rprint._console.print(*args, **kwargs)

defprint2(*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.
 '''
frombuiltinsimportprintasbuiltin_print
ifnothasattr(print2, '_newlined'):
print2._newlined=True
end=kwargs.pop('end', '\n')
ifend.endswith('\n') andnotprint2._newlined:
builtin_print()
print2._newlined=Trueifend.endswith('\n') elseFalse
builtin_print(*args, **kwargs, end=end)

defarray_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
foridx, seginenumerate(a):
ifidx==0: # first segment
s+=f'{seg}'
range_start=seg
else:
ifseg-prev_seg==1:
ifidx==len(a) -1: # last segment
s+=f'{dash}{seg}'
else:
ifprev_seg>range_start:
s+=f'{dash}{prev_seg}'
s+=f'{sep}{seg}'
range_start=seg
prev_seg=seg
returns

defpluralize(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).
 '''
ifcount==1:
returnword
else:
ifalt:
returnalt
else:
returnword+suffix

defcompare_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
fromrichimportprintasrprint

defprint(*args, **kwargs):
ifnotmute:
rprint(*args, **kwargs)

equal=True
type_changed=False

iftype(value_old) !=type(value):
type_changed=True
equal=False
elifisinstance(value, dict):
forkey, vinvalue.items():
ifkeynotinvalue_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)
forkey, vinvalue_old.items():
ifkeynotinvalue:
equal=False
print(f'{print_prefix}.{key}: [red]Removed:[/red]', v)
returnequal
elifisinstance(value, list):
foriinrange(min(len(value), len(value_old))):
equal&=compare_obj(value_old[i], value[i], print_prefix=f'{print_prefix}[{i}]', mute=mute)
iflen(value_old) <len(value):
equal=False
foriinrange(len(value_old), len(value)):
print(f'{print_prefix}[{i}]: [green]Added:[/green]', value[i])
eliflen(value_old) >len(value):
equal=False
foriinrange(len(value), len(value_old)):
print(f'{print_prefix}[{i}]: [red]Removed:[/red]',value_old[i])
returnequal

elifisinstance(value, str):
equal=re.sub(r'\s+',' ', value_old) ==re.sub(r'\s+',' ', value)
else:
equal=value_old==value
ifnotequal:
print(f'{print_prefix}:', end='')
s=str(value_old) +str(value)
iflen(s) <60andnot'\n'ins:
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)
returnequal

defnormalize(s):
'''Normalize a string by replacing full-width characters with half-width characters
 but keep the full-width dash (～) as it is.'''
returnunicodedata.normalize('NFKC', s.replace('～', '$dash$')).replace('$dash$', '～')

# ==================== datetime related ====================
defparse_to_shortdate(date_str, fmt=None):
# pip install python-dateutil
fromdateutilimportparser

current_year=datetime.now().year

iffmtisNone: fmt='%y%m%d'
ifisinstance(date_str, datetime):
returndate_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})']
forpatterninpatterns:
# 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.
ifm:=re.search(pattern, date_str) orre.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'] orcurrent_year}-{m['m']}-{m['d']}"
break
try:
returnparser.parse(date_str, yearfirst=True).strftime(fmt)
except:
return""


classMyTime:
def__init__(self, t=None):
importpytz

self.tz=pytz.timezone('Asia/Tokyo')

iftisNone:
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):
ifformat_type=='obj':
returndt
elifformat_type=="short":
returndt.strftime('%y%m%d_%H%M%S')
elifformat_type=="pretty":
returndt.strftime('%Y-%m-%d (%a) %H:%M:%S')
else:
raiseValueError("Invalid format_type. Choose from 'obj', 'short', 'pretty'.")

deflocal(self, format_type="obj"):
returnself._format_time(self.local_time, format_type)

defjst(self, format_type="obj"):
returnself._format_time(self.jst_time, format_type)

defnaive(self, format_type="obj"):
returnself._format_time(self.naive_time, format_type)

# deprecated, will be removed in the future
defget_current_time(now=None):
# pip install pytz
importpytz

tz=pytz.timezone('Asia/Tokyo')

ifnowisNone:
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
deftd_format(td_object_or_sec, long_form=True):
ifisinstance(td_object_or_sec, int) orisinstance(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=[]
forperiod_name, period_secondsinperiods:
ifseconds>period_seconds:
period_value , seconds=divmod(seconds, period_seconds)
iflong_form:
has_s='s'ifperiod_value>1else''
strings.append(f"{period_value}{period_name}{has_s}")
else:
strings.append(f"{period_value}{period_name[0:1]}")
iflong_form:
return", ".join(strings)
else:
return"".join(strings)

# deprecated, just use MyTime(dt).jst()
defto_jp_time(dt, input_timezone=None):
# pip install python-dateutil pytz
fromdateutilimportparser
frompytzimporttimezone

ifisinstance(dt, str):
dt=parser.parse(dt)
ifinput_timezone:
dt=timezone(input_timezone).localize(dt)
returndt.astimezone(timezone('asia/tokyo'))

# ==================== performance related ====================
deftic():
global_start_time
_start_time=time.time()

deftac(print=True):
global_start_time
t=time.time() -_start_time
ifprint:
builtins.print(f'Time passed: {t:.2f} s')
returnt

# a decorator to time a function
deftimeme(func):
defwrapper(*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")
returnresult
returnwrapper

# ==================== file related ====================
defdump_json(mydict, filename, **kwargs):
filename=Path(filename)
iffilename.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)
withfilename.open('w', encoding='utf-8') asf:
json.dump(mydict, f, **default_kwargs)

defload_json(filename, encoding='utf-8'):
filename=Path(filename)
withfilename.open('r', encoding=encoding) asf:
data=json.load(f)
returndata

defdump_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)
ifprettify:
text=soup.prettify()
else:
text=str(soup)
withfilename.open('w', encoding=encoding) asf:
f.write(text)

defdump_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=''
ifisinstance(data[0], dict):
headers=data[0].keys()
ifprint_header:
s+='\t'.join(str(h) forhinheaders) +'\n'
forrowindata:
s+='\t'.join([str(row.get(h, '')) forhinheaders]) +'\n'
else:
ifprint_header:
print('[W] No header provided. Use default header [0, 1, 2, ...]')
s+='\t'.join([str(i) foriinrange(len(data[0]))]) +'\n'
forrowindata:
s+='\t'.join([str(i) foriinrow]) +'\n'
ifverbose:
print(s)
filename=Path(filename)
filename.parent.mkdir(parents=True, exist_ok=True)
filename.write_text(s, encoding='utf8')


defget_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.
iffile_filterisNoneandpath_filterisNone:
defquick_scan(directory):
forentryinos.scandir(directory):
ifentry.is_file():
yieldentry
elifrecursiveandentry.is_dir(follow_symlinks=False):
yieldfromquick_scan(entry.path)
return [Path(f) forfinquick_scan(directory)]
# else, use pathlib.iterdir and just dynamically change the list. The speed is basically the same (slow).
file_list= []
forxindirectory.iterdir():
ifx.is_file():
ifnotfile_filterorfile_filter(x):
file_list.append(x)
elifrecursiveandx.is_dir() and (notpath_filterorpath_filter(x)):
file_list.extend(get_files(x, recursive=recursive, file_filter=file_filter, path_filter=path_filter))
returnfile_list

defremove_empty_folders(directory, remove_root=True): #Including root.
directory=Path(directory)
try:
assertdirectory.is_dir()
forxindirectory.iterdir():
ifx.is_dir():
remove_empty_folders(x, remove_root=True)
ifremove_rootandnotlist(directory.iterdir()):
directory.rmdir()
exceptPermissionErrorase:
print('Error:', e)

defensure_nonexist(f):
'''
 Ensure the file does not exist. If it does, rename it to filename_2, filename_3, etc.
 '''
i=2
stem=f.stem
ifm:=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.
ifint(m[2]) <10andlen(m[2]) ==1:
stem=m[1]
i=int(m[2]) +1
whilef.exists():
f=f.with_name(f'{stem}_{i}{f.suffix}')
i=i+1
returnf

defensure_path_exists(path):
'''
 ensure the path exists.
 '''
path=Path(path)
ifnotpath.exists():
raiseFileNotFoundError(f'Path {path} does not exist!')
returnpath

defquickmd5(f):
hasher=hashlib.md5()
file_size=f.stat().st_size
buffer_size=1024*1024# 1MB

withf.open('rb') asf:
iffile_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)

returnf"{file_size}_{hasher.hexdigest()}"# this way is more readable than just return the hexdigest.

defmove_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
 """
ifnotsrc.exists():
raiseFileNotFoundError(f"The source file {src} does not exist.")
ifsrc==dst:
raiseValueError(f"Source and destination are the same: {src}")
ifdst.exists():
ifhash_method(src) ==hash_method(dst):
print(f'[W] {src.name} is a duplicate. Remove.')
src.unlink()
return
else:
ifconflict=='skip':
print(f'[W] Destination file {dst} already exists and hash does not match. Skip.')
return
elifconflict=='error':
raiseFileExistsError(f"Destination file {dst} already exists and hash does not match.")
elifconflict=='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_
ifverbose:
ifsrc.parent==dst.parent:
print(f"Rename {src.name} to {dst.name}.")
elifsrc.name==dst.name:
print(f'Move {src.name} into {dst.parent}')
else:
print(f'Move {src.name} to {dst}')
shutil.move(src, dst)

defbatch_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= [fforf, _inrenamings]
dst_files= [f.with_name(name) forf, nameinrenamings]

iflen(set(files)) !=len(files):
print('[E] There are duplicate files in the list! Please check.')
return

iflen(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
fornew_findst_files:
ifnew_f.exists() andnew_fnotinfiles:
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= []
forf, nameinrenamings:
iffindst_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))

forf, nameinreal_renamings:
f.rename(f.with_name(name))

# ==================== network related ====================
defrequests_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
importrequests
fromrequests.adaptersimportHTTPAdapter
fromurllib3.util.retryimportRetry

session=sessionorrequests.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)
returnsession

defget(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
 """
frombs4importBeautifulSoup

ifnotsession:
session=requests_retry_session()
r=session.get(url, cookies=cookies, headers=headers, timeout=timeout)
ifencoding:
returnBeautifulSoup(r.content, parser, from_encoding=encoding)
else:
returnBeautifulSoup(r.content, parser)

defget_webname(url):
returnunquote(url.split('?')[0].split('/')[-1])

defload_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)]>
 """
fromhttp.cookiejarimportMozillaCookieJar
fromrequests.cookiesimportRequestsCookieJar, create_cookie
importbrowsercookie
# pip install browsercookie

defconvert(cj):
cookies=RequestsCookieJar()
forcookieincj:
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)
returncookies

ifm:=re.search(r'^(chrome|firefox|edge)(/.+)?', str(s), re.IGNORECASE):
domain_name=m[2].lstrip('/') ifm[2] elseNone
ifm[1].lower() =='chrome':
cj=browsercookie.chrome()
elifm[1].lower() =='firefox':
cj=browsercookie.firefox()
elifm[1].lower() =='edge':
cj=browsercookie.edge()
ifdomain_name:
cj= [cookieforcookieincjifdomain_nameincookie.domain]
returnconvert(cj)

ifPath(s).exists():
cj=MozillaCookieJar(s)
cj.load(ignore_expires=True, ignore_discard=True)
forcookieincj:
ifcookie.expires==0:
cookie.expires=int(time.time()+86400)
returnconvert(cj)

ifre.search(r'^(.+?):\s*(.+?)', str(s)):
cookies=RequestsCookieJar()
fork, vinre.findall(r'(.+?):\s*([^;]+?)(?:;|$)', s):
cookies.set(k, v)
returncookies

raiseValueError(f'Invalid cookie string: {s}')

defdownload(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.
 """
frommimetypesimportguess_extension
# it uses requests_retry_session, so
# pip install requests

ifdupenotin ['skip', 'overwrite', 'rename', 'skip_same_size']:
raiseValueError(f'[Error] Invalid dupe method: {dupe} (must be either skip, overwrite, rename or skip_same_size).')

defprint(s, verbose_level, only=False):
if (onlyandverbose==verbose_level) or (notonlyandverbose>=verbose_level):
builtins.print(s)

defhas_valid_suffix(f):
# common suffixes
iff.suffix.lower() in ['.jpg', '.png', '.gif', '.webp', '.jpeg', '.bmp', '.svg', '.ico', '.mp4', '.mkv', '.webm', '.heic', '.pdf']:
returnTrue
iff.suffix.lower() in ['.php', '']:
returnFalse
# if the suffix is too long, has a space, etc., we assume it is not a valid suffix
iflen(f.suffix.lower()) >5or' 'inf.suffix.lower():
returnFalse
returnTrue

defreplace_suffix(f, content_type):

importbuiltins
importjson
importos
importre
importsys
importtime
importhashlib
importshutil
importunicodedata
fromdatetimeimportdatetime
frompathlibimportPath
fromurllib.parseimportunquote

# 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. ====================
defto_list(a):
returnaifisinstance(a, list) oraisNoneelse [a]

defflatten(x):
fromcollections.abcimportIterable
ifisinstance(x, Iterable) andnotisinstance(x, str):
return [aforiinxforainflatten(i)]
else:
return [x]

defprint_cmd(cmd, prefix='', actually_print=True):
commands_text_form= [f'"{c}"'ifre.search(r'[ ?]', str(c)) elsestr(c) forcincmd]
output=prefix+' '.join(commands_text_form)
ifactually_print:
print(output)
returnoutput

defcopy(data):
# pip install pyperclip
importpyperclip
pyperclip.copy(data)

defget_clipboard_data():
# pip install pyperclip
importpyperclip
returnpyperclip.paste()

defsafeify(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.
 """

assertisinstance(name, str), f'Name must be a string, not {type(name)}'

template= {'\\': '＼', '/': '／', ':': '：', '*': '＊', '?': '？', '"': '＂', '<': '＜', '>': '＞', '|': '｜','\n':'','\r':'','\t':''}
ifignore_backslash:
template.pop('\\', None)

forillegalintemplate:
name=name.replace(illegal, template[illegal])
returnname

defformat_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
importwcwidth

ifsisNone:
s=''
else:
s=str(s)
ifnotwidth:
returns
output=''
length=0
forcharins:
size=wcwidth.wcswidth(char)
iflength+size>width:
break
output+=char
length+=size
ifalign=='left':
returnoutput+padding*(width-length)
ifalign=='right':
returnpadding*(width-length) +output
ifalign=='center':
left_space= (width-length)//2
right_space=width-length-left_space
returnpadding*left_space+output+padding*right_space

classTable():
# pip install wcwidth

def__init__(self, rows=None, headers=None, max_width=100) ->None:
ifrowsandnotheaders:
self.headers=rows[0]
self.data=rows[1:]
elifheaders:
self.headers=headers
ifrows:
self.data=rows
else:
self.data= []
else:
raiseException('No header or data given!')
self.max_width=max_width

defrows(self):
forrowinself.data:
row_dict= {}
fori, headerinenumerate(self.headers):
row_dict[header] =row[i]
yieldrow_dict

defsearch(self, keywords, single=True):
matched_rows= []
forrowinself.data:
matched=True
forkey, valueinkeywords.items():
assertkeyinself.headers
col_idx=self.headers.index(key)
ifnotvalue: # Ignore empty values
continue
ifnot ((str(row[col_idx]) ==str(value)) or (row[col_idx] ==value)):
matched=False
break
ifmatched:
matched_rows.append(row)
ifsingle:
break
ifmatched_rows:
ifsingleandlen(matched_rows) ==1:
returnmatched_rows[0]
else:
returnmatched_rows
else:
return []

defsort(self, header_name_or_index, reverse=False):
ifisinstance(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=lambdax: x[col_idx], reverse=reverse)

defappend(self, new_data):
d= [''] *len(self.headers)
forkey, valueinnew_data.items():
ifnotkeyinself.headers: # silently ignore keys that are not in headers
continue
col_idx=self.headers.index(key)
d[col_idx] =value
self.data.append(d)

defprint(self, formats=None, custom_print=None):
importwcwidth

deffmt(value, str_format):
try:
returnstr_format.format(value)
exceptException:
returnvalue

defcalc_width(col_format, idx):
maxw=sum(wcwidth.wcwidth(c) forcinstr(self.headers[idx]))
forrowinself.data:
ifidx>len(row) -1:
continue
w=sum(wcwidth.wcwidth(c) forcinfmt(row[idx], col_format['str_format']))
ifw>maxw:
maxw=w
maxw=min(maxw, col_format['max_width'])
col_format['width'] =maxw

defprint_row(row, header_mode=False, custom_print=custom_print):
parts= []
foridxinrange(len(row)):
col_format=col_formats[idx]
# override str format back to nothing for headers
str_format="{}"ifheader_modeelsecol_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)
ifcustom_print:
custom_print(line)
else:
print(line)

col_formats=dict()
foridxinrange(len(self.headers)):
col_format= {
"align": "left",
"str_format": "{}",
"max_width": self.max_width
 }
ifformats:
ifidxinformats:
col_format.update(formats[idx])
elifself.headers[idx] informats:
col_format.update(formats[self.headers[idx]])
ifnot'width'incol_format:
calc_width(col_format, idx)
col_formats[idx] =col_format

print_row(self.headers, header_mode=True)
forrowinself.data:
print_row(row)

# return some useful info
total_width=sum(col_format['width'] forcol_formatincol_formats.values()) +2* (len(col_formats) -1)
return {'total_width': total_width, 'col_formats': col_formats}

defsave(self, f):
s=''
f=Path(f)
s+='\t'.join(self.headers) +'\n'
forrowinself.data:
s+='\t'.join([str(cell) forcellinrow]) +'\n'
f.write_text(s, encoding='utf8')

defmulti_col_print(data, columns=5):
'''Print a list of data in a table with columns'''
# pip install rich
fromrich.consoleimportConsole
fromrich.tableimportTable

table=Table(show_header=False, box=None)
for_inrange(columns):
table.add_column(style='cyan')
# add data to table vertically
row_count=len(data) //columns+1
foriinrange(row_count):
row=data[i::row_count]
table.add_row(*row)
console=Console()
console.print(table)

defrprint(*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
fromrich.consoleimportConsole
ifnothasattr(rprint, '_console'):
rprint._console=Console()
rprint._console.print(*args, **kwargs)

defprint2(*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.
 '''
frombuiltinsimportprintasbuiltin_print
ifnothasattr(print2, '_newlined'):
print2._newlined=True
end=kwargs.pop('end', '\n')
ifend.endswith('\n') andnotprint2._newlined:
builtin_print()
print2._newlined=Trueifend.endswith('\n') elseFalse
builtin_print(*args, **kwargs, end=end)

defarray_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
foridx, seginenumerate(a):
ifidx==0: # first segment
s+=f'{seg}'
range_start=seg
else:
ifseg-prev_seg==1:
ifidx==len(a) -1: # last segment
s+=f'{dash}{seg}'
else:
ifprev_seg>range_start:
s+=f'{dash}{prev_seg}'
s+=f'{sep}{seg}'
range_start=seg
prev_seg=seg
returns

defpluralize(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).
 '''
ifcount==1:
returnword
else:
ifalt:
returnalt
else:
returnword+suffix

defcompare_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
fromrichimportprintasrprint

defprint(*args, **kwargs):
ifnotmute:
rprint(*args, **kwargs)

equal=True
type_changed=False

iftype(value_old) !=type(value):
type_changed=True
equal=False
elifisinstance(value, dict):
forkey, vinvalue.items():
ifkeynotinvalue_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)
forkey, vinvalue_old.items():
ifkeynotinvalue:
equal=False
print(f'{print_prefix}.{key}: [red]Removed:[/red]', v)
returnequal
elifisinstance(value, list):
foriinrange(min(len(value), len(value_old))):
equal&=compare_obj(value_old[i], value[i], print_prefix=f'{print_prefix}[{i}]', mute=mute)
iflen(value_old) <len(value):
equal=False
foriinrange(len(value_old), len(value)):
print(f'{print_prefix}[{i}]: [green]Added:[/green]', value[i])
eliflen(value_old) >len(value):
equal=False
foriinrange(len(value), len(value_old)):
print(f'{print_prefix}[{i}]: [red]Removed:[/red]',value_old[i])
returnequal

elifisinstance(value, str):
equal=re.sub(r'\s+',' ', value_old) ==re.sub(r'\s+',' ', value)
else:
equal=value_old==value
ifnotequal:
print(f'{print_prefix}:', end='')
s=str(value_old) +str(value)
iflen(s) <60andnot'\n'ins:
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)
returnequal

defnormalize(s):
'''Normalize a string by replacing full-width characters with half-width characters
 but keep the full-width dash (～) as it is.'''
returnunicodedata.normalize('NFKC', s.replace('～', '$dash$')).replace('$dash$', '～')

# ==================== datetime related ====================
defparse_to_shortdate(date_str, fmt=None):
# pip install python-dateutil
fromdateutilimportparser

current_year=datetime.now().year

iffmtisNone: fmt='%y%m%d'
ifisinstance(date_str, datetime):
returndate_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})']
forpatterninpatterns:
# 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.
ifm:=re.search(pattern, date_str) orre.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'] orcurrent_year}-{m['m']}-{m['d']}"
break
try:
returnparser.parse(date_str, yearfirst=True).strftime(fmt)
except:
return""


classMyTime:
def__init__(self, t=None):
importpytz

self.tz=pytz.timezone('Asia/Tokyo')

iftisNone:
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):
ifformat_type=='obj':
returndt
elifformat_type=="short":
returndt.strftime('%y%m%d_%H%M%S')
elifformat_type=="pretty":
returndt.strftime('%Y-%m-%d (%a) %H:%M:%S')
else:
raiseValueError("Invalid format_type. Choose from 'obj', 'short', 'pretty'.")

deflocal(self, format_type="obj"):
returnself._format_time(self.local_time, format_type)

defjst(self, format_type="obj"):
returnself._format_time(self.jst_time, format_type)

defnaive(self, format_type="obj"):
returnself._format_time(self.naive_time, format_type)

# deprecated, will be removed in the future
defget_current_time(now=None):
# pip install pytz
importpytz

tz=pytz.timezone('Asia/Tokyo')

ifnowisNone:
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
deftd_format(td_object_or_sec, long_form=True):
ifisinstance(td_object_or_sec, int) orisinstance(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=[]
forperiod_name, period_secondsinperiods:
ifseconds>period_seconds:
period_value , seconds=divmod(seconds, period_seconds)
iflong_form:
has_s='s'ifperiod_value>1else''
strings.append(f"{period_value}{period_name}{has_s}")
else:
strings.append(f"{period_value}{period_name[0:1]}")
iflong_form:
return", ".join(strings)
else:
return"".join(strings)

# deprecated, just use MyTime(dt).jst()
defto_jp_time(dt, input_timezone=None):
# pip install python-dateutil pytz
fromdateutilimportparser
frompytzimporttimezone

ifisinstance(dt, str):
dt=parser.parse(dt)
ifinput_timezone:
dt=timezone(input_timezone).localize(dt)
returndt.astimezone(timezone('asia/tokyo'))

# ==================== performance related ====================
deftic():
global_start_time
_start_time=time.time()

deftac(print=True):
global_start_time
t=time.time() -_start_time
ifprint:
builtins.print(f'Time passed: {t:.2f} s')
returnt

# a decorator to time a function
deftimeme(func):
defwrapper(*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")
returnresult
returnwrapper

# ==================== file related ====================
defdump_json(mydict, filename, **kwargs):
filename=Path(filename)
iffilename.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)
withfilename.open('w', encoding='utf-8') asf:
json.dump(mydict, f, **default_kwargs)

defload_json(filename, encoding='utf-8'):
filename=Path(filename)
withfilename.open('r', encoding=encoding) asf:
data=json.load(f)
returndata

defdump_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)
ifprettify:
text=soup.prettify()
else:
text=str(soup)
withfilename.open('w', encoding=encoding) asf:
f.write(text)

defdump_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=''
ifisinstance(data[0], dict):
headers=data[0].keys()
ifprint_header:
s+='\t'.join(str(h) forhinheaders) +'\n'
forrowindata:
s+='\t'.join([str(row.get(h, '')) forhinheaders]) +'\n'
else:
ifprint_header:
print('[W] No header provided. Use default header [0, 1, 2, ...]')
s+='\t'.join([str(i) foriinrange(len(data[0]))]) +'\n'
forrowindata:
s+='\t'.join([str(i) foriinrow]) +'\n'
ifverbose:
print(s)
filename=Path(filename)
filename.parent.mkdir(parents=True, exist_ok=True)
filename.write_text(s, encoding='utf8')


defget_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.
iffile_filterisNoneandpath_filterisNone:
defquick_scan(directory):
forentryinos.scandir(directory):
ifentry.is_file():
yieldentry
elifrecursiveandentry.is_dir(follow_symlinks=False):
yieldfromquick_scan(entry.path)
return [Path(f) forfinquick_scan(directory)]
# else, use pathlib.iterdir and just dynamically change the list. The speed is basically the same (slow).
file_list= []
forxindirectory.iterdir():
ifx.is_file():
ifnotfile_filterorfile_filter(x):
file_list.append(x)
elifrecursiveandx.is_dir() and (notpath_filterorpath_filter(x)):
file_list.extend(get_files(x, recursive=recursive, file_filter=file_filter, path_filter=path_filter))
returnfile_list

defremove_empty_folders(directory, remove_root=True): #Including root.
directory=Path(directory)
try:
assertdirectory.is_dir()
forxindirectory.iterdir():
ifx.is_dir():
remove_empty_folders(x, remove_root=True)
ifremove_rootandnotlist(directory.iterdir()):
directory.rmdir()
exceptPermissionErrorase:
print('Error:', e)

defensure_nonexist(f):
'''
 Ensure the file does not exist. If it does, rename it to filename_2, filename_3, etc.
 '''
i=2
stem=f.stem
ifm:=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.
ifint(m[2]) <10andlen(m[2]) ==1:
stem=m[1]
i=int(m[2]) +1
whilef.exists():
f=f.with_name(f'{stem}_{i}{f.suffix}')
i=i+1
returnf

defensure_path_exists(path):
'''
 ensure the path exists.
 '''
path=Path(path)
ifnotpath.exists():
raiseFileNotFoundError(f'Path {path} does not exist!')
returnpath

defquickmd5(f):
hasher=hashlib.md5()
file_size=f.stat().st_size
buffer_size=1024*1024# 1MB

withf.open('rb') asf:
iffile_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)

returnf"{file_size}_{hasher.hexdigest()}"# this way is more readable than just return the hexdigest.

defmove_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
 """
ifnotsrc.exists():
raiseFileNotFoundError(f"The source file {src} does not exist.")
ifsrc==dst:
raiseValueError(f"Source and destination are the same: {src}")
ifdst.exists():
ifhash_method(src) ==hash_method(dst):
print(f'[W] {src.name} is a duplicate. Remove.')
src.unlink()
return
else:
ifconflict=='skip':
print(f'[W] Destination file {dst} already exists and hash does not match. Skip.')
return
elifconflict=='error':
raiseFileExistsError(f"Destination file {dst} already exists and hash does not match.")
elifconflict=='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_
ifverbose:
ifsrc.parent==dst.parent:
print(f"Rename {src.name} to {dst.name}.")
elifsrc.name==dst.name:
print(f'Move {src.name} into {dst.parent}')
else:
print(f'Move {src.name} to {dst}')
shutil.move(src, dst)

defbatch_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= [fforf, _inrenamings]
dst_files= [f.with_name(name) forf, nameinrenamings]

iflen(set(files)) !=len(files):
print('[E] There are duplicate files in the list! Please check.')
return

iflen(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
fornew_findst_files:
ifnew_f.exists() andnew_fnotinfiles:
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= []
forf, nameinrenamings:
iffindst_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))

forf, nameinreal_renamings:
f.rename(f.with_name(name))

# ==================== network related ====================
defrequests_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
importrequests
fromrequests.adaptersimportHTTPAdapter
fromurllib3.util.retryimportRetry

session=sessionorrequests.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)
returnsession

defget(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
 """
frombs4importBeautifulSoup

ifnotsession:
session=requests_retry_session()
r=session.get(url, cookies=cookies, headers=headers, timeout=timeout)
ifencoding:
returnBeautifulSoup(r.content, parser, from_encoding=encoding)
else:
returnBeautifulSoup(r.content, parser)

defget_webname(url):
returnunquote(url.split('?')[0].split('/')[-1])

defload_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)]>
 """
fromhttp.cookiejarimportMozillaCookieJar
fromrequests.cookiesimportRequestsCookieJar, create_cookie
importbrowsercookie
# pip install browsercookie

defconvert(cj):
cookies=RequestsCookieJar()
forcookieincj:
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)
returncookies

ifm:=re.search(r'^(chrome|firefox|edge)(/.+)?', str(s), re.IGNORECASE):
domain_name=m[2].lstrip('/') ifm[2] elseNone
ifm[1].lower() =='chrome':
cj=browsercookie.chrome()
elifm[1].lower() =='firefox':
cj=browsercookie.firefox()
elifm[1].lower() =='edge':
cj=browsercookie.edge()
ifdomain_name:
cj= [cookieforcookieincjifdomain_nameincookie.domain]
returnconvert(cj)

ifPath(s).exists():
cj=MozillaCookieJar(s)
cj.load(ignore_expires=True, ignore_discard=True)
forcookieincj:
ifcookie.expires==0:
cookie.expires=int(time.time()+86400)
returnconvert(cj)

ifre.search(r'^(.+?):\s*(.+?)', str(s)):
cookies=RequestsCookieJar()
fork, vinre.findall(r'(.+?):\s*([^;]+?)(?:;|$)', s):
cookies.set(k, v)
returncookies

raiseValueError(f'Invalid cookie string: {s}')

defdownload(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.
 """
frommimetypesimportguess_extension
# it uses requests_retry_session, so
# pip install requests

ifdupenotin ['skip', 'overwrite', 'rename', 'skip_same_size']:
raiseValueError(f'[Error] Invalid dupe method: {dupe} (must be either skip, overwrite, rename or skip_same_size).')

defprint(s, verbose_level, only=False):
if (onlyandverbose==verbose_level) or (notonlyandverbose>=verbose_level):
builtins.print(s)

defhas_valid_suffix(f):
# common suffixes
iff.suffix.lower() in ['.jpg', '.png', '.gif', '.webp', '.jpeg', '.bmp', '.svg', '.ico', '.mp4', '.mkv', '.webm', '.heic', '.pdf']:
returnTrue
iff.suffix.lower() in ['.php', '']:
returnFalse
# if the suffix is too long, has a space, etc., we assume it is not a valid suffix
iflen(f.suffix.lower()) >5or' 'inf.suffix.lower():
returnFalse
returnTrue

defreplace_suffix(f, content_type):

importbuiltins
importjson
importos
importre
importsys
importtime
importhashlib
importshutil
importunicodedata
fromdatetimeimportdatetime
frompathlibimportPath
fromurllib.parseimportunquote

# 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. ====================
defto_list(a):
returnaifisinstance(a, list) oraisNoneelse [a]

defflatten(x):
fromcollections.abcimportIterable
ifisinstance(x, Iterable) andnotisinstance(x, str):
return [aforiinxforainflatten(i)]
else:
return [x]

defprint_cmd(cmd, prefix='', actually_print=True):
commands_text_form= [f'"{c}"'ifre.search(r'[ ?]', str(c)) elsestr(c) forcincmd]
output=prefix+' '.join(commands_text_form)
ifactually_print:
print(output)
returnoutput

defcopy(data):
# pip install pyperclip
importpyperclip
pyperclip.copy(data)

defget_clipboard_data():
# pip install pyperclip
importpyperclip
returnpyperclip.paste()

defsafeify(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.
 """

assertisinstance(name, str), f'Name must be a string, not {type(name)}'

template= {'\\': '＼', '/': '／', ':': '：', '*': '＊', '?': '？', '"': '＂', '<': '＜', '>': '＞', '|': '｜','\n':'','\r':'','\t':''}
ifignore_backslash:
template.pop('\\', None)

forillegalintemplate:
name=name.replace(illegal, template[illegal])
returnname

defformat_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
importwcwidth

ifsisNone:
s=''
else:
s=str(s)
ifnotwidth:
returns
output=''
length=0
forcharins:
size=wcwidth.wcswidth(char)
iflength+size>width:
break
output+=char
length+=size
ifalign=='left':
returnoutput+padding*(width-length)
ifalign=='right':
returnpadding*(width-length) +output
ifalign=='center':
left_space= (width-length)//2
right_space=width-length-left_space
returnpadding*left_space+output+padding*right_space

classTable():
# pip install wcwidth

def__init__(self, rows=None, headers=None, max_width=100) ->None:
ifrowsandnotheaders:
self.headers=rows[0]
self.data=rows[1:]
elifheaders:
self.headers=headers
ifrows:
self.data=rows
else:
self.data= []
else:
raiseException('No header or data given!')
self.max_width=max_width

defrows(self):
forrowinself.data:
row_dict= {}
fori, headerinenumerate(self.headers):
row_dict[header] =row[i]
yieldrow_dict

defsearch(self, keywords, single=True):
matched_rows= []
forrowinself.data:
matched=True
forkey, valueinkeywords.items():
assertkeyinself.headers
col_idx=self.headers.index(key)
ifnotvalue: # Ignore empty values
continue
ifnot ((str(row[col_idx]) ==str(value)) or (row[col_idx] ==value)):
matched=False
break
ifmatched:
matched_rows.append(row)
ifsingle:
break
ifmatched_rows:
ifsingleandlen(matched_rows) ==1:
returnmatched_rows[0]
else:
returnmatched_rows
else:
return []

defsort(self, header_name_or_index, reverse=False):
ifisinstance(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=lambdax: x[col_idx], reverse=reverse)

defappend(self, new_data):
d= [''] *len(self.headers)
forkey, valueinnew_data.items():
ifnotkeyinself.headers: # silently ignore keys that are not in headers
continue
col_idx=self.headers.index(key)
d[col_idx] =value
self.data.append(d)

defprint(self, formats=None, custom_print=None):
importwcwidth

deffmt(value, str_format):
try:
returnstr_format.format(value)
exceptException:
returnvalue

defcalc_width(col_format, idx):
maxw=sum(wcwidth.wcwidth(c) forcinstr(self.headers[idx]))
forrowinself.data:
ifidx>len(row) -1:
continue
w=sum(wcwidth.wcwidth(c) forcinfmt(row[idx], col_format['str_format']))
ifw>maxw:
maxw=w
maxw=min(maxw, col_format['max_width'])
col_format['width'] =maxw

defprint_row(row, header_mode=False, custom_print=custom_print):
parts= []
foridxinrange(len(row)):
col_format=col_formats[idx]
# override str format back to nothing for headers
str_format="{}"ifheader_modeelsecol_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)
ifcustom_print:
custom_print(line)
else:
print(line)

col_formats=dict()
foridxinrange(len(self.headers)):
col_format= {
"align": "left",
"str_format": "{}",
"max_width": self.max_width
 }
ifformats:
ifidxinformats:
col_format.update(formats[idx])
elifself.headers[idx] informats:
col_format.update(formats[self.headers[idx]])
ifnot'width'incol_format:
calc_width(col_format, idx)
col_formats[idx] =col_format

print_row(self.headers, header_mode=True)
forrowinself.data:
print_row(row)

# return some useful info
total_width=sum(col_format['width'] forcol_formatincol_formats.values()) +2* (len(col_formats) -1)
return {'total_width': total_width, 'col_formats': col_formats}

defsave(self, f):
s=''
f=Path(f)
s+='\t'.join(self.headers) +'\n'
forrowinself.data:
s+='\t'.join([str(cell) forcellinrow]) +'\n'
f.write_text(s, encoding='utf8')

defmulti_col_print(data, columns=5):
'''Print a list of data in a table with columns'''
# pip install rich
fromrich.consoleimportConsole
fromrich.tableimportTable

table=Table(show_header=False, box=None)
for_inrange(columns):
table.add_column(style='cyan')
# add data to table vertically
row_count=len(data) //columns+1
foriinrange(row_count):
row=data[i::row_count]
table.add_row(*row)
console=Console()
console.print(table)

defrprint(*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
fromrich.consoleimportConsole
ifnothasattr(rprint, '_console'):
rprint._console=Console()
rprint._console.print(*args, **kwargs)

defprint2(*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.
 '''
frombuiltinsimportprintasbuiltin_print
ifnothasattr(print2, '_newlined'):
print2._newlined=True
end=kwargs.pop('end', '\n')
ifend.endswith('\n') andnotprint2._newlined:
builtin_print()
print2._newlined=Trueifend.endswith('\n') elseFalse
builtin_print(*args, **kwargs, end=end)

defarray_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
foridx, seginenumerate(a):
ifidx==0: # first segment
s+=f'{seg}'
range_start=seg
else:
ifseg-prev_seg==1:
ifidx==len(a) -1: # last segment
s+=f'{dash}{seg}'
else:
ifprev_seg>range_start:
s+=f'{dash}{prev_seg}'
s+=f'{sep}{seg}'
range_start=seg
prev_seg=seg
returns

defpluralize(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).
 '''
ifcount==1:
returnword
else:
ifalt:
returnalt
else:
returnword+suffix

defcompare_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
fromrichimportprintasrprint

defprint(*args, **kwargs):
ifnotmute:
rprint(*args, **kwargs)

equal=True
type_changed=False

iftype(value_old) !=type(value):
type_changed=True
equal=False
elifisinstance(value, dict):
forkey, vinvalue.items():
ifkeynotinvalue_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)
forkey, vinvalue_old.items():
ifkeynotinvalue:
equal=False
print(f'{print_prefix}.{key}: [red]Removed:[/red]', v)
returnequal
elifisinstance(value, list):
foriinrange(min(len(value), len(value_old))):
equal&=compare_obj(value_old[i], value[i], print_prefix=f'{print_prefix}[{i}]', mute=mute)
iflen(value_old) <len(value):
equal=False
foriinrange(len(value_old), len(value)):
print(f'{print_prefix}[{i}]: [green]Added:[/green]', value[i])
eliflen(value_old) >len(value):
equal=False
foriinrange(len(value), len(value_old)):
print(f'{print_prefix}[{i}]: [red]Removed:[/red]',value_old[i])
returnequal

elifisinstance(value, str):
equal=re.sub(r'\s+',' ', value_old) ==re.sub(r'\s+',' ', value)
else:
equal=value_old==value
ifnotequal:
print(f'{print_prefix}:', end='')
s=str(value_old) +str(value)
iflen(s) <60andnot'\n'ins:
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)
returnequal

defnormalize(s):
'''Normalize a string by replacing full-width characters with half-width characters
 but keep the full-width dash (～) as it is.'''
returnunicodedata.normalize('NFKC', s.replace('～', '$dash$')).replace('$dash$', '～')

# ==================== datetime related ====================
defparse_to_shortdate(date_str, fmt=None):
# pip install python-dateutil
fromdateutilimportparser

current_year=datetime.now().year

iffmtisNone: fmt='%y%m%d'
ifisinstance(date_str, datetime):
returndate_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})']
forpatterninpatterns:
# 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.
ifm:=re.search(pattern, date_str) orre.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'] orcurrent_year}-{m['m']}-{m['d']}"
break
try:
returnparser.parse(date_str, yearfirst=True).strftime(fmt)
except:
return""


classMyTime:
def__init__(self, t=None):
importpytz

self.tz=pytz.timezone('Asia/Tokyo')

iftisNone:
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):
ifformat_type=='obj':
returndt
elifformat_type=="short":
returndt.strftime('%y%m%d_%H%M%S')
elifformat_type=="pretty":
returndt.strftime('%Y-%m-%d (%a) %H:%M:%S')
else:
raiseValueError("Invalid format_type. Choose from 'obj', 'short', 'pretty'.")

deflocal(self, format_type="obj"):
returnself._format_time(self.local_time, format_type)

defjst(self, format_type="obj"):
returnself._format_time(self.jst_time, format_type)

defnaive(self, format_type="obj"):
returnself._format_time(self.naive_time, format_type)

# deprecated, will be removed in the future
defget_current_time(now=None):
# pip install pytz
importpytz

tz=pytz.timezone('Asia/Tokyo')

ifnowisNone:
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
deftd_format(td_object_or_sec, long_form=True):
ifisinstance(td_object_or_sec, int) orisinstance(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=[]
forperiod_name, period_secondsinperiods:
ifseconds>period_seconds:
period_value , seconds=divmod(seconds, period_seconds)
iflong_form:
has_s='s'ifperiod_value>1else''
strings.append(f"{period_value}{period_name}{has_s}")
else:
strings.append(f"{period_value}{period_name[0:1]}")
iflong_form:
return", ".join(strings)
else:
return"".join(strings)

# deprecated, just use MyTime(dt).jst()
defto_jp_time(dt, input_timezone=None):
# pip install python-dateutil pytz
fromdateutilimportparser
frompytzimporttimezone

ifisinstance(dt, str):
dt=parser.parse(dt)
ifinput_timezone:
dt=timezone(input_timezone).localize(dt)
returndt.astimezone(timezone('asia/tokyo'))

# ==================== performance related ====================
deftic():
global_start_time
_start_time=time.time()

deftac(print=True):
global_start_time
t=time.time() -_start_time
ifprint:
builtins.print(f'Time passed: {t:.2f} s')
returnt

# a decorator to time a function
deftimeme(func):
defwrapper(*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")
returnresult
returnwrapper

# ==================== file related ====================
defdump_json(mydict, filename, **kwargs):
filename=Path(filename)
iffilename.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)
withfilename.open('w', encoding='utf-8') asf:
json.dump(mydict, f, **default_kwargs)

defload_json(filename, encoding='utf-8'):
filename=Path(filename)
withfilename.open('r', encoding=encoding) asf:
data=json.load(f)
returndata

defdump_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)
ifprettify:
text=soup.prettify()
else:
text=str(soup)
withfilename.open('w', encoding=encoding) asf:
f.write(text)

defdump_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=''
ifisinstance(data[0], dict):
headers=data[0].keys()
ifprint_header:
s+='\t'.join(str(h) forhinheaders) +'\n'
forrowindata:
s+='\t'.join([str(row.get(h, '')) forhinheaders]) +'\n'
else:
ifprint_header:
print('[W] No header provided. Use default header [0, 1, 2, ...]')
s+='\t'.join([str(i) foriinrange(len(data[0]))]) +'\n'
forrowindata:
s+='\t'.join([str(i) foriinrow]) +'\n'
ifverbose:
print(s)
filename=Path(filename)
filename.parent.mkdir(parents=True, exist_ok=True)
filename.write_text(s, encoding='utf8')


defget_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.
iffile_filterisNoneandpath_filterisNone:
defquick_scan(directory):
forentryinos.scandir(directory):
ifentry.is_file():
yieldentry
elifrecursiveandentry.is_dir(follow_symlinks=False):
yieldfromquick_scan(entry.path)
return [Path(f) forfinquick_scan(directory)]
# else, use pathlib.iterdir and just dynamically change the list. The speed is basically the same (slow).
file_list= []
forxindirectory.iterdir():
ifx.is_file():
ifnotfile_filterorfile_filter(x):
file_list.append(x)
elifrecursiveandx.is_dir() and (notpath_filterorpath_filter(x)):
file_list.extend(get_files(x, recursive=recursive, file_filter=file_filter, path_filter=path_filter))
returnfile_list

defremove_empty_folders(directory, remove_root=True): #Including root.
directory=Path(directory)
try:
assertdirectory.is_dir()
forxindirectory.iterdir():
ifx.is_dir():
remove_empty_folders(x, remove_root=True)
ifremove_rootandnotlist(directory.iterdir()):
directory.rmdir()
exceptPermissionErrorase:
print('Error:', e)

defensure_nonexist(f):
'''
 Ensure the file does not exist. If it does, rename it to filename_2, filename_3, etc.
 '''
i=2
stem=f.stem
ifm:=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.
ifint(m[2]) <10andlen(m[2]) ==1:
stem=m[1]
i=int(m[2]) +1
whilef.exists():
f=f.with_name(f'{stem}_{i}{f.suffix}')
i=i+1
returnf

defensure_path_exists(path):
'''
 ensure the path exists.
 '''
path=Path(path)
ifnotpath.exists():
raiseFileNotFoundError(f'Path {path} does not exist!')
returnpath

defquickmd5(f):
hasher=hashlib.md5()
file_size=f.stat().st_size
buffer_size=1024*1024# 1MB

withf.open('rb') asf:
iffile_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)

returnf"{file_size}_{hasher.hexdigest()}"# this way is more readable than just return the hexdigest.

defmove_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
 """
ifnotsrc.exists():
raiseFileNotFoundError(f"The source file {src} does not exist.")
ifsrc==dst:
raiseValueError(f"Source and destination are the same: {src}")
ifdst.exists():
ifhash_method(src) ==hash_method(dst):
print(f'[W] {src.name} is a duplicate. Remove.')
src.unlink()
return
else:
ifconflict=='skip':
print(f'[W] Destination file {dst} already exists and hash does not match. Skip.')
return
elifconflict=='error':
raiseFileExistsError(f"Destination file {dst} already exists and hash does not match.")
elifconflict=='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_
ifverbose:
ifsrc.parent==dst.parent:
print(f"Rename {src.name} to {dst.name}.")
elifsrc.name==dst.name:
print(f'Move {src.name} into {dst.parent}')
else:
print(f'Move {src.name} to {dst}')
shutil.move(src, dst)

defbatch_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= [fforf, _inrenamings]
dst_files= [f.with_name(name) forf, nameinrenamings]

iflen(set(files)) !=len(files):
print('[E] There are duplicate files in the list! Please check.')
return

iflen(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
fornew_findst_files:
ifnew_f.exists() andnew_fnotinfiles:
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= []
forf, nameinrenamings:
iffindst_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))

forf, nameinreal_renamings:
f.rename(f.with_name(name))

# ==================== network related ====================
defrequests_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
importrequests
fromrequests.adaptersimportHTTPAdapter
fromurllib3.util.retryimportRetry

session=sessionorrequests.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)
returnsession

defget(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
 """
frombs4importBeautifulSoup

ifnotsession:
session=requests_retry_session()
r=session.get(url, cookies=cookies, headers=headers, timeout=timeout)
ifencoding:
returnBeautifulSoup(r.content, parser, from_encoding=encoding)
else:
returnBeautifulSoup(r.content, parser)

defget_webname(url):
returnunquote(url.split('?')[0].split('/')[-1])

defload_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)]>
 """
fromhttp.cookiejarimportMozillaCookieJar
fromrequests.cookiesimportRequestsCookieJar, create_cookie
importbrowsercookie
# pip install browsercookie

defconvert(cj):
cookies=RequestsCookieJar()
forcookieincj:
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)
returncookies

ifm:=re.search(r'^(chrome|firefox|edge)(/.+)?', str(s), re.IGNORECASE):
domain_name=m[2].lstrip('/') ifm[2] elseNone
ifm[1].lower() =='chrome':
cj=browsercookie.chrome()
elifm[1].lower() =='firefox':
cj=browsercookie.firefox()
elifm[1].lower() =='edge':
cj=browsercookie.edge()
ifdomain_name:
cj= [cookieforcookieincjifdomain_nameincookie.domain]
returnconvert(cj)

ifPath(s).exists():
cj=MozillaCookieJar(s)
cj.load(ignore_expires=True, ignore_discard=True)
forcookieincj:
ifcookie.expires==0:
cookie.expires=int(time.time()+86400)
returnconvert(cj)

ifre.search(r'^(.+?):\s*(.+?)', str(s)):
cookies=RequestsCookieJar()
fork, vinre.findall(r'(.+?):\s*([^;]+?)(?:;|$)', s):
cookies.set(k, v)
returncookies

raiseValueError(f'Invalid cookie string: {s}')

defdownload(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.
 """
frommimetypesimportguess_extension
# it uses requests_retry_session, so
# pip install requests

ifdupenotin ['skip', 'overwrite', 'rename', 'skip_same_size']:
raiseValueError(f'[Error] Invalid dupe method: {dupe} (must be either skip, overwrite, rename or skip_same_size).')

defprint(s, verbose_level, only=False):
if (onlyandverbose==verbose_level) or (notonlyandverbose>=verbose_level):
builtins.print(s)

defhas_valid_suffix(f):
# common suffixes
iff.suffix.lower() in ['.jpg', '.png', '.gif', '.webp', '.jpeg', '.bmp', '.svg', '.ico', '.mp4', '.mkv', '.webm', '.heic', '.pdf']:
returnTrue
iff.suffix.lower() in ['.php', '']:
returnFalse
# if the suffix is too long, has a space, etc., we assume it is not a valid suffix
iflen(f.suffix.lower()) >5or' 'inf.suffix.lower():
returnFalse
returnTrue

defreplace_suffix(f, content_type):

importbuiltins
importjson
importos
importre
importsys
importtime
importhashlib
importshutil
importunicodedata
fromdatetimeimportdatetime
frompathlibimportPath
fromurllib.parseimportunquote

# 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. ====================
defto_list(a):
returnaifisinstance(a, list) oraisNoneelse [a]

defflatten(x):
fromcollections.abcimportIterable
ifisinstance(x, Iterable) andnotisinstance(x, str):
return [aforiinxforainflatten(i)]
else:
return [x]

defprint_cmd(cmd, prefix='', actually_print=True):
commands_text_form= [f'"{c}"'ifre.search(r'[ ?]', str(c)) elsestr(c) forcincmd]
output=prefix+' '.join(commands_text_form)
ifactually_print:
print(output)
returnoutput

defcopy(data):
# pip install pyperclip
importpyperclip
pyperclip.copy(data)

defget_clipboard_data():
# pip install pyperclip
importpyperclip
returnpyperclip.paste()

defsafeify(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.
 """

assertisinstance(name, str), f'Name must be a string, not {type(name)}'

template= {'\\': '＼', '/': '／', ':': '：', '*': '＊', '?': '？', '"': '＂', '<': '＜', '>': '＞', '|': '｜','\n':'','\r':'','\t':''}
ifignore_backslash:
template.pop('\\', None)

forillegalintemplate:
name=name.replace(illegal, template[illegal])
returnname

defformat_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
importwcwidth

ifsisNone:
s=''
else:
s=str(s)
ifnotwidth:
returns
output=''
length=0
forcharins:
size=wcwidth.wcswidth(char)
iflength+size>width:
break
output+=char
length+=size
ifalign=='left':
returnoutput+padding*(width-length)
ifalign=='right':
returnpadding*(width-length) +output
ifalign=='center':
left_space= (width-length)//2
right_space=width-length-left_space
returnpadding*left_space+output+padding*right_space

classTable():
# pip install wcwidth

def__init__(self, rows=None, headers=None, max_width=100) ->None:
ifrowsandnotheaders:
self.headers=rows[0]
self.data=rows[1:]
elifheaders:
self.headers=headers
ifrows:
self.data=rows
else:
self.data= []
else:
raiseException('No header or data given!')
self.max_width=max_width

defrows(self):
forrowinself.data:
row_dict= {}
fori, headerinenumerate(self.headers):
row_dict[header] =row[i]
yieldrow_dict

defsearch(self, keywords, single=True):
matched_rows= []
forrowinself.data:
matched=True
forkey, valueinkeywords.items():
assertkeyinself.headers
col_idx=self.headers.index(key)
ifnotvalue: # Ignore empty values
continue
ifnot ((str(row[col_idx]) ==str(value)) or (row[col_idx] ==value)):
matched=False
break
ifmatched:
matched_rows.append(row)
ifsingle:
break
ifmatched_rows:
ifsingleandlen(matched_rows) ==1:
returnmatched_rows[0]
else:
returnmatched_rows
else:
return []

defsort(self, header_name_or_index, reverse=False):
ifisinstance(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=lambdax: x[col_idx], reverse=reverse)

defappend(self, new_data):
d= [''] *len(self.headers)
forkey, valueinnew_data.items():
ifnotkeyinself.headers: # silently ignore keys that are not in headers
continue
col_idx=self.headers.index(key)
d[col_idx] =value
self.data.append(d)

defprint(self, formats=None, custom_print=None):
importwcwidth

deffmt(value, str_format):
try:
returnstr_format.format(value)
exceptException:
returnvalue

defcalc_width(col_format, idx):
maxw=sum(wcwidth.wcwidth(c) forcinstr(self.headers[idx]))
forrowinself.data:
ifidx>len(row) -1:
continue
w=sum(wcwidth.wcwidth(c) forcinfmt(row[idx], col_format['str_format']))
ifw>maxw:
maxw=w
maxw=min(maxw, col_format['max_width'])
col_format['width'] =maxw

defprint_row(row, header_mode=False, custom_print=custom_print):
parts= []
foridxinrange(len(row)):
col_format=col_formats[idx]
# override str format back to nothing for headers
str_format="{}"ifheader_modeelsecol_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)
ifcustom_print:
custom_print(line)
else:
print(line)

col_formats=dict()
foridxinrange(len(self.headers)):
col_format= {
"align": "left",
"str_format": "{}",
"max_width": self.max_width
 }
ifformats:
ifidxinformats:
col_format.update(formats[idx])
elifself.headers[idx] informats:
col_format.update(formats[self.headers[idx]])
ifnot'width'incol_format:
calc_width(col_format, idx)
col_formats[idx] =col_format

print_row(self.headers, header_mode=True)
forrowinself.data:
print_row(row)

# return some useful info
total_width=sum(col_format['width'] forcol_formatincol_formats.values()) +2* (len(col_formats) -1)
return {'total_width': total_width, 'col_formats': col_formats}

defsave(self, f):
s=''
f=Path(f)
s+='\t'.join(self.headers) +'\n'
forrowinself.data:
s+='\t'.join([str(cell) forcellinrow]) +'\n'
f.write_text(s, encoding='utf8')

defmulti_col_print(data, columns=5):
'''Print a list of data in a table with columns'''
# pip install rich
fromrich.consoleimportConsole
fromrich.tableimportTable

table=Table(show_header=False, box=None)
for_inrange(columns):
table.add_column(style='cyan')
# add data to table vertically
row_count=len(data) //columns+1
foriinrange(row_count):
row=data[i::row_count]
table.add_row(*row)
console=Console()
console.print(table)

defrprint(*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
fromrich.consoleimportConsole
ifnothasattr(rprint, '_console'):
rprint._console=Console()
rprint._console.print(*args, **kwargs)

defprint2(*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.
 '''
frombuiltinsimportprintasbuiltin_print
ifnothasattr(print2, '_newlined'):
print2._newlined=True
end=kwargs.pop('end', '\n')
ifend.endswith('\n') andnotprint2._newlined:
builtin_print()
print2._newlined=Trueifend.endswith('\n') elseFalse
builtin_print(*args, **kwargs, end=end)

defarray_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
foridx, seginenumerate(a):
ifidx==0: # first segment
s+=f'{seg}'
range_start=seg
else:
ifseg-prev_seg==1:
ifidx==len(a) -1: # last segment
s+=f'{dash}{seg}'
else:
ifprev_seg>range_start:
s+=f'{dash}{prev_seg}'
s+=f'{sep}{seg}'
range_start=seg
prev_seg=seg
returns

defpluralize(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).
 '''
ifcount==1:
returnword
else:
ifalt:
returnalt
else:
returnword+suffix

defcompare_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
fromrichimportprintasrprint

defprint(*args, **kwargs):
ifnotmute:
rprint(*args, **kwargs)

equal=True
type_changed=False

iftype(value_old) !=type(value):
type_changed=True
equal=False
elifisinstance(value, dict):
forkey, vinvalue.items():
ifkeynotinvalue_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)
forkey, vinvalue_old.items():
ifkeynotinvalue:
equal=False
print(f'{print_prefix}.{key}: [red]Removed:[/red]', v)
returnequal
elifisinstance(value, list):
foriinrange(min(len(value), len(value_old))):
equal&=compare_obj(value_old[i], value[i], print_prefix=f'{print_prefix}[{i}]', mute=mute)
iflen(value_old) <len(value):
equal=False
foriinrange(len(value_old), len(value)):
print(f'{print_prefix}[{i}]: [green]Added:[/green]', value[i])
eliflen(value_old) >len(value):
equal=False
foriinrange(len(value), len(value_old)):
print(f'{print_prefix}[{i}]: [red]Removed:[/red]',value_old[i])
returnequal

elifisinstance(value, str):
equal=re.sub(r'\s+',' ', value_old) ==re.sub(r'\s+',' ', value)
else:
equal=value_old==value
ifnotequal:
print(f'{print_prefix}:', end='')
s=str(value_old) +str(value)
iflen(s) <60andnot'\n'ins:
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)
returnequal

defnormalize(s):
'''Normalize a string by replacing full-width characters with half-width characters
 but keep the full-width dash (～) as it is.'''
returnunicodedata.normalize('NFKC', s.replace('～', '$dash$')).replace('$dash$', '～')

# ==================== datetime related ====================
defparse_to_shortdate(date_str, fmt=None):
# pip install python-dateutil
fromdateutilimportparser

current_year=datetime.now().year

iffmtisNone: fmt='%y%m%d'
ifisinstance(date_str, datetime):
returndate_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})']
forpatterninpatterns:
# 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.
ifm:=re.search(pattern, date_str) orre.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'] orcurrent_year}-{m['m']}-{m['d']}"
break
try:
returnparser.parse(date_str, yearfirst=True).strftime(fmt)
except:
return""


classMyTime:
def__init__(self, t=None):
importpytz

self.tz=pytz.timezone('Asia/Tokyo')

iftisNone:
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):
ifformat_type=='obj':
returndt
elifformat_type=="short":
returndt.strftime('%y%m%d_%H%M%S')
elifformat_type=="pretty":
returndt.strftime('%Y-%m-%d (%a) %H:%M:%S')
else:
raiseValueError("Invalid format_type. Choose from 'obj', 'short', 'pretty'.")

deflocal(self, format_type="obj"):
returnself._format_time(self.local_time, format_type)

defjst(self, format_type="obj"):
returnself._format_time(self.jst_time, format_type)

defnaive(self, format_type="obj"):
returnself._format_time(self.naive_time, format_type)

# deprecated, will be removed in the future
defget_current_time(now=None):
# pip install pytz
importpytz

tz=pytz.timezone('Asia/Tokyo')

ifnowisNone:
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
deftd_format(td_object_or_sec, long_form=True):
ifisinstance(td_object_or_sec, int) orisinstance(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=[]
forperiod_name, period_secondsinperiods:
ifseconds>period_seconds:
period_value , seconds=divmod(seconds, period_seconds)
iflong_form:
has_s='s'ifperiod_value>1else''
strings.append(f"{period_value}{period_name}{has_s}")
else:
strings.append(f"{period_value}{period_name[0:1]}")
iflong_form:
return", ".join(strings)
else:
return"".join(strings)

# deprecated, just use MyTime(dt).jst()
defto_jp_time(dt, input_timezone=None):
# pip install python-dateutil pytz
fromdateutilimportparser
frompytzimporttimezone

ifisinstance(dt, str):
dt=parser.parse(dt)
ifinput_timezone:
dt=timezone(input_timezone).localize(dt)
returndt.astimezone(timezone('asia/tokyo'))

# ==================== performance related ====================
deftic():
global_start_time
_start_time=time.time()

deftac(print=True):
global_start_time
t=time.time() -_start_time
ifprint:
builtins.print(f'Time passed: {t:.2f} s')
returnt

# a decorator to time a function
deftimeme(func):
defwrapper(*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")
returnresult
returnwrapper

# ==================== file related ====================
defdump_json(mydict, filename, **kwargs):
filename=Path(filename)
iffilename.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)
withfilename.open('w', encoding='utf-8') asf:
json.dump(mydict, f, **default_kwargs)

defload_json(filename, encoding='utf-8'):
filename=Path(filename)
withfilename.open('r', encoding=encoding) asf:
data=json.load(f)
returndata

defdump_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)
ifprettify:
text=soup.prettify()
else:
text=str(soup)
withfilename.open('w', encoding=encoding) asf:
f.write(text)

defdump_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=''
ifisinstance(data[0], dict):
headers=data[0].keys()
ifprint_header:
s+='\t'.join(str(h) forhinheaders) +'\n'
forrowindata:
s+='\t'.join([str(row.get(h, '')) forhinheaders]) +'\n'
else:
ifprint_header:
print('[W] No header provided. Use default header [0, 1, 2, ...]')
s+='\t'.join([str(i) foriinrange(len(data[0]))]) +'\n'
forrowindata:
s+='\t'.join([str(i) foriinrow]) +'\n'
ifverbose:
print(s)
filename=Path(filename)
filename.parent.mkdir(parents=True, exist_ok=True)
filename.write_text(s, encoding='utf8')


defget_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.
iffile_filterisNoneandpath_filterisNone:
defquick_scan(directory):
forentryinos.scandir(directory):
ifentry.is_file():
yieldentry
elifrecursiveandentry.is_dir(follow_symlinks=False):
yieldfromquick_scan(entry.path)
return [Path(f) forfinquick_scan(directory)]
# else, use pathlib.iterdir and just dynamically change the list. The speed is basically the same (slow).
file_list= []
forxindirectory.iterdir():
ifx.is_file():
ifnotfile_filterorfile_filter(x):
file_list.append(x)
elifrecursiveandx.is_dir() and (notpath_filterorpath_filter(x)):
file_list.extend(get_files(x, recursive=recursive, file_filter=file_filter, path_filter=path_filter))
returnfile_list

defremove_empty_folders(directory, remove_root=True): #Including root.
directory=Path(directory)
try:
assertdirectory.is_dir()
forxindirectory.iterdir():
ifx.is_dir():
remove_empty_folders(x, remove_root=True)
ifremove_rootandnotlist(directory.iterdir()):
directory.rmdir()
exceptPermissionErrorase:
print('Error:', e)

defensure_nonexist(f):
'''
 Ensure the file does not exist. If it does, rename it to filename_2, filename_3, etc.
 '''
i=2
stem=f.stem
ifm:=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.
ifint(m[2]) <10andlen(m[2]) ==1:
stem=m[1]
i=int(m[2]) +1
whilef.exists():
f=f.with_name(f'{stem}_{i}{f.suffix}')
i=i+1
returnf

defensure_path_exists(path):
'''
 ensure the path exists.
 '''
path=Path(path)
ifnotpath.exists():
raiseFileNotFoundError(f'Path {path} does not exist!')
returnpath

defquickmd5(f):
hasher=hashlib.md5()
file_size=f.stat().st_size
buffer_size=1024*1024# 1MB

withf.open('rb') asf:
iffile_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)

returnf"{file_size}_{hasher.hexdigest()}"# this way is more readable than just return the hexdigest.

defmove_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
 """
ifnotsrc.exists():
raiseFileNotFoundError(f"The source file {src} does not exist.")
ifsrc==dst:
raiseValueError(f"Source and destination are the same: {src}")
ifdst.exists():
ifhash_method(src) ==hash_method(dst):
print(f'[W] {src.name} is a duplicate. Remove.')
src.unlink()
return
else:
ifconflict=='skip':
print(f'[W] Destination file {dst} already exists and hash does not match. Skip.')
return
elifconflict=='error':
raiseFileExistsError(f"Destination file {dst} already exists and hash does not match.")
elifconflict=='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_
ifverbose:
ifsrc.parent==dst.parent:
print(f"Rename {src.name} to {dst.name}.")
elifsrc.name==dst.name:
print(f'Move {src.name} into {dst.parent}')
else:
print(f'Move {src.name} to {dst}')
shutil.move(src, dst)

defbatch_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= [fforf, _inrenamings]
dst_files= [f.with_name(name) forf, nameinrenamings]

iflen(set(files)) !=len(files):
print('[E] There are duplicate files in the list! Please check.')
return

iflen(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
fornew_findst_files:
ifnew_f.exists() andnew_fnotinfiles:
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= []
forf, nameinrenamings:
iffindst_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))

forf, nameinreal_renamings:
f.rename(f.with_name(name))

# ==================== network related ====================
defrequests_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
importrequests
fromrequests.adaptersimportHTTPAdapter
fromurllib3.util.retryimportRetry

session=sessionorrequests.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)
returnsession

defget(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
 """
frombs4importBeautifulSoup

ifnotsession:
session=requests_retry_session()
r=session.get(url, cookies=cookies, headers=headers, timeout=timeout)
ifencoding:
returnBeautifulSoup(r.content, parser, from_encoding=encoding)
else:
returnBeautifulSoup(r.content, parser)

defget_webname(url):
returnunquote(url.split('?')[0].split('/')[-1])

defload_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)]>
 """
fromhttp.cookiejarimportMozillaCookieJar
fromrequests.cookiesimportRequestsCookieJar, create_cookie
importbrowsercookie
# pip install browsercookie

defconvert(cj):
cookies=RequestsCookieJar()
forcookieincj:
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)
returncookies

ifm:=re.search(r'^(chrome|firefox|edge)(/.+)?', str(s), re.IGNORECASE):
domain_name=m[2].lstrip('/') ifm[2] elseNone
ifm[1].lower() =='chrome':
cj=browsercookie.chrome()
elifm[1].lower() =='firefox':
cj=browsercookie.firefox()
elifm[1].lower() =='edge':
cj=browsercookie.edge()
ifdomain_name:
cj= [cookieforcookieincjifdomain_nameincookie.domain]
returnconvert(cj)

ifPath(s).exists():
cj=MozillaCookieJar(s)
cj.load(ignore_expires=True, ignore_discard=True)
forcookieincj:
ifcookie.expires==0:
cookie.expires=int(time.time()+86400)
returnconvert(cj)

ifre.search(r'^(.+?):\s*(.+?)', str(s)):
cookies=RequestsCookieJar()
fork, vinre.findall(r'(.+?):\s*([^;]+?)(?:;|$)', s):
cookies.set(k, v)
returncookies

raiseValueError(f'Invalid cookie string: {s}')

defdownload(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.
 """
frommimetypesimportguess_extension
# it uses requests_retry_session, so
# pip install requests

ifdupenotin ['skip', 'overwrite', 'rename', 'skip_same_size']:
raiseValueError(f'[Error] Invalid dupe method: {dupe} (must be either skip, overwrite, rename or skip_same_size).')

defprint(s, verbose_level, only=False):
if (onlyandverbose==verbose_level) or (notonlyandverbose>=verbose_level):
builtins.print(s)

defhas_valid_suffix(f):
# common suffixes
iff.suffix.lower() in ['.jpg', '.png', '.gif', '.webp', '.jpeg', '.bmp', '.svg', '.ico', '.mp4', '.mkv', '.webm', '.heic', '.pdf']:
returnTrue
iff.suffix.lower() in ['.php', '']:
returnFalse
# if the suffix is too long, has a space, etc., we assume it is not a valid suffix
iflen(f.suffix.lower()) >5or' 'inf.suffix.lower():
returnFalse
returnTrue

defreplace_suffix(f, content_type):

importbuiltins
importjson
importos
importre
importsys
importtime
importhashlib
importshutil
importunicodedata
fromdatetimeimportdatetime
frompathlibimportPath
fromurllib.parseimportunquote

# 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. ====================
defto_list(a):
returnaifisinstance(a, list) oraisNoneelse [a]

defflatten(x):
fromcollections.abcimportIterable
ifisinstance(x, Iterable) andnotisinstance(x, str):
return [aforiinxforainflatten(i)]
else:
return [x]

defprint_cmd(cmd, prefix='', actually_print=True):
commands_text_form= [f'"{c}"'ifre.search(r'[ ?]', str(c)) elsestr(c) forcincmd]
output=prefix+' '.join(commands_text_form)
ifactually_print:
print(output)
returnoutput

defcopy(data):
# pip install pyperclip
importpyperclip
pyperclip.copy(data)

defget_clipboard_data():
# pip install pyperclip
importpyperclip
returnpyperclip.paste()

defsafeify(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.
 """

assertisinstance(name, str), f'Name must be a string, not {type(name)}'

template= {'\\': '＼', '/': '／', ':': '：', '*': '＊', '?': '？', '"': '＂', '<': '＜', '>': '＞', '|': '｜','\n':'','\r':'','\t':''}
ifignore_backslash:
template.pop('\\', None)

forillegalintemplate:
name=name.replace(illegal, template[illegal])
returnname

defformat_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
importwcwidth

ifsisNone:
s=''
else:
s=str(s)
ifnotwidth:
returns
output=''
length=0
forcharins:
size=wcwidth.wcswidth(char)
iflength+size>width:
break
output+=char
length+=size
ifalign=='left':
returnoutput+padding*(width-length)
ifalign=='right':
returnpadding*(width-length) +output
ifalign=='center':
left_space= (width-length)//2
right_space=width-length-left_space
returnpadding*left_space+output+padding*right_space

classTable():
# pip install wcwidth

def__init__(self, rows=None, headers=None, max_width=100) ->None:
ifrowsandnotheaders:
self.headers=rows[0]
self.data=rows[1:]
elifheaders:
self.headers=headers
ifrows:
self.data=rows
else:
self.data= []
else:
raiseException('No header or data given!')
self.max_width=max_width

defrows(self):
forrowinself.data:
row_dict= {}
fori, headerinenumerate(self.headers):
row_dict[header] =row[i]
yieldrow_dict

defsearch(self, keywords, single=True):
matched_rows= []
forrowinself.data:
matched=True
forkey, valueinkeywords.items():
assertkeyinself.headers
col_idx=self.headers.index(key)
ifnotvalue: # Ignore empty values
continue
ifnot ((str(row[col_idx]) ==str(value)) or (row[col_idx] ==value)):
matched=False
break
ifmatched:
matched_rows.append(row)
ifsingle:
break
ifmatched_rows:
ifsingleandlen(matched_rows) ==1:
returnmatched_rows[0]
else:
returnmatched_rows
else:
return []

defsort(self, header_name_or_index, reverse=False):
ifisinstance(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=lambdax: x[col_idx], reverse=reverse)

defappend(self, new_data):
d= [''] *len(self.headers)
forkey, valueinnew_data.items():
ifnotkeyinself.headers: # silently ignore keys that are not in headers
continue
col_idx=self.headers.index(key)
d[col_idx] =value
self.data.append(d)

defprint(self, formats=None, custom_print=None):
importwcwidth

deffmt(value, str_format):
try:
returnstr_format.format(value)
exceptException:
returnvalue

defcalc_width(col_format, idx):
maxw=sum(wcwidth.wcwidth(c) forcinstr(self.headers[idx]))
forrowinself.data:
ifidx>len(row) -1:
continue
w=sum(wcwidth.wcwidth(c) forcinfmt(row[idx], col_format['str_format']))
ifw>maxw:
maxw=w
maxw=min(maxw, col_format['max_width'])
col_format['width'] =maxw

defprint_row(row, header_mode=False, custom_print=custom_print):
parts= []
foridxinrange(len(row)):
col_format=col_formats[idx]
# override str format back to nothing for headers
str_format="{}"ifheader_modeelsecol_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)
ifcustom_print:
custom_print(line)
else:
print(line)

col_formats=dict()
foridxinrange(len(self.headers)):
col_format= {
"align": "left",
"str_format": "{}",
"max_width": self.max_width
 }
ifformats:
ifidxinformats:
col_format.update(formats[idx])
elifself.headers[idx] informats:
col_format.update(formats[self.headers[idx]])
ifnot'width'incol_format:
calc_width(col_format, idx)
col_formats[idx] =col_format

print_row(self.headers, header_mode=True)
forrowinself.data:
print_row(row)

# return some useful info
total_width=sum(col_format['width'] forcol_formatincol_formats.values()) +2* (len(col_formats) -1)
return {'total_width': total_width, 'col_formats': col_formats}

defsave(self, f):
s=''
f=Path(f)
s+='\t'.join(self.headers) +'\n'
forrowinself.data:
s+='\t'.join([str(cell) forcellinrow]) +'\n'
f.write_text(s, encoding='utf8')

defmulti_col_print(data, columns=5):
'''Print a list of data in a table with columns'''
# pip install rich
fromrich.consoleimportConsole
fromrich.tableimportTable

table=Table(show_header=False, box=None)
for_inrange(columns):
table.add_column(style='cyan')
# add data to table vertically
row_count=len(data) //columns+1
foriinrange(row_count):
row=data[i::row_count]
table.add_row(*row)
console=Console()
console.print(table)

defrprint(*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
fromrich.consoleimportConsole
ifnothasattr(rprint, '_console'):
rprint._console=Console()
rprint._console.print(*args, **kwargs)

defprint2(*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.
 '''
frombuiltinsimportprintasbuiltin_print
ifnothasattr(print2, '_newlined'):
print2._newlined=True
end=kwargs.pop('end', '\n')
ifend.endswith('\n') andnotprint2._newlined:
builtin_print()
print2._newlined=Trueifend.endswith('\n') elseFalse
builtin_print(*args, **kwargs, end=end)

defarray_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
foridx, seginenumerate(a):
ifidx==0: # first segment
s+=f'{seg}'
range_start=seg
else:
ifseg-prev_seg==1:
ifidx==len(a) -1: # last segment
s+=f'{dash}{seg}'
else:
ifprev_seg>range_start:
s+=f'{dash}{prev_seg}'
s+=f'{sep}{seg}'
range_start=seg
prev_seg=seg
returns

defpluralize(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).
 '''
ifcount==1:
returnword
else:
ifalt:
returnalt
else:
returnword+suffix

defcompare_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
fromrichimportprintasrprint

defprint(*args, **kwargs):
ifnotmute:
rprint(*args, **kwargs)

equal=True
type_changed=False

iftype(value_old) !=type(value):
type_changed=True
equal=False
elifisinstance(value, dict):
forkey, vinvalue.items():
ifkeynotinvalue_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)
forkey, vinvalue_old.items():
ifkeynotinvalue:
equal=False
print(f'{print_prefix}.{key}: [red]Removed:[/red]', v)
returnequal
elifisinstance(value, list):
foriinrange(min(len(value), len(value_old))):
equal&=compare_obj(value_old[i], value[i], print_prefix=f'{print_prefix}[{i}]', mute=mute)
iflen(value_old) <len(value):
equal=False
foriinrange(len(value_old), len(value)):
print(f'{print_prefix}[{i}]: [green]Added:[/green]', value[i])
eliflen(value_old) >len(value):
equal=False
foriinrange(len(value), len(value_old)):
print(f'{print_prefix}[{i}]: [red]Removed:[/red]',value_old[i])
returnequal

elifisinstance(value, str):
equal=re.sub(r'\s+',' ', value_old) ==re.sub(r'\s+',' ', value)
else:
equal=value_old==value
ifnotequal:
print(f'{print_prefix}:', end='')
s=str(value_old) +str(value)
iflen(s) <60andnot'\n'ins:
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)
returnequal

defnormalize(s):
'''Normalize a string by replacing full-width characters with half-width characters
 but keep the full-width dash (～) as it is.'''
returnunicodedata.normalize('NFKC', s.replace('～', '$dash$')).replace('$dash$', '～')

# ==================== datetime related ====================
defparse_to_shortdate(date_str, fmt=None):
# pip install python-dateutil
fromdateutilimportparser

current_year=datetime.now().year

iffmtisNone: fmt='%y%m%d'
ifisinstance(date_str, datetime):
returndate_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})']
forpatterninpatterns:
# 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.
ifm:=re.search(pattern, date_str) orre.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'] orcurrent_year}-{m['m']}-{m['d']}"
break
try:
returnparser.parse(date_str, yearfirst=True).strftime(fmt)
except:
return""


classMyTime:
def__init__(self, t=None):
importpytz

self.tz=pytz.timezone('Asia/Tokyo')

iftisNone:
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):
ifformat_type=='obj':
returndt
elifformat_type=="short":
returndt.strftime('%y%m%d_%H%M%S')
elifformat_type=="pretty":
returndt.strftime('%Y-%m-%d (%a) %H:%M:%S')
else:
raiseValueError("Invalid format_type. Choose from 'obj', 'short', 'pretty'.")

deflocal(self, format_type="obj"):
returnself._format_time(self.local_time, format_type)

defjst(self, format_type="obj"):
returnself._format_time(self.jst_time, format_type)

defnaive(self, format_type="obj"):
returnself._format_time(self.naive_time, format_type)

# deprecated, will be removed in the future
defget_current_time(now=None):
# pip install pytz
importpytz

tz=pytz.timezone('Asia/Tokyo')

ifnowisNone:
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
deftd_format(td_object_or_sec, long_form=True):
ifisinstance(td_object_or_sec, int) orisinstance(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=[]
forperiod_name, period_secondsinperiods:
ifseconds>period_seconds:
period_value , seconds=divmod(seconds, period_seconds)
iflong_form:
has_s='s'ifperiod_value>1else''
strings.append(f"{period_value}{period_name}{has_s}")
else:
strings.append(f"{period_value}{period_name[0:1]}")
iflong_form:
return", ".join(strings)
else:
return"".join(strings)

# deprecated, just use MyTime(dt).jst()
defto_jp_time(dt, input_timezone=None):
# pip install python-dateutil pytz
fromdateutilimportparser
frompytzimporttimezone

ifisinstance(dt, str):
dt=parser.parse(dt)
ifinput_timezone:
dt=timezone(input_timezone).localize(dt)
returndt.astimezone(timezone('asia/tokyo'))

# ==================== performance related ====================
deftic():
global_start_time
_start_time=time.time()

deftac(print=True):
global_start_time
t=time.time() -_start_time
ifprint:
builtins.print(f'Time passed: {t:.2f} s')
returnt

# a decorator to time a function
deftimeme(func):
defwrapper(*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")
returnresult
returnwrapper

# ==================== file related ====================
defdump_json(mydict, filename, **kwargs):
filename=Path(filename)
iffilename.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)
withfilename.open('w', encoding='utf-8') asf:
json.dump(mydict, f, **default_kwargs)

defload_json(filename, encoding='utf-8'):
filename=Path(filename)
withfilename.open('r', encoding=encoding) asf:
data=json.load(f)
returndata

defdump_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)
ifprettify:
text=soup.prettify()
else:
text=str(soup)
withfilename.open('w', encoding=encoding) asf:
f.write(text)

defdump_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=''
ifisinstance(data[0], dict):
headers=data[0].keys()
ifprint_header:
s+='\t'.join(str(h) forhinheaders) +'\n'
forrowindata:
s+='\t'.join([str(row.get(h, '')) forhinheaders]) +'\n'
else:
ifprint_header:
print('[W] No header provided. Use default header [0, 1, 2, ...]')
s+='\t'.join([str(i) foriinrange(len(data[0]))]) +'\n'
forrowindata:
s+='\t'.join([str(i) foriinrow]) +'\n'
ifverbose:
print(s)
filename=Path(filename)
filename.parent.mkdir(parents=True, exist_ok=True)
filename.write_text(s, encoding='utf8')


defget_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.
iffile_filterisNoneandpath_filterisNone:
defquick_scan(directory):
forentryinos.scandir(directory):
ifentry.is_file():
yieldentry
elifrecursiveandentry.is_dir(follow_symlinks=False):
yieldfromquick_scan(entry.path)
return [Path(f) forfinquick_scan(directory)]
# else, use pathlib.iterdir and just dynamically change the list. The speed is basically the same (slow).
file_list= []
forxindirectory.iterdir():
ifx.is_file():
ifnotfile_filterorfile_filter(x):
file_list.append(x)
elifrecursiveandx.is_dir() and (notpath_filterorpath_filter(x)):
file_list.extend(get_files(x, recursive=recursive, file_filter=file_filter, path_filter=path_filter))
returnfile_list

defremove_empty_folders(directory, remove_root=True): #Including root.
directory=Path(directory)
try:
assertdirectory.is_dir()
forxindirectory.iterdir():
ifx.is_dir():
remove_empty_folders(x, remove_root=True)
ifremove_rootandnotlist(directory.iterdir()):
directory.rmdir()
exceptPermissionErrorase:
print('Error:', e)

defensure_nonexist(f):
'''
 Ensure the file does not exist. If it does, rename it to filename_2, filename_3, etc.
 '''
i=2
stem=f.stem
ifm:=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.
ifint(m[2]) <10andlen(m[2]) ==1:
stem=m[1]
i=int(m[2]) +1
whilef.exists():
f=f.with_name(f'{stem}_{i}{f.suffix}')
i=i+1
returnf

defensure_path_exists(path):
'''
 ensure the path exists.
 '''
path=Path(path)
ifnotpath.exists():
raiseFileNotFoundError(f'Path {path} does not exist!')
returnpath

defquickmd5(f):
hasher=hashlib.md5()
file_size=f.stat().st_size
buffer_size=1024*1024# 1MB

withf.open('rb') asf:
iffile_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)

returnf"{file_size}_{hasher.hexdigest()}"# this way is more readable than just return the hexdigest.

defmove_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
 """
ifnotsrc.exists():
raiseFileNotFoundError(f"The source file {src} does not exist.")
ifsrc==dst:
raiseValueError(f"Source and destination are the same: {src}")
ifdst.exists():
ifhash_method(src) ==hash_method(dst):
print(f'[W] {src.name} is a duplicate. Remove.')
src.unlink()
return
else:
ifconflict=='skip':
print(f'[W] Destination file {dst} already exists and hash does not match. Skip.')
return
elifconflict=='error':
raiseFileExistsError(f"Destination file {dst} already exists and hash does not match.")
elifconflict=='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_
ifverbose:
ifsrc.parent==dst.parent:
print(f"Rename {src.name} to {dst.name}.")
elifsrc.name==dst.name:
print(f'Move {src.name} into {dst.parent}')
else:
print(f'Move {src.name} to {dst}')
shutil.move(src, dst)

defbatch_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= [fforf, _inrenamings]
dst_files= [f.with_name(name) forf, nameinrenamings]

iflen(set(files)) !=len(files):
print('[E] There are duplicate files in the list! Please check.')
return

iflen(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
fornew_findst_files:
ifnew_f.exists() andnew_fnotinfiles:
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= []
forf, nameinrenamings:
iffindst_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))

forf, nameinreal_renamings:
f.rename(f.with_name(name))

# ==================== network related ====================
defrequests_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
importrequests
fromrequests.adaptersimportHTTPAdapter
fromurllib3.util.retryimportRetry

session=sessionorrequests.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)
returnsession

defget(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
 """
frombs4importBeautifulSoup

ifnotsession:
session=requests_retry_session()
r=session.get(url, cookies=cookies, headers=headers, timeout=timeout)
ifencoding:
returnBeautifulSoup(r.content, parser, from_encoding=encoding)
else:
returnBeautifulSoup(r.content, parser)

defget_webname(url):
returnunquote(url.split('?')[0].split('/')[-1])

defload_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)]>
 """
fromhttp.cookiejarimportMozillaCookieJar
fromrequests.cookiesimportRequestsCookieJar, create_cookie
importbrowsercookie
# pip install browsercookie

defconvert(cj):
cookies=RequestsCookieJar()
forcookieincj:
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)
returncookies

ifm:=re.search(r'^(chrome|firefox|edge)(/.+)?', str(s), re.IGNORECASE):
domain_name=m[2].lstrip('/') ifm[2] elseNone
ifm[1].lower() =='chrome':
cj=browsercookie.chrome()
elifm[1].lower() =='firefox':
cj=browsercookie.firefox()
elifm[1].lower() =='edge':
cj=browsercookie.edge()
ifdomain_name:
cj= [cookieforcookieincjifdomain_nameincookie.domain]
returnconvert(cj)

ifPath(s).exists():
cj=MozillaCookieJar(s)
cj.load(ignore_expires=True, ignore_discard=True)
forcookieincj:
ifcookie.expires==0:
cookie.expires=int(time.time()+86400)
returnconvert(cj)

ifre.search(r'^(.+?):\s*(.+?)', str(s)):
cookies=RequestsCookieJar()
fork, vinre.findall(r'(.+?):\s*([^;]+?)(?:;|$)', s):
cookies.set(k, v)
returncookies

raiseValueError(f'Invalid cookie string: {s}')

defdownload(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.
 """
frommimetypesimportguess_extension
# it uses requests_retry_session, so
# pip install requests

ifdupenotin ['skip', 'overwrite', 'rename', 'skip_same_size']:
raiseValueError(f'[Error] Invalid dupe method: {dupe} (must be either skip, overwrite, rename or skip_same_size).')

defprint(s, verbose_level, only=False):
if (onlyandverbose==verbose_level) or (notonlyandverbose>=verbose_level):
builtins.print(s)

defhas_valid_suffix(f):
# common suffixes
iff.suffix.lower() in ['.jpg', '.png', '.gif', '.webp', '.jpeg', '.bmp', '.svg', '.ico', '.mp4', '.mkv', '.webm', '.heic', '.pdf']:
returnTrue
iff.suffix.lower() in ['.php', '']:
returnFalse
# if the suffix is too long, has a space, etc., we assume it is not a valid suffix
iflen(f.suffix.lower()) >5or' 'inf.suffix.lower():
returnFalse
returnTrue

defreplace_suffix(f, content_type):

importbuiltins
importjson
importos
importre
importsys
importtime
importhashlib
importshutil
importunicodedata
fromdatetimeimportdatetime
frompathlibimportPath
fromurllib.parseimportunquote

# 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. ====================
defto_list(a):
returnaifisinstance(a, list) oraisNoneelse [a]

defflatten(x):
fromcollections.abcimportIterable
ifisinstance(x, Iterable) andnotisinstance(x, str):
return [aforiinxforainflatten(i)]
else:
return [x]

defprint_cmd(cmd, prefix='', actually_print=True):
commands_text_form= [f'"{c}"'ifre.search(r'[ ?]', str(c)) elsestr(c) forcincmd]
output=prefix+' '.join(commands_text_form)
ifactually_print:
print(output)
returnoutput

defcopy(data):
# pip install pyperclip
importpyperclip
pyperclip.copy(data)

defget_clipboard_data():
# pip install pyperclip
importpyperclip
returnpyperclip.paste()

defsafeify(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.
 """

assertisinstance(name, str), f'Name must be a string, not {type(name)}'

template= {'\\': '＼', '/': '／', ':': '：', '*': '＊', '?': '？', '"': '＂', '<': '＜', '>': '＞', '|': '｜','\n':'','\r':'','\t':''}
ifignore_backslash:
template.pop('\\', None)

forillegalintemplate:
name=name.replace(illegal, template[illegal])
returnname

defformat_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
importwcwidth

ifsisNone:
s=''
else:
s=str(s)
ifnotwidth:
returns
output=''
length=0
forcharins:
size=wcwidth.wcswidth(char)
iflength+size>width:
break
output+=char
length+=size
ifalign=='left':
returnoutput+padding*(width-length)
ifalign=='right':
returnpadding*(width-length) +output
ifalign=='center':
left_space= (width-length)//2
right_space=width-length-left_space
returnpadding*left_space+output+padding*right_space

classTable():
# pip install wcwidth

def__init__(self, rows=None, headers=None, max_width=100) ->None:
ifrowsandnotheaders:
self.headers=rows[0]
self.data=rows[1:]
elifheaders:
self.headers=headers
ifrows:
self.data=rows
else:
self.data= []
else:
raiseException('No header or data given!')
self.max_width=max_width

defrows(self):
forrowinself.data:
row_dict= {}
fori, headerinenumerate(self.headers):
row_dict[header] =row[i]
yieldrow_dict

defsearch(self, keywords, single=True):
matched_rows= []
forrowinself.data:
matched=True
forkey, valueinkeywords.items():
assertkeyinself.headers
col_idx=self.headers.index(key)
ifnotvalue: # Ignore empty values
continue
ifnot ((str(row[col_idx]) ==str(value)) or (row[col_idx] ==value)):
matched=False
break
ifmatched:
matched_rows.append(row)
ifsingle:
break
ifmatched_rows:
ifsingleandlen(matched_rows) ==1:
returnmatched_rows[0]
else:
returnmatched_rows
else:
return []

defsort(self, header_name_or_index, reverse=False):
ifisinstance(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=lambdax: x[col_idx], reverse=reverse)

defappend(self, new_data):
d= [''] *len(self.headers)
forkey, valueinnew_data.items():
ifnotkeyinself.headers: # silently ignore keys that are not in headers
continue
col_idx=self.headers.index(key)
d[col_idx] =value
self.data.append(d)

defprint(self, formats=None, custom_print=None):
importwcwidth

deffmt(value, str_format):
try:
returnstr_format.format(value)
exceptException:
returnvalue

defcalc_width(col_format, idx):
maxw=sum(wcwidth.wcwidth(c) forcinstr(self.headers[idx]))
forrowinself.data:
ifidx>len(row) -1:
continue
w=sum(wcwidth.wcwidth(c) forcinfmt(row[idx], col_format['str_format']))
ifw>maxw:
maxw=w
maxw=min(maxw, col_format['max_width'])
col_format['width'] =maxw

defprint_row(row, header_mode=False, custom_print=custom_print):
parts= []
foridxinrange(len(row)):
col_format=col_formats[idx]
# override str format back to nothing for headers
str_format="{}"ifheader_modeelsecol_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)
ifcustom_print:
custom_print(line)
else:
print(line)

col_formats=dict()
foridxinrange(len(self.headers)):
col_format= {
"align": "left",
"str_format": "{}",
"max_width": self.max_width
 }
ifformats:
ifidxinformats:
col_format.update(formats[idx])
elifself.headers[idx] informats:
col_format.update(formats[self.headers[idx]])
ifnot'width'incol_format:
calc_width(col_format, idx)
col_formats[idx] =col_format

print_row(self.headers, header_mode=True)
forrowinself.data:
print_row(row)

# return some useful info
total_width=sum(col_format['width'] forcol_formatincol_formats.values()) +2* (len(col_formats) -1)
return {'total_width': total_width, 'col_formats': col_formats}

defsave(self, f):
s=''
f=Path(f)
s+='\t'.join(self.headers) +'\n'
forrowinself.data:
s+='\t'.join([str(cell) forcellinrow]) +'\n'
f.write_text(s, encoding='utf8')

defmulti_col_print(data, columns=5):
'''Print a list of data in a table with columns'''
# pip install rich
fromrich.consoleimportConsole
fromrich.tableimportTable

table=Table(show_header=False, box=None)
for_inrange(columns):
table.add_column(style='cyan')
# add data to table vertically
row_count=len(data) //columns+1
foriinrange(row_count):
row=data[i::row_count]
table.add_row(*row)
console=Console()
console.print(table)

defrprint(*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
fromrich.consoleimportConsole
ifnothasattr(rprint, '_console'):
rprint._console=Console()
rprint._console.print(*args, **kwargs)

defprint2(*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.
 '''
frombuiltinsimportprintasbuiltin_print
ifnothasattr(print2, '_newlined'):
print2._newlined=True
end=kwargs.pop('end', '\n')
ifend.endswith('\n') andnotprint2._newlined:
builtin_print()
print2._newlined=Trueifend.endswith('\n') elseFalse
builtin_print(*args, **kwargs, end=end)

defarray_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
foridx, seginenumerate(a):
ifidx==0: # first segment
s+=f'{seg}'
range_start=seg
else:
ifseg-prev_seg==1:
ifidx==len(a) -1: # last segment
s+=f'{dash}{seg}'
else:
ifprev_seg>range_start:
s+=f'{dash}{prev_seg}'
s+=f'{sep}{seg}'
range_start=seg
prev_seg=seg
returns

defpluralize(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).
 '''
ifcount==1:
returnword
else:
ifalt:
returnalt
else:
returnword+suffix

defcompare_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
fromrichimportprintasrprint

defprint(*args, **kwargs):
ifnotmute:
rprint(*args, **kwargs)

equal=True
type_changed=False

iftype(value_old) !=type(value):
type_changed=True
equal=False
elifisinstance(value, dict):
forkey, vinvalue.items():
ifkeynotinvalue_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)
forkey, vinvalue_old.items():
ifkeynotinvalue:
equal=False
print(f'{print_prefix}.{key}: [red]Removed:[/red]', v)
returnequal
elifisinstance(value, list):
foriinrange(min(len(value), len(value_old))):
equal&=compare_obj(value_old[i], value[i], print_prefix=f'{print_prefix}[{i}]', mute=mute)
iflen(value_old) <len(value):
equal=False
foriinrange(len(value_old), len(value)):
print(f'{print_prefix}[{i}]: [green]Added:[/green]', value[i])
eliflen(value_old) >len(value):
equal=False
foriinrange(len(value), len(value_old)):
print(f'{print_prefix}[{i}]: [red]Removed:[/red]',value_old[i])
returnequal

elifisinstance(value, str):
equal=re.sub(r'\s+',' ', value_old) ==re.sub(r'\s+',' ', value)
else:
equal=value_old==value
ifnotequal:
print(f'{print_prefix}:', end='')
s=str(value_old) +str(value)
iflen(s) <60andnot'\n'ins:
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)
returnequal

defnormalize(s):
'''Normalize a string by replacing full-width characters with half-width characters
 but keep the full-width dash (～) as it is.'''
returnunicodedata.normalize('NFKC', s.replace('～', '$dash$')).replace('$dash$', '～')

# ==================== datetime related ====================
defparse_to_shortdate(date_str, fmt=None):
# pip install python-dateutil
fromdateutilimportparser

current_year=datetime.now().year

iffmtisNone: fmt='%y%m%d'
ifisinstance(date_str, datetime):
returndate_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})']
forpatterninpatterns:
# 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.
ifm:=re.search(pattern, date_str) orre.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'] orcurrent_year}-{m['m']}-{m['d']}"
break
try:
returnparser.parse(date_str, yearfirst=True).strftime(fmt)
except:
return""


classMyTime:
def__init__(self, t=None):
importpytz

self.tz=pytz.timezone('Asia/Tokyo')

iftisNone:
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):
ifformat_type=='obj':
returndt
elifformat_type=="short":
returndt.strftime('%y%m%d_%H%M%S')
elifformat_type=="pretty":
returndt.strftime('%Y-%m-%d (%a) %H:%M:%S')
else:
raiseValueError("Invalid format_type. Choose from 'obj', 'short', 'pretty'.")

deflocal(self, format_type="obj"):
returnself._format_time(self.local_time, format_type)

defjst(self, format_type="obj"):
returnself._format_time(self.jst_time, format_type)

defnaive(self, format_type="obj"):
returnself._format_time(self.naive_time, format_type)

# deprecated, will be removed in the future
defget_current_time(now=None):
# pip install pytz
importpytz

tz=pytz.timezone('Asia/Tokyo')

ifnowisNone:
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
deftd_format(td_object_or_sec, long_form=True):
ifisinstance(td_object_or_sec, int) orisinstance(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=[]
forperiod_name, period_secondsinperiods:
ifseconds>period_seconds:
period_value , seconds=divmod(seconds, period_seconds)
iflong_form:
has_s='s'ifperiod_value>1else''
strings.append(f"{period_value}{period_name}{has_s}")
else:
strings.append(f"{period_value}{period_name[0:1]}")
iflong_form:
return", ".join(strings)
else:
return"".join(strings)

# deprecated, just use MyTime(dt).jst()
defto_jp_time(dt, input_timezone=None):
# pip install python-dateutil pytz
fromdateutilimportparser
frompytzimporttimezone

ifisinstance(dt, str):
dt=parser.parse(dt)
ifinput_timezone:
dt=timezone(input_timezone).localize(dt)
returndt.astimezone(timezone('asia/tokyo'))

# ==================== performance related ====================
deftic():
global_start_time
_start_time=time.time()

deftac(print=True):
global_start_time
t=time.time() -_start_time
ifprint:
builtins.print(f'Time passed: {t:.2f} s')
returnt

# a decorator to time a function
deftimeme(func):
defwrapper(*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")
returnresult
returnwrapper

# ==================== file related ====================
defdump_json(mydict, filename, **kwargs):
filename=Path(filename)
iffilename.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)
withfilename.open('w', encoding='utf-8') asf:
json.dump(mydict, f, **default_kwargs)

defload_json(filename, encoding='utf-8'):
filename=Path(filename)
withfilename.open('r', encoding=encoding) asf:
data=json.load(f)
returndata

defdump_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)
ifprettify:
text=soup.prettify()
else:
text=str(soup)
withfilename.open('w', encoding=encoding) asf:
f.write(text)

defdump_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=''
ifisinstance(data[0], dict):
headers=data[0].keys()
ifprint_header:
s+='\t'.join(str(h) forhinheaders) +'\n'
forrowindata:
s+='\t'.join([str(row.get(h, '')) forhinheaders]) +'\n'
else:
ifprint_header:
print('[W] No header provided. Use default header [0, 1, 2, ...]')
s+='\t'.join([str(i) foriinrange(len(data[0]))]) +'\n'
forrowindata:
s+='\t'.join([str(i) foriinrow]) +'\n'
ifverbose:
print(s)
filename=Path(filename)
filename.parent.mkdir(parents=True, exist_ok=True)
filename.write_text(s, encoding='utf8')


defget_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.
iffile_filterisNoneandpath_filterisNone:
defquick_scan(directory):
forentryinos.scandir(directory):
ifentry.is_file():
yieldentry
elifrecursiveandentry.is_dir(follow_symlinks=False):
yieldfromquick_scan(entry.path)
return [Path(f) forfinquick_scan(directory)]
# else, use pathlib.iterdir and just dynamically change the list. The speed is basically the same (slow).
file_list= []
forxindirectory.iterdir():
ifx.is_file():
ifnotfile_filterorfile_filter(x):
file_list.append(x)
elifrecursiveandx.is_dir() and (notpath_filterorpath_filter(x)):
file_list.extend(get_files(x, recursive=recursive, file_filter=file_filter, path_filter=path_filter))
returnfile_list

defremove_empty_folders(directory, remove_root=True): #Including root.
directory=Path(directory)
try:
assertdirectory.is_dir()
forxindirectory.iterdir():
ifx.is_dir():
remove_empty_folders(x, remove_root=True)
ifremove_rootandnotlist(directory.iterdir()):
directory.rmdir()
exceptPermissionErrorase:
print('Error:', e)

defensure_nonexist(f):
'''
 Ensure the file does not exist. If it does, rename it to filename_2, filename_3, etc.
 '''
i=2
stem=f.stem
ifm:=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.
ifint(m[2]) <10andlen(m[2]) ==1:
stem=m[1]
i=int(m[2]) +1
whilef.exists():
f=f.with_name(f'{stem}_{i}{f.suffix}')
i=i+1
returnf

defensure_path_exists(path):
'''
 ensure the path exists.
 '''
path=Path(path)
ifnotpath.exists():
raiseFileNotFoundError(f'Path {path} does not exist!')
returnpath

defquickmd5(f):
hasher=hashlib.md5()
file_size=f.stat().st_size
buffer_size=1024*1024# 1MB

withf.open('rb') asf:
iffile_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)

returnf"{file_size}_{hasher.hexdigest()}"# this way is more readable than just return the hexdigest.

defmove_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
 """
ifnotsrc.exists():
raiseFileNotFoundError(f"The source file {src} does not exist.")
ifsrc==dst:
raiseValueError(f"Source and destination are the same: {src}")
ifdst.exists():
ifhash_method(src) ==hash_method(dst):
print(f'[W] {src.name} is a duplicate. Remove.')
src.unlink()
return
else:
ifconflict=='skip':
print(f'[W] Destination file {dst} already exists and hash does not match. Skip.')
return
elifconflict=='error':
raiseFileExistsError(f"Destination file {dst} already exists and hash does not match.")
elifconflict=='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_
ifverbose:
ifsrc.parent==dst.parent:
print(f"Rename {src.name} to {dst.name}.")
elifsrc.name==dst.name:
print(f'Move {src.name} into {dst.parent}')
else:
print(f'Move {src.name} to {dst}')
shutil.move(src, dst)

defbatch_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= [fforf, _inrenamings]
dst_files= [f.with_name(name) forf, nameinrenamings]

iflen(set(files)) !=len(files):
print('[E] There are duplicate files in the list! Please check.')
return

iflen(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
fornew_findst_files:
ifnew_f.exists() andnew_fnotinfiles:
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= []
forf, nameinrenamings:
iffindst_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))

forf, nameinreal_renamings:
f.rename(f.with_name(name))

# ==================== network related ====================
defrequests_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
importrequests
fromrequests.adaptersimportHTTPAdapter
fromurllib3.util.retryimportRetry

session=sessionorrequests.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)
returnsession

defget(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
 """
frombs4importBeautifulSoup

ifnotsession:
session=requests_retry_session()
r=session.get(url, cookies=cookies, headers=headers, timeout=timeout)
ifencoding:
returnBeautifulSoup(r.content, parser, from_encoding=encoding)
else:
returnBeautifulSoup(r.content, parser)

defget_webname(url):
returnunquote(url.split('?')[0].split('/')[-1])

defload_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)]>
 """
fromhttp.cookiejarimportMozillaCookieJar
fromrequests.cookiesimportRequestsCookieJar, create_cookie
importbrowsercookie
# pip install browsercookie

defconvert(cj):
cookies=RequestsCookieJar()
forcookieincj:
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)
returncookies

ifm:=re.search(r'^(chrome|firefox|edge)(/.+)?', str(s), re.IGNORECASE):
domain_name=m[2].lstrip('/') ifm[2] elseNone
ifm[1].lower() =='chrome':
cj=browsercookie.chrome()
elifm[1].lower() =='firefox':
cj=browsercookie.firefox()
elifm[1].lower() =='edge':
cj=browsercookie.edge()
ifdomain_name:
cj= [cookieforcookieincjifdomain_nameincookie.domain]
returnconvert(cj)

ifPath(s).exists():
cj=MozillaCookieJar(s)
cj.load(ignore_expires=True, ignore_discard=True)
forcookieincj:
ifcookie.expires==0:
cookie.expires=int(time.time()+86400)
returnconvert(cj)

ifre.search(r'^(.+?):\s*(.+?)', str(s)):
cookies=RequestsCookieJar()
fork, vinre.findall(r'(.+?):\s*([^;]+?)(?:;|$)', s):
cookies.set(k, v)
returncookies

raiseValueError(f'Invalid cookie string: {s}')

defdownload(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.
 """
frommimetypesimportguess_extension
# it uses requests_retry_session, so
# pip install requests

ifdupenotin ['skip', 'overwrite', 'rename', 'skip_same_size']:
raiseValueError(f'[Error] Invalid dupe method: {dupe} (must be either skip, overwrite, rename or skip_same_size).')

defprint(s, verbose_level, only=False):
if (onlyandverbose==verbose_level) or (notonlyandverbose>=verbose_level):
builtins.print(s)

defhas_valid_suffix(f):
# common suffixes
iff.suffix.lower() in ['.jpg', '.png', '.gif', '.webp', '.jpeg', '.bmp', '.svg', '.ico', '.mp4', '.mkv', '.webm', '.heic', '.pdf']:
returnTrue
iff.suffix.lower() in ['.php', '']:
returnFalse
# if the suffix is too long, has a space, etc., we assume it is not a valid suffix
iflen(f.suffix.lower()) >5or' 'inf.suffix.lower():
returnFalse
returnTrue

defreplace_suffix(f, content_type):

importbuiltins
importjson
importos
importre
importsys
importtime
importhashlib
importshutil
importunicodedata
fromdatetimeimportdatetime
frompathlibimportPath
fromurllib.parseimportunquote

# 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. ====================
defto_list(a):
returnaifisinstance(a, list) oraisNoneelse [a]

defflatten(x):
fromcollections.abcimportIterable
ifisinstance(x, Iterable) andnotisinstance(x, str):
return [aforiinxforainflatten(i)]
else:
return [x]

defprint_cmd(cmd, prefix='', actually_print=True):
commands_text_form= [f'"{c}"'ifre.search(r'[ ?]', str(c)) elsestr(c) forcincmd]
output=prefix+' '.join(commands_text_form)
ifactually_print:
print(output)
returnoutput

defcopy(data):
# pip install pyperclip
importpyperclip
pyperclip.copy(data)

defget_clipboard_data():
# pip install pyperclip
importpyperclip
returnpyperclip.paste()

defsafeify(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.
 """

assertisinstance(name, str), f'Name must be a string, not {type(name)}'

template= {'\\': '＼', '/': '／', ':': '：', '*': '＊', '?': '？', '"': '＂', '<': '＜', '>': '＞', '|': '｜','\n':'','\r':'','\t':''}
ifignore_backslash:
template.pop('\\', None)

forillegalintemplate:
name=name.replace(illegal, template[illegal])
returnname

defformat_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
importwcwidth

ifsisNone:
s=''
else:
s=str(s)
ifnotwidth:
returns
output=''
length=0
forcharins:
size=wcwidth.wcswidth(char)
iflength+size>width:
break
output+=char
length+=size
ifalign=='left':
returnoutput+padding*(width-length)
ifalign=='right':
returnpadding*(width-length) +output
ifalign=='center':
left_space= (width-length)//2
right_space=width-length-left_space
returnpadding*left_space+output+padding*right_space

classTable():
# pip install wcwidth

def__init__(self, rows=None, headers=None, max_width=100) ->None:
ifrowsandnotheaders:
self.headers=rows[0]
self.data=rows[1:]
elifheaders:
self.headers=headers
ifrows:
self.data=rows
else:
self.data= []
else:
raiseException('No header or data given!')
self.max_width=max_width

defrows(self):
forrowinself.data:
row_dict= {}
fori, headerinenumerate(self.headers):
row_dict[header] =row[i]
yieldrow_dict

defsearch(self, keywords, single=True):
matched_rows= []
forrowinself.data:
matched=True
forkey, valueinkeywords.items():
assertkeyinself.headers
col_idx=self.headers.index(key)
ifnotvalue: # Ignore empty values
continue
ifnot ((str(row[col_idx]) ==str(value)) or (row[col_idx] ==value)):
matched=False
break
ifmatched:
matched_rows.append(row)
ifsingle:
break
ifmatched_rows:
ifsingleandlen(matched_rows) ==1:
returnmatched_rows[0]
else:
returnmatched_rows
else:
return []

defsort(self, header_name_or_index, reverse=False):
ifisinstance(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=lambdax: x[col_idx], reverse=reverse)

defappend(self, new_data):
d= [''] *len(self.headers)
forkey, valueinnew_data.items():
ifnotkeyinself.headers: # silently ignore keys that are not in headers
continue
col_idx=self.headers.index(key)
d[col_idx] =value
self.data.append(d)

defprint(self, formats=None, custom_print=None):
importwcwidth

deffmt(value, str_format):
try:
returnstr_format.format(value)
exceptException:
returnvalue

defcalc_width(col_format, idx):
maxw=sum(wcwidth.wcwidth(c) forcinstr(self.headers[idx]))
forrowinself.data:
ifidx>len(row) -1:
continue
w=sum(wcwidth.wcwidth(c) forcinfmt(row[idx], col_format['str_format']))
ifw>maxw:
maxw=w
maxw=min(maxw, col_format['max_width'])
col_format['width'] =maxw

defprint_row(row, header_mode=False, custom_print=custom_print):
parts= []
foridxinrange(len(row)):
col_format=col_formats[idx]
# override str format back to nothing for headers
str_format="{}"ifheader_modeelsecol_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)
ifcustom_print:
custom_print(line)
else:
print(line)

col_formats=dict()
foridxinrange(len(self.headers)):
col_format= {
"align": "left",
"str_format": "{}",
"max_width": self.max_width
 }
ifformats:
ifidxinformats:
col_format.update(formats[idx])
elifself.headers[idx] informats:
col_format.update(formats[self.headers[idx]])
ifnot'width'incol_format:
calc_width(col_format, idx)
col_formats[idx] =col_format

print_row(self.headers, header_mode=True)
forrowinself.data:
print_row(row)

# return some useful info
total_width=sum(col_format['width'] forcol_formatincol_formats.values()) +2* (len(col_formats) -1)
return {'total_width': total_width, 'col_formats': col_formats}

defsave(self, f):
s=''
f=Path(f)
s+='\t'.join(self.headers) +'\n'
forrowinself.data:
s+='\t'.join([str(cell) forcellinrow]) +'\n'
f.write_text(s, encoding='utf8')

defmulti_col_print(data, columns=5):
'''Print a list of data in a table with columns'''
# pip install rich
fromrich.consoleimportConsole
fromrich.tableimportTable

table=Table(show_header=False, box=None)
for_inrange(columns):
table.add_column(style='cyan')
# add data to table vertically
row_count=len(data) //columns+1
foriinrange(row_count):
row=data[i::row_count]
table.add_row(*row)
console=Console()
console.print(table)

defrprint(*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
fromrich.consoleimportConsole
ifnothasattr(rprint, '_console'):
rprint._console=Console()
rprint._console.print(*args, **kwargs)

defprint2(*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.
 '''
frombuiltinsimportprintasbuiltin_print
ifnothasattr(print2, '_newlined'):
print2._newlined=True
end=kwargs.pop('end', '\n')
ifend.endswith('\n') andnotprint2._newlined:
builtin_print()
print2._newlined=Trueifend.endswith('\n') elseFalse
builtin_print(*args, **kwargs, end=end)

defarray_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
foridx, seginenumerate(a):
ifidx==0: # first segment
s+=f'{seg}'
range_start=seg
else:
ifseg-prev_seg==1:
ifidx==len(a) -1: # last segment
s+=f'{dash}{seg}'
else:
ifprev_seg>range_start:
s+=f'{dash}{prev_seg}'
s+=f'{sep}{seg}'
range_start=seg
prev_seg=seg
returns

defpluralize(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).
 '''
ifcount==1:
returnword
else:
ifalt:
returnalt
else:
returnword+suffix

defcompare_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
fromrichimportprintasrprint

defprint(*args, **kwargs):
ifnotmute:
rprint(*args, **kwargs)

equal=True
type_changed=False

iftype(value_old) !=type(value):
type_changed=True
equal=False
elifisinstance(value, dict):
forkey, vinvalue.items():
ifkeynotinvalue_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)
forkey, vinvalue_old.items():
ifkeynotinvalue:
equal=False
print(f'{print_prefix}.{key}: [red]Removed:[/red]', v)
returnequal
elifisinstance(value, list):
foriinrange(min(len(value), len(value_old))):
equal&=compare_obj(value_old[i], value[i], print_prefix=f'{print_prefix}[{i}]', mute=mute)
iflen(value_old) <len(value):
equal=False
foriinrange(len(value_old), len(value)):
print(f'{print_prefix}[{i}]: [green]Added:[/green]', value[i])
eliflen(value_old) >len(value):
equal=False
foriinrange(len(value), len(value_old)):
print(f'{print_prefix}[{i}]: [red]Removed:[/red]',value_old[i])
returnequal

elifisinstance(value, str):
equal=re.sub(r'\s+',' ', value_old) ==re.sub(r'\s+',' ', value)
else:
equal=value_old==value
ifnotequal:
print(f'{print_prefix}:', end='')
s=str(value_old) +str(value)
iflen(s) <60andnot'\n'ins:
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)
returnequal

defnormalize(s):
'''Normalize a string by replacing full-width characters with half-width characters
 but keep the full-width dash (～) as it is.'''
returnunicodedata.normalize('NFKC', s.replace('～', '$dash$')).replace('$dash$', '～')

# ==================== datetime related ====================
defparse_to_shortdate(date_str, fmt=None):
# pip install python-dateutil
fromdateutilimportparser

current_year=datetime.now().year

iffmtisNone: fmt='%y%m%d'
ifisinstance(date_str, datetime):
returndate_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})']
forpatterninpatterns:
# 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.
ifm:=re.search(pattern, date_str) orre.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'] orcurrent_year}-{m['m']}-{m['d']}"
break
try:
returnparser.parse(date_str, yearfirst=True).strftime(fmt)
except:
return""


classMyTime:
def__init__(self, t=None):
importpytz

self.tz=pytz.timezone('Asia/Tokyo')

iftisNone:
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):
ifformat_type=='obj':
returndt
elifformat_type=="short":
returndt.strftime('%y%m%d_%H%M%S')
elifformat_type=="pretty":
returndt.strftime('%Y-%m-%d (%a) %H:%M:%S')
else:
raiseValueError("Invalid format_type. Choose from 'obj', 'short', 'pretty'.")

deflocal(self, format_type="obj"):
returnself._format_time(self.local_time, format_type)

defjst(self, format_type="obj"):
returnself._format_time(self.jst_time, format_type)

defnaive(self, format_type="obj"):
returnself._format_time(self.naive_time, format_type)

# deprecated, will be removed in the future
defget_current_time(now=None):
# pip install pytz
importpytz

tz=pytz.timezone('Asia/Tokyo')

ifnowisNone:
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
deftd_format(td_object_or_sec, long_form=True):
ifisinstance(td_object_or_sec, int) orisinstance(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=[]
forperiod_name, period_secondsinperiods:
ifseconds>period_seconds:
period_value , seconds=divmod(seconds, period_seconds)
iflong_form:
has_s='s'ifperiod_value>1else''
strings.append(f"{period_value}{period_name}{has_s}")
else:
strings.append(f"{period_value}{period_name[0:1]}")
iflong_form:
return", ".join(strings)
else:
return"".join(strings)

# deprecated, just use MyTime(dt).jst()
defto_jp_time(dt, input_timezone=None):
# pip install python-dateutil pytz
fromdateutilimportparser
frompytzimporttimezone

ifisinstance(dt, str):
dt=parser.parse(dt)
ifinput_timezone:
dt=timezone(input_timezone).localize(dt)
returndt.astimezone(timezone('asia/tokyo'))

# ==================== performance related ====================
deftic():
global_start_time
_start_time=time.time()

deftac(print=True):
global_start_time
t=time.time() -_start_time
ifprint:
builtins.print(f'Time passed: {t:.2f} s')
returnt

# a decorator to time a function
deftimeme(func):
defwrapper(*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")
returnresult
returnwrapper

# ==================== file related ====================
defdump_json(mydict, filename, **kwargs):
filename=Path(filename)
iffilename.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)
withfilename.open('w', encoding='utf-8') asf:
json.dump(mydict, f, **default_kwargs)

defload_json(filename, encoding='utf-8'):
filename=Path(filename)
withfilename.open('r', encoding=encoding) asf:
data=json.load(f)
returndata

defdump_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)
ifprettify:
text=soup.prettify()
else:
text=str(soup)
withfilename.open('w', encoding=encoding) asf:
f.write(text)

defdump_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=''
ifisinstance(data[0], dict):
headers=data[0].keys()
ifprint_header:
s+='\t'.join(str(h) forhinheaders) +'\n'
forrowindata:
s+='\t'.join([str(row.get(h, '')) forhinheaders]) +'\n'
else:
ifprint_header:
print('[W] No header provided. Use default header [0, 1, 2, ...]')
s+='\t'.join([str(i) foriinrange(len(data[0]))]) +'\n'
forrowindata:
s+='\t'.join([str(i) foriinrow]) +'\n'
ifverbose:
print(s)
filename=Path(filename)
filename.parent.mkdir(parents=True, exist_ok=True)
filename.write_text(s, encoding='utf8')


defget_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.
iffile_filterisNoneandpath_filterisNone:
defquick_scan(directory):
forentryinos.scandir(directory):
ifentry.is_file():
yieldentry
elifrecursiveandentry.is_dir(follow_symlinks=False):
yieldfromquick_scan(entry.path)
return [Path(f) forfinquick_scan(directory)]
# else, use pathlib.iterdir and just dynamically change the list. The speed is basically the same (slow).
file_list= []
forxindirectory.iterdir():
ifx.is_file():
ifnotfile_filterorfile_filter(x):
file_list.append(x)
elifrecursiveandx.is_dir() and (notpath_filterorpath_filter(x)):
file_list.extend(get_files(x, recursive=recursive, file_filter=file_filter, path_filter=path_filter))
returnfile_list

defremove_empty_folders(directory, remove_root=True): #Including root.
directory=Path(directory)
try:
assertdirectory.is_dir()
forxindirectory.iterdir():
ifx.is_dir():
remove_empty_folders(x, remove_root=True)
ifremove_rootandnotlist(directory.iterdir()):
directory.rmdir()
exceptPermissionErrorase:
print('Error:', e)

defensure_nonexist(f):
'''
 Ensure the file does not exist. If it does, rename it to filename_2, filename_3, etc.
 '''
i=2
stem=f.stem
ifm:=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.
ifint(m[2]) <10andlen(m[2]) ==1:
stem=m[1]
i=int(m[2]) +1
whilef.exists():
f=f.with_name(f'{stem}_{i}{f.suffix}')
i=i+1
returnf

defensure_path_exists(path):
'''
 ensure the path exists.
 '''
path=Path(path)
ifnotpath.exists():
raiseFileNotFoundError(f'Path {path} does not exist!')
returnpath

defquickmd5(f):
hasher=hashlib.md5()
file_size=f.stat().st_size
buffer_size=1024*1024# 1MB

withf.open('rb') asf:
iffile_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)

returnf"{file_size}_{hasher.hexdigest()}"# this way is more readable than just return the hexdigest.

defmove_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
 """
ifnotsrc.exists():
raiseFileNotFoundError(f"The source file {src} does not exist.")
ifsrc==dst:
raiseValueError(f"Source and destination are the same: {src}")
ifdst.exists():
ifhash_method(src) ==hash_method(dst):
print(f'[W] {src.name} is a duplicate. Remove.')
src.unlink()
return
else:
ifconflict=='skip':
print(f'[W] Destination file {dst} already exists and hash does not match. Skip.')
return
elifconflict=='error':
raiseFileExistsError(f"Destination file {dst} already exists and hash does not match.")
elifconflict=='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_
ifverbose:
ifsrc.parent==dst.parent:
print(f"Rename {src.name} to {dst.name}.")
elifsrc.name==dst.name:
print(f'Move {src.name} into {dst.parent}')
else:
print(f'Move {src.name} to {dst}')
shutil.move(src, dst)

defbatch_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= [fforf, _inrenamings]
dst_files= [f.with_name(name) forf, nameinrenamings]

iflen(set(files)) !=len(files):
print('[E] There are duplicate files in the list! Please check.')
return

iflen(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
fornew_findst_files:
ifnew_f.exists() andnew_fnotinfiles:
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= []
forf, nameinrenamings:
iffindst_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))

forf, nameinreal_renamings:
f.rename(f.with_name(name))

# ==================== network related ====================
defrequests_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
importrequests
fromrequests.adaptersimportHTTPAdapter
fromurllib3.util.retryimportRetry

session=sessionorrequests.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)
returnsession

defget(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
 """
frombs4importBeautifulSoup

ifnotsession:
session=requests_retry_session()
r=session.get(url, cookies=cookies, headers=headers, timeout=timeout)
ifencoding:
returnBeautifulSoup(r.content, parser, from_encoding=encoding)
else:
returnBeautifulSoup(r.content, parser)

defget_webname(url):
returnunquote(url.split('?')[0].split('/')[-1])

defload_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)]>
 """
fromhttp.cookiejarimportMozillaCookieJar
fromrequests.cookiesimportRequestsCookieJar, create_cookie
importbrowsercookie
# pip install browsercookie

defconvert(cj):
cookies=RequestsCookieJar()
forcookieincj:
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)
returncookies

ifm:=re.search(r'^(chrome|firefox|edge)(/.+)?', str(s), re.IGNORECASE):
domain_name=m[2].lstrip('/') ifm[2] elseNone
ifm[1].lower() =='chrome':
cj=browsercookie.chrome()
elifm[1].lower() =='firefox':
cj=browsercookie.firefox()
elifm[1].lower() =='edge':
cj=browsercookie.edge()
ifdomain_name:
cj= [cookieforcookieincjifdomain_nameincookie.domain]
returnconvert(cj)

ifPath(s).exists():
cj=MozillaCookieJar(s)
cj.load(ignore_expires=True, ignore_discard=True)
forcookieincj:
ifcookie.expires==0:
cookie.expires=int(time.time()+86400)
returnconvert(cj)

ifre.search(r'^(.+?):\s*(.+?)', str(s)):
cookies=RequestsCookieJar()
fork, vinre.findall(r'(.+?):\s*([^;]+?)(?:;|$)', s):
cookies.set(k, v)
returncookies

raiseValueError(f'Invalid cookie string: {s}')

defdownload(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.
 """
frommimetypesimportguess_extension
# it uses requests_retry_session, so
# pip install requests

ifdupenotin ['skip', 'overwrite', 'rename', 'skip_same_size']:
raiseValueError(f'[Error] Invalid dupe method: {dupe} (must be either skip, overwrite, rename or skip_same_size).')

defprint(s, verbose_level, only=False):
if (onlyandverbose==verbose_level) or (notonlyandverbose>=verbose_level):
builtins.print(s)

defhas_valid_suffix(f):
# common suffixes
iff.suffix.lower() in ['.jpg', '.png', '.gif', '.webp', '.jpeg', '.bmp', '.svg', '.ico', '.mp4', '.mkv', '.webm', '.heic', '.pdf']:
returnTrue
iff.suffix.lower() in ['.php', '']:
returnFalse
# if the suffix is too long, has a space, etc., we assume it is not a valid suffix
iflen(f.suffix.lower()) >5or' 'inf.suffix.lower():
returnFalse
returnTrue

defreplace_suffix(f, content_type):

importbuiltins
importjson
importos
importre
importsys
importtime
importhashlib
importshutil
importunicodedata
fromdatetimeimportdatetime
frompathlibimportPath
fromurllib.parseimportunquote

# 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. ====================
defto_list(a):
returnaifisinstance(a, list) oraisNoneelse [a]

defflatten(x):
fromcollections.abcimportIterable
ifisinstance(x, Iterable) andnotisinstance(x, str):
return [aforiinxforainflatten(i)]
else:
return [x]

defprint_cmd(cmd, prefix='', actually_print=True):
commands_text_form= [f'"{c}"'ifre.search(r'[ ?]', str(c)) elsestr(c) forcincmd]
output=prefix+' '.join(commands_text_form)
ifactually_print:
print(output)
returnoutput

defcopy(data):
# pip install pyperclip
importpyperclip
pyperclip.copy(data)

defget_clipboard_data():
# pip install pyperclip
importpyperclip
returnpyperclip.paste()

defsafeify(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.
 """

assertisinstance(name, str), f'Name must be a string, not {type(name)}'

template= {'\\': '＼', '/': '／', ':': '：', '*': '＊', '?': '？', '"': '＂', '<': '＜', '>': '＞', '|': '｜','\n':'','\r':'','\t':''}
ifignore_backslash:
template.pop('\\', None)

forillegalintemplate:
name=name.replace(illegal, template[illegal])
returnname

defformat_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
importwcwidth

ifsisNone:
s=''
else:
s=str(s)
ifnotwidth:
returns
output=''
length=0
forcharins:
size=wcwidth.wcswidth(char)
iflength+size>width:
break
output+=char
length+=size
ifalign=='left':
returnoutput+padding*(width-length)
ifalign=='right':
returnpadding*(width-length) +output
ifalign=='center':
left_space= (width-length)//2
right_space=width-length-left_space
returnpadding*left_space+output+padding*right_space

classTable():
# pip install wcwidth

def__init__(self, rows=None, headers=None, max_width=100) ->None:
ifrowsandnotheaders:
self.headers=rows[0]
self.data=rows[1:]
elifheaders:
self.headers=headers
ifrows:
self.data=rows
else:
self.data= []
else:
raiseException('No header or data given!')
self.max_width=max_width

defrows(self):
forrowinself.data:
row_dict= {}
fori, headerinenumerate(self.headers):
row_dict[header] =row[i]
yieldrow_dict

defsearch(self, keywords, single=True):
matched_rows= []
forrowinself.data:
matched=True
forkey, valueinkeywords.items():
assertkeyinself.headers
col_idx=self.headers.index(key)
ifnotvalue: # Ignore empty values
continue
ifnot ((str(row[col_idx]) ==str(value)) or (row[col_idx] ==value)):
matched=False
break
ifmatched:
matched_rows.append(row)
ifsingle:
break
ifmatched_rows:
ifsingleandlen(matched_rows) ==1:
returnmatched_rows[0]
else:
returnmatched_rows
else:
return []

defsort(self, header_name_or_index, reverse=False):
ifisinstance(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=lambdax: x[col_idx], reverse=reverse)

defappend(self, new_data):
d= [''] *len(self.headers)
forkey, valueinnew_data.items():
ifnotkeyinself.headers: # silently ignore keys that are not in headers
continue
col_idx=self.headers.index(key)
d[col_idx] =value
self.data.append(d)

defprint(self, formats=None, custom_print=None):
importwcwidth

deffmt(value, str_format):
try:
returnstr_format.format(value)
exceptException:
returnvalue

defcalc_width(col_format, idx):
maxw=sum(wcwidth.wcwidth(c) forcinstr(self.headers[idx]))
forrowinself.data:
ifidx>len(row) -1:
continue
w=sum(wcwidth.wcwidth(c) forcinfmt(row[idx], col_format['str_format']))
ifw>maxw:
maxw=w
maxw=min(maxw, col_format['max_width'])
col_format['width'] =maxw

defprint_row(row, header_mode=False, custom_print=custom_print):
parts= []
foridxinrange(len(row)):
col_format=col_formats[idx]
# override str format back to nothing for headers
str_format="{}"ifheader_modeelsecol_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)
ifcustom_print:
custom_print(line)
else:
print(line)

col_formats=dict()
foridxinrange(len(self.headers)):
col_format= {
"align": "left",
"str_format": "{}",
"max_width": self.max_width
 }
ifformats:
ifidxinformats:
col_format.update(formats[idx])
elifself.headers[idx] informats:
col_format.update(formats[self.headers[idx]])
ifnot'width'incol_format:
calc_width(col_format, idx)
col_formats[idx] =col_format

print_row(self.headers, header_mode=True)
forrowinself.data:
print_row(row)

# return some useful info
total_width=sum(col_format['width'] forcol_formatincol_formats.values()) +2* (len(col_formats) -1)
return {'total_width': total_width, 'col_formats': col_formats}

defsave(self, f):
s=''
f=Path(f)
s+='\t'.join(self.headers) +'\n'
forrowinself.data:
s+='\t'.join([str(cell) forcellinrow]) +'\n'
f.write_text(s, encoding='utf8')

defmulti_col_print(data, columns=5):
'''Print a list of data in a table with columns'''
# pip install rich
fromrich.consoleimportConsole
fromrich.tableimportTable

table=Table(show_header=False, box=None)
for_inrange(columns):
table.add_column(style='cyan')
# add data to table vertically
row_count=len(data) //columns+1
foriinrange(row_count):
row=data[i::row_count]
table.add_row(*row)
console=Console()
console.print(table)

defrprint(*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
fromrich.consoleimportConsole
ifnothasattr(rprint, '_console'):
rprint._console=Console()
rprint._console.print(*args, **kwargs)

defprint2(*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.
 '''
frombuiltinsimportprintasbuiltin_print
ifnothasattr(print2, '_newlined'):
print2._newlined=True
end=kwargs.pop('end', '\n')
ifend.endswith('\n') andnotprint2._newlined:
builtin_print()
print2._newlined=Trueifend.endswith('\n') elseFalse
builtin_print(*args, **kwargs, end=end)

defarray_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
foridx, seginenumerate(a):
ifidx==0: # first segment
s+=f'{seg}'
range_start=seg
else:
ifseg-prev_seg==1:
ifidx==len(a) -1: # last segment
s+=f'{dash}{seg}'
else:
ifprev_seg>range_start:
s+=f'{dash}{prev_seg}'
s+=f'{sep}{seg}'
range_start=seg
prev_seg=seg
returns

defpluralize(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).
 '''
ifcount==1:
returnword
else:
ifalt:
returnalt
else:
returnword+suffix

defcompare_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
fromrichimportprintasrprint

defprint(*args, **kwargs):
ifnotmute:
rprint(*args, **kwargs)

equal=True
type_changed=False

iftype(value_old) !=type(value):
type_changed=True
equal=False
elifisinstance(value, dict):
forkey, vinvalue.items():
ifkeynotinvalue_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)
forkey, vinvalue_old.items():
ifkeynotinvalue:
equal=False
print(f'{print_prefix}.{key}: [red]Removed:[/red]', v)
returnequal
elifisinstance(value, list):
foriinrange(min(len(value), len(value_old))):
equal&=compare_obj(value_old[i], value[i], print_prefix=f'{print_prefix}[{i}]', mute=mute)
iflen(value_old) <len(value):
equal=False
foriinrange(len(value_old), len(value)):
print(f'{print_prefix}[{i}]: [green]Added:[/green]', value[i])
eliflen(value_old) >len(value):
equal=False
foriinrange(len(value), len(value_old)):
print(f'{print_prefix}[{i}]: [red]Removed:[/red]',value_old[i])
returnequal

elifisinstance(value, str):
equal=re.sub(r'\s+',' ', value_old) ==re.sub(r'\s+',' ', value)
else:
equal=value_old==value
ifnotequal:
print(f'{print_prefix}:', end='')
s=str(value_old) +str(value)
iflen(s) <60andnot'\n'ins:
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)
returnequal

defnormalize(s):
'''Normalize a string by replacing full-width characters with half-width characters
 but keep the full-width dash (～) as it is.'''
returnunicodedata.normalize('NFKC', s.replace('～', '$dash$')).replace('$dash$', '～')

# ==================== datetime related ====================
defparse_to_shortdate(date_str, fmt=None):
# pip install python-dateutil
fromdateutilimportparser

current_year=datetime.now().year

iffmtisNone: fmt='%y%m%d'
ifisinstance(date_str, datetime):
returndate_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})']
forpatterninpatterns:
# 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.
ifm:=re.search(pattern, date_str) orre.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'] orcurrent_year}-{m['m']}-{m['d']}"
break
try:
returnparser.parse(date_str, yearfirst=True).strftime(fmt)
except:
return""


classMyTime:
def__init__(self, t=None):
importpytz

self.tz=pytz.timezone('Asia/Tokyo')

iftisNone:
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):
ifformat_type=='obj':
returndt
elifformat_type=="short":
returndt.strftime('%y%m%d_%H%M%S')
elifformat_type=="pretty":
returndt.strftime('%Y-%m-%d (%a) %H:%M:%S')
else:
raiseValueError("Invalid format_type. Choose from 'obj', 'short', 'pretty'.")

deflocal(self, format_type="obj"):
returnself._format_time(self.local_time, format_type)

defjst(self, format_type="obj"):
returnself._format_time(self.jst_time, format_type)

defnaive(self, format_type="obj"):
returnself._format_time(self.naive_time, format_type)

# deprecated, will be removed in the future
defget_current_time(now=None):
# pip install pytz
importpytz

tz=pytz.timezone('Asia/Tokyo')

ifnowisNone:
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
deftd_format(td_object_or_sec, long_form=True):
ifisinstance(td_object_or_sec, int) orisinstance(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=[]
forperiod_name, period_secondsinperiods:
ifseconds>period_seconds:
period_value , seconds=divmod(seconds, period_seconds)
iflong_form:
has_s='s'ifperiod_value>1else''
strings.append(f"{period_value}{period_name}{has_s}")
else:
strings.append(f"{period_value}{period_name[0:1]}")
iflong_form:
return", ".join(strings)
else:
return"".join(strings)

# deprecated, just use MyTime(dt).jst()
defto_jp_time(dt, input_timezone=None):
# pip install python-dateutil pytz
fromdateutilimportparser
frompytzimporttimezone

ifisinstance(dt, str):
dt=parser.parse(dt)
ifinput_timezone:
dt=timezone(input_timezone).localize(dt)
returndt.astimezone(timezone('asia/tokyo'))

# ==================== performance related ====================
deftic():
global_start_time
_start_time=time.time()

deftac(print=True):
global_start_time
t=time.time() -_start_time
ifprint:
builtins.print(f'Time passed: {t:.2f} s')
returnt

# a decorator to time a function
deftimeme(func):
defwrapper(*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")
returnresult
returnwrapper

# ==================== file related ====================
defdump_json(mydict, filename, **kwargs):
filename=Path(filename)
iffilename.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)
withfilename.open('w', encoding='utf-8') asf:
json.dump(mydict, f, **default_kwargs)

defload_json(filename, encoding='utf-8'):
filename=Path(filename)
withfilename.open('r', encoding=encoding) asf:
data=json.load(f)
returndata

defdump_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)
ifprettify:
text=soup.prettify()
else:
text=str(soup)
withfilename.open('w', encoding=encoding) asf:
f.write(text)

defdump_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=''
ifisinstance(data[0], dict):
headers=data[0].keys()
ifprint_header:
s+='\t'.join(str(h) forhinheaders) +'\n'
forrowindata:
s+='\t'.join([str(row.get(h, '')) forhinheaders]) +'\n'
else:
ifprint_header:
print('[W] No header provided. Use default header [0, 1, 2, ...]')
s+='\t'.join([str(i) foriinrange(len(data[0]))]) +'\n'
forrowindata:
s+='\t'.join([str(i) foriinrow]) +'\n'
ifverbose:
print(s)
filename=Path(filename)
filename.parent.mkdir(parents=True, exist_ok=True)
filename.write_text(s, encoding='utf8')


defget_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.
iffile_filterisNoneandpath_filterisNone:
defquick_scan(directory):
forentryinos.scandir(directory):
ifentry.is_file():
yieldentry
elifrecursiveandentry.is_dir(follow_symlinks=False):
yieldfromquick_scan(entry.path)
return [Path(f) forfinquick_scan(directory)]
# else, use pathlib.iterdir and just dynamically change the list. The speed is basically the same (slow).
file_list= []
forxindirectory.iterdir():
ifx.is_file():
ifnotfile_filterorfile_filter(x):
file_list.append(x)
elifrecursiveandx.is_dir() and (notpath_filterorpath_filter(x)):
file_list.extend(get_files(x, recursive=recursive, file_filter=file_filter, path_filter=path_filter))
returnfile_list

defremove_empty_folders(directory, remove_root=True): #Including root.
directory=Path(directory)
try:
assertdirectory.is_dir()
forxindirectory.iterdir():
ifx.is_dir():
remove_empty_folders(x, remove_root=True)
ifremove_rootandnotlist(directory.iterdir()):
directory.rmdir()
exceptPermissionErrorase:
print('Error:', e)

defensure_nonexist(f):
'''
 Ensure the file does not exist. If it does, rename it to filename_2, filename_3, etc.
 '''
i=2
stem=f.stem
ifm:=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.
ifint(m[2]) <10andlen(m[2]) ==1:
stem=m[1]
i=int(m[2]) +1
whilef.exists():
f=f.with_name(f'{stem}_{i}{f.suffix}')
i=i+1
returnf

defensure_path_exists(path):
'''
 ensure the path exists.
 '''
path=Path(path)
ifnotpath.exists():
raiseFileNotFoundError(f'Path {path} does not exist!')
returnpath

defquickmd5(f):
hasher=hashlib.md5()
file_size=f.stat().st_size
buffer_size=1024*1024# 1MB

withf.open('rb') asf:
iffile_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)

returnf"{file_size}_{hasher.hexdigest()}"# this way is more readable than just return the hexdigest.

defmove_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
 """
ifnotsrc.exists():
raiseFileNotFoundError(f"The source file {src} does not exist.")
ifsrc==dst:
raiseValueError(f"Source and destination are the same: {src}")
ifdst.exists():
ifhash_method(src) ==hash_method(dst):
print(f'[W] {src.name} is a duplicate. Remove.')
src.unlink()
return
else:
ifconflict=='skip':
print(f'[W] Destination file {dst} already exists and hash does not match. Skip.')
return
elifconflict=='error':
raiseFileExistsError(f"Destination file {dst} already exists and hash does not match.")
elifconflict=='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_
ifverbose:
ifsrc.parent==dst.parent:
print(f"Rename {src.name} to {dst.name}.")
elifsrc.name==dst.name:
print(f'Move {src.name} into {dst.parent}')
else:
print(f'Move {src.name} to {dst}')
shutil.move(src, dst)

defbatch_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= [fforf, _inrenamings]
dst_files= [f.with_name(name) forf, nameinrenamings]

iflen(set(files)) !=len(files):
print('[E] There are duplicate files in the list! Please check.')
return

iflen(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
fornew_findst_files:
ifnew_f.exists() andnew_fnotinfiles:
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= []
forf, nameinrenamings:
iffindst_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))

forf, nameinreal_renamings:
f.rename(f.with_name(name))

# ==================== network related ====================
defrequests_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
importrequests
fromrequests.adaptersimportHTTPAdapter
fromurllib3.util.retryimportRetry

session=sessionorrequests.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)
returnsession

defget(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
 """
frombs4importBeautifulSoup

ifnotsession:
session=requests_retry_session()
r=session.get(url, cookies=cookies, headers=headers, timeout=timeout)
ifencoding:
returnBeautifulSoup(r.content, parser, from_encoding=encoding)
else:
returnBeautifulSoup(r.content, parser)

defget_webname(url):
returnunquote(url.split('?')[0].split('/')[-1])

defload_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)]>
 """
fromhttp.cookiejarimportMozillaCookieJar
fromrequests.cookiesimportRequestsCookieJar, create_cookie
importbrowsercookie
# pip install browsercookie

defconvert(cj):
cookies=RequestsCookieJar()
forcookieincj:
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)
returncookies

ifm:=re.search(r'^(chrome|firefox|edge)(/.+)?', str(s), re.IGNORECASE):
domain_name=m[2].lstrip('/') ifm[2] elseNone
ifm[1].lower() =='chrome':
cj=browsercookie.chrome()
elifm[1].lower() =='firefox':
cj=browsercookie.firefox()
elifm[1].lower() =='edge':
cj=browsercookie.edge()
ifdomain_name:
cj= [cookieforcookieincjifdomain_nameincookie.domain]
returnconvert(cj)

ifPath(s).exists():
cj=MozillaCookieJar(s)
cj.load(ignore_expires=True, ignore_discard=True)
forcookieincj:
ifcookie.expires==0:
cookie.expires=int(time.time()+86400)
returnconvert(cj)

ifre.search(r'^(.+?):\s*(.+?)', str(s)):
cookies=RequestsCookieJar()
fork, vinre.findall(r'(.+?):\s*([^;]+?)(?:;|$)', s):
cookies.set(k, v)
returncookies

raiseValueError(f'Invalid cookie string: {s}')

defdownload(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.
 """
frommimetypesimportguess_extension
# it uses requests_retry_session, so
# pip install requests

ifdupenotin ['skip', 'overwrite', 'rename', 'skip_same_size']:
raiseValueError(f'[Error] Invalid dupe method: {dupe} (must be either skip, overwrite, rename or skip_same_size).')

defprint(s, verbose_level, only=False):
if (onlyandverbose==verbose_level) or (notonlyandverbose>=verbose_level):
builtins.print(s)

defhas_valid_suffix(f):
# common suffixes
iff.suffix.lower() in ['.jpg', '.png', '.gif', '.webp', '.jpeg', '.bmp', '.svg', '.ico', '.mp4', '.mkv', '.webm', '.heic', '.pdf']:
returnTrue
iff.suffix.lower() in ['.php', '']:
returnFalse
# if the suffix is too long, has a space, etc., we assume it is not a valid suffix
iflen(f.suffix.lower()) >5or' 'inf.suffix.lower():
returnFalse
returnTrue

defreplace_suffix(f, content_type):