added service copy script #5
Conversation
| source = parse_args().source | ||
| dest = parse_args().dest | ||
| ems = get_em(source) | ||
| em_paths = find_interlace(ems) |
There was a problem hiding this comment.
Looking at this again, I recommended the wrong approach. Because we complete each entire stage one at a time (e.g. transcode all the files, then upload all the files), it makes recovering from errors harder. It would be better to work each em through the whole process one at a time. That way if it errors out half way through, at least the first half of the files are already uploaded.
To do this, and this is a larger rewrite, the code should accept and pass single items instead of lists of items.
And then here it would be
for em in ems:
find_interlace
make_ffmpeg_cmd
run_ffmpeg
make_rclone_cmd
run_rclone
better yet combine all of these into a single function.
def create_sc_and_upload():
find_interlace
....
run_rclone
That would set us up for multiprocessing.
|
|
||
| def main(): | ||
| source = parse_args().source | ||
| dest = parse_args().dest |
There was a problem hiding this comment.
you only needs to call parse_args once. It will return an object that has all of the parsed arguments.
idiomatically, I see this like this
args = parse_args()
get_em(args.source)
| return path | ||
|
|
||
| # def rclone_remote(p: str) -> Path: | ||
| # if not re.match(r'*:*', p): |
There was a problem hiding this comment.
the * (zero or more of) should be replaced by .+, (any character, 1 or more of them)
| source = path | ||
| ems = [] | ||
| for x in source.rglob("*_em.*"): | ||
| if not str(x).endswith('mov'): |
There was a problem hiding this comment.
Path objects have the property .suffix for getting the extension.
Because it looks like we'll have a few more extensions to work with, you can see if that extension is in our list of acceptable extensions with the following.
if not x.suffix in ['mov', 'mp4', ...]:
We're also going to have audio files in here. I'm not sure how we'll fit them into the overall flow of this code. Something to think about.
There was a problem hiding this comment.
might need to be ['.mov', '.mp4', ...]
| added because ffmpeg did not like the directory not already existing | ||
| ''' | ||
| dest = base / 'ServiceCopies' | ||
| subprocess.run(['mkdir', f'{dest}']) |
There was a problem hiding this comment.
Path objects have a method for making a directory .mkdir(). When possible use a native Python method instead of subprocessing out. Subprocessing out can be the source of hard to solve bugs.
This line should be replaced by dest.mkdir()
There was a problem hiding this comment.
I was going to say that you don't need to f-string the dest variable, but now I realize that this is a clever way to deal with the whole, "Is a Path object interpreted as a string or not?" problem.
| # PosixPath('test_ems/dir_3/data/EditMaster/sample_dig_3_em.mov') | ||
| base = em_path.parent.parent | ||
| ''' | ||
| May be useful to include a check for service copy directory already existing. |
There was a problem hiding this comment.
Yes, this would be especially useful for partial failures when we're transcoding. Although, we should decide if we want to overwrite the initial batch of SCs created. Also, I'm not sure how to incorporate the effect of a path existing into this function, since we'd want to skip multiple other functions. if sc.exists(): should probably be down in main()
In addition, it would be useful to move all the SC prep work to another function that gets called before this one. Otherwise, it's a little confusing why there are directories being made in a function with the main purpose of building an ffmpeg command.
def make_ffmpeg_cmd(em_path, interlacing, sc_path) -> list[str]:
No description provided.