cache.py 2.3 KB

12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970717273747576777879808182838485868788899091
  1. import tempfile
  2. import os
  3. import time
  4. import hashlib
  5. import shutil
  6. cache_dir = os.path.join(tempfile.gettempdir(), "cache")
  7. os.makedirs(cache_dir, exist_ok=True)
  8. time_filename = "update_time"
  9. max_cache = 5
  10. def deterministic_hash(obj):
  11. hash_object = hashlib.sha256()
  12. hash_object.update(str(obj).encode())
  13. return hash_object.hexdigest()[0:20]
  14. def get_dirs():
  15. dirs = [
  16. os.path.join(cache_dir, dir)
  17. for dir in os.listdir(cache_dir)
  18. if os.path.isdir(os.path.join(cache_dir, dir))
  19. ]
  20. return dirs
  21. def get_time(dir):
  22. try:
  23. timefile = os.path.join(dir, time_filename)
  24. t = float(open(timefile, encoding="utf-8").read())
  25. return t
  26. except FileNotFoundError:
  27. # handle the error as needed, for now we'll just return a default value
  28. return float(
  29. "inf"
  30. ) # This ensures that this directory will be the first to be removed if required
  31. def write_time(dir):
  32. timefile = os.path.join(dir, time_filename)
  33. t = time.time()
  34. print(t, file=open(timefile, "w", encoding="utf-8"), end="")
  35. def argmin(iterable):
  36. return min(enumerate(iterable), key=lambda x: x[1])[0]
  37. def remove_extra():
  38. dirs = get_dirs()
  39. for dir in dirs:
  40. if not os.path.isdir(
  41. dir
  42. ): # This line might be redundant now, as get_dirs() ensures only directories are returned
  43. os.remove(dir)
  44. try:
  45. get_time(dir)
  46. except BaseException:
  47. shutil.rmtree(dir)
  48. while True:
  49. dirs = get_dirs()
  50. if len(dirs) <= max_cache:
  51. break
  52. times = [get_time(dir) for dir in dirs]
  53. arg = argmin(times)
  54. shutil.rmtree(dirs[arg])
  55. def is_cached(hash_key):
  56. dir = os.path.join(cache_dir, hash_key)
  57. return os.path.exists(dir)
  58. def create_cache(hash_key):
  59. dir = os.path.join(cache_dir, hash_key)
  60. os.makedirs(dir, exist_ok=True)
  61. write_time(dir)
  62. def load_paragraph(hash_key, hash_key_paragraph):
  63. filename = os.path.join(cache_dir, hash_key, hash_key_paragraph)
  64. if os.path.exists(filename):
  65. return open(filename, encoding="utf-8").read()
  66. else:
  67. return None
  68. def write_paragraph(hash_key, hash_key_paragraph, paragraph):
  69. filename = os.path.join(cache_dir, hash_key, hash_key_paragraph)
  70. print(paragraph, file=open(filename, "w", encoding="utf-8"), end="")