king
9c58b022e2
- ERP 셸: 좌측 사이드바 + 상단 헤더 + 콘텐츠 영역(erp_base.html)
- 분기: 슈퍼관리자 → 업무모듈 카드, 일반사용자 → ERP 메인(erp_home.html)
- 신규 라우트: /modules (슈퍼관리자 모듈 카드 재진입)
- 개인경비 모듈: app/modules/expense/ (라우터/저장소/템플릿)
- JSON 저장소(ExpenseStore) + PostgreSQL 저장소(ExpenseDBStore)
- 팩토리: EXPENSE_DB_URL 있으면 DB, 없으면 JSON 폴백
- DB: scripts/sql/expense_db_init.sql (멱등 DDL, postgres-db 컨테이너)
- expense_db, 역할 expense_app, 테이블 expense_items, 인덱스/트리거
- 마이그레이션: scripts/migrate_expense_json_to_db.py (멱등, --dry-run)
- 의존성: psycopg[binary,pool]>=3.2
- 환경변수: EXPENSE_DB_URL 추가
- 문서: CLAUDE.md 작업 기준 + docs/{PROJECT_OVERVIEW,SERVER_ARCHITECTURE,DATABASES,DEPLOYMENT}.md
- main-app 배포 경로 /opt/www/main 고정 명시
- 위험 명령(DROP/TRUNCATE/rm -rf/volume 삭제) 사용자 승인 정책
117 lines
3.6 KiB
Python
117 lines
3.6 KiB
Python
#!/usr/bin/env python3
|
|
"""JSON 저장소 → expense_db 마이그레이션.
|
|
|
|
사용법:
|
|
EXPENSE_DB_URL=postgresql://expense_app:<pwd>@<host>:5432/expense_db \
|
|
python scripts/migrate_expense_json_to_db.py \
|
|
--json /data/expense.json \
|
|
[--dry-run]
|
|
|
|
원칙:
|
|
- 멱등(idempotent): id 가 같으면 INSERT ... ON CONFLICT DO NOTHING.
|
|
- JSON 원본은 건드리지 않는다. 검증 완료 후 사용자가 직접 보존/이관.
|
|
- 실패 시 트랜잭션 롤백 후 종료.
|
|
|
|
위험 경고:
|
|
- 본 스크립트는 운영 데이터에 쓰기 작업을 한다. 백업/덤프 후 실행할 것.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
import os
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
REQUIRED_KEYS = {
|
|
"id",
|
|
"owner",
|
|
"spent_at",
|
|
"category",
|
|
"method",
|
|
"merchant",
|
|
"amount",
|
|
"memo",
|
|
"status",
|
|
}
|
|
|
|
|
|
def load_items(path: Path) -> list[dict]:
|
|
with path.open("r", encoding="utf-8") as f:
|
|
data = json.load(f)
|
|
items = data.get("items") if isinstance(data, dict) else None
|
|
if not isinstance(items, list):
|
|
raise SystemExit(f"형식 오류: {path} 의 'items' 가 리스트가 아닙니다.")
|
|
return items
|
|
|
|
|
|
def main() -> int:
|
|
parser = argparse.ArgumentParser()
|
|
parser.add_argument("--json", required=True, help="원본 expense.json 경로")
|
|
parser.add_argument("--dry-run", action="store_true", help="DB 에 쓰지 않고 검증만")
|
|
args = parser.parse_args()
|
|
|
|
dsn = os.environ.get("EXPENSE_DB_URL", "").strip()
|
|
if not dsn:
|
|
raise SystemExit("EXPENSE_DB_URL 환경변수가 설정되지 않았습니다.")
|
|
|
|
json_path = Path(args.json)
|
|
if not json_path.exists():
|
|
raise SystemExit(f"파일을 찾을 수 없습니다: {json_path}")
|
|
|
|
items = load_items(json_path)
|
|
print(f"읽음: {len(items)} 건 ({json_path})")
|
|
|
|
bad = [i for i in items if not REQUIRED_KEYS.issubset(i.keys())]
|
|
if bad:
|
|
print(f"경고: 필드 누락 {len(bad)}건 — 누락 키는 기본값 채움", file=sys.stderr)
|
|
|
|
if args.dry_run:
|
|
print("dry-run 완료. DB 에 쓰지 않았습니다.")
|
|
return 0
|
|
|
|
import psycopg # type: ignore
|
|
|
|
inserted = skipped = 0
|
|
with psycopg.connect(dsn, autocommit=False) as conn:
|
|
with conn.cursor() as cur:
|
|
for it in items:
|
|
row = (
|
|
it["id"],
|
|
str(it.get("owner", "")).lower().strip(),
|
|
it.get("spent_at") or None,
|
|
it.get("category") or "기타",
|
|
it.get("method") or "법인카드",
|
|
it.get("merchant") or "",
|
|
int(it.get("amount") or 0),
|
|
it.get("memo") or "",
|
|
it.get("status") or "작성중",
|
|
it.get("created_at"),
|
|
it.get("updated_at"),
|
|
)
|
|
cur.execute(
|
|
"""
|
|
INSERT INTO expense_items
|
|
(id, owner, spent_at, category, method, merchant,
|
|
amount, memo, status, created_at, updated_at)
|
|
VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,
|
|
COALESCE(%s::timestamptz, now()),
|
|
COALESCE(%s::timestamptz, now()))
|
|
ON CONFLICT (id) DO NOTHING
|
|
""",
|
|
row,
|
|
)
|
|
if cur.rowcount == 1:
|
|
inserted += 1
|
|
else:
|
|
skipped += 1
|
|
conn.commit()
|
|
|
|
print(f"완료: insert={inserted}, skip(중복)={skipped}")
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|