o2o-site-AEO/solution/backend/services/ops_service.py
Mina Choi 46b74b310b [feat] solution/backend,frontend: 개발자 전용 사이트관리·유저관리 — 조회 전용 경량 화면
admin 앱을 새 메뉴로 키우려면 새 도메인이 필요해서, solution 앱에 DEVELOPER 게이트 하나로 얹었다.
메뉴 문자열은 사장님 번들에도 실리지만 데이터 접근은 백엔드 게이트가 막는다.

- router/v1/ops: GET /v1/ops/sites · /v1/ops/users (RequireDeveloper)
- services/ops_service.py, crud/site_crud.list_all_sites, crud/user_crud.list_users
- frontend: OpsSitesPage · OpsUsersPage, AppShell 에 role===DEVELOPER 일 때만 두 줄
- api/generated: orval 코드젠 결과

프론트 tsc·eslint 통과, openapi 코드젠 성공. DB 조회는 미검증(DEVLOG 2026-09-23)

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
2026-09-28 15:53:02 +09:00

90 lines
4.0 KiB
Python

from fastapi import Depends
from common.database.db_session_manager import DB_SESSION_MNG
from common.database.model.models import places, users
from common.enums import DBWRType, ErrorType, UserRole
from common.models.gmodel import PageParams
from crud.site_crud import ISiteCRUD, SiteCRUD
from crud.user_crud import IUserCRUD, UserCRUD
from router.v1.ops.protocol import OpsSiteData, OpsUserData, Res_OpsSites, Res_OpsUsers
class OpsService:
"""내부 운영(개발자) 전용 조회 — 회사 스코프 없는 전 계정 사이트·유저 목록.
admin/frontend 의 사업장 화면(사실 검증)과 역할이 다르다 — 여기는 발행 인프라 상태와
계정 상태만 본다. 게이트는 라우터가 RequireDeveloper 로 건다(여기서는 다시 확인하지 않는다)."""
def __init__(self, site_crud: ISiteCRUD = Depends(SiteCRUD), user_crud: IUserCRUD = Depends(UserCRUD)):
self.site_crud = site_crud
self.user_crud = user_crud
async def list_sites(self, pg: PageParams) -> Res_OpsSites:
res = Res_OpsSites(page=pg.page, size=pg.size)
err_type, rows, total = await DB_SESSION_MNG.execute_lambda(
places.DBType(), DBWRType.DB_READ.value,
lambda s: self.site_crud.list_all_sites(s, pg.skip, pg.size),
)
if err_type != ErrorType.SUCCESS:
res.result.SetResult(err_type)
return res
res.sites = [
self._site_row(place, site, built_at, photo, login_id, email, name)
for place, site, built_at, photo, login_id, email, name in rows
]
res.total = total
return res
@staticmethod
def _site_row(place, site, built_at, primary_photo_url, owner_login_id, owner_email, owner_name) -> OpsSiteData:
# ★ 재빌드 판별은 site_service._my_site_row 와 같은 규칙이어야 한다 — 다르면 화면마다 다른 답을 한다.
changed = place.content_updated_at
ever_published = site is not None and getattr(site, "published_at", None) is not None
thumbnail_url = getattr(site, "thumbnail_url", None) or (primary_photo_url if ever_published else None)
return OpsSiteData(
place_id=place.place_id,
name=place.name,
category=place.category,
place_status=place.status,
created_at=place.created_at,
site_id=getattr(site, "site_id", None),
status=getattr(site, "status", None),
domain=getattr(site, "domain", None),
published_at=getattr(site, "published_at", None),
thumbnail_url=thumbnail_url,
needs_rebuild=bool(site is not None and changed and (built_at is None or changed > built_at)),
owner_login_id=owner_login_id,
owner_email=owner_email,
owner_name=owner_name,
)
async def list_users(self, pg: PageParams, search: str | None = None) -> Res_OpsUsers:
"""USER/OWNER 만 — 개발자 계정은 목록에서 뺀다(UserRole 주석: 고객사에 존재를 노출하지 않는다)."""
res = Res_OpsUsers(page=pg.page, size=pg.size)
roles = [UserRole.USER.value, UserRole.OWNER.value]
err_type, rows, total = await DB_SESSION_MNG.execute_lambda(
users.DBType(), DBWRType.DB_READ.value,
lambda s: self.user_crud.list_users(s, roles, pg.skip, pg.size, search),
)
if err_type != ErrorType.SUCCESS:
res.result.SetResult(err_type)
return res
res.users = [
OpsUserData(
user_id=user.user_id,
login_id=user.id,
name=user.name,
email=user.email,
contact_number=user.contact_number,
role=user.role,
status=user.status,
provider=user.provider,
created_at=user.created_at,
last_accessed_at=user.last_accessed_at,
place_count=place_count,
)
for user, place_count in rows
]
res.total = total
return res