75 lines
1.6 KiB
Python
75 lines
1.6 KiB
Python
import logging
|
|
import enum
|
|
import dataclasses
|
|
import dataclasses_json
|
|
import multiprocessing
|
|
import traceback
|
|
import subprocess
|
|
import os
|
|
import sys
|
|
import json
|
|
from typing import (
|
|
Any,
|
|
Literal,
|
|
Optional,
|
|
Iterable,
|
|
)
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
#logging.getLogger().setLevel(logging.INFO)
|
|
|
|
class tiktok_config_t:
|
|
@dataclasses_json.dataclass_json
|
|
@dataclasses.dataclass
|
|
class res_t:
|
|
project_root: str=''
|
|
cache: str=''
|
|
videos: str=''
|
|
audios: str=''
|
|
celery_broker: str=''
|
|
celery_result_backend: str=''
|
|
celery_imports: Iterable[str]=tuple()
|
|
|
|
def tiktok_config() -> tiktok_config_t.res_t:
|
|
res = tiktok_config_t.res_t(
|
|
project_root=os.path.abspath(
|
|
os.path.join(
|
|
os.path.dirname(__file__),
|
|
'..', '..', '..',
|
|
),
|
|
),
|
|
)
|
|
|
|
res.celery_broker = 'redis://redis:6379/1'
|
|
res.celery_result_backend = 'redis://redis:6379/2'
|
|
res.celery_imports = ['python.tasks.tiktok.tasks']
|
|
res.cache = os.path.join(
|
|
res.project_root,
|
|
'tmp/cache/tiktok',
|
|
)
|
|
res.videos = os.path.join(
|
|
res.cache,
|
|
'videos',
|
|
)
|
|
res.audios = os.path.join(
|
|
res.cache,
|
|
'audios',
|
|
)
|
|
|
|
os.makedirs(res.videos, exist_ok=True)
|
|
os.makedirs(res.audios, exist_ok=True)
|
|
|
|
return res
|
|
|
|
def logger_setup(name: str) -> logging.Logger:
|
|
logger = logging.getLogger(name)
|
|
if len(logger.handlers) == 0:
|
|
handler = logging.StreamHandler(sys.stderr)
|
|
logger.addHandler(handler)
|
|
logger.setLevel(logging.INFO)
|
|
|
|
return logger
|
|
|
|
logger = logger_setup(__name__)
|