Spaces:
Build error
Build error
| """GitHub operations β read, write, list files in the agent's repo.""" | |
| import base64 | |
| from datetime import datetime, timezone | |
| from github import Github, GithubException | |
| import config | |
| def _client(): | |
| return Github(config.GITHUB_TOKEN) | |
| def _repo(): | |
| return _client().get_repo(config.GITHUB_REPO) | |
| # ββ List ββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ | |
| def list_files(folder: str) -> list[str]: | |
| """Return a list of file paths inside `folder`.""" | |
| try: | |
| contents = _repo().get_contents(folder) | |
| return [c.path for c in contents] | |
| except GithubException as e: | |
| return [f"ERROR: {e.data.get('message', str(e))}"] | |
| # ββ Read ββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ | |
| def read_file(path: str) -> str: | |
| """Return decoded text content of `path`.""" | |
| try: | |
| file = _repo().get_contents(path) | |
| return base64.b64decode(file.content).decode("utf-8") | |
| except GithubException as e: | |
| return f"ERROR: {e.data.get('message', str(e))}" | |
| # ββ Write / Update ββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ | |
| def write_file(path: str, content: str, message: str = "") -> str: | |
| """Create or update `path` with `content`. Returns commit SHA.""" | |
| repo = _repo() | |
| commit_msg = message or f"savvy: update {path}" | |
| try: | |
| existing = repo.get_contents(path) | |
| result = repo.update_file(path, commit_msg, content, existing.sha) | |
| except GithubException: | |
| # File doesn't exist yet | |
| result = repo.create_file(path, commit_msg, content) | |
| return result["commit"].sha | |
| # ββ Convenience helpers βββββββββββββββββββββββββββββββββββββββββββββββββββββββ | |
| def save_memory(chat_id: int | str, user_msg: str, agent_reply: str) -> str: | |
| """Append a conversation turn to memory/{chat_id}.md""" | |
| ts = datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M:%S UTC") | |
| path = f"{config.MEMORY_PATH}/{chat_id}.md" | |
| existing = read_file(path) | |
| if existing.startswith("ERROR"): | |
| existing = f"# Memory β chat {chat_id}\n\n" | |
| block = f"\n---\n**{ts}**\n\n**User:** {user_msg}\n\n**Savvy:** {agent_reply}\n" | |
| write_file(path, existing + block, f"memory: chat {chat_id}") | |
| return path | |
| def save_research(query: str, result: str) -> str: | |
| """Save a web search result to research/ with timestamp in filename.""" | |
| ts = datetime.now(timezone.utc).strftime("%Y-%m-%d_%H-%M-%S") | |
| safe_q = query[:60].replace(" ", "_").replace("/", "-") | |
| path = f"{config.RESEARCH_PATH}/{ts}_{safe_q}.md" | |
| content = f"# Research: {query}\n\n_Saved: {ts} UTC_\n\n{result}\n" | |
| write_file(path, content, f"research: {query[:60]}") | |
| return path | |