feat(cupang): 출고 삭제 시 구글 시트도 동기화

- GoogleSheetsWriter.delete_sheet 추가 (없으면 no-op, 마지막 남은 시트면
  삭제 대신 내용만 비움 — 구글이 마지막 시트 삭제를 거부)
- 건별 삭제(soft/hard)와 날짜 전체 삭제 후 해당 출고일 시트를 동기화:
  남은 출고가 있으면 다시 기록, 없으면 시트 삭제
- 시트 설정 조회를 _sheet_target 으로 공통화, 결과는 로그로 남김
- 건별 삭제 후 기본 이동 경로를 그 출고일 달력으로

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
2026-09-01 16:17:32 +09:00
parent 4a260cb161
commit ba3b3a5ad8
3 changed files with 101 additions and 18 deletions
+37
View File
@@ -153,6 +153,43 @@ class GoogleSheetsWriter:
).execute()
return int(res["replies"][0]["addSheet"]["properties"]["sheetId"])
def delete_sheet(self, *, spreadsheet_id: str, title: str) -> dict[str, Any]:
"""이름이 정확히 일치하는 시트 1장을 삭제한다.
- 없으면 아무것도 하지 않는다(missing).
- 문서에 시트가 그 하나뿐이면 구글이 삭제를 거부하므로,
대신 내용을 비운다(cleared).
"""
if not self.enabled:
raise RuntimeError(self.reason or "Google Sheets 미설정")
meta = self._service.spreadsheets().get(spreadsheetId=spreadsheet_id).execute()
sheets = meta.get("sheets", [])
target = None
for sh in sheets:
props = sh.get("properties", {})
if props.get("title") == title:
target = int(props.get("sheetId"))
break
if target is None:
return {"ok": True, "action": "missing", "title": title}
if len(sheets) <= 1:
self._service.spreadsheets().batchUpdate(
spreadsheetId=spreadsheet_id,
body={"requests": [
{"unmergeCells": {"range": {"sheetId": target}}},
{"updateCells": {"range": {"sheetId": target}, "fields": "*"}},
]},
).execute()
return {"ok": True, "action": "cleared", "title": title}
self._service.spreadsheets().batchUpdate(
spreadsheetId=spreadsheet_id,
body={"requests": [{"deleteSheet": {"sheetId": target}}]},
).execute()
return {"ok": True, "action": "deleted", "title": title}
def write_table(
self,
*,
+63 -17
View File
@@ -243,6 +243,53 @@ def _shipments_for_date(store: Any, ship_date: str) -> list[dict[str, Any]]:
return out
def _sheet_target(store: Any) -> tuple[Any, str, dict[str, Any] | None]:
"""(writer, spreadsheet_id, skip 사유) — 설정이 없으면 skip 정보를 돌려준다."""
import os # noqa: WPS433
from app.integrations.google_sheets import get_writer # noqa: WPS433
spreadsheet_id = (os.getenv("CUPANG_SHEET_ID") or "").strip()
if not spreadsheet_id:
return None, "", {"ok": False, "skipped": True, "reason": "CUPANG_SHEET_ID 미설정"}
writer = get_writer()
if not writer.enabled:
return None, "", {"ok": False, "skipped": True, "reason": writer.reason}
return writer, spreadsheet_id, None
def _sync_google_sheet_after_delete(store: Any, ship_date: str) -> dict[str, Any]:
"""출고 삭제 후 구글 시트를 현재 상태에 맞춘다.
- 그날 남은 출고가 없으면 해당 날짜 시트를 삭제
- 남아 있으면 남은 내용으로 다시 기록
실패해도 DB 삭제는 이미 끝났으므로 예외를 밖으로 던지지 않는다.
"""
from .export import sheet_title # noqa: WPS433
writer, spreadsheet_id, skip = _sheet_target(store)
if skip:
return skip
if _shipments_for_date(store, ship_date):
return _push_to_google_sheet(store, ship_date)
try:
res = writer.delete_sheet(spreadsheet_id=spreadsheet_id, title=sheet_title(ship_date))
except Exception as exc: # noqa: BLE001 - 사유만 남긴다
return {"ok": False, "skipped": False, "reason": f"{type(exc).__name__}: {exc}"}
return {"ok": True, "skipped": False, **res}
def _log_sheet(action: str, ship_date: str, res: dict[str, Any]) -> None:
print( # noqa: T201 - 컨테이너 로그 확인용(비밀값 없음)
f"[cupang] sheet {action} date={ship_date} ok={res.get('ok')} "
f"skipped={res.get('skipped')} detail={res.get('action', '')} "
f"reason={res.get('reason', '')}",
flush=True,
)
def _push_to_google_sheet(store: Any, ship_date: str) -> dict[str, Any]:
"""확정한 출고일을 Google 스프레드시트에 시트 1장으로 기록.
@@ -250,22 +297,14 @@ def _push_to_google_sheet(store: Any, ship_date: str) -> dict[str, Any]:
- 인증: 서비스 계정(app/integrations/google_sheets.py). 미설정이면 조용히 skip.
- 실패해도 출고 묶음 저장은 이미 끝났으므로 예외를 밖으로 던지지 않는다.
"""
import os # noqa: WPS433
from app.integrations.google_sheets import get_writer # noqa: WPS433
from .export import ( # noqa: WPS433
COL_FIRST, COL_LAST, FIRST_DATA_ROW, HEADER_BG, HEADER_ROW, PALLET_BG,
TITLE_ROW, WIDTHS_PX, build_table, sheet_title,
)
spreadsheet_id = (os.getenv("CUPANG_SHEET_ID") or "").strip()
if not spreadsheet_id:
return {"ok": False, "skipped": True, "reason": "CUPANG_SHEET_ID 미설정"}
writer = get_writer()
if not writer.enabled:
return {"ok": False, "skipped": True, "reason": writer.reason}
writer, spreadsheet_id, skip = _sheet_target(store)
if skip:
return skip
shipments = _shipments_for_date(store, ship_date)
if not shipments:
@@ -410,12 +449,17 @@ async def delete(
store = _store(request)
if store is None:
raise HTTPException(status_code=503, detail="cupang_db 미설정")
ship = store.get_shipment(shipment_id=shipment_id, with_lines=False)
ship_date = str((ship or {}).get("ship_date") or "")
try:
store.soft_delete(shipment_id=shipment_id)
except KeyError:
raise HTTPException(status_code=404, detail="출고 묶음을 찾을 수 없습니다.")
if ship_date:
_log_sheet("sync", ship_date, _sync_google_sheet_after_delete(store, ship_date))
return RedirectResponse(
url=_safe_next(next, f"/cupang/{shipment_id}"), status_code=303
url=_safe_next(next, f"/cupang/?date={ship_date}" if ship_date else "/cupang/"),
status_code=303,
)
@@ -429,10 +473,14 @@ async def hard_delete(
store = _store(request)
if store is None:
raise HTTPException(status_code=503, detail="cupang_db 미설정")
ship = store.get_shipment(shipment_id=shipment_id, with_lines=False)
ship_date = str((ship or {}).get("ship_date") or "")
try:
store.hard_delete(shipment_id=shipment_id)
except KeyError:
raise HTTPException(status_code=404, detail="출고 묶음을 찾을 수 없습니다.")
if ship_date:
_log_sheet("sync", ship_date, _sync_google_sheet_after_delete(store, ship_date))
return RedirectResponse(url="/cupang/", status_code=303)
@@ -461,6 +509,8 @@ async def day_delete(
store.soft_delete(shipment_id=ship["id"])
except KeyError:
continue
# 그날 출고가 모두 사라졌으면 구글 시트의 해당 날짜 시트도 지운다.
_log_sheet("sync", day, _sync_google_sheet_after_delete(store, day))
return RedirectResponse(
url=_safe_next(next, f"/cupang/?date={day}"), status_code=303
)
@@ -1428,11 +1478,7 @@ async def box_calc_confirm(
# 구글 스프레드시트에 출고일 시트 기록(설정돼 있을 때만).
sheet = _push_to_google_sheet(store, ship_date)
print( # noqa: T201 - 컨테이너 로그로 확인용(비밀값 없음)
f"[cupang] sheet push date={ship_date} ok={sheet.get('ok')} "
f"skipped={sheet.get('skipped')} reason={sheet.get('reason', '')}",
flush=True,
)
_log_sheet("push", ship_date, sheet)
return JSONResponse({"created": created, "ship_date": ship_date, "sheet": sheet})