from __future__ import annotations
import logging
import re
from collections import deque
from contextlib import ExitStack
from datetime import datetime
from pathlib import Path
from typing import List, Type, Any, Dict, Callable, Union
import dateparser
import dpath.util
from dateutil import parser
from fuzzywuzzy import process
from http_requester.requester import Requester
from pydantic import HttpUrl, BaseModel, Extra, root_validator, validator
from .creds import get_sharefile_credentials
from .helpers import to_snake, extract_attributes, get_key, to_pascal
from .models import ListModel, Collection
this = Path(__file__)
logger = logging.getLogger(f"logger.{this.stem}")
__all__ = [
'Event',
'MainClass',
'File',
'Folder',
'FavoriteFolder',
'ParentFolder',
'TemplateFoler',
'ProductionFolder'
]
sf_creds = get_sharefile_credentials()
BASE_URL = 'https://welocalize.sf-api.com/sf/v3'
SF_REQUESTER = Requester(
BASE_URL,
creds=sf_creds
)
class MetaConfig(BaseModel):
class Config:
arbitrary_types_allowed = True
extra = Extra.allow
alias_generator = to_pascal
allow_population_by_field_name = True
class ConfigModel(MetaConfig):
keys: List[str] = ['name', 'id']
def __repr__(self):
return f"{self.__class__.__name__}"
def __str__(self):
return repr(self)
def __getitem__(self, key):
return self.__getattribute__(to_snake(key))
def __getattr__(self, key):
return self.__getattribute__(to_snake(key))
def __setattr__(self, key, value):
return object.__setattr__(self, to_snake(key), value)
@root_validator
def validate_attributes(cls, values):
logger.debug(f"BEFORE [ConfigModel] [{cls.__name__}] root_validator {values=}")
if values.get('attributes') is None:
attributes = extract_attributes(values)
values.update(
{
'attributes': attributes
}
)
for key, value in values.items():
if klasses := value:
if isinstance(klasses, list) and all(isinstance(klass, BaseModel) for klass in klasses):
logger.debug(f"[ConfigModel] [{cls.__name__}] {key=} {klasses=}")
klass = klasses[0].__class__
values.update(
{
key: klass.s(*klasses)
}
)
if values.get('requester') is None:
values.update(
{
'requester': SF_REQUESTER
}
)
logger.debug(f"AFTER [ConfigModel] [{cls.__name__}] root_validator {values=}")
return values
@classmethod
def construct_list_class(cls):
def __init__(
self,
*data,
keys_override: List[str] = None,
transform_func: Callable[[str], str] = None
):
super(ListModel, self).__init__(
klass=cls,
klasses=data
)
def index_attributes(
klasses,
keys_override: List[str] = None,
transform_func: Callable[[str], str] = None
):
keys = keys_override if keys_override else cls.__fields__.get('keys').default
attributes = {}
for klass in klasses:
for key in keys:
if attr := get_key(klass, key=key, transform_func=transform_func):
attributes.setdefault(
attr, []
).append(klass)
return {
name: _klass[0] if len(_klass) == 1 else _klass
for name, _klass in attributes.items()
if name not in self.__dict__
}
for name, attr in index_attributes(
klasses=self.klasses,
keys_override=keys_override,
transform_func=transform_func
).items():
if isinstance(attr, list) and all(
isinstance(item, cls) for item in attr
):
SubList = type(
f"{cls.__name__}s",
(ListModel, MetaConfig),
{
'__annotations__': {
'klass': Type[cls],
'klasses': List[cls]
},
'__module__': cls.__module__
}
)
attr = SubList(cls, *attr)
for list_name, list_attr in index_attributes(
klasses=attr
).items():
object.__setattr__(attr, list_name, list_attr)
object.__setattr__(self, name, attr)
return type(
f"{cls.__name__}s",
(ListModel, MetaConfig),
{
'__annotations__': {
'klass': Type[cls],
'klasses': List[cls]
},
'__module__': cls.__module__,
'__init__': __init__
}
)
@classmethod
def s(
cls,
*klasses,
keys_override: List[str] = None,
transform_func: Callable[[str], str] = None
):
ClassList = cls.construct_list_class()
return ClassList(
*klasses,
keys_override=keys_override,
transform_func=transform_func
)
class Event(ConfigModel):
keys: List[str] = ['upload_file_name']
parent_id: str = None
additional_info: str = None
event_id: str = None
user_id: str = None
time_stamp: datetime = None
item_type: str = None
browser: str = None
path: str = None
first_name: str = None
last_name: str = None
email: str = None
full_name: str = None
full_name_short: str = None
account_id: str = None
ip_address: str = None
city: str = None
country: str = None
account_id: str = None
url: HttpUrl = None
odata_type: str = None
odata_metadata: HttpUrl = None
upload_file_name: str = None
upload_item: Any = None
requester: Requester = None
def json(self, exclude: dict = None, by_alias: bool = True):
exclude = exclude or {
'requester': ...,
'keys': ...,
'attributes': ...,
'upload_file_name': ...,
'upload_item': ...
}
return super().json(
exclude=exclude,
by_alias=by_alias
)
@property
def upload_product(self):
return self.additional_info.split('/')[5]
@property
def new_project_name(self):
project_name = self.additional_info.split('/')[6]
for section in self.additional_info.split('/')[7:-1]:
if to_snake(section) not in ['from_hcl', 'to_welocalize', 'for_translation']:
for part in re.split(r'\s|\_|\-', section):
if part.lower() not in project_name.lower():
project_name = ' '.join((project_name, part))
return project_name
@property
def sql(self):
return (
self.event_id,
self.time_stamp,
self.parent_id,
self.path,
self.upload_file_name,
self.full_name,
self.email
)
@validator('upload_file_name', pre=True, always=True)
def extract_upload_file_name(cls, v, values):
return values.get('additional_info').split('/')[-1]
@validator('upload_item', pre=True, always=True)
def get_upload_item(cls, v, values) -> File:
if not v:
parent
Python库 | sharepyle-1.0.0.dev2.tar.gz
版权申诉
37 浏览量
2022-04-15
09:54:45
上传
评论
收藏 11KB GZ 举报
sharepyle-1.0.0.dev2.tar.gz （9个子文件） 
sharepyle-1.0.0.dev2 
PKG-INFO 550B pyproject.toml 488B sharepyle models.py 4KB creds.py 8KB listener.py 2KB helpers.py 2KB __init__.py 104B entities.py 32KB setup.py 770B资源评论
