Spaces:
Sleeping
Sleeping
| """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, | |
| ) | |