Spaces:
Sleeping
Sleeping
File size: 5,799 Bytes
526950e | 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 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 | """HF Dataset I/O for the per-faculty notes store.
Companion to state_io.py; handles a different set of files:
- notes/<login>.md
- notes/archive/<login>.<YYYY-MM-DD>[-NN].md
- notes_summarizer_log.jsonl
Lazy imports of huggingface_hub via re-exported wrappers so tests can patch
this module's symbols and so this module imports without huggingface_hub
installed (local dev).
"""
from __future__ import annotations
import json
import logging
from pathlib import Path
from scripts.lib.turns_merge import is_not_found_error as _is_not_found_error
logger = logging.getLogger(__name__)
# Re-exported so tests can monkeypatch this module.
is_not_found_error = _is_not_found_error
def hf_hub_download(*args, **kwargs):
"""Re-export so tests can patch this module's reference."""
from huggingface_hub import hf_hub_download as _real
return _real(*args, **kwargs)
def HfApi(*args, **kwargs):
"""Re-export so tests can patch this module's reference."""
from huggingface_hub import HfApi as _real
return _real(*args, **kwargs)
def _notes_path_in_repo(login: str) -> str:
return f"notes/{login}.md"
_SUMMARIZER_LOG_PATH = "notes_summarizer_log.jsonl"
def fetch_user_notes(*, login: str, repo_id: str, token: str) -> str:
"""Return the current notes body for `login` from the dataset.
Returns the empty string when the file does not exist (first run for a
user, or notes/<login>.md never created).
"""
try:
local = hf_hub_download(
repo_id=repo_id,
filename=_notes_path_in_repo(login),
repo_type="dataset",
token=token,
)
except Exception as e:
if is_not_found_error(e):
return ""
raise
return Path(local).read_text(encoding="utf-8")
def fetch_summarizer_log(*, repo_id: str, token: str) -> list[dict]:
"""Return the existing summarizer audit rows. Empty list when missing."""
try:
local = hf_hub_download(
repo_id=repo_id,
filename=_SUMMARIZER_LOG_PATH,
repo_type="dataset",
token=token,
)
except Exception as e:
if is_not_found_error(e):
return []
raise
rows: list[dict] = []
for line in Path(local).read_text(encoding="utf-8").splitlines():
if not line.strip():
continue
try:
rows.append(json.loads(line))
except json.JSONDecodeError:
logger.warning("notes_io: skipping malformed audit row")
return rows
def archive_path_for(*, login: str, ts: str) -> str:
"""Return the repo-relative archive path for a given login + ts.
The base form ignores collisions; callers using a local working dir
should use next_archive_path_local() to pick a -NN suffix when needed.
"""
date_part = ts[:10] # YYYY-MM-DD slice of an ISO 8601 timestamp
return f"notes/archive/{login}.{date_part}.md"
def next_archive_path_local(*, archive_dir: Path, login: str, ts: str) -> Path:
"""Return a non-colliding local archive path for the given login + ts.
If <login>.<date>.md exists in archive_dir, picks <login>.<date>-NN.md
starting from -01.
"""
date_part = ts[:10]
base = archive_dir / f"{login}.{date_part}.md"
if not base.exists():
return base
for n in range(1, 100):
candidate = archive_dir / f"{login}.{date_part}-{n:02d}.md"
if not candidate.exists():
return candidate
raise RuntimeError(
f"notes_io: archive collision could not be resolved for {login} on {date_part}"
)
def push_notes_update(
*,
repo_id: str,
token: str,
login: str,
new_notes_body: str,
archive_repo_path: str,
archive_body: str,
summarizer_log_rows: list[dict],
commit_message: str,
) -> None:
"""Upload a coherent notes-update transaction: new notes file, archive
of the prior version, and the updated summarizer log. All three commit
together so the dataset never has an inconsistent intermediate state."""
api = HfApi()
# Three independent upload_file calls — the HF API does not support
# multi-file atomic commits, so we accept this and order writes such
# that a partial failure leaves recoverable state (notes_summarizer_log
# last, so a failed run is replayable by the next cron tick).
api.upload_file(
path_or_fileobj=archive_body.encode("utf-8"),
path_in_repo=archive_repo_path,
repo_id=repo_id,
repo_type="dataset",
token=token,
commit_message=f"{commit_message} (archive)",
)
api.upload_file(
path_or_fileobj=new_notes_body.encode("utf-8"),
path_in_repo=_notes_path_in_repo(login),
repo_id=repo_id,
repo_type="dataset",
token=token,
commit_message=f"{commit_message} (notes)",
)
log_body = "\n".join(json.dumps(r) for r in summarizer_log_rows) + "\n"
api.upload_file(
path_or_fileobj=log_body.encode("utf-8"),
path_in_repo=_SUMMARIZER_LOG_PATH,
repo_id=repo_id,
repo_type="dataset",
token=token,
commit_message=f"{commit_message} (log)",
)
def push_summarizer_log_only(
*,
repo_id: str,
token: str,
summarizer_log_rows: list[dict],
commit_message: str,
) -> None:
"""Upload only the summarizer log (used for no-op rows and failure rows
where the notes file did not change)."""
api = HfApi()
log_body = "\n".join(json.dumps(r) for r in summarizer_log_rows) + "\n"
api.upload_file(
path_or_fileobj=log_body.encode("utf-8"),
path_in_repo=_SUMMARIZER_LOG_PATH,
repo_id=repo_id,
repo_type="dataset",
token=token,
commit_message=commit_message,
)
|