(base_path, task_id, level=2)
| 39 | |
| 40 | |
| 41 | def get_task_log_path(base_path, task_id, level=2): |
| 42 | task_id = str(task_id) |
| 43 | try: |
| 44 | uuid.UUID(task_id) |
| 45 | except: |
| 46 | return os.path.join(PROJECT_DIR, 'data', 'caution.txt') |
| 47 | |
| 48 | rel_path = os.path.join(*task_id[:level], task_id + '.log') |
| 49 | path = os.path.join(base_path, rel_path) |
| 50 | make_dirs(os.path.dirname(path), exist_ok=True) |
| 51 | return path |
| 52 | |
| 53 | |
| 54 | def get_celery_task_log_path(task_id): |
no test coverage detected