jprdonnelly/twitter-api-client
1
0
Fork 0
mirror of https://github.com/trevorhobenshield/twitter-api-client.git synced 2025-12-25 02:03:15 -05:00
Code Issues Packages Projects Releases Wiki Activity

refactoring

Browse Source
This commit is contained in:
trevor hobenshield
2023-06-29 09:11:08 -07:00
parent 3469ba25a2
commit 9c7b045f83
6 changed files with 45 additions and 51 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
scripts/update.py
View File

@@ -36,7 +36,7 @@ JS_FILES_MAP = Path('js.json')
JS_FILES = Path('js')
OPERATIONS = Path('operations')
JS_FILES.mkdir(exist_ok=True, parents=True)
logging.config.dictConfig(LOGGER_CONFIG)
logging.config.dictConfig(LOG_CONFIG)
logger = logging.getLogger('twitter')

2
twitter/__version__.py
View File

@@ -1,5 +1,5 @@
__title__ = "twitter-api-client"
__description__ = "Implementation of Twitter's v1, v2, and GraphQL APIs."
__version__ = "0.10.2"
__version__ = "0.10.3"
__author__ = "Trevor Hobenshield"
__license__ = "MIT"

28
twitter/account.py
View File

@@ -38,12 +38,12 @@ if platform.system() != 'Windows':
class Account:
def __init__(self, email: str = None, username: str = None, password: str = None, session: Client = None, **kwargs):
self.logger = self._init_logger(kwargs.get('log_config', False))
self.session = self._validate_session(email, username, password, session, **kwargs)
self.save = kwargs.get('save', True)
self.debug = kwargs.get('debug', 0)
self.gql_api = 'https://twitter.com/i/api/graphql'
self.v1_api = 'https://api.twitter.com/1.1'
self.logger = self._init_logger(**kwargs)
self.session = self._validate_session(email, username, password, session, **kwargs)
def gql(self, method: str, operation: tuple, variables: dict, features: dict = Operation.default_features) -> dict:
qid, op = operation
@@ -592,22 +592,20 @@ class Account:
r = self.session.post(url, headers=get_headers(self.session), json=params)
return r
@staticmethod
def _init_logger(cfg: dict) -> Logger:
if cfg:
logging.config.dictConfig(cfg)
else:
logging.config.dictConfig(LOGGER_CONFIG)
def _init_logger(self, **kwargs) -> Logger:
if kwargs.get('debug'):
cfg = kwargs.get('log_config')
logging.config.dictConfig(cfg or LOG_CONFIG)
# only support one logger
logger_name = list(LOGGER_CONFIG['loggers'].keys())[0]
# only support one logger
logger_name = list(LOG_CONFIG['loggers'].keys())[0]
# set level of all other loggers to ERROR
for name in logging.root.manager.loggerDict:
if name != logger_name:
logging.getLogger(name).setLevel(logging.ERROR)
# set level of all other loggers to ERROR
for name in logging.root.manager.loggerDict:
if name != logger_name:
logging.getLogger(name).setLevel(logging.ERROR)
return logging.getLogger(logger_name)
return logging.getLogger(logger_name)
@staticmethod
def _validate_session(*args, **kwargs):

2
twitter/constants.py
View File

@@ -19,7 +19,7 @@ WHITE = '\x1b[37m'
BOLD = '\x1b[1m'
RESET = '\x1b[0m'
LOGGER_CONFIG = {
LOG_CONFIG = {
'version': 1,
'disable_existing_loggers': False,
'formatters': {

32
twitter/scraper.py
View File

@@ -32,13 +32,13 @@ if platform.system() != 'Windows':
class Scraper:
def __init__(self, email: str = None, username: str = None, password: str = None, session: Client = None, **kwargs):
self.guest = False
self.logger = self._init_logger(kwargs.get('log_config', False))
self.session = self._validate_session(email, username, password, session, **kwargs)
self.debug = kwargs.get('debug', 0)
self.save = kwargs.get('save', True)
self.debug = kwargs.get('debug', 0)
self.pbar = kwargs.get('pbar', True)
self.out_path = Path('data')
self.guest = False
self.logger = self._init_logger(**kwargs)
self.session = self._validate_session(email, username, password, session, **kwargs)
def users(self, screen_names: list[str], **kwargs) -> list[dict]:
"""
@@ -770,22 +770,20 @@ class Scraper:
spaces = self.spaces(rooms=rooms)
return asyncio.run(process(spaces))
@staticmethod
def _init_logger(cfg: dict) -> Logger:
if cfg:
logging.config.dictConfig(cfg)
else:
logging.config.dictConfig(LOGGER_CONFIG)
def _init_logger(self, **kwargs) -> Logger:
if kwargs.get('debug'):
cfg = kwargs.get('log_config')
logging.config.dictConfig(cfg or LOG_CONFIG)
# only support one logger
logger_name = list(LOGGER_CONFIG['loggers'].keys())[0]
# only support one logger
logger_name = list(LOG_CONFIG['loggers'].keys())[0]
# set level of all other loggers to ERROR
for name in logging.root.manager.loggerDict:
if name != logger_name:
logging.getLogger(name).setLevel(logging.ERROR)
# set level of all other loggers to ERROR
for name in logging.root.manager.loggerDict:
if name != logger_name:
logging.getLogger(name).setLevel(logging.ERROR)
return logging.getLogger(logger_name)
return logging.getLogger(logger_name)
def _validate_session(self, *args, **kwargs):
email, username, password, session = args

30
twitter/search.py
View File

@@ -36,11 +36,11 @@ if platform.system() != 'Windows':
class Search:
def __init__(self, email: str = None, username: str = None, password: str = None, session: Client = None, **kwargs):
self.logger = self._init_logger(kwargs.get('log_config', False))
self.session = self._validate_session(email, username, password, session, **kwargs)
self.api = 'https://api.twitter.com/2/search/adaptive.json?'
self.save = kwargs.get('save', True)
self.debug = kwargs.get('debug', 0)
self.api = 'https://api.twitter.com/2/search/adaptive.json?'
self.logger = self._init_logger(**kwargs)
self.session = self._validate_session(email, username, password, session, **kwargs)
def run(self, *args, out: str = 'data', **kwargs):
out_path = self.make_output_dirs(out)
@@ -134,22 +134,20 @@ class Search:
(p / 'final').mkdir(parents=True, exist_ok=True)
return p
@staticmethod
def _init_logger(cfg: dict) -> Logger:
if cfg:
logging.config.dictConfig(cfg)
else:
logging.config.dictConfig(LOGGER_CONFIG)
def _init_logger(self, **kwargs) -> Logger:
if kwargs.get('debug'):
cfg = kwargs.get('log_config')
logging.config.dictConfig(cfg or LOG_CONFIG)
# only support one logger
logger_name = list(LOGGER_CONFIG['loggers'].keys())[0]
# only support one logger
logger_name = list(LOG_CONFIG['loggers'].keys())[0]
# set level of all other loggers to ERROR
for name in logging.root.manager.loggerDict:
if name != logger_name:
logging.getLogger(name).setLevel(logging.ERROR)
# set level of all other loggers to ERROR
for name in logging.root.manager.loggerDict:
if name != logger_name:
logging.getLogger(name).setLevel(logging.ERROR)
return logging.getLogger(logger_name)
return logging.getLogger(logger_name)
@staticmethod
def _validate_session(*args, **kwargs):