Repository navigation
Expand file tree
/
Copy patharchive_old.py
More file actions
102 lines (80 loc) · 2.76 KB
/
Copy patharchive_old.py
File metadata and controls
102 lines (80 loc) · 2.76 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
"""
Script to compress old log directories.
- Run from repository root
- Looks under ./log-files
- Directory names assumed to be YYYY-MM-DD_HH-MM-SS, optionally suffixed with -<run id>
- Directories older than 6 months are archived to .tar.xz
- Assumes `tar` and `xz` are available
"""
from pathlib import Path
from datetime import datetime, timedelta
from typing import List, Optional
import subprocess
import tempfile
LOG_ROOT = Path("log-files")
DEFAULT_DAYS = 182
def get_cutoff(days: int = DEFAULT_DAYS) -> datetime:
"""Return the cutoff datetime; directories older than this should be archived."""
return datetime.now() - timedelta(days=days)
def parse_timestamp(name: str) -> Optional[datetime]:
"""
Parse a directory name of the form YYYY-MM-DD_HH-MM-SS(-<run id>).
Return a datetime on success, or None if the leading 19 characters do not match.
"""
try:
return datetime.strptime(name[:19], "%Y-%m-%d_%H-%M-%S")
except ValueError:
return None
def find_old_dirs(root: Path, cutoff: datetime) -> List[Path]:
"""
Return directories under `root` which contain index.html, and whose
names parse as timestamps and are strictly older than `cutoff`.
"""
if not root.is_dir():
return []
results: List[Path] = []
for p in sorted(root.iterdir()):
if not (p.is_dir() and (p / 'index.html').is_file()):
continue
ts = parse_timestamp(p.name)
if ts is None:
continue
if ts < cutoff:
results.append(p)
return results
def compress_dir(src: Path) -> None:
"""
Compress `src` directory to src.tar.xz using `tar | xz -9`.
Skip if the archive already exists.
"""
dest = src.with_name(src.name + ".tar.xz")
if dest.exists():
return
with tempfile.NamedTemporaryFile(dir=str(dest.parent), delete=False) as tf:
tmp = Path(tf.name)
tar_cmd = ["tar", "-C", str(src.parent), "-c", src.name]
xz_cmd = ["xz", "-9", "-c"]
try:
with open(tmp, "wb") as fout:
p1 = subprocess.Popen(tar_cmd, stdout=subprocess.PIPE)
p2 = subprocess.Popen(xz_cmd, stdin=p1.stdout, stdout=fout)
p1.stdout.close()
p2.communicate()
if p1.wait() != 0 or p2.returncode != 0:
raise RuntimeError("tar/xz failed")
tmp.replace(dest)
except Exception:
tmp.unlink(missing_ok=True)
raise
def main() -> None:
cutoff = get_cutoff()
old_dirs = find_old_dirs(LOG_ROOT, cutoff)
for src in old_dirs:
print(f"Compressing {src} ...", end="", flush=True)
try:
compress_dir(src)
print(" done")
except Exception as e:
print(f" ERROR: {e}")
if __name__ == "__main__":
main()