Spaces:
Sleeping
Sleeping
File size: 4,670 Bytes
3c5146b 386a178 3c5146b 386a178 3c5146b 386a178 3c5146b 386a178 3c5146b 386a178 3c5146b 386a178 3c5146b 386a178 3c5146b | 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 | """Google Sheets integration — exports RACI and Risk Register as shareable spreadsheets."""
import asyncio
import os
from core.planner import ERROR_MESSAGES, ProjectPlan, RACIEntry
SCOPES = ["https://www.googleapis.com/auth/spreadsheets"]
def _build_service():
"""Build Google Sheets API service. Return None if credentials unavailable."""
try:
from google.oauth2.credentials import Credentials
from googleapiclient.discovery import build
except ImportError:
return None
client_id = os.environ.get("GOOGLE_CLIENT_ID")
client_secret = os.environ.get("GOOGLE_CLIENT_SECRET")
if not client_id or not client_secret:
return None
try:
credentials = Credentials(
token=None,
client_id=client_id,
client_secret=client_secret,
scopes=SCOPES,
)
return build("sheets", "v4", credentials=credentials)
except Exception:
return None
def _raci_rows(plan: ProjectPlan) -> list[list]:
"""Build rows for RACI sheet using plan.raci entries."""
# Collect all unique teams in order of appearance
teams: list[str] = []
for entry in plan.raci:
for team in (entry.responsible + [entry.accountable] + entry.consulted + entry.informed):
if team and team not in teams:
teams.append(team)
header = ["Task"] + teams
rows = [header]
for entry in plan.raci:
row = [entry.task]
for team in teams:
roles: list[str] = []
if team in entry.responsible:
roles.append("R")
if team == entry.accountable:
roles.append("A")
if team in entry.consulted:
roles.append("C")
if team in entry.informed:
roles.append("I")
row.append("/".join(roles) if roles else "")
rows.append(row)
return rows
def _risks_rows(plan: ProjectPlan) -> list[list]:
"""Build rows for Risk Register using plan.risks entries."""
header = ["Phase", "Description", "Likelihood", "Impact", "Mitigation"]
rows = [header]
for risk in plan.risks:
rows.append([
risk.phase_id or "General",
risk.description,
risk.likelihood,
risk.impact,
risk.mitigation,
])
return rows
async def export_raci(plan: ProjectPlan) -> dict:
"""
Create a new Google Spreadsheet with RACI data.
Returns {"success": bool, "url": str | None, "error": str | None}
"""
service = _build_service()
if service is None:
return {"success": False, "url": None, "error": ERROR_MESSAGES["google_auth"]}
try:
rows = _raci_rows(plan)
def _create() -> str:
body = {"properties": {"title": "ProjectCraft — RACI Matrix"}}
spreadsheet = service.spreadsheets().create(body=body, fields="spreadsheetId").execute()
sheet_id = spreadsheet["spreadsheetId"]
service.spreadsheets().values().update(
spreadsheetId=sheet_id,
range="Sheet1!A1",
valueInputOption="RAW",
body={"values": rows},
).execute()
return f"https://docs.google.com/spreadsheets/d/{sheet_id}"
url = await asyncio.to_thread(_create)
return {"success": True, "url": url, "error": None}
except Exception as exc:
return {"success": False, "url": None, "error": str(exc)}
async def export_risks(plan: ProjectPlan) -> dict:
"""
Create a new Google Spreadsheet with Risk Register data.
Returns {"success": bool, "url": str | None, "error": str | None}
"""
service = _build_service()
if service is None:
return {"success": False, "url": None, "error": ERROR_MESSAGES["google_auth"]}
try:
rows = _risks_rows(plan)
def _create() -> str:
body = {"properties": {"title": "ProjectCraft — Risk Register"}}
spreadsheet = service.spreadsheets().create(body=body, fields="spreadsheetId").execute()
sheet_id = spreadsheet["spreadsheetId"]
service.spreadsheets().values().update(
spreadsheetId=sheet_id,
range="Sheet1!A1",
valueInputOption="RAW",
body={"values": rows},
).execute()
return f"https://docs.google.com/spreadsheets/d/{sheet_id}"
url = await asyncio.to_thread(_create)
return {"success": True, "url": url, "error": None}
except Exception as exc:
return {"success": False, "url": None, "error": str(exc)}
|