fxreader.online/freelance-project-34-marketing-blog
2
0
Fork 0
Code Issues 2 Pull Requests Actions Packages Projects Releases Wiki Activity

[~] Refactor

Browse Source
This commit is contained in:
Siarhei Siniak 2024-07-07 08:59:39 +03:00
parent 98f99cc470
commit bd883d810c
1 changed files with 40 additions and 11 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

51
python/tasks/tiktok/tasks.py
View File

@ -132,7 +132,7 @@ def tiktok_video_fetch(
method = tiktok_video_fetch_t.method_t(method_str) method = tiktok_video_fetch_t.method_t(method_str)
if method is None: if method is None:
method = tiktok_video_fetch_t.method_t.tikcdn_io_curl method = tiktok_video_fetch_t.method_t.pyktok
if method == tiktok_video_fetch_t.method_t.pyktok: if method == tiktok_video_fetch_t.method_t.pyktok:
import pyktok import pyktok
@ -241,20 +241,43 @@ def tiktok_videos_fetch(
return stats return stats
class tiktok_videos_process_t:
@dataclasses_json.dataclass_json
@dataclasses.dataclass
class res_t:
@dataclasses_json.dataclass_json
@dataclasses.dataclass
class stats_t:
saved: int=0
total: int=0
skipped: int=0
error: int=0
@dataclasses_json.dataclass_json
@dataclasses.dataclass
class video_t:
meta: Optional[dict[str, Any]]=None
processed_path: Optional[str]=None
stats: stats_t=dataclasses.field(default_factory=stats_t)
videos: Iterable[video_t]=dataclasses.field(default_factory=list)
@shared_task() @shared_task()
def tiktok_videos_process(meta: Iterable[dict[str, Any]]) -> dict[str, Any]: def tiktok_videos_process(meta: Iterable[dict[str, Any]]) -> dict[str, Any]:
import tqdm import tqdm
stats = dict(
saved=0, res = tiktok_videos_process_t.res_t(
total=0, videos=[],
skipped=0,
error=0,
) )
song = audio_get() song = audio_get()
for o in tqdm.tqdm(meta): for o in tqdm.tqdm(meta):
stats['total'] += 1 res.stats.total += 1
res.videos.append(tiktok_videos_process_t.res_t.video_t())
res.videos[-1].meta = o
path = os.path.join( path = os.path.join(
o['result_dir'], o['result_dir'],
@ -267,8 +290,11 @@ def tiktok_videos_process(meta: Iterable[dict[str, Any]]) -> dict[str, Any]:
processed_path = path_parts[0] + '-proc' + path_parts[1] processed_path = path_parts[0] + '-proc' + path_parts[1]
processed_path_tmp = path_parts[0] + '-proc.tmp' + path_parts[1] processed_path_tmp = path_parts[0] + '-proc.tmp' + path_parts[1]
if os.path.exists(processed_path):
res.videos[-1].processed_path = processed_path
if not os.path.exists(path) or os.path.exists(processed_path): if not os.path.exists(path) or os.path.exists(processed_path):
stats['skipped'] += 1 res.stats.skipped += 1
continue continue
if os.path.exists(processed_path_tmp): if os.path.exists(processed_path_tmp):
@ -299,16 +325,19 @@ def tiktok_videos_process(meta: Iterable[dict[str, Any]]) -> dict[str, Any]:
os.rename(processed_path_tmp, processed_path) os.rename(processed_path_tmp, processed_path)
stats['saved'] += 1 if os.path.exists(processed_path):
res.videos[-1].processed_path = processed_path
res.stats.saved += 1
except KeyboardInterrupt: except KeyboardInterrupt:
break break
except: except:
logger.error(json.dumps(dict( logger.error(json.dumps(dict(
msg=traceback.format_exc(), msg=traceback.format_exc(),
))) )))
stats['error'] += 1 res.stats.error += 1
return stats return res
class audio_get_t: class audio_get_t:
@dataclasses_json.dataclass_json @dataclasses_json.dataclass_json