Spaces:
Paused
Paused
| #!/usr/bin/python3 | |
| # -*- coding: utf-8 -*- | |
| """ | |
| https://pypi.org/project/bypy/ | |
| https://github.com/houtianze/bypy | |
| """ | |
| import argparse | |
| import json | |
| import logging | |
| import shutil | |
| import tempfile | |
| from pathlib import Path | |
| import bypy | |
| from bypy import const | |
| from project_settings import project_path, environment | |
| from toolbox.design_patterns.singleton import ParamsSingleton | |
| logger = logging.getLogger("toolbox") | |
| class BaiduNetdiskClient(ParamsSingleton): | |
| def __init__(self, configdir: str = None): | |
| if not self._initialized: | |
| self.credentials = None | |
| if configdir is None: | |
| configdir = Path(tempfile.gettempdir()) / "baidu_netdisk/configdir" | |
| if configdir.exists(): | |
| shutil.rmtree(configdir.as_posix()) | |
| self.configdir = Path(configdir) | |
| self.token_path = self.configdir / const.TokenFileName | |
| self._bypy_client: bypy.ByPy = None | |
| self.bypy_login() | |
| self._initialized = True | |
| def bypy_client(self): | |
| if self._bypy_client is None: | |
| raise AssertionError(f"bypy not login yet!") | |
| return self._bypy_client | |
| def check_login(self): | |
| if self._bypy_client is None: | |
| return False | |
| status_code = self._bypy_client.info() | |
| if status_code == 0: | |
| flag = True | |
| else: | |
| flag = False | |
| return flag | |
| def make_bypy_login_config(self): | |
| if self.configdir.exists(): | |
| shutil.rmtree(self.configdir.as_posix()) | |
| if self._bypy_client is None: | |
| logger.info(f"login by configdir: {self.configdir.as_posix()}") | |
| self._bypy_client = bypy.ByPy( | |
| configdir=self.configdir.as_posix(), | |
| # debug=1, | |
| # verbose=1, | |
| ) | |
| # print tokens | |
| with open(self.token_path.as_posix(), "r", encoding="utf-8") as f: | |
| js = json.load(f) | |
| js = json.dumps(js, ensure_ascii=False) | |
| print(js) | |
| return None | |
| def bypy_login(self): | |
| if not self.token_path.exists(): | |
| return False | |
| # print tokens | |
| # with open(self.token_path.as_posix(), "r", encoding="utf-8") as f: | |
| # js = json.load(f) | |
| # js = json.dumps(js, ensure_ascii=False) | |
| # print(js) | |
| # exit(0) | |
| if self._bypy_client is None: | |
| logger.info(f"login by configdir: {self.configdir.as_posix()}") | |
| self._bypy_client = bypy.ByPy( | |
| configdir=self.configdir.as_posix(), | |
| # debug=1, | |
| # verbose=1, | |
| ) | |
| self._bypy_client.info() | |
| return None | |
| def login_with_credentials_info(self, credentials_info: dict): | |
| self.credentials = credentials_info | |
| self.token_path.parent.mkdir(parents=True, exist_ok=True) | |
| with open(self.token_path.as_posix(), "w", encoding="utf-8") as f: | |
| json.dump(credentials_info, f, ensure_ascii=False, indent=4) | |
| self.bypy_login() | |
| return True | |
| def get_args(): | |
| parser = argparse.ArgumentParser() | |
| parser.add_argument( | |
| "--key_of_credentials", | |
| default="baidu_netdisk_honeytian_credentials", | |
| type=str, | |
| ) | |
| parser.add_argument( | |
| "--config_dir", | |
| default=(project_path / "dotenv/baidu_netdisk/config_dir").as_posix(), | |
| type=str, | |
| ) | |
| args = parser.parse_args() | |
| return args | |
| def main(): | |
| args = get_args() | |
| import log | |
| from project_settings import environment, project_path, log_directory, time_zone_info | |
| log.setup_size_rotating(log_directory=log_directory, tz_info=time_zone_info) | |
| client = BaiduNetdiskClient( | |
| configdir=args.config_dir | |
| ) | |
| client.make_bypy_login_config() | |
| flag = client.check_login() | |
| print(f"flag: {flag}") | |
| # credentials_info = environment.get(key=args.key_of_credentials, dtype=json.loads) | |
| # client.login_with_credentials_info(credentials_info=credentials_info) | |
| # flag = client.check_login() | |
| # print(f"flag: {flag}") | |
| return | |
| if __name__ == "__main__": | |
| main() | |