video_decode/utils/general.py

import os
import re
from datetime import datetime, timedelta
from pathlib import Path
from random import randint
from typing import Any, List, Literal, Optional, Tuple


_STR_DIGITAL_MAP = {
    'k': 1e3,
    '千': 1e3,
    'w': 1e4,
    '万': 1e4,
    '亿': 1e8,
}

def get_root_dir() -> str:
    """获取项目根目录的绝对路径"""
    current_path = Path(__file__).resolve()
    root_path = current_path.parent if not current_path.is_dir() else current_path
    while True:
        if 'requirements.txt' in os.listdir(root_path):
            return str(root_path.absolute())
        root_path = root_path.parent


def get_abs_path(relative_path: str) -> str:
    """获取目标文件或目录的相对路径在系统中的绝对路径"""
    return os.path.join(get_root_dir(), relative_path)




def get_now_ts(length: Literal[10, 13] = 13) -> int:
    """获取当前时间的10位或13位时间戳"""
    now_ts = datetime.now().timestamp()
    if length == 10:
        return int(now_ts)
    else:
        return int(now_ts * 1000)


def pascal_to_snake(pascal_str: str) -> str:
    """将Pascal字符串转为蛇形字符串"""
    snake_str = re.sub(r'([a-z])([A-Z])', r'\1_\2', pascal_str)
    return snake_str.lower()


def snake_to_pascal(snake_str: str) -> str:
    """将蛇形字符串转为Pascal字符串"""
    return ''.join([item.title() for item in snake_str.split('_')])


def generate_task_id():
    """生成task_id"""
    now = datetime.now().strftime('%Y%m%d%H%M%S%f')[:-3]
    return f'{now}{randint(100000, 999999)}'


def get_x_day_ago_timestamp(x_days_ago: int = 0, length: Literal[13, 10] = 13):
    now = datetime.now()
    yesterday = now - timedelta(days=x_days_ago)
    yesterday_midnight = yesterday.replace(hour=0, minute=0, second=0, microsecond=0)
    if length == 10:
        return int(yesterday_midnight.timestamp())
    else:
        return int(yesterday_midnight.timestamp() * 1000)


def get_file_size(file_path):
    with open(file_path, 'rb') as f:
        return len(f.read())


def cookie_str_to_json(cookie: str) -> dict:
    components = cookie.strip().split(';')
    short_parsed_request = {component.split('=')[0].strip(): component.split('=')[1].strip() for component in components if
                            '=' in component}
    return short_parsed_request

def get_day_timestamp(day):
    now = datetime.now()
    yesterday = now - timedelta(days=day)
    yesterday_midnight = yesterday.replace(hour=0, minute=0, second=0, microsecond=0)
    timestamp_yesterday_midnight_ms = int(yesterday_midnight.timestamp())

    return timestamp_yesterday_midnight_ms


def str_digital_to_int(v: str) -> int:
    """将带有+、k、w字符的数据类字符串转为int"""
    v = v.rstrip('+')
    unit = v[-1]
    if unit in _STR_DIGITAL_MAP:
        v = float(v.rstrip(unit)) * _STR_DIGITAL_MAP[unit]
    else:
        v = float(v)
    return int(v)

def datetime_to_timestamp(time_str:str):
    dt = datetime.strptime(time_str, "%Y-%m-%d %H:%M:%S")
    # 转换为时间戳
    timestamp = int(dt.timestamp())*1000
    return timestamp

if __name__ == '__main__':
    print(datetime_to_timestamp("2024-10-30 19:05:56"))