This commit is contained in:
2026-06-07 17:48:09 +07:00
parent de55110b19
commit dfa0e720ac
12 changed files with 354 additions and 0 deletions

View File

@@ -0,0 +1,5 @@
SPREADSHEET_ID=your_spreadsheet_id_here
SHEET_RANGE=Sheet1!A2:E
CREDENTIALS_FILE=credentials.json
TOKEN_FILE=token.json
README_PATH=../README.md

10
BABA_YAGA_Updater/.gitignore vendored Normal file
View File

@@ -0,0 +1,10 @@
.env
credentials.json
token.json
__pycache__/
*.py[cod]
*$py.class
.venv/
venv/
ENV/
env/

View File

@@ -0,0 +1,21 @@
import os
from dotenv import load_dotenv
# Load .env file if it exists
load_dotenv()
class Settings:
# Google Sheets Configuration
SPREADSHEET_ID = os.getenv("SPREADSHEET_ID", "")
SHEET_RANGE = os.getenv("SHEET_RANGE", "Sheet1!A2:E") # Default range
CREDENTIALS_FILE = os.getenv("CREDENTIALS_FILE", "credentials.json")
TOKEN_FILE = os.getenv("TOKEN_FILE", "token.json")
# README Configuration
README_PATH = os.getenv("README_PATH", "../README.md")
# Section Markers
START_MARKER = "<!-- START_UPDATES -->"
END_MARKER = "<!-- END_UPDATES -->"
settings = Settings()

View File

@@ -0,0 +1,12 @@
from pydantic import BaseModel
from typing import Optional
class Task(BaseModel):
category: str
task_name: str
status: str
progress: str # e.g., "75%" or "In Progress"
notes: Optional[str] = ""
class ProgressReport(BaseModel):
tasks: list[Task]

38
BABA_YAGA_Updater/main.py Normal file
View File

@@ -0,0 +1,38 @@
import sys
from services.gsheet_client import GSheetClient
from mappers.sheet_mapper import SheetMapper
from mappers.markdown_builder import MarkdownBuilder
from services.readme_editor import ReadmeEditor
def main():
try:
print("🚀 Starting README update from Google Sheets...")
# 1. Fetch data from Google Sheets
client = GSheetClient()
raw_rows = client.fetch_data()
if not raw_rows:
print("⚠️ No data found in the spreadsheet or error occurred.")
return
# 2. Map raw rows to Core Models
report = SheetMapper.map_rows_to_report(raw_rows)
print(f"✅ Parsed {len(report.tasks)} tasks.")
# 3. Build Markdown content
markdown_content = MarkdownBuilder.build_table(report)
# 4. Update README.md
editor = ReadmeEditor()
if editor.update_section(markdown_content):
print("🎉 README.md successfully updated!")
else:
print("❌ Failed to update README.md.")
except Exception as e:
print(f"💥 An unexpected error occurred: {e}")
sys.exit(1)
if __name__ == "__main__":
main()

View File

@@ -0,0 +1,24 @@
from core.models import ProgressReport
class MarkdownBuilder:
@staticmethod
def build_table(report: ProgressReport) -> str:
if not report.tasks:
return "_No tasks updated._"
header = "| Category | Task | Status | Progress | Notes |\n"
separator = "| :--- | :--- | :--- | :--- | :--- |\n"
rows = []
for task in report.tasks:
# Format status with some icons if possible, or just plain text
status_text = task.status
if "done" in status_text.lower() or "complete" in status_text.lower():
status_text = f"{status_text}"
elif "progress" in status_text.lower():
status_text = f"🔄 {status_text}"
row = f"| {task.category} | {task.task_name} | {status_text} | {task.progress} | {task.notes} |"
rows.append(row)
return header + separator + "\n".join(rows)

View File

@@ -0,0 +1,20 @@
from core.models import Task, ProgressReport
class SheetMapper:
@staticmethod
def map_rows_to_report(rows: list[list]) -> ProgressReport:
tasks = []
for row in rows:
# Ensure the row has enough columns, fill missing with empty strings
padded_row = row + [""] * (5 - len(row))
task = Task(
category=padded_row[0],
task_name=padded_row[1],
status=padded_row[2],
progress=padded_row[3],
notes=padded_row[4]
)
tasks.append(task)
return ProgressReport(tasks=tasks)

View File

@@ -0,0 +1,5 @@
google-api-python-client
google-auth-httplib2
google-auth-oauthlib
python-dotenv
pydantic

View File

@@ -0,0 +1,52 @@
import os
from google.auth.transport.requests import Request
from google.oauth2.credentials import Credentials
from google_auth_oauthlib.flow import InstalledAppFlow
from googleapiclient.discovery import build
from googleapiclient.errors import HttpError
from config.settings import settings
class GSheetClient:
SCOPES = ['https://www.googleapis.com/auth/spreadsheets.readonly']
def __init__(self):
self.creds = self._authenticate()
def _authenticate(self):
creds = None
# The file token.json stores the user's access and refresh tokens
if os.path.exists(settings.TOKEN_FILE):
creds = Credentials.from_authorized_user_file(settings.TOKEN_FILE, self.SCOPES)
# If there are no (valid) credentials available, let the user log in.
if not creds or not creds.valid:
if creds and creds.expired and creds.refresh_token:
creds.refresh(Request())
else:
if not os.path.exists(settings.CREDENTIALS_FILE):
raise FileNotFoundError(f"Credentials file not found at {settings.CREDENTIALS_FILE}. Please download it from Google Cloud Console.")
flow = InstalledAppFlow.from_client_secrets_file(
settings.CREDENTIALS_FILE, self.SCOPES)
creds = flow.run_local_server(port=0)
# Save the credentials for the next run
with open(settings.TOKEN_FILE, 'w') as token:
token.write(creds.to_json())
return creds
def fetch_data(self):
try:
service = build('sheets', 'v4', credentials=self.creds)
sheet = service.spreadsheets()
result = sheet.values().get(
spreadsheetId=settings.SPREADSHEET_ID,
range=settings.SHEET_RANGE
).execute()
return result.get('values', [])
except HttpError as err:
print(f"An error occurred: {err}")
return []

View File

@@ -0,0 +1,30 @@
import re
from config.settings import settings
class ReadmeEditor:
def __init__(self, file_path: str = None):
self.file_path = file_path or settings.README_PATH
def update_section(self, new_content: str):
with open(self.file_path, 'r', encoding='utf-8') as f:
content = f.read()
# Regex to find content between markers
pattern = re.compile(
f"({re.escape(settings.START_MARKER)})(.*?)({re.escape(settings.END_MARKER)})",
re.DOTALL
)
if not pattern.search(content):
print(f"Markers {settings.START_MARKER} and {settings.END_MARKER} not found in {self.file_path}")
return False
updated_content = pattern.sub(
f"\\1\n\n{new_content}\n\n\\3",
content
)
with open(self.file_path, 'w', encoding='utf-8') as f:
f.write(updated_content)
return True