<commit_before>import sys
import os
import hashlib
import requests
from urllib.parse import urlparse
from pathlib import Path, PosixPath
from homura import download
import shutil

from .. import get_logger

logger = get_logger(__file__)

if sys.platform.startswith("win"):
    default_cache_dir = Path(os.getenv("LOCALAPPDATA")) / "kbdgen" / "cache"
elif sys.platform.startswith("darwin"):
    default_cache_dir = Path(os.getenv("HOME")) / "Library" / "Caches" / "kbdgen"
else:
    default_cache_dir = Path(os.getenv("HOME")) / ".cache" / "kbdgen"

class FileCache:
    def __init__(self, cache_dir=default_cache_dir):
        self.cache_dir = Path(cache_dir)
        self.ensure_cache_exists()

    def ensure_cache_exists(self):
        if not self.cache_dir.exists():
            self.cache_dir.mkdir(exist_ok=True)

    def is_cached_valid(self, filename: str, sha256sum: str) -> bool:
        candidate = self.cache_dir / filename
        if not candidate.exists():
            return False
        if sha256sum is None:
            return True
        m = hashlib.sha256()
        with candidate.open('rb') as f:
            m.update(f.read())
        new_sum = m.hexdigest()
        logger.debug("SHA256: %s", new_sum)
        return new_sum == sha256sum

    def save_directory_tree(self, id: str, basepath: str, tree: str):
        src = Path(basepath) / tree
        target = self.cache_dir / id / tree
        shutil.rmtree(target, ignore_errors=True)
        target.mkdir()
        shutil.copytree(src, target)

    def inject_directory_tree(self, id: str, tree: str, base_target: str) -> bool:
        src = self.cache_dir / id / tree
        if not src.exists():
            return False
        target = Path(base_target) / Path(tree).parent
        os.makedirs(str(target), exist_ok=True)
        shutil.rmtree(target, ignore_errors=True)
        logger.debug("Copying '%s' to '%s'" % (src, target))
        
        shutil.copytree(src, target)
        return True

    def download(self, raw_url: str, sha256sum: str) -> str:
        url = urlparse(raw_url)
        filename = PosixPath(url.path).name
        candidate = self.cache_dir / filename
        if self.is_cached_valid(filename, sha256sum):
            return candidate
        logger.info("Downloading '%s'…" % filename)
        download(url=raw_url, path=str(candidate))
        if not self.is_cached_valid(filename, sha256sum):
            raise Exception("Cached file '%s' has failed integrity checks." % filename)
        return candidate

    def download_latest_from_github(self, repo: str, branch: str="master") -> str:
        repo_meta = requests.get("https://api.github.com/repos/{repo}/commits/HEAD?branch={branch}".format(
            repo=repo,
            branch=branch
        )).json()

        sha = repo_meta["sha"]
        filename = "%s-%s.tgz" % (repo.replace("/", "-"), sha)
        candidate = self.cache_dir / filename
        if self.is_cached_valid(filename, None):
            return candidate
        download_url = "https://api.github.com/repos/{repo}/tarball/{branch}".format(repo=repo, branch=branch)
        logger.debug("Download URL: %s" % download_url)
        download(url=download_url, path=str(candidate))
        return candidate
<commit_msg>Fix incomplete files on Ctrl+C when caching<commit_after>import sys
import os
import hashlib
import requests
from urllib.parse import urlparse
from pathlib import Path, PosixPath
from homura import download
import tempfile
import shutil

from .. import get_logger

logger = get_logger(__file__)

if sys.platform.startswith("win"):
    default_cache_dir = Path(os.getenv("LOCALAPPDATA")) / "kbdgen" / "cache"
elif sys.platform.startswith("darwin"):
    default_cache_dir = Path(os.getenv("HOME")) / "Library" / "Caches" / "kbdgen"
else:
    default_cache_dir = Path(os.getenv("HOME")) / ".cache" / "kbdgen"

class FileCache:
    def __init__(self, cache_dir=default_cache_dir):
        self.cache_dir = Path(cache_dir)
        self.ensure_cache_exists()

    def ensure_cache_exists(self):
        if not self.cache_dir.exists():
            self.cache_dir.mkdir(exist_ok=True)

    def is_cached_valid(self, filename: str, sha256sum: str) -> bool:
        candidate = self.cache_dir / filename
        if not candidate.exists():
            return False
        if sha256sum is None:
            return True
        m = hashlib.sha256()
        with candidate.open('rb') as f:
            m.update(f.read())
        new_sum = m.hexdigest()
        logger.debug("SHA256: %s", new_sum)
        return new_sum == sha256sum

    def save_directory_tree(self, id: str, basepath: str, tree: str):
        src = Path(basepath) / tree
        target = self.cache_dir / id / tree
        shutil.rmtree(target, ignore_errors=True)
        target.mkdir()
        shutil.copytree(src, target)

    def inject_directory_tree(self, id: str, tree: str, base_target: str) -> bool:
        src = self.cache_dir / id / tree
        if not src.exists():
            return False
        target = Path(base_target) / Path(tree).parent
        os.makedirs(str(target), exist_ok=True)
        shutil.rmtree(target, ignore_errors=True)
        logger.debug("Copying '%s' to '%s'" % (src, target))
        
        shutil.copytree(src, target)
        return True

    def download(self, raw_url: str, sha256sum: str) -> str:
        url = urlparse(raw_url)
        filename = PosixPath(url.path).name
        candidate = self.cache_dir / filename
        if self.is_cached_valid(filename, sha256sum):
            return candidate
        logger.info("Downloading '%s'…" % filename)
        download(url=raw_url, path=str(candidate))
        if not self.is_cached_valid(filename, sha256sum):
            raise Exception("Cached file '%s' has failed integrity checks." % filename)
        return candidate

    def download_latest_from_github(self, repo: str, branch: str="master") -> str:
        repo_meta = requests.get("https://api.github.com/repos/{repo}/commits/HEAD?branch={branch}".format(
            repo=repo,
            branch=branch
        )).json()

        sha = repo_meta["sha"]
        filename = "%s-%s.tgz" % (repo.replace("/", "-"), sha)
        candidate = self.cache_dir / filename
        if self.is_cached_valid(filename, None):
            return candidate
        download_url = "https://api.github.com/repos/{repo}/tarball/{branch}".format(repo=repo, branch=branch)
        logger.debug("Download URL: %s" % download_url)
        with tempfile.TemporaryDirectory() as tmpdir:
            fp = os.path.join(tmpdir, filename)
            download(url=download_url, path=fp)
            Path(fp).rename(candidate)
        return candidate

{'fraction_non_alphanumeric': 0.1211759438889718, 'fraction_numerical': 0.007163109983584539, 'mean_word_length': 2.678375411635565, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 11, 'https://': 4, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 20, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '4712990', 'n_tokens_mistral': 2091, 'n_tokens_neox': 2044, 'n_words': 470}