Source code for hf3fs.fuse

import os
from pathlib import PosixPath

from hf3fs_pyclient import HF3FS_SUPER_MAGIC

HF3FS_IOC_GET_MOUNT_NAME = 2149607424
HF3FS_IOC_GET_PATH_OFFSET = 2147772417
HF3FS_IOC_GET_MAGIC_NUM = 2147772418

HF3FS_IOC_RECURSIVE_RM = 2147772426

def serverPath(p):
    '''
    从完整路径获取 client 接受的路径名

    Args:
        p: 待解析的路径名

    Examples:

    .. code-block:: python

        import hf3fs.fuse
        hf3fs.fuse.serverPath('/hf3fs-jd/aaa/../cpu/abc/def')
    '''
    np = os.path.normpath(os.path.abspath(p))
    return os.path.join('/', *PosixPath(np).parts[3:])

def mountName(p):
    '''
    从完整路径获取 mount name

    Args:
        p: 待解析的路径名

    Examples:
    
    .. code-block:: python
    
        import hf3fs.fuse
        hf3fs.fuse.mountName('/hf3fs-jd/aaa/../cpu/abc/def')
    '''
    np = os.path.normpath(os.path.abspath(p))
    return PosixPath(np).parts[2]

def checkOnHf3fs(p):
    '''
    判断路径是否在 hf3fs 上

    Args:
        p: 待解析的路径名

    Examples:
    
    .. code-block:: python
    
        import hf3fs.fuse
        hf3fs.fuse.checkOnHf3fs('/hf3fs-jd/aaa/../cpu/abc/def')
    '''
    np = os.path.normpath(os.path.abspath(p))
    parts = PosixPath(np).parts
    if len(parts) >= 3 and parts[1] == 'hf3fs-jd':
        return True
    else:
        return False

def getToken():
    '''
    从个人 weka 目录下获取 hf3fs token，token 存放在 /weka-jd/prod/private/$USER/hf3fs.token
    '''
    import getpass
    user = getpass.getuser()
    default_path = f'/weka-jd/prod/private/{user}/hf3fs.token'
    path = os.getenv('HF3FS_TOKEN_FILE', default_path)
    path = path.replace("${USER}", user)
    with open(path) as f:
        token = f.readline().strip()
    return token