feat(analyzer): move compute_md5 to shared library
This commit is contained in:
parent
c0bddf83a8
commit
be9f36dbdc
|
@ -1,9 +1,9 @@
|
||||||
import hashlib
|
|
||||||
from datetime import timedelta
|
from datetime import timedelta
|
||||||
from pathlib import Path
|
from pathlib import Path
|
||||||
from typing import Any, Dict
|
from typing import Any, Dict
|
||||||
|
|
||||||
import mutagen
|
import mutagen
|
||||||
|
from libretime_shared.files import compute_md5
|
||||||
from loguru import logger
|
from loguru import logger
|
||||||
|
|
||||||
|
|
||||||
|
@ -108,18 +108,3 @@ def analyze_metadata(filepath_: str, metadata: Dict[str, Any]):
|
||||||
continue
|
continue
|
||||||
|
|
||||||
return metadata
|
return metadata
|
||||||
|
|
||||||
|
|
||||||
def compute_md5(filepath: Path) -> str:
|
|
||||||
"""
|
|
||||||
Compute a file md5sum.
|
|
||||||
"""
|
|
||||||
with filepath.open("rb") as file:
|
|
||||||
buffer = hashlib.md5() # nosec
|
|
||||||
while True:
|
|
||||||
blob = file.read(8192)
|
|
||||||
if not blob:
|
|
||||||
break
|
|
||||||
buffer.update(blob)
|
|
||||||
|
|
||||||
return buffer.hexdigest()
|
|
||||||
|
|
|
@ -2,7 +2,7 @@ from pathlib import Path
|
||||||
|
|
||||||
import pytest
|
import pytest
|
||||||
|
|
||||||
from libretime_analyzer.pipeline.analyze_metadata import analyze_metadata, compute_md5
|
from libretime_analyzer.pipeline.analyze_metadata import analyze_metadata
|
||||||
|
|
||||||
from ..fixtures import FILE_INVALID_DRM, FILE_INVALID_TXT, FILES_TAGGED
|
from ..fixtures import FILE_INVALID_DRM, FILE_INVALID_TXT, FILES_TAGGED
|
||||||
|
|
||||||
|
@ -48,7 +48,3 @@ def test_analyze_metadata_unparsable_file():
|
||||||
"hidden": False,
|
"hidden": False,
|
||||||
"md5": "4d5e4b1c8e8febbd31fa9ce7f088beae",
|
"md5": "4d5e4b1c8e8febbd31fa9ce7f088beae",
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
||||||
def test_compute_md5():
|
|
||||||
assert compute_md5(FILE_INVALID_TXT) == "4d5e4b1c8e8febbd31fa9ce7f088beae"
|
|
||||||
|
|
|
@ -0,0 +1,17 @@
|
||||||
|
import hashlib
|
||||||
|
from pathlib import Path
|
||||||
|
|
||||||
|
|
||||||
|
def compute_md5(filepath: Path) -> str:
|
||||||
|
"""
|
||||||
|
Compute a file md5sum.
|
||||||
|
"""
|
||||||
|
with filepath.open("rb") as file:
|
||||||
|
buffer = hashlib.md5() # nosec
|
||||||
|
while True:
|
||||||
|
blob = file.read(8192)
|
||||||
|
if not blob:
|
||||||
|
break
|
||||||
|
buffer.update(blob)
|
||||||
|
|
||||||
|
return buffer.hexdigest()
|
|
@ -0,0 +1,10 @@
|
||||||
|
from pathlib import Path
|
||||||
|
|
||||||
|
from libretime_shared.files import compute_md5
|
||||||
|
|
||||||
|
|
||||||
|
def test_compute_md5(tmp_path: Path) -> None:
|
||||||
|
tmp_file = tmp_path / "somefile.txt"
|
||||||
|
tmp_file.write_text("some test")
|
||||||
|
|
||||||
|
assert compute_md5(tmp_file) == "f1b75ac7689ff88e1ecc40c84b115785"
|
Loading…
Reference in New Issue