106 lines
2.4 KiB
Python
106 lines
2.4 KiB
Python
import json
|
|
from os.path import exists
|
|
from typing import Dict, List
|
|
|
|
from utils import id_to_url, url_to_id
|
|
|
|
db_filename = "db.json"
|
|
|
|
|
|
def init_db():
|
|
file_exists = exists(db_filename)
|
|
if not file_exists:
|
|
print("Initializing json file database")
|
|
with open(db_filename, "w") as db_file:
|
|
db_file.write("{}")
|
|
|
|
|
|
init_db()
|
|
|
|
|
|
def _load_db():
|
|
with open(db_filename, "r") as db_file:
|
|
db = json.loads(db_file.read())
|
|
return db
|
|
|
|
|
|
def _save_db(db):
|
|
with open(db_filename, "w") as db_file:
|
|
db_file.write(json.dumps(db, indent=4))
|
|
|
|
|
|
def get_latest_sync_time() -> int:
|
|
db = _load_db()
|
|
return db.get("latest_sync_time", 0)
|
|
|
|
|
|
def set_latest_sync_time(le_date) -> None:
|
|
db = _load_db()
|
|
db["latest_sync_time"] = le_date
|
|
_save_db(db)
|
|
|
|
|
|
def add_discovered_url(file_url, originating_mm_post) -> List[str]:
|
|
db = _load_db()
|
|
files = db.get("files", {})
|
|
file_id = url_to_id(file_url)
|
|
if file_id not in files:
|
|
files[file_id] = {
|
|
"originating_mm_post_id": originating_mm_post["id"],
|
|
"originating_mm_post_channel_id": originating_mm_post["channel_id"],
|
|
"source_url": file_url,
|
|
}
|
|
db["files"] = files
|
|
_save_db(db)
|
|
return files
|
|
|
|
|
|
def get_files() -> List[str]:
|
|
db = _load_db()
|
|
files = db.get("files", {})
|
|
return files
|
|
|
|
|
|
def set_local_file_path(file_id, local_file_path):
|
|
db = _load_db()
|
|
file = db["files"][file_id]
|
|
file["local_file_path"] = local_file_path
|
|
_save_db(db)
|
|
return file
|
|
|
|
|
|
def mark_file_valid(file_id, metadata) -> (bool, Dict):
|
|
"""
|
|
Return: Boolean that indicates if the file is newly identified as valid.
|
|
"""
|
|
db = _load_db()
|
|
|
|
file = db["files"][file_id]
|
|
new_file = False
|
|
if "valid" not in file or not file["valid"]:
|
|
new_file = True
|
|
file["valid"] = True
|
|
file["metadata"] = metadata
|
|
|
|
_save_db(db)
|
|
|
|
return new_file, file
|
|
|
|
|
|
def mark_file_invalid(file_id) -> (bool, Dict):
|
|
"""
|
|
Return: Boolean that indicates if the state is changed.
|
|
True means it was valid but is now invalid or it was unknown and is know invalid
|
|
False means it was invalid and is still invalid.
|
|
"""
|
|
db = _load_db()
|
|
file = db["files"][file_id]
|
|
changed = False
|
|
if "valid" not in file or file["valid"]:
|
|
changed = True
|
|
file["valid"] = False
|
|
|
|
_save_db(db)
|
|
|
|
return changed, file
|